obj/

collection.rs

//! `Collection<T>` — typed handle to one collection's primary
//! B-tree.

use std::borrow::Cow;
use std::collections::{HashMap, HashSet, VecDeque};
use std::marker::PhantomData;
use std::ops::Bound;
use std::sync::{Arc, Mutex, MutexGuard};

use obj_core::btree::BTree;
use obj_core::codec::{decode, encode};
use obj_core::pager::page::PageId;
use obj_core::pager::Pager;
use obj_core::platform::FileHandle;
use obj_core::{Catalog, CollectionDescriptor, Document, Error, Id, Result};

/// Boxed iterator alias used by [`Collection::lookup`] and
/// [`Collection::index_range`]. The iterator borrows from the
/// enclosing transaction; the `'tx` lifetime is bound on the
/// `Collection<T>` it was obtained from.
pub type IndexIter<'a, Item> = Box<dyn Iterator<Item = Result<Item>> + Send + 'a>;

/// Per-batch refill size for [`IterIndexRange`]. The iterator yields
/// one `(user_key, T)` pair at a time but pulls index B-tree entries
/// in fixed-size chunks so the per-step pager-lock acquisition cost
/// amortises across many `next()` calls. Power-of-ten Rule 3: the
/// buffer is fixed-size — at ~8 bytes/key plus the trailing 8-byte
/// id suffix that's ~4 KiB peak for the staged batch. The buffer
/// does NOT scale with the range's total size.
const ITER_INDEX_RANGE_BATCH: usize = 256;

/// Per-call cap on the bounded `HashSet<Id>` used by
/// [`Collection::count_distinct_ids_in_range`] to count unique
/// document `Id`s under an `Each` index. Power-of-ten Rule 3: the
/// distinct set is allocation-bounded; exceeding the cap surfaces
/// [`obj_core::Error::DistinctCountExceeded`] rather than chewing
/// arbitrary memory. The user can narrow the range via
/// `.index_range(...)` to fit inside the budget.
pub const MAX_DISTINCT_IDS: usize = 100_000;

use crate::txn::{lock_catalog, ReadTxn, WriteTxn};

/// #90 — per-transaction descriptor cache. Maps collection name to
/// its LIVE [`CollectionDescriptor`] (with `next_id`, `primary_root`,
/// and every index `root_page_id` advanced IN-MEMORY across the
/// transaction). This is the single mid-txn source of truth for those
/// roots; [`crate::WriteTxn::commit`] flushes each entry back to the
/// catalog exactly once. Shared (via `Arc`) between the [`WriteTxn`]
/// and every [`Collection`] handle opened on it, so two handles of the
/// same collection observe the same advancing descriptor.
pub(crate) type DescriptorCache = Arc<Mutex<HashMap<String, CollectionDescriptor>>>;

/// Construct an empty [`DescriptorCache`].
#[must_use]
pub(crate) fn new_descriptor_cache() -> DescriptorCache {
    Arc::new(Mutex::new(HashMap::new()))
}

/// Acquire the descriptor-cache mutex; map poison into `Busy` (Rule
/// 7 — no panic on a poisoned lock).
pub(crate) fn lock_descriptors(
    cache: &Mutex<HashMap<String, CollectionDescriptor>>,
) -> Result<MutexGuard<'_, HashMap<String, CollectionDescriptor>>> {
    cache.lock().map_err(|_| Error::Busy {
        kind: obj_core::LockKind::WriterInProcess,
    })
}

/// Resolve the live cached descriptor for `name`, lazily loading it
/// from the catalog B-tree on first touch in this transaction. After
/// the first load the catalog tree is NEVER re-read mid-txn for this
/// collection — every subsequent read/advance goes through the cache
/// entry, so the unique pre-check and every index-tree open observe
/// the in-memory-advanced roots (#90 load-bearing invariant).
///
/// Returns a mutable borrow into the cache so callers bump `next_id`,
/// advance `primary_root`, and let `apply_doc_change` advance index
/// roots in place — all without a per-doc `Catalog::update`.
pub(crate) fn cached_descriptor_mut<'g>(
    cache: &'g mut HashMap<String, CollectionDescriptor>,
    pager: &mut Pager<FileHandle>,
    catalog: &Catalog<FileHandle>,
    name: &str,
) -> Result<&'g mut CollectionDescriptor> {
    if !cache.contains_key(name) {
        let descriptor = catalog_get_required(pager, catalog, name)?;
        cache.insert(name.to_owned(), descriptor);
    }
    cache.get_mut(name).ok_or(Error::Corruption { page_id: 0 })
}

/// Typed handle to a collection.
///
/// Construct via [`crate::WriteTxn::collection`] (lazy-create) or
/// [`crate::ReadTxn::collection`] (read-only; errors if absent), or
/// via [`crate::Db::collection`] for a one-shot read-only handle
/// bound to a runtime collection name (M11 #94 — Phase 1B).
///
/// All methods take `&self` because the underlying state lives
/// behind mutexes on the parent transaction; the handle itself is
/// stateless beyond the descriptor it caches.
pub struct Collection<'tx, T: Document> {
    /// `Mode::Write` carries the [`WriteTxn`] reference;
    /// `Mode::Read` the [`ReadTxn`] reference; `Mode::Lazy` carries
    /// a `&Db` and opens a private read transaction on each method
    /// call.  The [`Collection`]'s lifetime is bound to whichever
    /// it was constructed from.
    mode: CollectionMode<'tx>,
    /// Collection name resolved at construction. For handles built
    /// via [`crate::WriteTxn::collection`] / [`crate::ReadTxn::collection`]
    /// this equals `T::COLLECTION`. For handles built via
    /// [`crate::Db::collection`] this is the caller-supplied runtime
    /// name (which may differ from `T::COLLECTION`, e.g.
    /// `"archive.orders"` against a type whose declared `COLLECTION`
    /// is `"orders"`).
    //
    // `collection_name` rather than the lint-suggested `name`
    // because `name` collides with `IndexDescriptor::name` /
    // `IndexSpec::name` throughout this module's internals.
    //
    // `Cow<'static, str>` (#84): the common `WriteTxn`/`ReadTxn`
    // open path resolves to `T::COLLECTION` — a `&'static str` —
    // so it can be stored as `Cow::Borrowed` with NO `to_owned()`
    // heap allocation per read. The runtime-name path
    // (`Db::collection`, `open_readonly_named` with a caller name)
    // stores `Cow::Owned`.
    #[allow(clippy::struct_field_names)]
    collection_name: Cow<'static, str>,
    /// Cached descriptor.  Populated at construction for `Write` /
    /// `Read` mode; never updated in place — a `Collection<T>`
    /// reflects the catalog row that existed when the handle was
    /// opened.  `update` / `delete` inside the same txn re-read the
    /// descriptor through the catalog lock to capture mutations
    /// from prior calls in the same txn (e.g. an `insert` that
    /// advanced `next_id`).
    ///
    /// For `Lazy` mode the descriptor is a sentinel — the real
    /// descriptor is loaded inside each method's private read
    /// transaction.
    descriptor: CollectionDescriptor,
    _phantom: PhantomData<fn() -> T>,
}

/// Backing-reference inside a [`Collection`].  Encodes whether
/// the txn is writable.
enum CollectionMode<'tx> {
    Write(WriteRef<'tx>),
    Read(ReadRef<'tx>),
    /// Read-only handle that opens a private read transaction on
    /// each method call. Constructed by [`crate::Db::collection`];
    /// the `'tx` lifetime is the borrow of `&Db`.
    Lazy(LazyRef<'tx>),
}

struct LazyRef<'db> {
    /// Borrowed `Db` — methods open one-shot read transactions on
    /// it. The `'db` lifetime keeps the borrow alive for as long
    /// as the [`Collection`] handle.
    db: &'db crate::Db,
}

struct WriteRef<'tx> {
    env: &'tx obj_core::TxnEnv<FileHandle>,
    catalog: Arc<Mutex<Catalog<FileHandle>>>,
    /// #90: shared per-txn descriptor cache (cloned from the owning
    /// [`WriteTxn`]). The single mid-txn source of truth for
    /// `next_id` / `primary_root` / index roots; flushed once at
    /// commit.
    descriptors: DescriptorCache,
}

struct ReadRef<'tx> {
    snapshot: &'tx obj_core::ReaderSnapshot<FileHandle>,
    env: &'tx obj_core::TxnEnv<FileHandle>,
}

impl<'tx, T: Document> Collection<'tx, T> {
    /// Open the collection on the write side, lazy-creating the
    /// catalog row + an empty primary B-tree on first call, and
    /// reconciling the type's declared `Document::indexes()` against
    /// the catalog's stored descriptors (M7 #57) on the first call
    /// per process per collection.
    pub(crate) fn open_or_create(tx: &'tx mut WriteTxn<'_>) -> Result<Self> {
        // Stage 1: lazy-create the collection row + an empty primary
        // B-tree if this is the first call for `T` against this
        // database. The returned descriptor is the pre-reconcile
        // shape; we re-read after reconciliation below because the
        // reconciler may have appended index descriptors.
        let _initial = ensure_collection::<T>(&tx.inner, &tx.catalog)?;
        reconcile_indexes_once::<T>(
            &tx.inner,
            &tx.catalog,
            &tx.reconciled,
            &mut tx.reconciled_staged,
        )?;
        // Re-read the descriptor in case the reconciler mutated the
        // index roster (the reconciler runs through `Catalog::update`
        // which rewrites the descriptor's `indexes` vector).
        let descriptor = reread_descriptor::<T>(&tx.inner, &tx.catalog)?;
        Ok(Self {
            mode: CollectionMode::Write(WriteRef {
                env: tx.inner.env(),
                catalog: Arc::clone(&tx.catalog),
                descriptors: Arc::clone(&tx.descriptors),
            }),
            // `T::COLLECTION` is `&'static str` (#84): borrow, don't
            // allocate.
            collection_name: Cow::Borrowed(T::COLLECTION),
            descriptor,
            _phantom: PhantomData,
        })
    }

    /// Open the collection on the read side.  Errors if the
    /// collection has not yet been registered AT THE SNAPSHOT'S
    /// PINNED LSN — a collection created by a concurrent writer
    /// AFTER the snapshot was pinned is invisible to this txn (M6
    /// #53). The descriptor is read by walking the catalog B+tree
    /// rooted at `snapshot.root_catalog()` (the value captured by
    /// `Pager::reader_snapshot` at pin time) using the snapshot-
    /// aware [`Catalog::lookup_via_snapshot`] free function — NOT
    /// via the writer's live `Catalog.tree.root`, which a concurrent
    /// writer may have COW-advanced past the snapshot.
    ///
    /// M11 #93: if `T::COLLECTION` is of the form
    /// `<namespace>.<name>`, the lookup dispatches against the
    /// attached database registered under `<namespace>` instead of
    /// the calling Db. The attached snapshot is the one pinned by
    /// `Db::read_transaction` at txn-begin; the attached env is
    /// passed through the same way.
    pub(crate) fn open_readonly(tx: &'tx ReadTxn<'_>) -> Result<Self> {
        // `T::COLLECTION` is `&'static str` — store it as
        // `Cow::Borrowed` (#84), avoiding the per-read `to_owned()`
        // heap allocation on the common typed-handle path.
        Self::open_readonly_named(tx, Cow::Borrowed(T::COLLECTION))
    }

    /// Open the collection on the read side against a caller-supplied
    /// runtime `name`. Like [`Self::open_readonly`] but the catalog
    /// lookup uses `name` instead of `T::COLLECTION` — required for
    /// [`crate::Db::collection`]'s Phase 1B accessor (M11 #94), which
    /// binds a runtime collection name (e.g. `"archive.orders"`) to
    /// the typed handle.
    ///
    /// Namespace dispatch (`<ns>.<tail>` → attached database) follows
    /// the same shape as [`Self::open_readonly`].
    pub(crate) fn open_readonly_named(
        tx: &'tx ReadTxn<'_>,
        name: Cow<'static, str>,
    ) -> Result<Self> {
        let (namespace, tail) = crate::db::split_namespace(&name);
        let (env, snapshot, lookup_name): (
            &'tx obj_core::TxnEnv<FileHandle>,
            &'tx obj_core::ReaderSnapshot<FileHandle>,
            &str,
        ) = match namespace {
            None => (tx.inner.env(), tx.inner.snapshot(), &name),
            Some(ns) => {
                let ctx = tx
                    .attached
                    .get(ns)
                    .ok_or_else(|| Error::CollectionNamespaceUnknown {
                        namespace: ns.to_owned(),
                    })?;
                (ctx.env.as_ref(), &ctx.snapshot, tail)
            }
        };
        let Some(descriptor) = read_descriptor_via_snapshot_named(env, snapshot, lookup_name)?
        else {
            return Err(Error::CollectionNotFound {
                name: name.into_owned(),
            });
        };
        Ok(Self {
            mode: CollectionMode::Read(ReadRef { snapshot, env }),
            // `name` is moved in unchanged — `Cow::Borrowed(&'static)`
            // for the typed-handle path (no alloc), `Cow::Owned` for
            // runtime-name handles.
            collection_name: name,
            descriptor,
            _phantom: PhantomData,
        })
    }

    /// Construct a deferred-lookup, read-only handle bound to a
    /// runtime collection name. Each method call opens a private
    /// read transaction on `db` and dispatches through
    /// [`Self::open_readonly_named`]. Construction is infallible —
    /// errors (missing collection, unknown namespace, etc.) surface
    /// at the first method call.
    ///
    /// Used by [`crate::Db::collection`] (Phase 1B, M11 #94).
    pub(crate) fn lazy(db: &'tx crate::Db, name: String) -> Self {
        Self {
            mode: CollectionMode::Lazy(LazyRef { db }),
            collection_name: Cow::Owned(name),
            // Sentinel descriptor — never read in Lazy mode; the
            // real descriptor is loaded inside each method's
            // private read transaction.
            descriptor: CollectionDescriptor::new(0, 0, 0),
            _phantom: PhantomData,
        }
    }

    /// Cached descriptor (`collection_id`, `primary_root`,
    /// `type_version`, `next_id` at handle-open time).
    #[must_use]
    pub fn descriptor(&self) -> &CollectionDescriptor {
        &self.descriptor
    }

    /// #90: the LIVE primary-tree root for a `Write`-mode handle.
    ///
    /// Prefers the per-txn descriptor cache (which carries every
    /// `primary_root` advance from this txn's prior writes) so a
    /// read-after-write on the same handle inside one transaction
    /// observes its own uncommitted inserts. Falls back to the
    /// handle's open-time `primary_root` if this collection has not
    /// yet been written in the txn (no cache entry).
    fn write_primary_root(&self, write: &WriteRef<'tx>) -> Result<u64> {
        let cache = lock_descriptors(&write.descriptors)?;
        Ok(cache
            .get(self.collection_name.as_ref())
            .map_or(self.descriptor.primary_root, |d| d.primary_root))
    }

    /// #90: the LIVE `root_page_id` for the named `Active` index on a
    /// `Write`-mode handle. Prefers the per-txn cache so a read after
    /// an index-mutating write in the same txn descends the advanced
    /// root; falls back to `fallback` (the handle's open-time
    /// descriptor entry) when the collection has no cache entry yet.
    fn write_index_root(
        &self,
        write: &WriteRef<'tx>,
        index_name: &str,
        fallback: u64,
    ) -> Result<u64> {
        let cache = lock_descriptors(&write.descriptors)?;
        let Some(descriptor) = cache.get(self.collection_name.as_ref()) else {
            return Ok(fallback);
        };
        let live = descriptor
            .indexes
            .iter()
            .find(|d| d.name == index_name && d.status == obj_core::IndexStatus::Active)
            .map_or(fallback, |d| d.root_page_id);
        Ok(live)
    }

    /// Insert `doc`.  Returns the freshly-allocated [`Id`].
    ///
    /// # Errors
    ///
    /// - [`Error::ReadOnly`] if the handle is read-only.
    /// - Pager / catalog / codec errors propagated.
    // `doc: T` is taken by value for caller ergonomics (the user
    // owns the value and gives it to the database).  The codec's
    // `encode(&T, ...)` takes a reference, so clippy sees the
    // function body as "owned but borrows only" — accepted as the
    // intended public-API shape.
    #[allow(clippy::needless_pass_by_value)]
    pub fn insert(&self, doc: T) -> Result<Id> {
        let write = self.write_or_err("insert")?;
        let name: &str = self.collection_name.as_ref();
        let mut pager = lock_pager(write.env)?;
        let catalog = lock_catalog(&write.catalog)?;
        // #90: the cached descriptor is the SOLE mid-txn source of
        // truth. `next_id` is an in-memory bump; `primary_root` and
        // index roots advance in the cache; NO per-doc
        // `Catalog::update` — the single flush is deferred to commit.
        let mut cache = lock_descriptors(&write.descriptors)?;
        let descriptor = cached_descriptor_mut(&mut cache, &mut pager, &catalog, name)?;
        let id = obj_core::id::bump_next_id(&mut descriptor.next_id, || name.to_owned())?;
        let bytes = encode(&doc, descriptor.collection_id)?;
        let key = id.to_be_bytes();
        let mut tree = btree_handle(&pager, descriptor.primary_root)?;
        tree.insert(&mut pager, &key, &bytes)?;
        descriptor.primary_root = tree.root().get();
        // Index maintenance lands inside the same WAL transaction as
        // the primary write, descending the cache's in-memory index
        // roots. A `UniqueConstraintViolation` here surfaces via `?`
        // and the surrounding `WriteTxn` rolls back atomically.
        crate::index_maint::apply_doc_change::<T>(&mut pager, descriptor, None, Some(&doc), id)?;
        Ok(id)
    }

    /// Fetch the document at `id`.
    ///
    /// On the write side this consults the pager (sees pending
    /// writes in the current txn).  On the read side it consults
    /// the snapshot's frozen view.
    ///
    /// # Lazy migration (M10 #84)
    ///
    /// If the on-disk record was written by an older
    /// `Document::VERSION` than the current `T::VERSION`, the codec
    /// walks the stored bytes through the schema registered for
    /// that version (see `T::historical_schemas()`) and dispatches
    /// the resulting structured `Dynamic` through `T::migrate`.
    /// **The migrated bytes are NOT written back to disk.** The
    /// next [`Collection::get`] re-reads the same v(n) bytes and
    /// re-runs migration. Only a subsequent
    /// [`Collection::update`](Self::update) /
    /// [`Collection::upsert`](Self::upsert) writes the document
    /// back, at which point the on-disk header records
    /// `T::VERSION`.
    ///
    /// This contract is what allows mixed-version reads to scale:
    /// a 10⁹-doc collection does not need to be batch-rewritten on
    /// schema upgrade. Power-of-ten Rule 7: every "migration ran"
    /// path returns the migrated `T`; no implicit write-back.
    ///
    /// # Errors
    ///
    /// Pager / B-tree / codec errors propagated. In particular:
    ///
    /// - [`Error::SchemaNotRegistered`](obj_core::Error::SchemaNotRegistered)
    ///   if the stored record carries a `type_version` for which
    ///   `T::historical_schemas()` has no entry.
    /// - [`Error::SchemaMigrationNotImplemented`](obj_core::Error::SchemaMigrationNotImplemented)
    ///   if the registered `T::migrate` returns the default error.
    pub fn get(&self, id: Id) -> Result<Option<T>> {
        if let Some(r) = self.dispatch_lazy(|c| c.get(id)) {
            return r;
        }
        let key = id.to_be_bytes();
        let Some(bytes) = self.read_or_write_env_for_get(&key)? else {
            return Ok(None);
        };
        Ok(Some(decode::<T>(&bytes, self.descriptor.collection_id)?))
    }

    /// Read the primary-tree value at `key` against the current
    /// `Write` / `Read` mode (Lazy mode is dispatched upstream of
    /// every public caller via [`Self::dispatch_lazy`]).
    fn read_or_write_env_for_get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        match &self.mode {
            CollectionMode::Write(write) => {
                // #90: descend the LIVE primary root so a get after an
                // insert/update in the SAME txn sees its own writes.
                let primary_root = self.write_primary_root(write)?;
                let mut pager = lock_pager(write.env)?;
                let tree = btree_handle(&pager, primary_root)?;
                tree.get(&mut pager, key)
            }
            CollectionMode::Read(read) => {
                snapshot_get_via_btree(read.snapshot, read.env, self.descriptor.primary_root, key)
            }
            CollectionMode::Lazy(_) => Err(Error::ReadOnly {
                operation: "internal: lazy-mode primary read",
            }),
        }
    }

    /// Apply `f` to the document at `id`, writing the mutated value
    /// back.
    ///
    /// # Errors
    ///
    /// - [`Error::ReadOnly`] on a read-side handle.
    /// - [`Error::DocumentNotFound`] if `id` is absent.
    /// - Pager / catalog / codec errors propagated.
    pub fn update<F>(&self, id: Id, f: F) -> Result<()>
    where
        F: FnOnce(&mut T),
    {
        let write = self.write_or_err("update")?;
        let name: &str = self.collection_name.as_ref();
        let mut pager = lock_pager(write.env)?;
        let catalog = lock_catalog(&write.catalog)?;
        // #90: resolve the live cached descriptor (sole mid-txn source
        // of truth); advance its roots in place, no per-doc flush.
        let mut cache = lock_descriptors(&write.descriptors)?;
        let descriptor = cached_descriptor_mut(&mut cache, &mut pager, &catalog, name)?;
        let key = id.to_be_bytes();
        let mut tree = btree_handle(&pager, descriptor.primary_root)?;
        let existing = tree.get(&mut pager, &key)?.ok_or(Error::DocumentNotFound {
            collection: T::COLLECTION,
            id: id.get(),
        })?;
        let old_value = decode::<T>(&existing, descriptor.collection_id)?;
        let mut new_value = decode::<T>(&existing, descriptor.collection_id)?;
        f(&mut new_value);
        let bytes = encode(&new_value, descriptor.collection_id)?;
        tree.delete(&mut pager, &key)?;
        tree.insert(&mut pager, &key, &bytes)?;
        descriptor.primary_root = tree.root().get();
        crate::index_maint::apply_doc_change::<T>(
            &mut pager,
            descriptor,
            Some(&old_value),
            Some(&new_value),
            id,
        )?;
        Ok(())
    }

    /// Delete the document at `id`.  Returns `true` if it existed.
    ///
    /// # Errors
    ///
    /// - [`Error::ReadOnly`] on a read-side handle.
    /// - Pager / catalog errors propagated.
    pub fn delete(&self, id: Id) -> Result<bool> {
        let write = self.write_or_err("delete")?;
        let name: &str = self.collection_name.as_ref();
        let mut pager = lock_pager(write.env)?;
        let catalog = lock_catalog(&write.catalog)?;
        // #90: cached descriptor is the sole mid-txn source of truth.
        let mut cache = lock_descriptors(&write.descriptors)?;
        let descriptor = cached_descriptor_mut(&mut cache, &mut pager, &catalog, name)?;
        let key = id.to_be_bytes();
        let mut tree = btree_handle(&pager, descriptor.primary_root)?;
        // Materialise the old document so the index-maintenance
        // path can compute the OLD key set and emit per-index
        // delete entries. If the row does not exist, we still
        // return `Ok(false)` for API back-compat (no index work).
        let old_value = match tree.get(&mut pager, &key)? {
            Some(bytes) => Some(decode::<T>(&bytes, descriptor.collection_id)?),
            None => None,
        };
        let removed = tree.delete(&mut pager, &key)?;
        descriptor.primary_root = tree.root().get();
        crate::index_maint::apply_doc_change::<T>(
            &mut pager,
            descriptor,
            old_value.as_ref(),
            None,
            id,
        )?;
        Ok(removed)
    }

    /// Insert-or-replace `doc` at `id`.
    ///
    /// # Errors
    ///
    /// - [`Error::ReadOnly`] on a read-side handle.
    /// - Pager / catalog / codec errors propagated.
    // `doc: T` passed by value for ergonomic ownership transfer;
    // codec takes `&T` so clippy reports a "borrow only" body.
    #[allow(clippy::needless_pass_by_value)]
    pub fn upsert(&self, id: Id, doc: T) -> Result<()> {
        let write = self.write_or_err("upsert")?;
        let name: &str = self.collection_name.as_ref();
        let mut pager = lock_pager(write.env)?;
        let catalog = lock_catalog(&write.catalog)?;
        // #90: cached descriptor is the sole mid-txn source of truth.
        let mut cache = lock_descriptors(&write.descriptors)?;
        let descriptor = cached_descriptor_mut(&mut cache, &mut pager, &catalog, name)?;
        let bytes = encode(&doc, descriptor.collection_id)?;
        let key = id.to_be_bytes();
        let mut tree = btree_handle(&pager, descriptor.primary_root)?;
        // Materialise the prior value (if any) so the index-
        // maintenance diff sees the OLD key set.
        let old_value = match tree.get(&mut pager, &key)? {
            Some(prior) => Some(decode::<T>(&prior, descriptor.collection_id)?),
            None => None,
        };
        let _ = tree.delete(&mut pager, &key)?;
        tree.insert(&mut pager, &key, &bytes)?;
        descriptor.primary_root = tree.root().get();
        crate::index_maint::apply_doc_change::<T>(
            &mut pager,
            descriptor,
            old_value.as_ref(),
            Some(&doc),
            id,
        )?;
        Ok(())
    }

    /// Look up the single document whose `index_name` key matches
    /// `key` under a `Unique` index.
    ///
    /// Errors with [`Error::IndexNotUnique`] if `index_name` resolves
    /// to a non-unique index — `find_unique` is *only* defined on
    /// `Unique` indexes. For `Standard` / `Each` / `Composite` use
    /// [`Self::lookup`] (which returns an iterator).
    ///
    /// Snapshot-aware: on a write-side handle the lookup sees the
    /// current txn's pending writes; on a read-side handle it sees
    /// the snapshot's frozen view.
    ///
    /// # Errors
    ///
    /// - [`Error::IndexNotFound`] if `index_name` is unknown / dropped.
    /// - [`Error::IndexNotUnique`] if the index is not `Unique`.
    /// - Pager / B-tree / codec errors propagated.
    pub fn find_unique(
        &self,
        index_name: &str,
        key: impl Into<obj_core::codec::Dynamic>,
    ) -> Result<Option<T>> {
        let key_dyn = key.into();
        if let Some(r) = self.dispatch_lazy(|c| c.find_unique(index_name, key_dyn.clone())) {
            return r;
        }
        let descriptor = self.active_index(index_name)?;
        if descriptor.kind != obj_core::IndexKind::Unique {
            return Err(Error::IndexNotUnique {
                collection: self.collection_name.clone().into_owned(),
                name: index_name.to_owned(),
            });
        }
        let encoded = index_key_for_lookup(descriptor, &[key_dyn])?;
        let id_bytes = self.index_get(descriptor, encoded.as_bytes())?;
        match id_bytes {
            Some(bytes) => match Id::from_be_bytes(&bytes) {
                Some(id) => self.get(id),
                None => Err(Error::Corruption { page_id: 0 }),
            },
            None => Ok(None),
        }
    }

    /// Yield every document whose `index_name` key matches `key`.
    /// Works on `Standard` / `Unique` / `Each` indexes. Returns
    /// `Err(Error::IndexKindMismatch)`-style guidance for
    /// `Composite` (use [`Self::index_range`] for tuple-shaped
    /// keys).
    ///
    /// The same document is yielded at most once even if it owns
    /// multiple matching entries — `Each` indexes can encode the
    /// same `id` under multiple element keys; we de-dup on emit.
    ///
    /// # Errors
    ///
    /// - [`Error::IndexNotFound`] if `index_name` is unknown / dropped.
    /// - Pager / B-tree / codec errors propagated.
    pub fn lookup(
        &self,
        index_name: &str,
        key: impl Into<obj_core::codec::Dynamic>,
    ) -> Result<IndexIter<'static, T>>
    where
        T: Send + 'static,
    {
        let key_dyn = key.into();
        if let Some(r) = self.dispatch_lazy(|c| c.lookup(index_name, key_dyn.clone())) {
            return r;
        }
        let descriptor = self.active_index(index_name)?;
        let encoded = index_key_for_lookup(descriptor, &[key_dyn])?;
        let ids = match descriptor.kind {
            obj_core::IndexKind::Unique => self.collect_unique(descriptor, encoded.as_bytes())?,
            obj_core::IndexKind::Standard
            | obj_core::IndexKind::Each
            | obj_core::IndexKind::Composite => {
                self.collect_nonunique_equal(descriptor, encoded.as_bytes())?
            }
            // `IndexKind` is `#[non_exhaustive]`; an unknown kind means
            // this build predates the on-disk index. Refuse rather than
            // return a wrong (possibly empty) id set.
            _ => return Err(Error::InvalidArgument("unsupported index kind")),
        };
        let resolved = self.resolve_unique_ids(ids)?;
        Ok(Box::new(resolved.into_iter().map(Ok)))
    }

    /// Yield `(user_key, doc)` pairs whose index key falls within
    /// `range`. The bounds are [`Dynamic`](obj_core::codec::Dynamic)
    /// values — the same ergonomic type [`crate::Query::index_range`]
    /// takes — encoded internally through the order-preserving field
    /// encoder ([`obj_core::index::encode_field`]); callers no longer
    /// hand-encode index-key bytes.
    ///
    /// For non-Unique kinds (`Standard` / `Each` / `Composite`) the
    /// bounds are widened internally so a user-facing
    /// `Included(x)..=Included(x)` range matches every entry whose
    /// user-key equals `x` even though the underlying B-tree key
    /// carries an `id_be8` suffix (see `docs/format.md` § Index key
    /// encoding § Range-bound widening (non-Unique kinds)).
    ///
    /// # Errors
    ///
    /// - [`Error::IndexNotFound`] if `index_name` is unknown / dropped.
    /// - [`obj_core::Error::Codec`] if a `Dynamic::String` bound
    ///   carries an embedded NUL byte (the order-preserving encoder
    ///   rejects those).
    /// - Pager / B-tree / codec errors propagated.
    pub fn index_range<R>(
        &self,
        index_name: &str,
        range: R,
    ) -> Result<IndexIter<'static, (Vec<u8>, T)>>
    where
        R: std::ops::RangeBounds<obj_core::codec::Dynamic>,
        T: Send + 'static,
    {
        let start = encode_dynamic_bound(range.start_bound())?;
        let end = encode_dynamic_bound(range.end_bound())?;
        self.index_range_encoded(index_name, start, end)
    }

    /// Encoded-bytes variant of [`Self::index_range`]. The bounds are
    /// already the order-preserving field encoding of the user's
    /// `Dynamic` value(s); this keeps the signature general for
    /// `Composite` "starts-with" scans and is the entry point the
    /// query layer / lazy-dispatch recursion call after they have
    /// done their own encoding.
    pub(crate) fn index_range_encoded(
        &self,
        index_name: &str,
        start_bound: std::ops::Bound<Vec<u8>>,
        end_bound: std::ops::Bound<Vec<u8>>,
    ) -> Result<IndexIter<'static, (Vec<u8>, T)>>
    where
        T: Send + 'static,
    {
        if let Some(r) = self.dispatch_lazy(|c| {
            c.index_range_encoded(index_name, start_bound.clone(), end_bound.clone())
        }) {
            return r;
        }
        let descriptor = self.active_index(index_name)?;
        let (start, end) =
            crate::index_bound::widen_bounds_for_kind(start_bound, end_bound, descriptor.kind);
        let entries = self.collect_range(descriptor, start, end)?;
        let descriptor_kind = descriptor.kind;
        let mut out: Vec<Result<(Vec<u8>, T)>> = Vec::with_capacity(entries.len());
        let mut emitted_ids: std::collections::HashSet<u64> = std::collections::HashSet::new();
        for (full_key, id_bytes_value) in entries {
            let Some(id) = Id::from_be_bytes(&id_bytes_value) else {
                out.push(Err(Error::Corruption { page_id: 0 }));
                continue;
            };
            // For Each indexes the same doc may appear multiple
            // times under different element keys — de-dup on
            // emission.
            if descriptor_kind == obj_core::IndexKind::Each && !emitted_ids.insert(id.get()) {
                continue;
            }
            // For non-unique kinds the B-tree key includes the
            // trailing 8-byte id suffix; strip it so the caller
            // sees only the user-key bytes.
            let user_key = strip_id_suffix(&full_key, descriptor_kind);
            match self.get(id) {
                Ok(Some(doc)) => out.push(Ok((user_key, doc))),
                Ok(None) => {
                    // Orphan index entry — surface as Corruption.
                    out.push(Err(Error::Corruption { page_id: 0 }));
                }
                Err(e) => out.push(Err(e)),
            }
        }
        Ok(Box::new(out.into_iter()))
    }

    /// Streaming variant of [`Self::index_range`] (Phase 7A perf
    /// pass, M14 #14). Yields `(user_key, T)` pairs lazily — the
    /// returned [`IterIndexRange`] decodes one `T` per `next()` call
    /// rather than building a `Vec<Result<(_, T)>>` of every match
    /// up front. The iterator borrows `&'a self`, so it must be
    /// consumed inside the lifetime of the enclosing
    /// [`crate::WriteTxn`] / [`crate::ReadTxn`] (or the
    /// [`crate::Db::collection`] handle, in Lazy mode).
    ///
    /// # When to prefer `iter_range` over `index_range`
    ///
    /// - **Memory.** `index_range` allocates `O(matches × sizeof(T))`
    ///   upfront; `iter_range` keeps a fixed-size [`VecDeque`] of
    ///   `(key, id)` pairs (`ITER_INDEX_RANGE_BATCH = 256` entries)
    ///   and decodes one `T` at a time. For a 100k-row range with
    ///   ~500-byte documents that's ~50 MB peak vs. a few KiB.
    /// - **Latency-to-first-row.** `index_range` decodes every
    ///   matching document before returning the iterator;
    ///   `iter_range` returns immediately after the first chunk
    ///   refill, so the first `next()` returns after one index walk
    ///   + one primary-tree `get` (rather than `N`).
    ///
    /// # When `index_range` is still the right answer
    ///
    /// `index_range` returns an `IndexIter<'static, _>` — it can
    /// escape the `read_transaction` / `transaction` closure that
    /// produced it. `iter_range` is bound to `&self`, so the
    /// iterator dies when the [`Collection`] handle dies. If you
    /// need to return the iterator to outer scope, stick with
    /// `index_range`.
    ///
    /// # Per-row `get`-back design choice
    ///
    /// Each `next()` yields `(user_key, T)` by calling
    /// [`Self::get`] under the hood — i.e. a SECOND B+tree descent
    /// per row (the first is the index range walk; the second is
    /// the primary-tree `get(id)`). This is intentional and
    /// inherited from `index_range`: the index leaf stores only
    /// the document `id` (8 bytes), not the document bytes. A
    /// future format-minor bump may add value-in-index storage to
    /// short-circuit the second descent; that work is pinned to
    /// post-1.0 (tracked as pit issue #16, "value-in-index
    /// storage to eliminate `index_range` double-decode").
    ///
    /// # Errors
    ///
    /// - [`Error::IndexNotFound`] if `index_name` is unknown / dropped.
    /// - Pager / B-tree / codec errors propagated at construction
    ///   and from each `next()` call.
    pub fn iter_range<'a, R>(&'a self, index_name: &str, range: R) -> Result<IterIndexRange<'a, T>>
    where
        R: std::ops::RangeBounds<obj_core::codec::Dynamic>,
        T: Send + 'static,
    {
        let start_bound = encode_dynamic_bound(range.start_bound())?;
        let end_bound = encode_dynamic_bound(range.end_bound())?;
        self.iter_range_encoded(index_name, start_bound, end_bound)
    }

    /// Encoded-bytes variant of [`Self::iter_range`]. Bounds are the
    /// order-preserving field encoding of the user's `Dynamic`
    /// value(s); used internally by the lazy-mode fallback path.
    fn iter_range_encoded<'a>(
        &'a self,
        index_name: &str,
        start_bound: Bound<Vec<u8>>,
        end_bound: Bound<Vec<u8>>,
    ) -> Result<IterIndexRange<'a, T>>
    where
        T: Send + 'static,
    {
        // Lazy mode falls back to `index_range`'s eager materialization
        // path so the index walk + per-row `get` share a single
        // snapshot. Streaming refills can't preserve that — each
        // refill would open a fresh txn and observe a different
        // snapshot. Lazy callers who want true streaming should open
        // an explicit `Db::read_transaction` and call `iter_range`
        // on the bound collection there.
        if matches!(self.mode, CollectionMode::Lazy(_)) {
            return self.iter_range_lazy_fallback(index_name, start_bound, end_bound);
        }
        let descriptor = self.active_index(index_name)?;
        // #90: in Write mode the iterator must walk the LIVE index
        // root (the per-txn cache's advanced value) so a streaming
        // scan opened after an in-txn index write sees its own
        // entries. In Read mode there is no write cache, so the
        // open-time descriptor root is authoritative.
        let index_root = match &self.mode {
            CollectionMode::Write(w) => {
                self.write_index_root(w, index_name, descriptor.root_page_id)?
            }
            _ => descriptor.root_page_id,
        };
        let (start, end) =
            crate::index_bound::widen_bounds_for_kind(start_bound, end_bound, descriptor.kind);
        // Stash the resumption marker as `Excluded(start)` only on
        // the first refill; afterwards the iterator overwrites it
        // with the last full_key it emitted. The initial `start`
        // bound is honoured by `refill` via the same `last_full_key`
        // → `Excluded(_)` translation.
        let initial_resume = match start {
            Bound::Included(k) => InitialResume::Included(k),
            Bound::Excluded(k) => InitialResume::Excluded(k),
            Bound::Unbounded => InitialResume::Unbounded,
        };
        Ok(IterIndexRange {
            coll: self,
            descriptor_kind: descriptor.kind,
            index_root,
            initial_resume: Some(initial_resume),
            last_full_key: None,
            end_bound: end,
            buffer: VecDeque::new(),
            emitted_ids: HashSet::new(),
            finished: false,
        })
    }

    /// Lazy-mode fallback for [`Self::iter_range`]: delegates to
    /// [`Self::index_range`] (which itself dispatches through a fresh
    /// read txn) and rehouses the eagerly-materialised entries into
    /// the streaming iterator's buffer as
    /// [`StagedEntry::Resolved`]. Power-of-ten Rule 4: keeping this
    /// isolated so the streaming path's `iter_range` body stays
    /// small.
    fn iter_range_lazy_fallback<'a>(
        &'a self,
        index_name: &str,
        start_bound: Bound<Vec<u8>>,
        end_bound: Bound<Vec<u8>>,
    ) -> Result<IterIndexRange<'a, T>>
    where
        T: Send + 'static,
    {
        let materialized = self.index_range_encoded(index_name, start_bound, end_bound)?;
        let mut buffer: VecDeque<Result<StagedEntry<T>>> = VecDeque::new();
        for item in materialized {
            match item {
                Ok((key, doc)) => buffer.push_back(Ok(StagedEntry::Resolved(key, doc))),
                Err(e) => buffer.push_back(Err(e)),
            }
        }
        Ok(IterIndexRange {
            coll: self,
            descriptor_kind: obj_core::IndexKind::Standard,
            index_root: 0,
            initial_resume: None,
            last_full_key: None,
            end_bound: Bound::Unbounded,
            buffer,
            emitted_ids: HashSet::new(),
            finished: true,
        })
    }

    /// Look up the `IndexKind` of an active index by name. Used by
    /// the M8 query layer to dispatch `Query::count` between the
    /// entry-count and distinct-id-count paths (M8 follow-up #72).
    ///
    /// # Errors
    ///
    /// - [`Error::IndexNotFound`] if `index_name` is unknown / dropped.
    pub(crate) fn index_kind(&self, index_name: &str) -> Result<obj_core::IndexKind> {
        Ok(self.active_index(index_name)?.kind)
    }

    /// Resolve `index_name` to an `Active` `IndexDescriptor` on the
    /// collection. Errors with [`Error::IndexNotFound`] if absent
    /// or `DroppedPending`.
    ///
    /// Returns a borrow into `self.descriptor.indexes` — no per-lookup
    /// clone (#84). Every caller uses the descriptor only within the
    /// enclosing `&self` borrow; `iter_range_encoded` copies the two
    /// `Copy` fields it needs (`kind`, `root_page_id`) into the
    /// returned iterator rather than holding the borrow.
    fn active_index(&self, index_name: &str) -> Result<&obj_core::IndexDescriptor> {
        let entry = self
            .descriptor
            .indexes
            .iter()
            .find(|d| d.name == index_name);
        match entry {
            Some(d) if d.status == obj_core::IndexStatus::Active => Ok(d),
            _ => Err(Error::IndexNotFound {
                collection: self.collection_name.clone().into_owned(),
                name: index_name.to_owned(),
            }),
        }
    }

    /// Single-key `get` on an index B-tree. Used by `find_unique`
    /// and by the Unique-kind branch of `lookup`.
    fn index_get(
        &self,
        descriptor: &obj_core::IndexDescriptor,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>> {
        match &self.mode {
            CollectionMode::Write(write) => {
                // #90: descend the LIVE index root from the per-txn
                // cache so an index read after an index-mutating write
                // in the SAME txn observes its own entries.
                let root_raw =
                    self.write_index_root(write, &descriptor.name, descriptor.root_page_id)?;
                let root = PageId::new(root_raw)
                    .ok_or(Error::InvalidArgument("index root_page_id is zero"))?;
                let mut pager = lock_pager(write.env)?;
                let tree = BTree::<FileHandle>::open(&pager, root)?;
                tree.get(&mut pager, key)
            }
            CollectionMode::Read(read) => {
                let root = PageId::new(descriptor.root_page_id)
                    .ok_or(Error::InvalidArgument("index root_page_id is zero"))?;
                let pager = lock_pager(read.env)?;
                BTree::<FileHandle>::get_via_snapshot(&pager, read.snapshot, root, key)
            }
            CollectionMode::Lazy(_) => Err(Error::ReadOnly {
                operation: "internal: lazy-mode index_get",
            }),
        }
    }

    /// Collect every `(full_key, value)` entry from an index B-tree
    /// whose key starts with `prefix`. For unique kinds the prefix
    /// is the entire key (one match max); for non-unique kinds we
    /// match every key whose first `prefix.len()` bytes equal
    /// `prefix` (the trailing `id_suffix` varies per doc).
    fn collect_nonunique_equal(
        &self,
        descriptor: &obj_core::IndexDescriptor,
        prefix: &[u8],
    ) -> Result<Vec<u64>> {
        let entries = self.collect_range(
            descriptor,
            std::ops::Bound::Included(prefix.to_vec()),
            // Upper bound is `prefix || 0xFF..` — every key whose
            // user-portion equals `prefix` falls in
            // `[prefix, prefix || u64::MAX]`.
            std::ops::Bound::Included(append_max_id(prefix)),
        )?;
        let mut ids = Vec::with_capacity(entries.len());
        let mut emitted: std::collections::HashSet<u64> = std::collections::HashSet::new();
        for (_full_key, value) in entries {
            let id = Id::from_be_bytes(&value).ok_or(Error::Corruption { page_id: 0 })?;
            if emitted.insert(id.get()) {
                ids.push(id.get());
            }
        }
        Ok(ids)
    }

    /// Collect a single id from a Unique index B-tree at `key`.
    fn collect_unique(
        &self,
        descriptor: &obj_core::IndexDescriptor,
        key: &[u8],
    ) -> Result<Vec<u64>> {
        match self.index_get(descriptor, key)? {
            Some(bytes) => Id::from_be_bytes(&bytes)
                .map(|id| vec![id.get()])
                .ok_or(Error::Corruption { page_id: 0 }),
            None => Ok(Vec::new()),
        }
    }

    /// Collect every `(full_key, value)` entry from an index B-tree
    /// whose key falls within `(start, end)`.
    fn collect_range(
        &self,
        descriptor: &obj_core::IndexDescriptor,
        start: std::ops::Bound<Vec<u8>>,
        end: std::ops::Bound<Vec<u8>>,
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        // Read mode pins every page read to the txn's snapshot (M12
        // #12): a concurrent writer's post-snapshot index entries must
        // not leak into a read txn's range/count. Write mode keeps the
        // live-pager scan so the txn observes its own uncommitted
        // index writes (it has no snapshot to pin against).
        match &self.mode {
            CollectionMode::Read(r) => {
                let root = PageId::new(descriptor.root_page_id)
                    .ok_or(Error::InvalidArgument("index root_page_id is zero"))?;
                let pager = lock_pager(r.env)?;
                let iter = BTree::<FileHandle>::range_via_snapshot(
                    &pager,
                    r.snapshot,
                    root,
                    (start, end),
                )?;
                let mut out = Vec::new();
                for step in iter {
                    out.push(step?);
                }
                Ok(out)
            }
            CollectionMode::Write(w) => {
                // #90: scan the LIVE index root from the per-txn cache
                // so an in-txn index scan sees its own uncommitted
                // entries (the open-time descriptor root may be stale).
                let root_raw =
                    self.write_index_root(w, &descriptor.name, descriptor.root_page_id)?;
                let root = PageId::new(root_raw)
                    .ok_or(Error::InvalidArgument("index root_page_id is zero"))?;
                let mut pager = lock_pager(w.env)?;
                let tree = BTree::<FileHandle>::open(&pager, root)?;
                let iter = tree.range(&mut pager, (start, end))?;
                let mut out = Vec::new();
                for step in iter {
                    out.push(step?);
                }
                Ok(out)
            }
            CollectionMode::Lazy(_) => Err(Error::ReadOnly {
                operation: "internal: lazy-mode collect_range",
            }),
        }
    }

    /// Resolve a `Vec<u64>` of `Id` integer values into concrete
    /// `T` documents via `self.get`. Preserves order; missing rows
    /// surface as `Error::Corruption` (orphan index entry).
    fn resolve_unique_ids(&self, ids: Vec<u64>) -> Result<Vec<T>> {
        let mut out = Vec::with_capacity(ids.len());
        for raw in ids {
            let id =
                Id::from_be_bytes(&raw.to_be_bytes()).ok_or(Error::Corruption { page_id: 0 })?;
            let doc = self.get(id)?.ok_or(Error::Corruption { page_id: 0 })?;
            out.push(doc);
        }
        Ok(out)
    }

    /// Count every entry in the primary tree WITHOUT decoding the
    /// documents. Used by the M8 [`crate::Query::count`] no-decode
    /// fast path; the iterator visits leaf pages and counts entries
    /// rather than running each through postcard.
    ///
    /// Power-of-ten Rule 2: bounded by the B+tree's `MAX_RANGE_NODES`
    /// budget (inherited from `BTree::range`).
    ///
    /// # Errors
    ///
    /// Pager / B-tree errors propagated.
    pub fn count_all(&self) -> Result<u64> {
        // Closure (rather than `Collection::count_all`) so the
        // higher-ranked lifetime on `dispatch_lazy`'s
        // `for<'a> FnOnce(&Collection<'a, T>) -> _` bound is
        // satisfied — `Collection::count_all` resolves to a single
        // lifetime fn-item type, which fails the HRTB check.
        #[allow(clippy::redundant_closure_for_method_calls)]
        if let Some(r) = self.dispatch_lazy(|c| c.count_all()) {
            return r;
        }
        // Read mode pins the full-tree scan to the txn's snapshot
        // (M12 #12) so a concurrent writer's post-snapshot inserts
        // cannot perturb the count; write mode keeps the live scan so
        // the txn sees its own uncommitted writes.
        match &self.mode {
            CollectionMode::Read(r) => {
                let pager = lock_pager(r.env)?;
                let pid = PageId::new(self.descriptor.primary_root)
                    .ok_or(Error::InvalidArgument("collection primary_root is zero"))?;
                let iter = BTree::<FileHandle>::range_via_snapshot(&pager, r.snapshot, pid, ..)?;
                count_range_iter(iter)
            }
            CollectionMode::Write(w) => {
                // #90: count the LIVE primary root so an in-txn count
                // reflects this txn's own uncommitted inserts/deletes.
                let root = self.write_primary_root(w)?;
                let mut pager = lock_pager(w.env)?;
                let tree = btree_handle(&pager, root)?;
                let iter = tree.range(&mut pager, ..)?;
                count_range_iter(iter)
            }
            CollectionMode::Lazy(_) => Err(Error::ReadOnly {
                operation: "internal: lazy-mode count_all",
            }),
        }
    }

    /// Count every entry whose encoded key falls inside `range` on
    /// the named index's B-tree, WITHOUT decoding any document. M8
    /// fast path for [`crate::Query::count`] when the source is an
    /// `index_range`.
    ///
    /// Returns the number of index B-tree entries — for an `Each`
    /// index that may exceed the document count (one doc emits
    /// multiple entries); for other kinds it equals the matching
    /// doc count.
    ///
    /// # Errors
    ///
    /// - [`Error::IndexNotFound`] if `index_name` is unknown / dropped.
    /// - Pager / B-tree errors propagated.
    pub fn count_index_range<R>(&self, index_name: &str, range: R) -> Result<u64>
    where
        R: std::ops::RangeBounds<obj_core::codec::Dynamic>,
    {
        let start = encode_dynamic_bound(range.start_bound())?;
        let end = encode_dynamic_bound(range.end_bound())?;
        self.count_index_range_encoded(index_name, start, end)
    }

    /// Encoded-bytes variant of [`Self::count_index_range`]. Bounds
    /// are the order-preserving field encoding of the user's
    /// `Dynamic` value(s); used by the query-layer count fast path.
    pub(crate) fn count_index_range_encoded(
        &self,
        index_name: &str,
        start_bound: std::ops::Bound<Vec<u8>>,
        end_bound: std::ops::Bound<Vec<u8>>,
    ) -> Result<u64> {
        if let Some(r) = self.dispatch_lazy(|c| {
            c.count_index_range_encoded(index_name, start_bound.clone(), end_bound.clone())
        }) {
            return r;
        }
        let descriptor = self.active_index(index_name)?;
        let (start, end) =
            crate::index_bound::widen_bounds_for_kind(start_bound, end_bound, descriptor.kind);
        let entries = self.collect_range(descriptor, start, end)?;
        // `entries.len()` is a `usize`; promote to `u64` carefully.
        // `usize` is at most 64 bits on every supported target.
        u64::try_from(entries.len()).map_err(|_| Error::BTreeInvariantViolated {
            reason: "index range entry count exceeds u64",
        })
    }

    /// Count distinct document `Id`s whose entries fall inside
    /// `range` on the named index's B-tree, WITHOUT decoding any
    /// document. For `Each` indexes this is the correct shape of
    /// the "how many docs match" question — `count_index_range`
    /// returns the entry count, which overshoots when a single doc
    /// contributes multiple entries.
    ///
    /// Implementation walks the index B-tree, parses the trailing
    /// 8-byte big-endian `Id` suffix from each non-unique key, and
    /// tracks the unique set in a bounded [`std::collections::HashSet`]
    /// capped at [`MAX_DISTINCT_IDS`]. Exceeding the cap surfaces
    /// [`Error::DistinctCountExceeded`] — the caller should narrow
    /// the range.
    ///
    /// # Per-kind semantics
    ///
    /// - `Standard`, `Composite`: equivalent to `count_index_range`
    ///   (one entry per doc by construction; the trailing-id-suffix
    ///   walk still produces the same total).
    /// - `Unique`: keys carry NO id suffix — the entry value is the
    ///   raw 8-byte `Id`; the walk reads the value instead.
    /// - `Each`: the dedup is meaningful — one doc may contribute
    ///   N entries under N distinct element keys.
    ///
    /// # Errors
    ///
    /// - [`Error::IndexNotFound`] if `index_name` is unknown / dropped.
    /// - [`Error::DistinctCountExceeded`] if the distinct set
    ///   exceeds [`MAX_DISTINCT_IDS`].
    /// - [`Error::Corruption`] if an entry's id suffix / value is
    ///   not parseable as an [`obj_core::Id`].
    /// - Pager / B-tree errors propagated.
    pub fn count_distinct_ids_in_range<R>(&self, index_name: &str, range: R) -> Result<u64>
    where
        R: std::ops::RangeBounds<obj_core::codec::Dynamic>,
    {
        let start = encode_dynamic_bound(range.start_bound())?;
        let end = encode_dynamic_bound(range.end_bound())?;
        self.count_distinct_ids_in_range_encoded(index_name, start, end)
    }

    /// Encoded-bytes variant of [`Self::count_distinct_ids_in_range`].
    /// Bounds are the order-preserving field encoding of the user's
    /// `Dynamic` value(s); used by the query-layer count fast path.
    pub(crate) fn count_distinct_ids_in_range_encoded(
        &self,
        index_name: &str,
        start_bound: std::ops::Bound<Vec<u8>>,
        end_bound: std::ops::Bound<Vec<u8>>,
    ) -> Result<u64> {
        if let Some(r) = self.dispatch_lazy(|c| {
            c.count_distinct_ids_in_range_encoded(
                index_name,
                start_bound.clone(),
                end_bound.clone(),
            )
        }) {
            return r;
        }
        let descriptor = self.active_index(index_name)?;
        let (start, end) =
            crate::index_bound::widen_bounds_for_kind(start_bound, end_bound, descriptor.kind);
        let entries = self.collect_range(descriptor, start, end)?;
        let mut distinct: HashSet<u64> = HashSet::new();
        for (full_key, value) in entries {
            let id = id_from_index_entry(&full_key, &value, descriptor.kind)?;
            if distinct.insert(id) && distinct.len() > MAX_DISTINCT_IDS {
                return Err(Error::DistinctCountExceeded {
                    limit: MAX_DISTINCT_IDS,
                });
            }
        }
        // `distinct.len()` is a `usize`; promote to `u64` carefully.
        // `usize` is at most 64 bits on every supported target.
        u64::try_from(distinct.len()).map_err(|_| Error::BTreeInvariantViolated {
            reason: "distinct id count exceeds u64",
        })
    }

    /// Materialise every `(Id, T)` pair in the collection.
    ///
    /// Implementation note: M6 returns an owned `Vec` rather than a
    /// streaming iterator because the B+tree range API borrows the
    /// pager, and threading that borrow through the mutex guards
    /// in the iterator chain is awkward.  M7+ may convert to a
    /// streaming shape once the index API is in place.
    ///
    /// # Errors
    ///
    /// Pager / B-tree / codec errors propagated.
    pub fn all(&self) -> Result<Vec<(Id, T)>> {
        // Closure (rather than `Collection::all`) for the same
        // HRTB reason documented on `count_all` above.
        #[allow(clippy::redundant_closure_for_method_calls)]
        if let Some(r) = self.dispatch_lazy(|c| c.all()) {
            return r;
        }
        match &self.mode {
            CollectionMode::Write(write) => {
                // #90: scan the LIVE primary root so an in-txn `all()`
                // includes this txn's own uncommitted inserts.
                let root = self.write_primary_root(write)?;
                let mut pager = lock_pager(write.env)?;
                scan_all::<T>(&mut pager, root, self.descriptor.collection_id)
            }
            CollectionMode::Read(read) => snapshot_scan_via_btree::<T>(
                read.snapshot,
                read.env,
                self.descriptor.primary_root,
                self.descriptor.collection_id,
            ),
            CollectionMode::Lazy(_) => Err(Error::ReadOnly {
                operation: "internal: lazy-mode all",
            }),
        }
    }

    fn write_or_err(&self, op: &'static str) -> Result<&WriteRef<'tx>> {
        match &self.mode {
            CollectionMode::Write(w) => Ok(w),
            CollectionMode::Read(_) | CollectionMode::Lazy(_) => {
                Err(Error::ReadOnly { operation: op })
            }
        }
    }

    /// If this handle is in `Lazy` mode, dispatch `body` through a
    /// fresh read transaction on the bound [`crate::Db`] — opening a
    /// transient [`Collection`] via [`Self::open_readonly_named`] and
    /// invoking the user-supplied closure on it. Returns `Some(_)`
    /// when the dispatch fires (the closure ran or the underlying
    /// open failed); returns `None` for non-Lazy handles so the
    /// caller can fall through to the existing logic.
    ///
    /// Power-of-ten Rule 4: keeps each public method's body small —
    /// the dispatch shim is one match arm instead of a per-method
    /// `if let CollectionMode::Lazy(_)` ladder.
    fn dispatch_lazy<R, F>(&self, body: F) -> Option<Result<R>>
    where
        F: FnOnce(&Collection<'_, T>) -> Result<R>,
    {
        match &self.mode {
            CollectionMode::Lazy(LazyRef { db }) => {
                // Clone the stored name into an owned `Cow` for the
                // transient handle opened inside the private read txn.
                // `into_owned` keeps the `'static` Cow bound the
                // `open_readonly_named` signature requires regardless
                // of whether the stored name was Borrowed or Owned.
                let name: Cow<'static, str> = Cow::Owned(self.collection_name.clone().into_owned());
                Some(db.read_transaction(move |tx| {
                    let coll = Collection::<T>::open_readonly_named(tx, name)?;
                    body(&coll)
                }))
            }
            _ => None,
        }
    }
}

// ---------- internal helpers --------------------------------------

fn lock_pager(
    env: &obj_core::TxnEnv<FileHandle>,
) -> Result<std::sync::MutexGuard<'_, Pager<FileHandle>>> {
    env.pager().lock().map_err(|_| Error::Busy {
        kind: obj_core::LockKind::WriterInProcess,
    })
}

/// Ensure `T::COLLECTION` exists in the catalog, lazy-creating an
/// empty primary B-tree on first call.  Used on the write side.
fn ensure_collection<T: Document>(
    inner: &obj_core::WriteTxn<'_, FileHandle>,
    catalog: &Arc<Mutex<Catalog<FileHandle>>>,
) -> Result<CollectionDescriptor> {
    let mut pager = inner.lock_pager()?;
    let mut catalog_guard = lock_catalog(catalog)?;
    if let Some(d) = catalog_guard.get(&mut pager, T::COLLECTION)? {
        return Ok(d);
    }
    let tree = BTree::<FileHandle>::empty(&mut pager)?;
    let descriptor = CollectionDescriptor::new(0, tree.root().get(), T::VERSION);
    let _id = catalog_guard.insert(&mut pager, T::COLLECTION, descriptor)?;
    catalog_guard
        .get(&mut pager, T::COLLECTION)?
        .ok_or(Error::Corruption { page_id: 0 })
}

/// Re-read the descriptor for `T::COLLECTION` after any catalog
/// mutation. Used by [`Collection::open_or_create`] after the
/// reconciler runs.
fn reread_descriptor<T: Document>(
    inner: &obj_core::WriteTxn<'_, FileHandle>,
    catalog: &Arc<Mutex<Catalog<FileHandle>>>,
) -> Result<CollectionDescriptor> {
    let mut pager = inner.lock_pager()?;
    let catalog_guard = lock_catalog(catalog)?;
    catalog_guard
        .get(&mut pager, T::COLLECTION)?
        .ok_or(Error::Corruption { page_id: 0 })
}

/// Reconcile `T::indexes()` against the catalog's stored descriptors
/// on the FIRST call per process per collection. Subsequent calls
/// observe the cache hit (in the shared `reconciled` set OR this txn's
/// `staged` set) and skip the catalog walk.
///
/// Reconciliation runs inside the user's WAL transaction so a
/// rolled-back txn leaves the catalog clean. If reconciliation
/// fails (e.g. `Error::IndexKindMismatch`), neither set is populated
/// so the next attempt re-runs the reconciler.
///
/// #93 — the skip-check is `shared ∪ staged`: a second handle of the
/// same collection within ONE txn still skips the (idempotent) catalog
/// walk via `staged`, but the name is recorded in the per-txn `staged`
/// set, NOT the shared set. [`crate::WriteTxn::commit`] promotes the
/// staged names into the shared set only after the WAL commit lands —
/// so a rolled-back lazy-create can never poison the shared cache into
/// skipping reconciliation on a later txn (the original bug).
fn reconcile_indexes_once<T: Document>(
    inner: &obj_core::WriteTxn<'_, FileHandle>,
    catalog: &Arc<Mutex<Catalog<FileHandle>>>,
    reconciled: &Arc<Mutex<HashSet<String>>>,
    staged: &mut HashSet<String>,
) -> Result<()> {
    // Fast path: already reconciled in a prior committed txn (shared)
    // or earlier in THIS txn (staged) — skip the catalog walk. Probe
    // `staged` first: it needs no lock and covers the repeat-handle
    // case; the shared probe takes the mutex only when `staged` misses.
    if staged.contains(T::COLLECTION) {
        return Ok(());
    }
    {
        let cache = lock_reconciled(reconciled)?;
        if cache.contains(T::COLLECTION) {
            return Ok(());
        }
    }
    let specs = T::indexes();
    // Validate specs before any catalog mutation so a bad spec
    // does not leave a half-reconciled catalog.
    for spec in &specs {
        spec.validate()?;
    }
    {
        let mut pager = inner.lock_pager()?;
        let mut catalog_guard = lock_catalog(catalog)?;
        let _post = catalog_guard.reconcile_indexes(&mut pager, T::COLLECTION, &specs)?;
    }
    // #93: stage in the PER-TXN set, not the shared one. Promotion to
    // the shared set is deferred to a successful `WriteTxn::commit`.
    staged.insert(T::COLLECTION.to_owned());
    Ok(())
}

fn lock_reconciled(
    reconciled: &Arc<Mutex<HashSet<String>>>,
) -> Result<std::sync::MutexGuard<'_, HashSet<String>>> {
    reconciled.lock().map_err(|_| Error::Busy {
        kind: obj_core::LockKind::WriterInProcess,
    })
}

/// Read-side descriptor lookup against a caller-supplied
/// collection name. M11 #93 introduced this byte-shape so the
/// namespace-aware [`Collection::open_readonly`] can perform the
/// catalog walk against either the calling Db's snapshot (no
/// namespace) or an attached Db's snapshot (with the namespace
/// prefix stripped).
fn read_descriptor_via_snapshot_named(
    env: &obj_core::TxnEnv<FileHandle>,
    snapshot: &obj_core::ReaderSnapshot<FileHandle>,
    name: &str,
) -> Result<Option<CollectionDescriptor>> {
    let pager = lock_pager(env)?;
    Catalog::<FileHandle>::lookup_via_snapshot(&pager, snapshot, name)
}

/// #83 (b): fused one-shot point read for [`crate::Db::get`].
///
/// Resolves the collection descriptor and the primary-tree value for
/// `id` under a SINGLE pager-mutex acquisition (see
/// [`snapshot_resolve_and_get`]), then decodes the bytes against the
/// resolved `collection_id`. This collapses the two back-to-back
/// pager locks the equivalent handle path pays (one to open the
/// handle / resolve the descriptor, one for the `get`).
///
/// Observably identical to
/// `tx.collection::<T>()?.get(id)` for the one-shot caller: the
/// descriptor lookup and the value get run on the same `ReadTxn`
/// snapshot, and a missing collection still surfaces as
/// [`Error::CollectionNotFound`]. Namespaced reads (`<ns>.<tail>`)
/// fall back to the handle path so the attached-snapshot dispatch is
/// unchanged.
pub(crate) fn fused_point_get<T: Document>(tx: &ReadTxn<'_>, id: Id) -> Result<Option<T>> {
    let (namespace, _tail) = crate::db::split_namespace(T::COLLECTION);
    if namespace.is_some() {
        // Namespaced: keep the existing attached-snapshot dispatch.
        return Collection::<T>::open_readonly(tx)?.get(id);
    }
    let key = id.to_be_bytes();
    let resolved =
        snapshot_resolve_and_get(tx.inner.snapshot(), tx.inner.env(), T::COLLECTION, &key)?;
    match resolved {
        Some((descriptor, bytes)) => Ok(Some(decode::<T>(&bytes, descriptor.collection_id)?)),
        None => Ok(None),
    }
}

/// Re-read the descriptor inside an already-locked pager + catalog
/// pair.  Surfaces a missing collection as `Error::Corruption`
/// because the caller has already opened a write txn against it.
fn catalog_get_required(
    pager: &mut Pager<FileHandle>,
    catalog: &Catalog<FileHandle>,
    name: &str,
) -> Result<CollectionDescriptor> {
    catalog
        .get(pager, name)?
        .ok_or(Error::Corruption { page_id: 0 })
}

fn btree_handle(pager: &Pager<FileHandle>, root: u64) -> Result<BTree<FileHandle>> {
    let root_pid =
        PageId::new(root).ok_or(Error::InvalidArgument("collection primary_root is zero"))?;
    BTree::<FileHandle>::open(pager, root_pid)
}

/// Drain a B-tree range iterator, counting entries WITHOUT retaining
/// their bytes. Shared by [`Collection::count_all`]'s live (write) and
/// snapshot-pinned (read) scan arms. Power-of-ten Rule 2: the
/// iterator carries its own `MAX_RANGE_NODES` budget; the `u64`
/// overflow check guards the count itself.
fn count_range_iter<I>(iter: I) -> Result<u64>
where
    I: Iterator<Item = Result<(Vec<u8>, Vec<u8>)>>,
{
    let mut n: u64 = 0;
    for step in iter {
        // Probe-only: drop the bytes the moment they decode.
        let _ = step?;
        n = n.checked_add(1).ok_or(Error::BTreeInvariantViolated {
            reason: "primary tree entry count exceeds u64",
        })?;
    }
    Ok(n)
}

// `persist_root` removed in M7 #58: every mutating method now
// routes through `index_maint::apply_doc_change`, which persists
// the descriptor (including the possibly-advanced `primary_root`)
// via `Catalog::update` after every per-index B-tree mutation.

/// Snapshot-consistent B-tree lookup (M6 #53).
///
/// Walks the primary B+tree rooted at `primary_root` using
/// [`obj_core::btree::BTree::get_via_snapshot`], which descends
/// through [`obj_core::ReaderSnapshot::read_page`] rather than the
/// live `Pager::read_page`. This bypasses the WAL `state.view` /
/// `state.pending` overlays — a concurrent writer's post-snapshot
/// COW commits cannot poison the reader's walk.
///
/// `primary_root` MUST be the descriptor's `primary_root` as-of
/// the snapshot's pinned LSN (i.e. the value read via
/// [`obj_core::Catalog::lookup_via_snapshot`] in
/// [`read_descriptor_via_snapshot`] above). Using the writer's
/// live `primary_root` would defeat the snapshot read.
fn snapshot_get_via_btree(
    snap: &obj_core::ReaderSnapshot<FileHandle>,
    env: &obj_core::TxnEnv<FileHandle>,
    primary_root: u64,
    key: &[u8],
) -> Result<Option<Vec<u8>>> {
    let pager = lock_pager(env)?;
    let root_pid = PageId::new(primary_root)
        .ok_or(Error::InvalidArgument("collection primary_root is zero"))?;
    obj_core::btree::BTree::<FileHandle>::get_via_snapshot(&pager, snap, root_pid, key)
}

/// #83 (b): fused single-lock read. Resolves the descriptor for
/// `name` and performs the primary-tree `get` for `key` under ONE
/// pager-mutex acquisition, against the SAME `snapshot` — collapsing
/// the descriptor-lookup lock (`read_descriptor_via_snapshot_named`)
/// and the value-get lock (`snapshot_get_via_btree`) that the
/// two-call handle path pays back-to-back.
///
/// Returns `(descriptor, value)` so the caller can `decode` against
/// the resolved `collection_id`. A missing collection surfaces as
/// `Err(CollectionNotFound)` (matching the handle path's open-time
/// contract for the one-shot caller); a present collection with no
/// entry for `key` surfaces as `Ok(None)`.
///
/// Power-of-ten: keeps poison → `Error::Busy` (Rule 7, via
/// `lock_pager`); ≤ 60 lines (Rule 4); `debug_assert`s that the
/// snapshot-resolved `primary_root` is the one fed to the get
/// (Rule 5).
fn snapshot_resolve_and_get(
    snap: &obj_core::ReaderSnapshot<FileHandle>,
    env: &obj_core::TxnEnv<FileHandle>,
    name: &str,
    key: &[u8],
) -> Result<Option<(CollectionDescriptor, Vec<u8>)>> {
    let pager = lock_pager(env)?;
    let Some(descriptor) = Catalog::<FileHandle>::lookup_via_snapshot(&pager, snap, name)? else {
        return Err(Error::CollectionNotFound {
            name: name.to_owned(),
        });
    };
    let root_pid = PageId::new(descriptor.primary_root)
        .ok_or(Error::InvalidArgument("collection primary_root is zero"))?;
    // Rule 5: the value-get MUST descend the descriptor's own
    // snapshot-time root — never a re-resolved or live root. A live
    // primary B-tree root is always page 1+, so a zero `collection_id`
    // here would mean we resolved a degenerate catalog row.
    debug_assert_eq!(
        root_pid.get(),
        descriptor.primary_root,
        "fused get must descend the snapshot-resolved primary_root",
    );
    let value =
        obj_core::btree::BTree::<FileHandle>::get_via_snapshot(&pager, snap, root_pid, key)?;
    Ok(value.map(|v| (descriptor, v)))
}

fn scan_all<T: Document>(
    pager: &mut Pager<FileHandle>,
    primary_root: u64,
    collection_id: u32,
) -> Result<Vec<(Id, T)>> {
    let tree = btree_handle(pager, primary_root)?;
    let iter = tree.range(pager, ..)?;
    let mut out = Vec::new();
    for entry in iter {
        let (key, value) = entry?;
        let id = Id::from_be_bytes(&key)
            .ok_or(Error::InvalidArgument("primary B-tree key is not an Id"))?;
        let doc = decode::<T>(&value, collection_id)?;
        out.push((id, doc));
    }
    Ok(out)
}

fn snapshot_scan_via_btree<T: Document>(
    _snap: &obj_core::ReaderSnapshot<FileHandle>,
    env: &obj_core::TxnEnv<FileHandle>,
    primary_root: u64,
    collection_id: u32,
) -> Result<Vec<(Id, T)>> {
    let mut pager = lock_pager(env)?;
    scan_all::<T>(&mut pager, primary_root, collection_id)
}

/// Encode the caller-supplied `Dynamic` value(s) into the bytes a
/// lookup against `descriptor` would use as a B-tree key. For
/// `Unique` indexes the result is the key bytes verbatim; for
/// non-unique kinds the lookup helpers extend with the per-doc
/// id suffix at scan time.
fn index_key_for_lookup(
    descriptor: &obj_core::IndexDescriptor,
    fields: &[obj_core::codec::Dynamic],
) -> Result<obj_core::index::EncodedIndexKey> {
    // Ref-based encode (#84): pass the descriptor's `kind` (Copy) and
    // `key_paths` BY REFERENCE — no transient `IndexSpec`, no
    // `name`/`key_paths` clone, no redundant `IndexSpec::validate` on
    // an already-validated on-disk descriptor. The byte output is
    // identical to the old `from_parts` + `encode_index_key` path
    // (see `encode_index_key_parts` and the byte-identity test).
    obj_core::index::encode_index_key_parts(descriptor.kind, &descriptor.key_paths, fields)
}

/// Encode a `Bound<&Dynamic>` into the index-key `Bound<Vec<u8>>` the
/// B-tree scan uses. Shared by the `Dynamic`-taking range methods on
/// [`Collection`].
///
/// A scalar `Dynamic` is encoded with the order-preserving field
/// encoder ([`obj_core::index::encode_field`]) — byte-identical to
/// what [`crate::Query::index_range`] produces, so a query and a
/// direct collection scan over the same scalar bound observe the
/// same entries. A [`Dynamic::Seq`](obj_core::codec::Dynamic::Seq)
/// bound is encoded as a composite key (the
/// [`COMPOSITE_TAG`](obj_core::index::COMPOSITE_TAG)-prefixed
/// concatenation of each element's field encoding) so a `Composite`
/// index can be range-scanned by a full tuple bound.
fn encode_dynamic_bound(
    b: std::ops::Bound<&obj_core::codec::Dynamic>,
) -> Result<std::ops::Bound<Vec<u8>>> {
    match b {
        std::ops::Bound::Included(v) => Ok(std::ops::Bound::Included(encode_bound_value(v)?)),
        std::ops::Bound::Excluded(v) => Ok(std::ops::Bound::Excluded(encode_bound_value(v)?)),
        std::ops::Bound::Unbounded => Ok(std::ops::Bound::Unbounded),
    }
}

/// Encode one `Dynamic` bound value into index-key bytes. Scalars go
/// through [`obj_core::index::encode_field`]; a `Seq` is encoded as a
/// composite tuple key. Power-of-ten Rule 4: kept separate so
/// [`encode_dynamic_bound`] stays a thin three-arm match.
fn encode_bound_value(v: &obj_core::codec::Dynamic) -> Result<Vec<u8>> {
    match v {
        obj_core::codec::Dynamic::Seq(fields) => {
            // `COMPOSITE_TAG || encode_field(f0) || encode_field(f1) ..`
            // is byte-identical to `encode_index_key`'s composite path
            // for the same fields — see `obj_core::index::key`.
            let mut out = vec![obj_core::index::COMPOSITE_TAG];
            for f in fields {
                out.extend_from_slice(obj_core::index::encode_field(f)?.as_bytes());
            }
            Ok(out)
        }
        _ => Ok(obj_core::index::encode_field(v)?.into_bytes()),
    }
}

/// Append 8 `0xFF` bytes to `prefix`. Used as the exclusive upper
/// bound of an equality lookup against a non-unique index: every
/// key with the same user-prefix is ≤ `prefix || 0xFF..` because
/// the trailing 8 bytes are an `Id` (`u64` BE).
fn append_max_id(prefix: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(prefix.len() + 8);
    out.extend_from_slice(prefix);
    out.extend_from_slice(&u64::MAX.to_be_bytes());
    out
}

/// Trim the trailing 8-byte id suffix off a non-unique index key.
/// For `Unique` keys the suffix is absent, so the full key is the
/// user portion.
fn strip_id_suffix(full_key: &[u8], kind: obj_core::IndexKind) -> Vec<u8> {
    match kind {
        obj_core::IndexKind::Unique => full_key.to_vec(),
        _ if full_key.len() >= 8 => full_key[..full_key.len() - 8].to_vec(),
        _ => full_key.to_vec(),
    }
}

/// Recover the `Id` (as a `u64`) from one index B-tree entry. For
/// non-unique kinds the id is the trailing 8 bytes of the KEY (the
/// suffix appended by the maintenance path); for `Unique` keys the
/// id is the VALUE. Used by
/// [`Collection::count_distinct_ids_in_range`].
fn id_from_index_entry(full_key: &[u8], value: &[u8], kind: obj_core::IndexKind) -> Result<u64> {
    // `Unique` indexes carry the id in the value; non-unique kinds
    // (Standard / Each / Composite) carry it as the trailing 8-byte
    // suffix of the key. The slicing here is O(1) — no per-entry
    // loop to bound (the outer walk's bound is the distinct-set cap).
    let bytes: &[u8] = if kind == obj_core::IndexKind::Unique {
        value
    } else {
        if full_key.len() < 8 {
            return Err(Error::Corruption { page_id: 0 });
        }
        &full_key[full_key.len() - 8..]
    };
    let id = Id::from_be_bytes(bytes).ok_or(Error::Corruption { page_id: 0 })?;
    Ok(id.get())
}

// =====================================================================
// Phase 7A (M14 #14) — streaming index range iterator
// =====================================================================

/// Resumption marker for [`IterIndexRange`]'s first refill. After the
/// first batch the iterator switches to `Excluded(last_emitted_full_key)`
/// for subsequent refills (the same shape `Db::iter_all` uses for the
/// primary tree).
enum InitialResume {
    Included(Vec<u8>),
    Excluded(Vec<u8>),
    Unbounded,
}

/// One entry in [`IterIndexRange`]'s pending buffer. Read/Write
/// modes stage `Pending(key, id)` and resolve the `T` lazily on
/// `next()`; Lazy mode pre-resolves under a single `read_transaction`
/// (to preserve snapshot consistency across the index walk + the
/// per-row primary `get`) and stages `Resolved(key, T)` directly.
enum StagedEntry<T> {
    Pending(Vec<u8>, Id),
    Resolved(Vec<u8>, T),
}

/// Streaming iterator returned by [`Collection::iter_range`]. Yields
/// `Result<(user_key_bytes, T)>` one row at a time; internally
/// refills a fixed-size `(user_key, Id)` buffer in batches of
/// `ITER_INDEX_RANGE_BATCH = 256` so the per-step pager-lock cost
/// amortises. Memory stays bounded at `O(batch × small_bytes +
/// distinct_ids)` regardless of the range's total size.
///
/// Held data: a `&'a Collection<'_, T>` borrow (the iterator is bound
/// to the lifetime of `Collection::iter_range`'s `&self` borrow), the
/// index's root page-id, the dedup set for `Each` indexes, the next-
/// chunk resumption marker, and the staged batch.
pub struct IterIndexRange<'a, T: Document> {
    coll: &'a Collection<'a, T>,
    descriptor_kind: obj_core::IndexKind,
    index_root: u64,
    /// First-refill marker — `None` after the iterator has emitted
    /// at least one chunk; subsequent refills use `last_full_key`.
    initial_resume: Option<InitialResume>,
    /// Last full B-tree key emitted by the most recent refill. Drives
    /// the `Excluded(_)` resumption bound for the next chunk.
    last_full_key: Option<Vec<u8>>,
    /// User-supplied end bound (already widened per index kind).
    end_bound: Bound<Vec<u8>>,
    /// Pre-staged entries from the most recent refill. `next()`
    /// pops from the front. Each entry is either `Pending(key, id)`
    /// (deferred get-back, the Read/Write streaming path) or
    /// `Resolved(key, T)` (eager get-back inside a single
    /// `read_transaction`, the Lazy-mode fallback).
    buffer: VecDeque<Result<StagedEntry<T>>>,
    /// Persistent de-dup set for `Each` indexes. Power-of-ten Rule
    /// 3: the set is intentionally unbounded — if the caller wants
    /// a hard cap they should use
    /// [`Collection::count_distinct_ids_in_range`] (which caps at
    /// [`MAX_DISTINCT_IDS`]); the iterator's correctness contract
    /// is per-row dedup across the whole range.
    emitted_ids: HashSet<u64>,
    finished: bool,
}

impl<T: Document> std::fmt::Debug for IterIndexRange<'_, T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("IterIndexRange")
            .field("descriptor_kind", &self.descriptor_kind)
            .field("index_root", &self.index_root)
            .field("buffer_len", &self.buffer.len())
            .field("emitted_ids_len", &self.emitted_ids.len())
            .field("finished", &self.finished)
            .finish_non_exhaustive()
    }
}

impl<T: Document> IterIndexRange<'_, T> {
    /// Refill `self.buffer` with up to [`ITER_INDEX_RANGE_BATCH`]
    /// `(user_key, Id)` pairs by walking the index B-tree from the
    /// current resumption marker. Sets `self.finished` when the
    /// underlying range scan yields fewer than the requested batch
    /// (i.e. it ran past the end bound).
    ///
    /// Power-of-ten Rule 7: per-step decode errors are pushed into
    /// the buffer as `Err(_)` so the caller observes them via
    /// `next()` rather than aborting iteration.
    fn refill(&mut self) -> Result<()> {
        let env = match &self.coll.mode {
            CollectionMode::Write(w) => w.env,
            CollectionMode::Read(r) => r.env,
            CollectionMode::Lazy(_) => {
                // Lazy mode falls back to the eager `index_range`
                // path in `iter_range` itself (see below). Reaching
                // refill in Lazy mode indicates an internal logic
                // error, NOT a recoverable corruption — surface as
                // a typed error rather than `unwrap`.
                return Err(Error::ReadOnly {
                    operation: "internal: iter_range refill in Lazy mode",
                });
            }
        };
        let root_pid = PageId::new(self.index_root)
            .ok_or(Error::InvalidArgument("index root_page_id is zero"))?;
        let start = self.next_start_bound();
        let end = clone_bound_ref(&self.end_bound);
        let mut pager = lock_pager(env)?;
        let tree = BTree::<FileHandle>::open(&pager, root_pid)?;
        let iter = tree.range(&mut pager, (start, end))?;
        let mut staged: VecDeque<Result<StagedEntry<T>>> =
            VecDeque::with_capacity(ITER_INDEX_RANGE_BATCH);
        let mut last_full: Option<Vec<u8>> = None;
        let mut consumed: usize = 0;
        for step in iter {
            if consumed >= ITER_INDEX_RANGE_BATCH {
                break;
            }
            consumed = consumed
                .checked_add(1)
                .ok_or(Error::BTreeInvariantViolated {
                    reason: "iter_range batch counter overflow",
                })?;
            self.stage_one(&mut staged, &mut last_full, step);
        }
        if consumed < ITER_INDEX_RANGE_BATCH {
            self.finished = true;
        }
        drop(pager);
        self.buffer.extend(staged);
        if let Some(k) = last_full {
            self.last_full_key = Some(k);
        }
        Ok(())
    }

    /// Process one B-tree step into the staged batch. Encapsulates
    /// the `Each`-dedup, the trailing-id-suffix strip, and the
    /// `Id::from_be_bytes` parse. Free helper so the refill body
    /// stays under the Rule-4 60-line ceiling.
    fn stage_one(
        &mut self,
        staged: &mut VecDeque<Result<StagedEntry<T>>>,
        last_full: &mut Option<Vec<u8>>,
        step: Result<(Vec<u8>, Vec<u8>)>,
    ) {
        let (full_key, id_bytes) = match step {
            Ok(kv) => kv,
            Err(e) => {
                staged.push_back(Err(e));
                return;
            }
        };
        *last_full = Some(full_key.clone());
        let Some(id) = Id::from_be_bytes(&id_bytes) else {
            staged.push_back(Err(Error::Corruption { page_id: 0 }));
            return;
        };
        if self.descriptor_kind == obj_core::IndexKind::Each && !self.emitted_ids.insert(id.get()) {
            // Same doc already emitted under a different element
            // key — skip without producing an output entry.
            return;
        }
        let user_key = strip_id_suffix(&full_key, self.descriptor_kind);
        staged.push_back(Ok(StagedEntry::Pending(user_key, id)));
    }

    /// Compute the start bound for the next refill: use
    /// `initial_resume` on the first call (consuming it), thereafter
    /// use `Excluded(last_full_key)`.
    fn next_start_bound(&mut self) -> Bound<Vec<u8>> {
        if let Some(initial) = self.initial_resume.take() {
            return match initial {
                InitialResume::Included(k) => Bound::Included(k),
                InitialResume::Excluded(k) => Bound::Excluded(k),
                InitialResume::Unbounded => Bound::Unbounded,
            };
        }
        match &self.last_full_key {
            Some(k) => Bound::Excluded(k.clone()),
            None => Bound::Unbounded,
        }
    }
}

impl<T: Document> Iterator for IterIndexRange<'_, T> {
    type Item = Result<(Vec<u8>, T)>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if let Some(staged) = self.buffer.pop_front() {
                return Some(self.resolve_one(staged));
            }
            if self.finished {
                return None;
            }
            if let Err(e) = self.refill() {
                // Latch the iterator shut on a refill failure
                // (lock acquisition, B-tree open, etc.). Surface
                // the error once, then return None on subsequent
                // calls — power-of-ten Rule 7.
                self.finished = true;
                return Some(Err(e));
            }
            // refill ran; the loop will pop or notice finished.
        }
    }
}

impl<T: Document> IterIndexRange<'_, T> {
    /// Resolve one staged entry into a `(user_key, T)` pair. For
    /// `Pending(_, id)` entries (the Read/Write streaming path),
    /// calls [`Collection::get`] to decode `T` on demand; for
    /// `Resolved(_, T)` entries (the Lazy-mode eager path), returns
    /// the already-decoded value. Orphan index entries (id missing
    /// in the primary tree) surface as [`Error::Corruption`],
    /// matching [`Collection::index_range`]'s existing contract.
    fn resolve_one(&self, staged: Result<StagedEntry<T>>) -> Result<(Vec<u8>, T)> {
        match staged? {
            StagedEntry::Pending(user_key, id) => match self.coll.get(id)? {
                Some(doc) => Ok((user_key, doc)),
                None => Err(Error::Corruption { page_id: 0 }),
            },
            StagedEntry::Resolved(user_key, doc) => Ok((user_key, doc)),
        }
    }
}

/// Clone a `&Bound<Vec<u8>>` into an owned `Bound<Vec<u8>>`. Takes a
/// borrowed owned bound (the shape `IterIndexRange::end_bound`
/// stores) and hands back an owned copy for the resumption walk.
fn clone_bound_ref(b: &Bound<Vec<u8>>) -> Bound<Vec<u8>> {
    match b {
        Bound::Included(v) => Bound::Included(v.clone()),
        Bound::Excluded(v) => Bound::Excluded(v.clone()),
        Bound::Unbounded => Bound::Unbounded,
    }
}