commonware_storage/qmdb/immutable/

compact.rs

//! An immutable authenticated db that discards historical operations, retaining only a witness
//! for each applied batch.
//!
//! Mirrors the API of [`crate::qmdb::immutable::Immutable`] (`new_batch -> merkleize ->
//! apply_batch -> commit / sync / start_sync`, pipelined batch chains, `StaleBatch` validation)
//! but is backed by the peak-only [`crate::merkle::compact`]. Because history is discarded,
//! the db has no `get` / `proof` / `bounds` methods. A merkleized batch can prove only its own
//! operations, and only until it is applied. Use the full variant for historical proofs.
//!
//! # Witness journal
//!
//! The witness journal holds a complete snapshot of every applied batch. [`Db::init`]
//! restores a retained applied state within its operation cap. [`Db::prune`] bounds the retained
//! history. Initialization restores the db's in-memory state from an entry. The Merkle is rebuilt
//! from the stored pinned nodes and operation, and the commit fields are decoded from the
//! operation. An entry that cannot rebuild surfaces as [`Error::DataCorrupted`]. Each retained
//! witness also serves compact sync at its size without retaining historical operations.
//!
//! # Inactivity floor
//!
//! Commits carry an inactivity floor for wire-format compatibility with
//! [`crate::qmdb::immutable::Immutable`]: the root is computed over the encoded operation
//! sequence, and that sequence must include the same floor to produce the same root as the
//! full variant. The floor has no effect on pruning or snapshot rebuilding here; all
//! historical in-memory state is discarded whenever a batch is applied.

use super::operation::Operation;
pub use crate::qmdb::compact::Config;
use crate::{
    Context,
    merkle::{Family, Location, Proof, batch, compact as compact_merkle},
    qmdb::{
        self, Error,
        any::value::ValueEncoding,
        chain::{self, Bounds, Commitment},
        compact::{batch as compact_batch, witness},
        operation::Key,
        sync::{CompactTarget, Request, Source, source},
    },
};
use commonware_codec::{Encode, EncodeShared, Read};
use commonware_cryptography::{Digest, Hasher};
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use commonware_runtime::Handle;
use core::marker::PhantomData;
use std::{
    collections::BTreeMap,
    sync::{Arc, Weak},
};

/// An immutable authenticated db that discards historical operations, retaining only a witness
/// for each applied batch.
pub struct Db<F, E, K, V, H, S: Strategy>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: Read,
{
    merkle: compact_merkle::Merkle<F, H::Digest, S>,
    last_commit_metadata: Option<V::Value>,
    inactivity_floor_loc: Location<F>,
    commit_codec_config: <Operation<F, K, V> as Read>::Cfg,
    witness: witness::Store<E, F, H::Digest>,
    _key: PhantomData<K>,
}

impl<F, E, K, V, H, S: Strategy> std::fmt::Debug for Db<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: Read,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Db")
            .field("size", &self.size())
            .field("inactivity_floor_loc", &self.inactivity_floor_loc())
            .finish_non_exhaustive()
    }
}

/// Result of merkleizing a batch.
type MerkleizeResult<F, D, K, V, S> = Result<Arc<MerkleizedBatch<F, D, K, V, S>>, Error<F>>;

/// A speculative batch for a compact immutable db.
#[allow(clippy::type_complexity)]
pub struct UnmerkleizedBatch<F, H, K, V, S: Strategy>
where
    F: Family,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
{
    merkle_batch: compact_merkle::UnmerkleizedBatch<F, H::Digest, S>,
    mutations: BTreeMap<K, V::Value>,
    parent: Option<Arc<MerkleizedBatch<F, H::Digest, K, V, S>>>,
    base: chain::Commitment<F, H::Digest>,
}

/// A speculative batch whose root digest has been computed.
#[derive(Clone)]
pub struct MerkleizedBatch<F: Family, D: Digest, K: Key, V: ValueEncoding, S: Strategy>
where
    Operation<F, K, V>: EncodeShared,
{
    pub(super) merkle_batch: Arc<batch::MerkleizedBatch<F, D, S>>,
    operations: Arc<Vec<Operation<F, K, V>>>,
    pub(super) commit_metadata: Option<V::Value>,
    pub(super) parent: Option<Weak<Self>>,
    pub(super) bounds: chain::Bounds<F, D>,
}

impl<F: Family, D: Digest, K: Key, V: ValueEncoding, S: Strategy> MerkleizedBatch<F, D, K, V, S>
where
    Operation<F, K, V>: EncodeShared,
{
    pub(super) fn ancestors(&self) -> impl Iterator<Item = Arc<Self>> + use<F, D, K, V, S> {
        chain::ancestors(self.parent.clone(), |batch| batch.parent.as_ref())
    }

    /// The [`Commitment`] this batch commits to.
    pub(super) const fn commitment(&self) -> Commitment<F, D> {
        self.bounds.tip
    }

    /// Return the root digest after this batch is applied.
    pub const fn root(&self) -> D {
        self.bounds.tip.root
    }

    /// Return the [`Bounds`] of the batch.
    pub const fn bounds(&self) -> &Bounds<F, D> {
        &self.bounds
    }

    /// Return the operations this batch appends to the log and the location of the first.
    #[allow(clippy::type_complexity)]
    pub fn operations(&self) -> (Location<F>, Arc<Vec<Operation<F, K, V>>>) {
        (self.bounds.base.size, Arc::clone(&self.operations))
    }

    /// Inclusion proof for the operations returned by [`Self::operations`], anchored at
    /// this batch's tip. The pair verifies against [`Self::root`] via
    /// [`crate::qmdb::verify_proof`]. Together with [`Self::pinned_nodes`] they verify via
    /// [`crate::qmdb::verify_proof_and_pinned_nodes`].
    ///
    /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be
    /// held. Nodes below the chain are read from `db`'s [Merkle store][crate::merkle::mem::Mem],
    /// which retains them at least until this batch's changes are applied (applying it or a
    /// descendant prunes the store to its frontier).
    ///
    /// # Errors
    ///
    /// Returns [`Error::StaleRead`] if `db` is not at a state of this batch's chain,
    /// [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or belongs to a
    /// dropped unapplied ancestor, and [`crate::merkle::Error::Empty`] if the batch has no
    /// operations (such as a batch from [`Db::to_batch`]).
    pub fn proof<E, H>(&self, db: &Db<F, E, K, V, H, S>) -> Result<Proof<F, D>, Error<F>>
    where
        E: Context,
        H: Hasher<Digest = D>,
        Operation<F, K, V>: Read,
    {
        let db = self.bounds.compatible(db, db.commitment())?;
        let inactive_peaks = F::inactive_peaks(self.bounds.tip.size, self.bounds.inactivity_floor);
        let hasher = qmdb::hasher::<H>();
        self.merkle_batch
            .range_proof(
                db.merkle.mem(),
                &hasher,
                self.bounds.base.size..self.bounds.tip.size,
                inactive_peaks,
            )
            .map_err(Into::into)
    }

    /// The Merkle frontier at the first operation returned by [`Self::operations`]
    /// ([`Family::nodes_to_pin`]), which lets a consumer holding only this batch's base rebuild
    /// compact state and replay the operations. The operations, [`Self::proof`], and pinned
    /// nodes verify against [`Self::root`] via [`crate::qmdb::verify_proof_and_pinned_nodes`].
    ///
    /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be
    /// held. Nodes below the chain are read from `db`'s [Merkle store][crate::merkle::mem::Mem],
    /// which retains them at least until this batch's changes are applied (applying it or a
    /// descendant prunes the store to its frontier).
    ///
    /// # Errors
    ///
    /// Returns [`Error::StaleRead`] if `db` is not at a state of this batch's chain, and
    /// [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or belongs to a
    /// dropped unapplied ancestor.
    pub fn pinned_nodes<E, H>(&self, db: &Db<F, E, K, V, H, S>) -> Result<Vec<D>, Error<F>>
    where
        E: Context,
        H: Hasher<Digest = D>,
        Operation<F, K, V>: Read,
    {
        let db = self.bounds.compatible(db, db.commitment())?;
        let base = db.merkle.mem();
        F::nodes_to_pin(self.bounds.base.size)
            .map(|pos| {
                self.merkle_batch
                    .get_node(pos)
                    .or_else(|| base.get_node(pos))
                    .ok_or(crate::merkle::Error::ElementPruned(pos))
            })
            .collect::<Result<Vec<_>, _>>()
            .map_err(Into::into)
    }

    /// Create a new speculative batch with this one as its parent.
    ///
    /// Every unapplied ancestor in the chain must be held until the child (or any descendant) is
    /// merkleized. Otherwise, `merkleize` returns [`Error::StaleBatch`].
    pub fn new_batch<H>(self: &Arc<Self>) -> UnmerkleizedBatch<F, H, K, V, S>
    where
        H: Hasher<Digest = D>,
    {
        UnmerkleizedBatch {
            merkle_batch: compact_merkle::UnmerkleizedBatch::wrap(self.merkle_batch.new_batch()),
            mutations: BTreeMap::new(),
            parent: Some(Arc::clone(self)),
            base: self.commitment(),
        }
    }
}

impl<F, H, K, V, S> UnmerkleizedBatch<F, H, K, V, S>
where
    F: Family,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    S: Strategy,
    Operation<F, K, V>: EncodeShared,
{
    pub(super) fn new<E>(db: &Db<F, E, K, V, H, S>, base: chain::Commitment<F, H::Digest>) -> Self
    where
        E: Context,
        Operation<F, K, V>: Read,
    {
        Self {
            merkle_batch: db.merkle.new_batch(),
            mutations: BTreeMap::new(),
            parent: None,
            base,
        }
    }

    /// The database boundary for this batch chain.
    ///
    /// A batch created from the database uses its base. A child inherits its parent's `db`.
    fn db(&self) -> Commitment<F, H::Digest> {
        self.parent
            .as_ref()
            .map_or(self.base, |parent| parent.bounds.db)
    }

    pub fn set(mut self, key: K, value: V::Value) -> Self {
        self.mutations.insert(key, value);
        self
    }

    /// Resolve mutations into operations, merkleize, and return the batch.
    ///
    /// `inactivity_floor` is threaded through the commit operation for wire-format parity with
    /// [`crate::qmdb::immutable::Immutable`]. It must be at least the floor this batch builds on
    /// (its parent's, or the database's) and at most the batch's commit location
    /// (`total_size - 1`). These bounds are validated, but the floor does not drive any local
    /// pruning or retention in this variant.
    ///
    /// # Errors
    ///
    /// - Returns [`Error::StaleBatch`] if `db` is not at a state of this batch's chain or an
    ///   unapplied ancestor was dropped.
    /// - Returns [`Error::FloorRegressed`] if `inactivity_floor` is below the floor this batch
    ///   builds on, and [`Error::FloorBeyondSize`] if it is past the commit location.
    #[tracing::instrument(
        name = "qmdb.immutable.compact.batch.merkleize",
        level = "info",
        skip_all
    )]
    pub async fn merkleize<E>(
        self,
        db: &Db<F, E, K, V, H, S>,
        metadata: Option<V::Value>,
        inactivity_floor: Location<F>,
    ) -> MerkleizeResult<F, H::Digest, K, V, S>
    where
        E: Context,
        Operation<F, K, V>: Read,
    {
        let live_ancestors: Vec<_> =
            chain::parent_and_ancestors(self.parent.as_ref(), |parent| parent.ancestors())
                .collect();
        let boundary = chain::effective_boundary(
            self.db(),
            live_ancestors.last().map(|oldest| oldest.bounds.base),
        );

        let ancestors: Vec<_> = live_ancestors
            .iter()
            .map(|batch| batch.commitment())
            .collect();
        let db = chain::merkleizable(db, db.commitment(), boundary, ancestors.iter().copied())?;

        let mut ops: Vec<Operation<F, K, V>> = Vec::with_capacity(self.mutations.len() + 1);
        for (key, value) in self.mutations {
            ops.push(Operation::Set(key, value));
        }
        ops.push(Operation::Commit(metadata.clone(), inactivity_floor));

        let operations = Arc::new(ops);
        let total_size = self.base.size + operations.len() as u64;
        chain::validate_merkleize_floor::<F>(
            self.parent.as_ref().map_or_else(
                || db.inactivity_floor_loc(),
                |parent| parent.bounds.inactivity_floor,
            ),
            inactivity_floor,
            total_size - 1,
        )?;
        let inactive_peaks = F::inactive_peaks(total_size, inactivity_floor);
        let (merkle, root) = compact_batch::merkleize_ops::<F, H, S, _>(
            &db.merkle,
            self.merkle_batch,
            Arc::clone(&operations),
            inactive_peaks,
        )
        .await?;

        // Hold ancestor batches until their operations and nodes have been captured.
        drop(live_ancestors);

        Ok(Arc::new(MerkleizedBatch {
            merkle_batch: merkle,
            operations,
            commit_metadata: metadata,
            parent: self.parent.as_ref().map(Arc::downgrade),
            bounds: chain::Bounds {
                base: self.base,
                db: boundary,
                tip: Commitment::new(total_size, root),
                ancestors,
                inactivity_floor,
            },
        }))
    }
}

impl<F, E, K, V, H, S> Db<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    S: Strategy,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: Read,
{
    /// Initialize from a retained witness.
    /// `Some(max_size)` selects the latest retained witness for at most `max_size` operations.
    /// `None` selects the latest retained state.
    ///
    /// Fresh storage receives a durable bootstrap commit and witness.
    #[boxed]
    pub async fn init(
        context: E,
        cfg: Config<<Operation<F, K, V> as Read>::Cfg, S>,
        max_size: Option<Location<F>>,
    ) -> Result<Self, Error<F>> {
        let Config {
            strategy,
            witness: witness_config,
            commit_codec_config,
        } = cfg;

        // Reconstruct and verify the selected witness before exposing any state derived from it.
        let mut merkle = compact_merkle::Merkle::new(strategy);
        let (witness, last_commit_op) = witness::init::<E, F, H, S, Operation<F, K, V>>(
            context.child("witness"),
            witness_config,
            max_size,
            &mut merkle,
            &commit_codec_config,
            Operation::<F, K, V>::Commit(None, Location::new(0))
                .encode()
                .to_vec(),
        )
        .await?;

        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
            return Err(Error::DataCorrupted("last operation was not a commit"));
        };

        Ok(Self {
            merkle,
            last_commit_metadata,
            inactivity_floor_loc,
            commit_codec_config,
            witness,
            _key: PhantomData,
        })
    }

    fn encode_commit_op(metadata: Option<V::Value>, inactivity_floor_loc: Location<F>) -> Vec<u8> {
        Operation::<F, K, V>::Commit(metadata, inactivity_floor_loc)
            .encode()
            .to_vec()
    }

    /// Build a compact db from state fetched by the sync engine.
    ///
    /// The imported witness lives only in memory until the first [`Self::apply_batch`],
    /// [`Self::commit`], [`Self::sync`], or [`Self::start_sync`]. Applying a batch replaces it with
    /// the newly applied journal checkpoint. A durability method journals it directly. Until one of
    /// those operations succeeds, prune is rejected.
    pub(crate) fn init_from_sync(
        strategy: S,
        journal: witness::Journal<E, F, H::Digest>,
        commit_codec_config: <Operation<F, K, V> as Read>::Cfg,
        last_commit_loc: Location<F>,
        pinned_nodes: Vec<H::Digest>,
        last_commit_op: Operation<F, K, V>,
    ) -> Result<Self, Error<F>> {
        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
            return Err(Error::UnexpectedData(last_commit_loc));
        };
        witness::validate_inactivity_floor(inactivity_floor_loc, last_commit_loc)?;

        let op_bytes = Self::encode_commit_op(last_commit_metadata.clone(), inactivity_floor_loc);
        let mut merkle =
            compact_merkle::Merkle::from_compact_state(strategy, last_commit_loc, pinned_nodes)?;
        let hasher = qmdb::hasher::<H>();
        merkle.append_leaf(&hasher, &op_bytes)?;
        let imported = witness::build_witness::<F, H, S>(&merkle, inactivity_floor_loc, op_bytes)?;
        merkle.prune_to_frontier();

        let witness = witness::Store::from_import(journal, imported);
        Ok(Self {
            merkle,
            last_commit_metadata,
            inactivity_floor_loc,
            commit_codec_config,
            witness,
            _key: PhantomData,
        })
    }

    /// Return the root of the db.
    pub const fn root(&self) -> H::Digest {
        self.witness.tip().root
    }

    /// Return the inactivity floor declared by the last committed batch.
    pub const fn inactivity_floor_loc(&self) -> Location<F> {
        self.inactivity_floor_loc
    }

    /// Return the location of the next operation appended to this db.
    pub const fn size(&self) -> Location<F> {
        self.witness.tip().size()
    }

    /// Get the metadata associated with the last commit.
    pub fn get_metadata(&self) -> Option<V::Value> {
        self.last_commit_metadata.clone()
    }

    /// Return the compact-sync target described by the current witness.
    ///
    /// This reflects the most recently applied batch. The target remains non-durable until a
    /// covering [`Self::commit`], [`Self::sync`], or [`Self::start_sync`] completes.
    pub const fn target(&self) -> CompactTarget<F, H::Digest> {
        self.witness.tip().target()
    }

    /// The [`Commitment`] for the database's current state.
    pub(crate) const fn commitment(&self) -> chain::Commitment<F, H::Digest> {
        chain::Commitment::new(self.size(), self.root())
    }

    /// Create a new speculative batch of operations with this database as its parent.
    pub fn new_batch(&self) -> UnmerkleizedBatch<F, H, K, V, S> {
        UnmerkleizedBatch::new(self, self.commitment())
    }

    /// Create an owned merkleized batch representing the current applied state.
    pub fn to_batch(&self) -> Arc<MerkleizedBatch<F, H::Digest, K, V, S>> {
        Arc::new(MerkleizedBatch {
            merkle_batch: self.merkle.to_batch(),
            operations: Arc::new(Vec::new()),
            commit_metadata: self.last_commit_metadata.clone(),
            parent: None,
            bounds: chain::Bounds::from_db(self.commitment(), self.inactivity_floor_loc),
        })
    }

    /// Check that `batch` can be applied to the database in its current state, without
    /// applying it.
    ///
    /// [`Self::apply_batch`] runs the same validation but consumes the database when it
    /// fails; callers that want to reject a bad batch and keep the handle can check first.
    pub fn validate_batch(
        &self,
        batch: &MerkleizedBatch<F, H::Digest, K, V, S>,
    ) -> Result<(), Error<F>> {
        batch.bounds.validate_apply_to(self.commitment())
    }

    /// Apply a merkleized batch to the database.
    ///
    /// The state is updated in memory and appended to the witness journal. Call [`Self::commit`]
    /// or [`Self::sync`], or await the handle returned by [`Self::start_sync`], to make the
    /// applied state durable.
    ///
    /// # Errors
    ///
    /// Returns [`Error::StaleBatch`] if the database is at neither the batch's recorded database
    /// boundary nor an ancestor's tip.
    #[tracing::instrument(
        name = "qmdb.immutable.compact.db.apply_batch",
        level = "info",
        skip_all
    )]
    pub async fn apply_batch(
        mut self,
        batch: Arc<MerkleizedBatch<F, H::Digest, K, V, S>>,
    ) -> Result<Self, Error<F>> {
        self.validate_batch(&batch)?;
        self.merkle.apply_batch(&batch.merkle_batch)?;
        self.last_commit_metadata = batch.commit_metadata.clone();
        self.inactivity_floor_loc = batch.bounds.inactivity_floor;
        let last_commit_metadata = self.last_commit_metadata.clone();
        let inactivity_floor_loc = self.inactivity_floor_loc;
        self.witness = self
            .witness
            .apply::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
            })
            .await?;
        assert_eq!(self.commitment(), batch.bounds.tip);
        Ok(self)
    }

    /// Begin durably persisting the current db state to disk.
    ///
    /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit],
    /// plus a best-effort attempt to bound the recovery needed on reopen. Use [Self::sync] to
    /// guarantee none is needed. A new sync waits for the prior sync before starting. Failures
    /// of the deferred durability work surface on the returned handle and the next durability
    /// operation.
    #[tracing::instrument(
        name = "qmdb.immutable.compact.db.start_sync",
        level = "info",
        skip_all
    )]
    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
        let last_commit_metadata = self.last_commit_metadata.clone();
        let inactivity_floor_loc = self.inactivity_floor_loc;
        let handle;
        (self.witness, handle) = self
            .witness
            .start_sync::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
            })
            .await?;
        Ok((self, handle))
    }

    /// Durably persist the current db state to disk. This is faster than [`Self::sync`] but
    /// reopen may need to replay the witness journal's tail to recover.
    #[tracing::instrument(name = "qmdb.immutable.compact.db.commit", level = "info", skip_all)]
    pub async fn commit(mut self) -> Result<Self, Error<F>> {
        let last_commit_metadata = self.last_commit_metadata.clone();
        let inactivity_floor_loc = self.inactivity_floor_loc;
        self.witness = self
            .witness
            .commit::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
            })
            .await?;
        Ok(self)
    }

    /// Durably persist the current db state to disk, also persisting journal metadata to
    /// minimize recovery work on reopen.
    #[tracing::instrument(name = "qmdb.immutable.compact.db.sync", level = "info", skip_all)]
    pub async fn sync(mut self) -> Result<Self, Error<F>> {
        let last_commit_metadata = self.last_commit_metadata.clone();
        let inactivity_floor_loc = self.inactivity_floor_loc;
        self.witness = self
            .witness
            .sync::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
            })
            .await?;
        Ok(self)
    }

    /// Drop witnesses for commits with fewer than `pruning_boundary` operations. Some witness below
    /// the boundary may survive.
    ///
    /// Bounded initialization and compact sync can only use witnesses that survive. The current
    /// commit's witness always survives. The prune is made durable before this method returns.
    ///
    /// # Errors
    ///
    /// Fails if a compact-sync import has not yet been applied to the witness journal.
    #[tracing::instrument(name = "qmdb.immutable.compact.db.prune", level = "info", skip_all)]
    pub async fn prune(mut self, pruning_boundary: Location<F>) -> Result<Self, Error<F>> {
        self.witness = self.witness.prune(pruning_boundary).await?;
        Ok(self)
    }

    /// Destroy all persisted state associated with this database.
    #[boxed]
    pub async fn destroy(self) -> Result<(), Error<F>> {
        self.witness.destroy().await?;
        Ok(())
    }
}

impl<F, E, K, V, H, S> Source for Db<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared + Read,
    S: Strategy,
{
    type Family = F;
    type Digest = H::Digest;
    type Op = Operation<F, K, V>;
    type Error = qmdb::Error<F>;

    async fn serve(&self, request: Request<F>) -> source::Result<Self> {
        let response = self
            .witness
            .compact_state::<H, S, Self::Op>(
                self.merkle.strategy(),
                &self.commit_codec_config,
                request,
            )
            .await?;
        Ok((response, None))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        journal::contiguous::{Contiguous as _, fixed, variable::Config as JournalConfig},
        merkle::{mmb, mmr},
        metadata::{Config as MetadataConfig, Metadata},
        qmdb::{
            any::value::FixedEncoding, compact::witness, verify_proof,
            verify_proof_and_pinned_nodes,
        },
    };
    use commonware_cryptography::{Sha256, sha256::Digest};
    use commonware_macros::test_traced;
    use commonware_parallel::Sequential;
    use commonware_runtime::{
        Blob as _, Runner as _, Storage as _, Supervisor as _,
        buffer::paged::CacheRef,
        deterministic,
        mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs},
    };
    use commonware_utils::{NZU16, NZU64, NZUsize, sequence::VecU64};
    use core::future::Future;
    use std::num::{NonZeroU16, NonZeroUsize};

    type TestDb<F> =
        Db<F, deterministic::Context, Digest, FixedEncoding<Digest>, Sha256, Sequential>;

    const WITNESS_PAGE_SIZE: NonZeroU16 = NZU16!(77);
    const WITNESS_PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9);

    fn witness_config(partition: &str) -> JournalConfig<()> {
        JournalConfig {
            partition: format!("{partition}-witness"),
            items_per_section: NZU64!(64),
            compression: None,
            codec_config: (),
            page_cache: CacheRef::new(WITNESS_PAGE_SIZE, WITNESS_PAGE_CACHE_SIZE),
            write_buffer: NZUsize!(1024),
            replay_buffer: NZUsize!(1024),
        }
    }

    async fn open_db<F: Family>(context: deterministic::Context, partition: &str) -> TestDb<F> {
        let witness_cfg = witness_config(partition);
        let cfg = Config {
            strategy: Sequential,
            witness: witness_cfg,
            commit_codec_config: (),
        };
        Db::init(context, cfg, None).await.unwrap()
    }

    async fn compact_merkleize_foreign_db_inner<F: Family>(context: deterministic::Context) {
        let db = open_db::<F>(context.child("db"), "immutable-merkleize-foreign-db").await;
        let foreign = open_db::<F>(
            context.child("foreign"),
            "immutable-merkleize-foreign-db-foreign",
        )
        .await;

        let batch = db
            .new_batch()
            .set(Sha256::fill(1), Sha256::fill(11))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let db = db.apply_batch(batch).await.unwrap();
        let batch = foreign
            .new_batch()
            .set(Sha256::fill(2), Sha256::fill(99))
            .merkleize(&foreign, None, Location::new(0))
            .await
            .unwrap();
        let foreign = foreign.apply_batch(batch).await.unwrap();
        assert_eq!(db.size(), foreign.size());
        assert_ne!(db.root(), foreign.root());

        let batch = db.new_batch().set(Sha256::fill(3), Sha256::fill(22));
        assert!(matches!(
            batch.merkleize(&foreign, None, Location::new(0)).await,
            Err(Error::StaleBatch)
        ));

        let parent = db
            .new_batch()
            .set(Sha256::fill(4), Sha256::fill(33))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let child = parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(5), Sha256::fill(44));
        assert!(matches!(
            child.merkleize(&foreign, None, Location::new(0)).await,
            Err(Error::StaleBatch)
        ));
        db.destroy().await.unwrap();
        foreign.destroy().await.unwrap();
    }

    async fn compact_merkleize_stale_sibling_inner<F: Family>(context: deterministic::Context) {
        let db = open_db::<F>(context.child("db"), "immutable-merkleize-stale-sibling").await;

        let direct = db.new_batch().set(Sha256::fill(1), Sha256::fill(11));
        let sibling = db
            .new_batch()
            .set(Sha256::fill(2), Sha256::fill(22))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let db = db.apply_batch(sibling).await.unwrap();
        assert!(matches!(
            direct.merkleize(&db, None, Location::new(0)).await,
            Err(Error::StaleBatch)
        ));

        let parent = db
            .new_batch()
            .set(Sha256::fill(3), Sha256::fill(33))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let child = parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(4), Sha256::fill(44));
        let sibling = db
            .new_batch()
            .set(Sha256::fill(5), Sha256::fill(55))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let db = db.apply_batch(sibling).await.unwrap();
        assert!(matches!(
            child.merkleize(&db, None, Location::new(0)).await,
            Err(Error::StaleBatch)
        ));

        db.destroy().await.unwrap();
    }

    async fn compact_merkleize_ancestor_states_inner<F: Family>(context: deterministic::Context) {
        let db = open_db::<F>(context.child("db"), "immutable-merkleize-ancestor-states").await;

        let grandparent = db
            .new_batch()
            .set(Sha256::fill(1), Sha256::fill(11))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let parent = grandparent
            .new_batch::<Sha256>()
            .set(Sha256::fill(2), Sha256::fill(22))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let pending = parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(3), Sha256::fill(33))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();

        let db = db.apply_batch(Arc::clone(&grandparent)).await.unwrap();
        let applied = parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(3), Sha256::fill(33))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        assert_eq!(pending.root(), applied.root());

        drop(grandparent);
        let retired = parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(3), Sha256::fill(33))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        assert_eq!(retired.bounds().db, db.commitment());
        assert_eq!(pending.root(), retired.root());

        let child = parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(3), Sha256::fill(33));
        let db = db.apply_batch(parent).await.unwrap();
        let child = child.merkleize(&db, None, Location::new(0)).await.unwrap();
        assert_eq!(pending.root(), child.root());
        let expected_root = child.root();
        let db = db.apply_batch(child).await.unwrap();
        assert_eq!(db.root(), expected_root);
        db.destroy().await.unwrap();
    }

    #[test_traced]
    fn test_compact_merkleize_foreign_db_mmr() {
        deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::<mmr::Family>);
    }

    #[test_traced]
    fn test_compact_merkleize_foreign_db_mmb() {
        deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::<mmb::Family>);
    }

    #[test_traced]
    fn test_compact_merkleize_stale_sibling_mmr() {
        deterministic::Runner::default()
            .start(compact_merkleize_stale_sibling_inner::<mmr::Family>);
    }

    #[test_traced]
    fn test_compact_merkleize_stale_sibling_mmb() {
        deterministic::Runner::default()
            .start(compact_merkleize_stale_sibling_inner::<mmb::Family>);
    }

    #[test_traced]
    fn test_compact_merkleize_ancestor_states_mmr() {
        deterministic::Runner::default()
            .start(compact_merkleize_ancestor_states_inner::<mmr::Family>);
    }

    #[test_traced]
    fn test_compact_merkleize_ancestor_states_mmb() {
        deterministic::Runner::default()
            .start(compact_merkleize_ancestor_states_inner::<mmb::Family>);
    }

    async fn open_bounded<F: Family>(
        context: deterministic::Context,
        witness_cfg: JournalConfig<()>,
        cap: Location<F>,
    ) -> Result<TestDb<F>, Error<F>> {
        let cfg = Config {
            strategy: Sequential,
            witness: witness_cfg,
            commit_codec_config: (),
        };
        Db::init(context, cfg, Some(cap)).await
    }

    /// A batch's proof and pinned nodes return `StaleRead` after a bounded initialization and
    /// another batch bring the database to the batch's base size with a different root.
    async fn proof_refused_after_off_chain_reopen<F: Family>(context: deterministic::Context) {
        let db = open_db::<F>(context.child("db"), "off-chain-proof").await;
        let mut seed = db.new_batch();
        for value in 1..=6 {
            seed = seed.set(Sha256::fill(value), Sha256::fill(value));
        }
        let seed = seed.merkleize(&db, None, Location::new(0)).await.unwrap();
        let db = db.apply_batch(seed).await.unwrap();
        let db = db.sync().await.unwrap();
        let batch = db
            .new_batch()
            .set(Sha256::fill(100), Sha256::fill(100))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let (start, ops) = batch.operations();
        let root = batch.root();
        let original_proof = batch.proof(&db).unwrap();
        let original_pins = batch.pinned_nodes(&db).unwrap();
        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
            &original_proof,
            start,
            &ops,
            &original_pins,
            &root
        ));
        drop(db);

        let db = open_bounded::<F>(
            context.child("reopen"),
            witness_config("off-chain-proof"),
            Location::new(1),
        )
        .await
        .unwrap();
        let mut other = db.new_batch();
        for value in 11..=16 {
            other = other.set(Sha256::fill(value), Sha256::fill(value));
        }
        let other = other.merkleize(&db, None, Location::new(0)).await.unwrap();
        let db = db.apply_batch(other).await.unwrap();
        assert_eq!(db.size(), start);
        assert!(matches!(
            batch
                .new_batch::<Sha256>()
                .merkleize(&db, None, Location::new(0))
                .await,
            Err(Error::StaleBatch)
        ));

        let proof = batch.proof(&db);
        let pins = batch.pinned_nodes(&db);
        assert!(
            matches!(proof, Err(Error::StaleRead)),
            "proof must return StaleRead at a state outside the batch's chain"
        );
        assert!(
            matches!(pins, Err(Error::StaleRead)),
            "pinned nodes must return StaleRead at a state outside the batch's chain"
        );
    }

    #[test]
    fn compact_mmr_proof_refused_after_off_chain_reopen() {
        deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::<mmr::Family>);
    }

    #[test]
    fn compact_mmb_proof_refused_after_off_chain_reopen() {
        deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::<mmb::Family>);
    }

    /// Batch artifacts (operations, range proof, pinned frontier) verify against the batch root,
    /// survive applying and dropping ancestors, and return `ElementPruned` once the batch itself is
    /// applied and the compact store is pruned past them.
    async fn compact_operations_and_proof_inner<F: Family>(context: deterministic::Context) {
        let db = open_db::<F>(context.child("db"), "immutable-operations-and-proof").await;
        let value = |key: u8| Sha256::fill(key.wrapping_add(100));

        // Seed committed state so the chain below starts above a pruned frontier.
        let mut seed = db.new_batch();
        for key in 1u8..=6 {
            seed = seed.set(Sha256::fill(key), value(key));
        }
        let seed = seed
            .merkleize(&db, Some(Sha256::fill(7)), Location::new(0))
            .await
            .unwrap();
        let db = db.apply_batch(seed).await.unwrap().sync().await.unwrap();
        let floor = db.size();
        assert_eq!(floor, Location::new(8));

        // A batch from `to_batch` has no operations to prove.
        assert!(matches!(
            db.to_batch().proof(&db),
            Err(Error::Merkle(crate::merkle::Error::Empty))
        ));

        // A two-deep unapplied chain. The child's artifacts read the parent's nodes through the
        // chain.
        let mut parent = db.new_batch();
        for key in 8u8..=12 {
            parent = parent.set(Sha256::fill(key), value(key));
        }
        let parent = parent
            .merkleize(&db, Some(Sha256::fill(13)), floor)
            .await
            .unwrap();
        let child = parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(14), value(14))
            .merkleize(&db, Some(Sha256::fill(15)), floor)
            .await
            .unwrap();

        // The operation suffix is the batch's own sets plus its commit, handed out zero-copy.
        let (child_start, child_ops) = child.operations();
        let (_, child_ops_again) = child.operations();
        let child_end = child.bounds().tip.size;
        assert!(Arc::ptr_eq(&child_ops, &child_ops_again));
        assert_eq!(child_start, parent.bounds().tip.size);
        assert_eq!(*child_start + child_ops.len() as u64, *child_end);
        assert_eq!(child_end, Location::new(16));
        assert!(matches!(
            child_ops.as_slice(),
            [Operation::Set(key, set_value), Operation::Commit(Some(metadata), operation_floor)]
                if key == &Sha256::fill(14)
                    && set_value == &value(14)
                    && metadata == &Sha256::fill(15)
                    && operation_floor == &floor
        ));

        // The proof is anchored at the batch tip and verifies with or without the pins.
        let child_root = child.root();
        let child_proof = child.proof(&db).unwrap();
        let child_pins = child.pinned_nodes(&db).unwrap();
        assert_eq!(child_proof.leaves, child_end);
        assert_eq!(
            child_proof.inactive_peaks,
            F::inactive_peaks(child_end, floor),
        );
        assert!(verify_proof::<Sha256, _, _>(
            &child_proof,
            child_start,
            &child_ops,
            &child_root
        ));
        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
            &child_proof,
            child_start,
            &child_ops,
            &child_pins,
            &child_root
        ));

        // The pins are order-sensitive.
        assert!(child_pins.len() > 1);
        let mut reordered_child_pins = child_pins.clone();
        reordered_child_pins.swap(0, 1);
        assert!(!verify_proof_and_pinned_nodes::<Sha256, _, _>(
            &child_proof,
            child_start,
            &child_ops,
            &reordered_child_pins,
            &child_root
        ));

        // Pipelined consumer: applying the parent prunes the store to the parent's tip, which is
        // exactly the frontier the child's base pins, so the artifacts survive dropping the
        // parent.
        let db = db.apply_batch(parent).await.unwrap();
        let child_proof_after = child.proof(&db).unwrap();
        assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins);
        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
            &child_proof_after,
            child_start,
            &child_ops,
            &child_pins,
            &child_root
        ));
        let applied_start = db.size();
        let db = db.apply_batch(child).await.unwrap();
        let child_range = applied_start..db.size();
        assert_eq!(child_range, child_start..child_end);

        // A commit-only batch proves exactly its commit operation.
        let commit_floor = db.size();
        let commit_only = db
            .new_batch()
            .merkleize(&db, Some(Sha256::fill(16)), commit_floor)
            .await
            .unwrap();
        let (commit_start, commit_ops) = commit_only.operations();
        let commit_end = commit_only.bounds().tip.size;
        let commit_root = commit_only.root();
        let commit_proof = commit_only.proof(&db).unwrap();
        let commit_pins = commit_only.pinned_nodes(&db).unwrap();
        assert_eq!(commit_start, commit_floor);
        assert!(matches!(
            commit_ops.as_slice(),
            [Operation::Commit(Some(metadata), operation_floor)]
                if metadata == &Sha256::fill(16) && operation_floor == &commit_floor
        ));
        assert_eq!(*commit_start + commit_ops.len() as u64, *commit_end);
        assert_eq!(commit_proof.leaves, commit_end);
        assert_eq!(
            commit_proof.inactive_peaks,
            F::inactive_peaks(commit_end, commit_floor)
        );
        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
            &commit_proof,
            commit_start,
            &commit_ops,
            &commit_pins,
            &commit_root
        ));
        let applied_start = db.size();
        let db = db.apply_batch(commit_only).await.unwrap();
        let commit_range = applied_start..db.size();
        assert_eq!(commit_range, commit_start..commit_end);
        let db = db.sync().await.unwrap();

        // Applying a batch built mid-mountain prunes its own artifacts. The accessors return
        // `ElementPruned` rather than a proof that fails to verify, while a child merkleized
        // before the apply still finds its base frontier in the store.
        let late_parent = db
            .new_batch()
            .set(Sha256::fill(17), value(17))
            .merkleize(&db, Some(Sha256::fill(18)), db.size())
            .await
            .unwrap();
        let late = late_parent
            .new_batch::<Sha256>()
            .set(Sha256::fill(19), value(19))
            .merkleize(&db, Some(Sha256::fill(20)), db.size())
            .await
            .unwrap();
        let db = db.apply_batch(Arc::clone(&late_parent)).await.unwrap();
        assert!(matches!(
            late_parent.proof(&db),
            Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
        ));
        assert!(matches!(
            late_parent.pinned_nodes(&db),
            Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
        ));
        drop(late_parent);

        let (late_start, late_ops) = late.operations();
        let late_root = late.root();
        let late_proof = late.proof(&db).unwrap();
        let late_pins = late.pinned_nodes(&db).unwrap();
        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
            &late_proof,
            late_start,
            &late_ops,
            &late_pins,
            &late_root
        ));
        let db = db.apply_batch(late).await.unwrap();

        db.destroy().await.unwrap();
    }

    #[test_traced]
    fn test_compact_operations_and_proof_mmr() {
        deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmr::Family>);
    }

    #[test_traced]
    fn test_compact_operations_and_proof_mmb() {
        deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmb::Family>);
    }

    /// Open the persisted witness journal directly so tests can corrupt the tip entry.
    async fn open_witness_journal(
        context: deterministic::Context,
        partition: &str,
    ) -> witness::Journal<deterministic::Context, mmr::Family, Digest> {
        let cfg = witness_config(partition);
        witness::Journal::init(context, cfg).await.unwrap()
    }

    /// A compact db over a delayed-sync storage backend.
    type DelayedDb = Db<
        mmr::Family,
        DelayedSyncContext<deterministic::Context>,
        Digest,
        FixedEncoding<Digest>,
        Sha256,
        Sequential,
    >;

    /// Open a [DelayedDb] whose blob syncs park on `pending`.
    ///
    /// Init durably persists the bootstrap witness, so while syncs park the returned future
    /// must be driven with `drive_pending_syncs` (or the mock unblocked first).
    fn open_delayed_db(
        context: deterministic::Context,
        partition: &str,
        pending: &PendingSyncs,
    ) -> impl Future<Output = Result<DelayedDb, Error<mmr::Family>>> {
        let witness_cfg = witness_config(partition);
        let cfg = Config {
            strategy: Sequential,
            witness: witness_cfg,
            commit_codec_config: (),
        };
        let context = DelayedSyncContext {
            inner: context,
            pending: pending.clone(),
        };
        DelayedDb::init(context, cfg, None)
    }

    /// Apply a single-key batch writing `key -> value` with `metadata`.
    async fn apply_set(db: DelayedDb, key: Digest, value: Digest, metadata: Digest) -> DelayedDb {
        let floor = db.inactivity_floor_loc();
        let batch = db
            .new_batch()
            .set(key, value)
            .merkleize(&db, Some(metadata), floor)
            .await
            .unwrap();
        db.apply_batch(batch).await.unwrap()
    }

    /// State persisted via an awaited start_sync handle is recovered on reopen.
    #[test_traced]
    fn test_compact_start_sync_recovery() {
        deterministic::Runner::default().start(|ctx| async move {
            let partition = "immutable-start-sync-recovery";
            let pending = PendingSyncs::default();
            pending.unblock();
            let mut db = open_delayed_db(ctx.child("delayed"), partition, &pending)
                .await
                .unwrap();
            let metadata = Sha256::fill(9u8);
            db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), metadata).await;

            let handle;
            (db, handle) = db.start_sync().await.unwrap();
            handle.await.unwrap();
            let root = db.root();
            drop(db);

            let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
                .await
                .unwrap();
            assert_eq!(db.root(), root);
            assert_eq!(db.get_metadata(), Some(metadata));
            db.destroy().await.unwrap();
        });
    }

    /// A sync begun by `start_sync` that fails in flight surfaces the error through both the
    /// returned handle and the next durability operation, even when that operation has nothing
    /// new to persist.
    #[test_traced]
    fn test_compact_start_sync_failure_propagates() {
        deterministic::Runner::default().start(|ctx| async move {
            let pending = PendingSyncs::default();
            pending.unblock();
            let mut db =
                open_delayed_db(ctx.child("delayed"), "immutable-start-sync-fail", &pending)
                    .await
                    .unwrap();
            db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), Sha256::fill(9u8)).await;

            // Arm all future syncs to resolve to an injected error.
            pending.arm_fail();

            let handle;
            (db, handle) = db.start_sync().await.unwrap();
            assert!(
                handle.await.is_err(),
                "the sync handle surfaces the failure"
            );
            let starts_before = pending.starts();

            // The witness entry was already appended, so this commit has nothing to stage.
            // It must still observe the retained failure rather than no-op.
            assert!(
                db.commit().await.is_err(),
                "the next durability op surfaces the failed in-flight sync"
            );
            assert_eq!(
                pending.starts(),
                starts_before,
                "the surfaced error is the retained failure, not a fresh sync's"
            );
        });
    }

    #[test_traced("INFO")]
    fn test_compact_stale_batch_rejected() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-stale").await;

            let key1 = Sha256::hash(&[&[1]]);
            let key2 = Sha256::hash(&[&[2]]);
            let value1 = Sha256::fill(10u8);
            let value2 = Sha256::fill(20u8);

            let batch_a = db
                .new_batch()
                .set(key1, value1)
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let batch_b = db
                .new_batch()
                .set(key2, value2)
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();

            let stale = db.new_batch().set(key2, value2);
            let expected_root = batch_a.root();
            let db = db.apply_batch(batch_a).await.unwrap();
            assert_eq!(db.root(), expected_root);

            // After the apply, merkleizing a batch built on the pre-apply state and applying the
            // merkleized sibling both return `StaleBatch`.
            assert!(matches!(
                stale.merkleize(&db, None, Location::new(0)).await,
                Err(Error::StaleBatch)
            ));
            assert!(matches!(
                db.apply_batch(batch_b).await,
                Err(Error::StaleBatch)
            ));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_delayed_merkleize_after_ancestor_apply() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-delayed-child").await;
            let key1 = Sha256::hash(&[&[1]]);
            let key2 = Sha256::hash(&[&[2]]);
            let key3 = Sha256::hash(&[&[3]]);
            let value1 = Sha256::fill(10u8);
            let value2 = Sha256::fill(20u8);
            let value3 = Sha256::fill(30u8);

            let a = db
                .new_batch()
                .set(key1, value1)
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let b = a
                .new_batch::<Sha256>()
                .set(key2, value2)
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let c = b.new_batch::<Sha256>().set(key3, value3);

            let db = db.apply_batch(a).await.unwrap();
            let c = c.merkleize(&db, None, Location::new(0)).await.unwrap();
            let expected_root = c.root();
            let db = db.apply_batch(c).await.unwrap();

            assert_eq!(db.root(), expected_root);
        });
    }

    /// `to_batch()` reflects the current applied state before it becomes durable.
    #[test_traced("INFO")]
    fn test_compact_to_batch_reflects_live_state() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-to-batch-live").await;

            let pre_apply_root = db.root();
            let pre_snapshot = db.to_batch();
            assert_eq!(
                pre_snapshot.root(),
                pre_apply_root,
                "snapshot before any mutation should match the live root"
            );

            let key = Sha256::hash(&[&[1]]);
            let value = Sha256::fill(10u8);
            let batch = db
                .new_batch()
                .set(key, value)
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap();

            // Observe the applied state before making it durable.
            let live_root = db.root();
            assert_ne!(
                live_root, pre_apply_root,
                "applying a non-empty batch must change the live root"
            );

            let snapshot = db.to_batch();
            assert_eq!(
                snapshot.root(),
                live_root,
                "to_batch().root() must match the live db.root() even before sync"
            );

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_stale_batch_chained() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-chained-stale").await;

            let common_parent = db
                .new_batch()
                .set(Sha256::hash(&[&[10]]), Sha256::fill(10u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let sibling_a = common_parent
                .new_batch::<Sha256>()
                .set(Sha256::hash(&[&[11]]), Sha256::fill(11u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let sibling_b = common_parent
                .new_batch::<Sha256>()
                .set(Sha256::hash(&[&[12]]), Sha256::fill(12u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(sibling_a).await.unwrap();
            assert!(matches!(
                db.validate_batch(&sibling_b),
                Err(Error::StaleBatch)
            ));

            let parent_a = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let parent_b = db
                .new_batch()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let child_b = parent_b
                .new_batch::<Sha256>()
                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();

            let db = db.apply_batch(parent_a).await.unwrap();
            assert!(matches!(
                db.validate_batch(&child_b),
                Err(Error::StaleBatch)
            ));
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_stale_parent_after_child_applied() {
        deterministic::Runner::default().start(|context| async move {
            let db =
                open_db::<mmr::Family>(context.child("db"), "immutable-child-before-parent").await;

            let parent = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let child = parent
                .new_batch::<Sha256>()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();

            let db = db.apply_batch(child).await.unwrap();
            assert!(matches!(
                db.apply_batch(parent).await,
                Err(Error::StaleBatch)
            ));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_sequential_commit_parent_then_child() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-parent-child").await;

            let parent = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let child = parent
                .new_batch::<Sha256>()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let expected_root = child.root();

            let db = db.apply_batch(parent).await.unwrap();
            let db = db.apply_batch(child).await.unwrap().sync().await.unwrap();

            assert_eq!(db.root(), expected_root);

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_floor_regressed() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-floor-regressed").await;

            let advance_floor = db.new_batch().set(Sha256::hash(&[&[1]]), Sha256::fill(1u8));
            let advance_floor = advance_floor
                .merkleize(&db, None, Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(advance_floor).await.unwrap();
            let db = db.sync().await.unwrap();

            let regressed = db
                .new_batch()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(0))
                .await;

            assert!(matches!(
                regressed,
                Err(Error::FloorRegressed(new, current))
                    if new == Location::new(0) && current == Location::new(1)
            ));
        });
    }

    // A chained batch whose tip floor is below its parent's floor must be rejected:
    // the parent's Commit participates in the per-commit monotonicity invariant even
    // before it is applied.
    #[test_traced("INFO")]
    fn test_compact_chained_floor_regression() {
        deterministic::Runner::default().start(|context| async move {
            let db =
                open_db::<mmr::Family>(context.child("db"), "immutable-regressed-ancestor-floor")
                    .await;

            let parent = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(1))
                .await
                .unwrap();
            let child = parent
                .new_batch::<Sha256>()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(0))
                .await;

            assert!(matches!(
                child,
                Err(Error::FloorRegressed(new, prev))
                    if new == Location::new(0) && prev == Location::new(1)
            ));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_restores_commit_metadata_and_floor() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-meta").await;

            let k1 = Sha256::hash(&[&[1]]);
            let v1 = Sha256::fill(11u8);
            let meta1 = Sha256::fill(0xaa);
            let floor1 = Location::new(0);
            let batch = db
                .new_batch()
                .set(k1, v1)
                .merkleize(&db, Some(meta1), floor1)
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let root_after_first = db.root();
            let size_after_first = db.size();

            let k2 = Sha256::hash(&[&[2]]);
            let v2 = Sha256::fill(22u8);
            let meta2 = Sha256::fill(0xbb);
            // Advance the floor to the commit of the first batch (loc 1).
            let floor2 = Location::new(1);
            let batch = db
                .new_batch()
                .set(k2, v2)
                .merkleize(&db, Some(meta2), floor2)
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            assert_eq!(db.get_metadata(), Some(meta2));
            assert_eq!(db.inactivity_floor_loc(), floor2);

            let db = {
                _ = db.sync().await.unwrap();
                open_bounded::<mmr::Family>(
                    context.child("cap"),
                    witness_config("immutable-rewind-meta"),
                    size_after_first,
                )
                .await
            }
            .unwrap();
            assert_eq!(db.root(), root_after_first);
            assert_eq!(db.get_metadata(), Some(meta1));
            assert_eq!(db.inactivity_floor_loc(), floor1);

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_persists_across_reopen() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-rewind-reopen";
            let meta1 = Sha256::fill(0xaa);
            let floor1 = Location::new(0);
            let meta2 = Sha256::fill(0xbb);
            let floor2 = Location::new(1);

            let root_after_first = {
                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
                    .merkleize(&db, Some(meta1), floor1)
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
                let root = db.root();
                let size_after_first = db.size();

                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8))
                    .merkleize(&db, Some(meta2), floor2)
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();

                let _db = {
                    _ = db.sync().await.unwrap();
                    open_bounded::<mmr::Family>(
                        context.child("cap"),
                        witness_config(partition),
                        size_after_first,
                    )
                    .await
                }
                .unwrap();
                root
            };

            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
            assert_eq!(db.root(), root_after_first);
            assert_eq!(db.get_metadata(), Some(meta1));
            assert_eq!(db.inactivity_floor_loc(), floor1);

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_commit_persists_across_reopen() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-commit-reopen";
            let meta1 = Sha256::fill(0xaa);
            let meta2 = Sha256::fill(0xbb);

            let root_after_second = {
                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
                    .merkleize(&db, Some(meta1), Location::new(0))
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();

                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8))
                    .merkleize(&db, Some(meta2), Location::new(1))
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
                db.root()
            };

            // Reopen recovers the committed tip even though the journal was never synced.
            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
            assert_eq!(db.root(), root_after_second);
            assert_eq!(db.get_metadata(), Some(meta2));
            assert_eq!(db.inactivity_floor_loc(), Location::new(1));
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_to_committed_entry_after_reopen() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-commit-rewind-reopen";
            let meta1 = Sha256::fill(0xaa);
            let meta2 = Sha256::fill(0xbb);

            let (root_a, size_a) = {
                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
                    .merkleize(&db, Some(meta1), Location::new(0))
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
                let root_a = db.root();
                let size_a = db.size();

                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8))
                    .merkleize(&db, Some(meta2), Location::new(1))
                    .await
                    .unwrap();
                let _db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
                (root_a, size_a)
            };

            // Both committed witnesses survive the crash. Reopen recovers the tip, and the earlier
            // commit remains a valid initialization target.
            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
            let db = {
                _ = db.sync().await.unwrap();
                open_bounded::<mmr::Family>(context.child("cap"), witness_config(partition), size_a)
                    .await
            }
            .unwrap();
            assert_eq!(db.root(), root_a);
            assert_eq!(db.get_metadata(), Some(meta1));
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_sync_after_commit() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-sync-after-commit";
            let meta = Sha256::fill(0xaa);

            let root = {
                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
                    .merkleize(&db, Some(meta), Location::new(0))
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap();

                // Commit persists witness data. Sync must also persist recovery metadata.
                let db = db.commit().await.unwrap();
                let db = db.sync().await.unwrap();
                db.root()
            };

            // Check the watermark before reopening can rebuild the offsets journal.
            let metadata = Metadata::<_, u64, VecU64>::init(
                context.child("checkpoint"),
                MetadataConfig {
                    partition: format!("{partition}-witness_offsets-metadata"),
                    codec_config: (),
                },
            )
            .await
            .unwrap();

            // Key 3 records the durable prefix: the bootstrap witness and the applied batch.
            assert_eq!(metadata.get(&3).copied().map(u64::from), Some(2));
            drop(metadata);

            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
            assert_eq!(db.root(), root);
            assert_eq!(db.get_metadata(), Some(meta));
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_reopen_rejects_tampered_witness() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-witness-tamper";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            drop(db);

            // Corrupt the entry structurally. An extra pinned node cannot rebuild the Merkle.
            let journal = open_witness_journal(context.child("tamper"), partition).await;
            let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
            pinned_nodes.push(Sha256::fill(0xff));
            witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;

            let cfg = Config {
                strategy: Sequential,
                witness: witness_config(partition),
                commit_codec_config: (),
            };
            let reopened =
                TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
            assert!(matches!(reopened, Err(Error::DataCorrupted(_))));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_rejects_corrupt_target_entry() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-corrupt-rewind-target";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let initialization_bound = db.target().size;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let tip_target = db.target();
            drop(db);

            // Corrupt the initialization target's entry (the journal holds bootstrap, target, tip).
            let mut journal = open_witness_journal(context.child("corrupt"), partition).await;
            journal = witness::tests::corrupt_entry(journal, 1, |entry| {
                entry.pinned_nodes.push(Sha256::fill(0xff));
            })
            .await;
            drop(journal);

            // The tip entry is intact, so reopen succeeds.
            let cfg = Config {
                strategy: Sequential,
                witness: witness_config(partition),
                commit_codec_config: (),
            };
            let reopened = TestDb::<mmr::Family>::init(context.child("reopen"), cfg, None)
                .await
                .unwrap();
            assert_eq!(reopened.target(), tip_target);

            // The corrupt entry fails the recovery before any truncation.
            assert!(matches!(
                {
                    drop(reopened);
                    open_bounded::<mmr::Family>(
                        context.child("cap"),
                        witness_config(partition),
                        initialization_bound,
                    )
                    .await
                },
                Err(Error::DataCorrupted(_))
            ));

            // The newer history survives: reopen still lands on the original tip.
            let cfg = Config {
                strategy: Sequential,
                witness: witness_config(partition),
                commit_codec_config: (),
            };
            let reopened = TestDb::<mmr::Family>::init(context.child("reopen2"), cfg, None)
                .await
                .unwrap();
            assert_eq!(reopened.target(), tip_target);
            reopened.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_reopen_rejects_interrupted_import() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-interrupted-import";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            drop(db);

            // Simulate a crash between an import's journal clear and its entry append: the
            // journal is empty but its size is nonzero.
            let journal = open_witness_journal(context.child("clear"), partition).await;
            let size = journal.size();
            let journal = journal.clear_to_size(size.max(1)).await.unwrap();
            drop(journal);

            // Reopen must fail rather than bootstrap a fresh db.
            let cfg = Config {
                strategy: Sequential,
                witness: witness_config(partition),
                commit_codec_config: (),
            };
            let reopened =
                TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
            assert!(matches!(
                reopened,
                Err(Error::DataCorrupted("witness journal has no tip"))
            ));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_reopen_rejects_commit_floor_beyond_tip() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-invalid-persisted-floor";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            drop(db);
            let oversized_floor = Location::new(10);

            // Overwrite the persisted commit op with a floor beyond its own commit location.
            let journal = open_witness_journal(context.child("tamper"), partition).await;
            let (_, size, pinned_nodes) = witness::tests::tip(&journal).await;
            let bad_op = Operation::<mmr::Family, Digest, FixedEncoding<Digest>>::Commit(
                Some(Sha256::fill(0xaa)),
                oversized_floor,
            )
            .encode()
            .to_vec();
            witness::tests::overwrite_tip(journal, bad_op, size, pinned_nodes).await;

            let cfg = Config {
                strategy: Sequential,
                witness: witness_config(partition),
                commit_codec_config: (),
            };
            let reopened =
                TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
            assert!(matches!(
                reopened,
                Err(Error::DataCorrupted("invalid compact witness"))
            ));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_reopen_rejects_tampered_pinned_nodes() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-pinned-nodes-tamper";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let tampered_target = db.target();
            drop(db);

            // Flip one pinned-node digest. There is no stored proof to cross-check against, so the
            // rebuild succeeds and yields a different root, the same way a bit-flipped replay
            // journal reopens with a different root.
            let journal = open_witness_journal(context.child("tamper"), partition).await;
            let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
            pinned_nodes[0] = Sha256::fill(0xff);
            witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;

            let cfg = Config {
                strategy: Sequential,
                witness: witness_config(partition),
                commit_codec_config: (),
            };
            let reopened = TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None)
                .await
                .unwrap();
            assert_ne!(reopened.target(), tampered_target);
            reopened.destroy().await.unwrap();
        });
    }

    /// A witness entry appended but not synced (a commit interrupted before its journal sync)
    /// must be dropped on reopen, recovering the last synced commit.
    #[test_traced("INFO")]
    fn test_compact_reopen_drops_unsynced_witness() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-witness-unsynced";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;

            // Commit state A.
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let target_a = db.target();
            drop(db);

            // Simulate the crash window: append an entry ahead of the tip without syncing it,
            // then drop the journal. The unsynced tail must not survive reopen.
            let journal = open_witness_journal(context.child("crash"), partition).await;
            let (op_bytes, mut size, pinned_nodes) = witness::tests::tip(&journal).await;
            size += 2;
            witness::tests::append_unsynced(journal, op_bytes, size, pinned_nodes).await;

            // Reopen must drop the unsynced entry and recover state A.
            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
            assert_eq!(reopened.target(), target_a);
            reopened.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_initialization_zero_and_above_end() {
        deterministic::Runner::default().start(|context| async move {
            let cfg = witness_config("immutable-caps");
            assert!(matches!(
                open_bounded::<mmr::Family>(context.child("zero"), cfg.clone(), Location::new(0))
                    .await,
                Err(Error::InvalidInitializationBound)
            ));
            let db = open_bounded::<mmr::Family>(
                context.child("fresh"),
                cfg.clone(),
                Location::new(100),
            )
            .await
            .unwrap();
            let root = db.root();
            assert_eq!(db.size(), Location::new(1));
            drop(db);
            let db = open_bounded::<mmr::Family>(context.child("above"), cfg, Location::new(100))
                .await
                .unwrap();
            assert_eq!(db.root(), root);
            assert_eq!(db.size(), Location::new(1));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_between_commits() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-between").await;
            let initial_root = db.root();

            // A multi-op commit jumps the committed size from 1 (bootstrap) to 4.
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let root_a = db.root();
            let size_a = db.size();
            assert_eq!(size_a, Location::new(4));

            // A second commit moves the size to 6.
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
                .merkleize(&db, Some(Sha256::fill(0xb1)), Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let cfg = witness_config("immutable-rewind-between");
            drop(db);
            for (target, size, root) in [(5, 4, root_a), (3, 1, initial_root), (2, 1, initial_root)]
            {
                let db = open_bounded::<mmr::Family>(
                    context.child("cap"),
                    cfg.clone(),
                    Location::new(target),
                )
                .await
                .unwrap();
                assert_eq!(db.size(), Location::new(size));
                assert_eq!(db.root(), root);
                drop(db);
                let db =
                    open_db::<mmr::Family>(context.child("restart"), "immutable-rewind-between")
                        .await;
                assert_eq!(db.root(), root);
            }
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_multiple_commits() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-rewind-multi";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;

            // Commit A, B, C, recording the state after A.
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let root_a = db.root();
            let size_a = db.size();
            let target_a = db.target();

            let mut db = db;
            for i in [2u8, 3] {
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[i]]), Sha256::fill(i))
                    .merkleize(&db, Some(Sha256::fill(i)), Location::new(0))
                    .await
                    .unwrap();
                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            }
            assert_ne!(db.root(), root_a);

            // Reopen two commits earlier.
            let db = {
                _ = db.sync().await.unwrap();
                open_bounded::<mmr::Family>(context.child("cap"), witness_config(partition), size_a)
                    .await
            }
            .unwrap();
            assert_eq!(db.root(), root_a);
            assert_eq!(db.size(), size_a);
            assert_eq!(db.get_metadata(), Some(Sha256::fill(0xa1)));
            assert_eq!(db.target(), target_a);
            drop(db);

            // The recovery is durable: reopen recovers state A.
            let db = open_db::<mmr::Family>(context.child("reopen"), partition).await;
            assert_eq!(db.root(), root_a);
            assert_eq!(db.target(), target_a);
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_preserves_current_state() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-noop").await;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let root = db.root();
            let size = db.size();

            let db = {
                _ = db.sync().await.unwrap();
                open_bounded::<mmr::Family>(
                    context.child("cap"),
                    witness_config("immutable-rewind-noop"),
                    size,
                )
                .await
            }
            .unwrap();
            assert_eq!(db.root(), root);
            assert_eq!(db.size(), size);
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_prune_then_bounded_initialization() {
        deterministic::Runner::default().start(|context| async move {
            // One entry per section so pruning takes effect at entry granularity (pruning is
            // section-aligned and never drops a partial section).
            let mut witness_cfg = witness_config("immutable-prune-rewind");
            witness_cfg.items_per_section = NZU64!(1);
            let cfg = Config {
                strategy: Sequential,
                witness: witness_cfg.clone(),
                commit_codec_config: (),
            };
            let mut db: TestDb<mmr::Family> =
                Db::init(context.child("db"), cfg, None).await.unwrap();

            // Commit A, B, C.
            let mut sizes = Vec::new();
            for i in [1u8, 2, 3] {
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[i]]), Sha256::fill(i))
                    .merkleize(&db, Some(Sha256::fill(i)), Location::new(0))
                    .await
                    .unwrap();
                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
                sizes.push(db.size());
            }

            // Reopen retained state B before the terminal check that pruned state A is unavailable.
            let db = db.prune(sizes[1]).await.unwrap().sync().await.unwrap();
            drop(db);
            let db = open_bounded::<mmr::Family>(
                context.child("retained"),
                witness_cfg.clone(),
                sizes[1],
            )
            .await
            .unwrap();
            assert_eq!(db.size(), sizes[1]);
            assert_eq!(db.get_metadata(), Some(Sha256::fill(2)));
            drop(db);

            assert!(matches!(
                open_bounded::<mmr::Family>(context.child("pruned"), witness_cfg, sizes[0]).await,
                Err(Error::HistoricalFloorPruned(_))
            ));
        });
    }

    /// Witness config holding one witness per section, so every commit occupies its own blobs.
    fn sectioned_witness_config(partition: &str) -> JournalConfig<()> {
        let mut cfg = witness_config(partition);
        cfg.items_per_section = NZU64!(1);
        cfg
    }

    /// Seed a db with the bootstrap witness plus `commits` synced metadata-only commits,
    /// returning the size and root after each commit. Every commit appends exactly its commit
    /// operation, so the witness at position `p` has size `p + 1`.
    async fn seed_witness_sections(
        context: deterministic::Context,
        witness_cfg: JournalConfig<()>,
        commits: u8,
    ) -> Vec<(Location<mmr::Family>, Digest)> {
        let cfg = Config {
            strategy: Sequential,
            witness: witness_cfg,
            commit_codec_config: (),
        };
        let mut db: TestDb<mmr::Family> = Db::init(context, cfg, None).await.unwrap();
        let mut states = Vec::new();
        for i in 1..=commits {
            let batch = db
                .new_batch()
                .merkleize(&db, Some(Sha256::fill(i)), db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            states.push((db.size(), db.root()));
        }
        states
    }

    /// Leave witness data blob `blob` with a partial trailing page, as a crash mid-write would.
    /// Paged tail recovery repairs (resizes and syncs) such a tail when the blob is opened.
    async fn tear_witness_data(
        context: &deterministic::Context,
        witness_cfg: &JournalConfig<()>,
        blob: u64,
    ) {
        // The variable journal keeps data blobs in `{partition}_data`, named by the big-endian
        // section index.
        let partition = format!("{}_data", witness_cfg.partition);
        let (handle, len) = context.open(&partition, &blob.to_be_bytes()).await.unwrap();
        handle.resize(len - 1).await.unwrap();
        handle.sync().await.unwrap();
    }

    /// Bounded init through a delayed-sync backend, returning the db and the durability calls
    /// initialization made.
    async fn open_bounded_counting(
        context: deterministic::Context,
        witness_cfg: JournalConfig<()>,
        cap: Location<mmr::Family>,
    ) -> (DelayedDb, usize) {
        // Arming counts every durability call from here on. The gate blocks the first call and
        // every later started sync parks. `drive_pending_syncs` releases them whenever
        // initialization stalls.
        let pending = PendingSyncs::default();
        pending.arm();
        let delayed = DelayedSyncContext {
            inner: context,
            pending: pending.clone(),
        };
        let cfg = Config {
            strategy: Sequential,
            witness: witness_cfg,
            commit_codec_config: (),
        };
        let db = drive_pending_syncs(&pending, DelayedDb::init(delayed, cfg, Some(cap)))
            .await
            .unwrap();
        (db, pending.calls())
    }

    /// Bounded initialization opens no witness section the bound discards, so a torn tail there
    /// costs no repair.
    #[test_traced]
    fn test_compact_bounded_initialization_ignores_discarded_witness_sections() {
        deterministic::Runner::default().start(|context| async move {
            // Build twin journals with one witness per section, so position `p` lives in data
            // blob `p`. Each holds the bootstrap witness (position 0, size 1), a synced commit
            // (position 1, size 2), and a commit without sync (position 2, size 3). The control
            // twin stays intact and sets the baseline durability count.
            let control_cfg = sectioned_witness_config("immutable-skip-discarded-control");
            let torn_cfg = sectioned_witness_config("immutable-skip-discarded-torn");
            let mut states = Vec::new();
            for (label, witness_cfg) in [("control", &control_cfg), ("torn", &torn_cfg)] {
                let context = context.child(label);
                states.push(
                    seed_witness_sections(context.child("seed"), witness_cfg.clone(), 1).await[0],
                );

                // Commit without syncing so the witness at position 2 lies above the recovery
                // watermark, where a torn tail is a crash shape rather than corruption.
                let cfg = Config {
                    strategy: Sequential,
                    witness: witness_cfg.clone(),
                    commit_codec_config: (),
                };
                let db: TestDb<mmr::Family> =
                    Db::init(context.child("extend"), cfg, None).await.unwrap();
                let batch = db
                    .new_batch()
                    .merkleize(&db, Some(Sha256::fill(2)), db.inactivity_floor_loc())
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap();
                drop(db.commit().await.unwrap());
            }

            // The twins share one history, so the synced commit's size and root are the expected
            // recovery for both.
            assert_eq!(states[0], states[1]);
            let (size, root) = states[0];

            // The witness at position 1 has size `size`, so the bound discards position 2 and
            // must not open it. Tear its data blob: repairing the tail would sync it.
            tear_witness_data(&context, &torn_cfg, 2).await;

            // Both twins recover the synced commit under the bound. They differ only in data
            // blob 2, so equal durability counts show the torn blob was not repaired.
            let (control, control_calls) =
                open_bounded_counting(context.child("control"), control_cfg, size).await;
            let (torn, torn_calls) =
                open_bounded_counting(context.child("torn"), torn_cfg, size).await;
            assert_eq!(control.size(), size);
            assert_eq!(control.root(), root);
            assert_eq!(torn.size(), size);
            assert_eq!(torn.root(), root);
            assert_eq!(
                torn_calls, control_calls,
                "the discarded torn section must not be repaired"
            );
            control.destroy().await.unwrap();
            torn.destroy().await.unwrap();
        });
    }

    /// A torn tail in a retained witness section is still repaired under a bound. The witness it
    /// held is lost and initialization falls back to the previous one.
    #[test_traced]
    fn test_compact_bounded_initialization_repairs_retained_witness_section() {
        deterministic::Runner::default().start(|context| async move {
            // Both witnesses share a section, so the bootstrap witness survives the torn tail
            // page and ends mid-page, where recovery must rewrite rather than only shrink.
            let clean_cfg = witness_config("immutable-repair-retained-clean");
            let torn_cfg = witness_config("immutable-repair-retained-torn");
            let mut states = Vec::new();
            for (label, witness_cfg) in [("clean", &clean_cfg), ("torn", &torn_cfg)] {
                // Fresh storage bootstraps the witness at position 0 (size 1), the state the torn
                // twin falls back to.
                let cfg = Config {
                    strategy: Sequential,
                    witness: witness_cfg.clone(),
                    commit_codec_config: (),
                };
                let db: TestDb<mmr::Family> =
                    Db::init(context.child(label), cfg, None).await.unwrap();
                let genesis = db.root();

                // Commit without syncing so the witness at position 1 lies above the recovery
                // watermark, where a torn tail is a crash shape rather than corruption.
                let batch = db
                    .new_batch()
                    .merkleize(&db, Some(Sha256::fill(1)), db.inactivity_floor_loc())
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
                states.push((genesis, db.size(), db.root()));
            }
            assert_eq!(states[0], states[1]);
            let (genesis, size, root) = states[0];

            // The torn twin's last page in section 0 holds the end of the witness at position 1,
            // so tearing it leaves that witness incomplete.
            tear_witness_data(&context, &torn_cfg, 0).await;

            // Section 0 lies below the bound, so both twins open it. The torn twin trims its
            // tail and republishes the journal without the lost witness.
            let (clean, clean_calls) =
                open_bounded_counting(context.child("clean"), clean_cfg, size).await;
            let (torn, torn_calls) =
                open_bounded_counting(context.child("torn"), torn_cfg, size).await;

            // The clean twin recovers the witness at position 1. The torn twin recovers the
            // bootstrap witness and spends extra durability calls on the repair.
            assert_eq!(clean.size(), size);
            assert_eq!(clean.root(), root);
            assert_eq!(torn.size(), Location::new(1));
            assert_eq!(torn.root(), genesis);
            assert!(
                torn_calls > clean_calls,
                "the retained torn section is repaired"
            );
            clean.destroy().await.unwrap();
            torn.destroy().await.unwrap();
        });
    }

    /// A compact-sync import lands at the journal end whatever its size, so afterwards witness
    /// positions can exceed sizes. Bounded initialization still selects by size, widening its
    /// view when the positions below the bound cannot settle the selection.
    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_after_import_below_positions() {
        deterministic::Runner::default().start(|context| async move {
            // A source state of size 2, captured from its witness journal. Its tip witness holds
            // the pinned nodes one operation below the commit, which a compact-sync boundary
            // response carries.
            let src_cfg = sectioned_witness_config("immutable-import-below-src");
            let (src_size, src_root) =
                seed_witness_sections(context.child("src"), src_cfg.clone(), 1).await[0];
            let journal =
                witness::Journal::<_, mmr::Family, Digest>::init(context.child("tip"), src_cfg)
                    .await
                    .unwrap();
            let (_, _, src_pinned) = witness::tests::tip(&journal).await;
            drop(journal);

            // The destination has used positions 0 through 3, so the import lands at position 4.
            let dst_cfg = sectioned_witness_config("immutable-import-below-dst");
            seed_witness_sections(context.child("dst"), dst_cfg.clone(), 3).await;

            // Import the source state from its last commit location, pinned nodes, and commit
            // operation. `seed_witness_sections` committed it with metadata `fill(1)` at floor 0,
            // so the rebuilt root matches the source root.
            let journal = witness::Journal::init(context.child("import"), dst_cfg.clone())
                .await
                .unwrap();
            let imported = TestDb::<mmr::Family>::init_from_sync(
                Sequential,
                journal,
                (),
                src_size - 1,
                src_pinned,
                Operation::Commit(Some(Sha256::fill(1)), Location::new(0)),
            )
            .unwrap();
            assert_eq!(imported.root(), src_root);

            // Committing applies the pending import. It clears the destination journal at its end
            // and appends the imported witness there, so its only entry has size 2 at position 4.
            drop(imported.commit().await.unwrap());
            let journal = witness::Journal::<_, mmr::Family, Digest>::init(
                context.child("placed"),
                dst_cfg.clone(),
            )
            .await
            .unwrap();
            assert_eq!(journal.bounds(), 4..5);
            drop(journal);

            // Reopening recovers the imported state. Three more commits occupy positions 5 through
            // 7 with sizes 3 through 5. `states` holds the size and root at positions 4 through 7.
            let cfg = Config {
                strategy: Sequential,
                witness: dst_cfg.clone(),
                commit_codec_config: (),
            };
            let mut db: TestDb<mmr::Family> =
                Db::init(context.child("reopen"), cfg, None).await.unwrap();
            assert_eq!(db.size(), src_size);
            let mut states = vec![(src_size, src_root)];
            for i in 2..=4u8 {
                let batch = db
                    .new_batch()
                    .merkleize(&db, Some(Sha256::fill(i)), db.inactivity_floor_loc())
                    .await
                    .unwrap();
                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
                states.push((db.size(), db.root()));
            }
            drop(db);

            // Open at each recorded size from the tip down. The cap-5 bounded view holds only
            // position 4 (size 2). It ends at the bound with a tip size below the bound, so
            // recovery widens and selects position 7 by size. Caps 4, 3, and 2 lie at or below the
            // retained start (position 4), so recovery opens unbounded and selects positions 6, 5,
            // and 4. Each open discards the witnesses above its selection, so the caps must
            // descend.
            for (size, root) in states.into_iter().rev() {
                let db = open_bounded::<mmr::Family>(
                    context.child("bounded").with_attribute("cap", *size),
                    dst_cfg.clone(),
                    size,
                )
                .await
                .unwrap();
                assert_eq!(db.size(), size);
                assert_eq!(db.root(), root);
            }

            // Only the imported witness remains, and its size 2 exceeds cap 1.
            assert!(matches!(
                open_bounded::<mmr::Family>(context.child("pruned"), dst_cfg, Location::new(1))
                    .await,
                Err(Error::HistoricalFloorPruned(pruned)) if pruned == Location::new(1)
            ));
        });
    }

    /// A bounded initialization whose selection fails leaves the witness offsets watermark
    /// acknowledging every synced witness.
    #[test_traced]
    fn test_compact_failed_bounded_selection_preserves_acknowledged_offsets() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-failed-bounded-offsets";
            let cap = Location::new(5);
            let witness_cfg = sectioned_witness_config(partition);

            // The witness journal lives in `{partition}-witness`. Its offsets journal, whose
            // checkpoint persists the recovery watermark, lives in `{partition}-witness_offsets`.
            let offsets_partition = format!("{partition}-witness_offsets");
            let cfg = Config {
                strategy: Sequential,
                witness: witness_cfg.clone(),
                commit_codec_config: (),
            };
            let mut db: TestDb<mmr::Family> =
                Db::init(context.child("seed"), cfg, None).await.unwrap();

            // The first batch appends six sets and a commit after the bootstrap commit, so the
            // witness at position 1 has size 8, above the cap used below.
            let mut batch = db.new_batch();
            for key in 1..=6u8 {
                batch = batch.set(Sha256::fill(key), Sha256::fill(key));
            }
            let batch = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
            db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let first_retained_size = db.size();
            assert!(first_retained_size > cap);

            // Eight synced empty commits occupy positions 2 through 9 with sizes 9 through 16.
            // One witness per section puts positions 1 through 4 in the sections the cap-5 view
            // opens, and positions 5 through 9 in sections it discards.
            for _ in 0..8 {
                let batch = db
                    .new_batch()
                    .merkleize(&db, None, db.inactivity_floor_loc())
                    .await
                    .unwrap();
                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            }
            let tip = db.size();

            // Pruning at the first commit removes the bootstrap section. The retained start
            // (position 1) lies below the cap while its size lies above it. A retained start at
            // or above the cap would open unbounded instead.
            let db = db
                .prune(first_retained_size)
                .await
                .unwrap()
                .sync()
                .await
                .unwrap();
            drop(db);

            // Sync acknowledged all ten witnesses (positions 0 through 9). The watermark lies above
            // the cap, so any lowering to the cap is visible below.
            let watermark = fixed::Journal::<_, u64>::persisted_watermark(
                context.child("before"),
                &offsets_partition,
            )
            .await
            .unwrap();
            assert_eq!(watermark, Some(10));

            // No compact-sync import put a smaller size at the journal end, so the bounded view
            // ends at the cap with a tip size above it and recovery stays bounded. The offsets open
            // clamps the watermark to the cap, so inspection anchors at the ceiling and leaves the
            // acknowledged offsets untouched. No retained witness fits under the cap, so selection
            // fails.
            assert!(matches!(
                open_bounded::<mmr::Family>(context.child("failed"), witness_cfg.clone(), cap)
                    .await,
                Err(Error::HistoricalFloorPruned(found)) if found == cap
            ));

            // Selection fails before publication, and inspection anchored at the ceiling skips the
            // offsets truncate, so the failed attempt leaves the durable watermark at 10.
            let watermark = fixed::Journal::<_, u64>::persisted_watermark(
                context.child("after"),
                &offsets_partition,
            )
            .await
            .unwrap();
            assert_eq!(watermark, Some(10));

            // Reinitializing at `tip` recovers every retained witness.
            let reopened = open_bounded::<mmr::Family>(context.child("retry"), witness_cfg, tip)
                .await
                .unwrap();
            assert_eq!(reopened.size(), tip);
            reopened.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_prune_past_tip_keeps_tip() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-prune-past-tip";
            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let target = db.target();

            // Prune with a boundary beyond the tip: the tip entry must survive.
            let boundary = db.size() + 100;
            let db = db.prune(boundary).await.unwrap();
            assert_eq!(db.target(), target);
            drop(db);

            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
            assert_eq!(reopened.target(), target);
            reopened.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_preserves_pre_advance_batch() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(
                context.child("db"),
                "immutable-rewind-preserves-pre-advance",
            )
            .await;

            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let size_after_first = db.size();

            // Merkleize a batch against the post-commit-A state.
            let held = db
                .new_batch()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();

            // Advance past that state and commit, then reopen at that state.
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let db = {
                _ = db.sync().await.unwrap();
                open_bounded::<mmr::Family>(
                    context.child("cap"),
                    witness_config("immutable-rewind-preserves-pre-advance"),
                    size_after_first,
                )
                .await
            }
            .unwrap();

            // The recovery restored the state that `held` was merkleized against, so it still
            // matches the Merkle size and applies cleanly.
            let db = db.apply_batch(held).await.unwrap();

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_noop_commit_after_commit() {
        deterministic::Runner::default().start(|context| async move {
            let db =
                open_db::<mmr::Family>(context.child("db"), "immutable-noop-after-commit").await;

            let k1 = Sha256::hash(&[&[1]]);
            let v1 = Sha256::fill(11u8);
            let k2 = Sha256::hash(&[&[2]]);
            let v2 = Sha256::fill(22u8);
            let batch = db
                .new_batch()
                .set(k1, v1)
                .set(k2, v2)
                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let root_after_first = db.root();
            let size_after_first = db.size();

            let db = db.sync().await.unwrap();
            assert_eq!(db.size(), size_after_first);
            assert_eq!(db.root(), root_after_first);

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_noop_commit_after_reopen() {
        deterministic::Runner::default().start(|context| async move {
            let partition = "immutable-noop-after-reopen";

            let (root_before_drop, size_before_drop) = {
                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
                let k1 = Sha256::hash(&[&[1]]);
                let v1 = Sha256::fill(11u8);
                let k2 = Sha256::hash(&[&[2]]);
                let v2 = Sha256::fill(22u8);
                let batch = db
                    .new_batch()
                    .set(k1, v1)
                    .set(k2, v2)
                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
                    .await
                    .unwrap();
                let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
                (db.root(), db.size())
            };

            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
            assert_eq!(db.root(), root_before_drop);
            assert_eq!(db.size(), size_before_drop);

            let db = db.sync().await.unwrap();
            assert_eq!(db.size(), size_before_drop);
            assert_eq!(db.root(), root_before_drop);

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_noop_commit_after_bounded_initialization() {
        deterministic::Runner::default().start(|context| async move {
            let db =
                open_db::<mmr::Family>(context.child("db"), "immutable-noop-after-rewind").await;

            let k1 = Sha256::hash(&[&[1]]);
            let v1 = Sha256::fill(11u8);
            let k2 = Sha256::hash(&[&[2]]);
            let v2 = Sha256::fill(22u8);
            let batch = db
                .new_batch()
                .set(k1, v1)
                .set(k2, v2)
                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let root_after_first = db.root();
            let size_after_first = db.size();

            let k3 = Sha256::hash(&[&[3]]);
            let v3 = Sha256::fill(33u8);
            let batch = db
                .new_batch()
                .set(k3, v3)
                .merkleize(&db, Some(Sha256::fill(0xbb)), Location::new(1))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();

            let db = {
                _ = db.sync().await.unwrap();
                open_bounded::<mmr::Family>(
                    context.child("cap"),
                    witness_config("immutable-noop-after-rewind"),
                    size_after_first,
                )
                .await
            }
            .unwrap();
            assert_eq!(db.size(), size_after_first);
            assert_eq!(db.root(), root_after_first);

            let db = db.sync().await.unwrap();
            assert_eq!(db.size(), size_after_first);
            assert_eq!(db.root(), root_after_first);

            db.destroy().await.unwrap();
        });
    }

    #[test_traced("INFO")]
    fn test_compact_bounded_initialization_makes_post_advance_batch_stale() {
        deterministic::Runner::default().start(|context| async move {
            let db =
                open_db::<mmr::Family>(context.child("db"), "immutable-rewind-makes-stale").await;

            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let size_after_first = db.size();

            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();
            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();

            // Merkleize a batch against the post-commit-B state, which the recovery will discard.
            let held = db
                .new_batch()
                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
                .merkleize(&db, None, Location::new(0))
                .await
                .unwrap();

            let db = {
                _ = db.sync().await.unwrap();
                open_bounded::<mmr::Family>(
                    context.child("cap"),
                    witness_config("immutable-rewind-makes-stale"),
                    size_after_first,
                )
                .await
            }
            .unwrap();

            // After recovery, mem.size reflects post-commit-A, but the held batch starts after
            // post-commit-B. Apply must be rejected with StaleBatch.
            assert!(matches!(db.apply_batch(held).await, Err(Error::StaleBatch)));
        });
    }

    #[test_traced("INFO")]
    fn test_compact_floor_beyond_size() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-floor-beyond").await;

            let batch = db.new_batch().merkleize(&db, None, Location::new(2)).await;

            assert!(matches!(
                batch,
                Err(Error::FloorBeyondSize(floor, tip))
                    if floor == Location::new(2) && tip == Location::new(1)
            ));
        });
    }

    // Merkleize returns `FloorBeyondSize` for a chained batch whose floor exceeds its own commit
    // location. The commit location includes the parent's operations.
    #[test_traced("INFO")]
    fn test_compact_chained_floor_beyond_commit() {
        deterministic::Runner::default().start(|context| async move {
            let db = open_db::<mmr::Family>(context.child("db"), "immutable-ancestor-floor-beyond")
                .await;

            // parent: set + commit at loc 2, floor=2.
            let parent = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
                .merkleize(&db, None, Location::new(2))
                .await
                .unwrap();

            // child: set + commit at loc 4, floor=5 (one past its commit).
            let child = parent
                .new_batch::<Sha256>()
                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
                .merkleize(&db, None, Location::new(5))
                .await;

            assert!(matches!(
                child,
                Err(Error::FloorBeyondSize(floor, commit))
                    if floor == Location::new(5) && commit == Location::new(4)
            ));
        });
    }
}