commonware_storage/qmdb/keyless/

mod.rs

//! The [Keyless] qmdb allows for append-only storage of data that can later be retrieved by its
//! location. Both fixed-size and variable-size values are supported via the [fixed] and [variable]
//! submodules.
//!
//! # Examples
//!
//! ```ignore
//! // Simple mode: apply a batch, then durably commit it.
//! let batch = db.new_batch().append(value);
//! let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await?;
//! let db = db.apply_batch(merkleized).await?.commit().await?;
//! ```
//!
//! ```ignore
//! // Several batches can be built on one parent before any is applied.
//! let floor = db.inactivity_floor_loc();
//! let parent = db.new_batch().append(value_a);
//! let parent = parent.merkleize(&db, None, floor).await?;
//!
//! let child_a = parent.new_batch();
//! let child_a = child_a.append(value_b);
//! let child_a = child_a.merkleize(&db, None, floor).await?;
//!
//! let child_b = parent.new_batch();
//! let child_b = child_b.append(value_c);
//! let child_b = child_b.merkleize(&db, None, floor).await?;
//!
//! let db = db.apply_batch(child_a).await?.commit().await?;
//! ```
//!
//! ```ignore
//! // Sequential commit: apply parent then child.
//! let floor = db.inactivity_floor_loc();
//! let parent = db.new_batch().append(value_a);
//! let parent_m = parent.merkleize(&db, None, floor).await?;
//! let child = parent_m.new_batch().append(value_b);
//! let child_m = child.merkleize(&db, None, floor).await?;
//!
//! let db = db.apply_batch(parent_m).await?;
//! let db = db.apply_batch(child_m).await?.commit().await?;
//! ```

use crate::{
    Context,
    journal::{
        authenticated,
        contiguous::{Contiguous, Mutable},
    },
    merkle::{Family, Location, Proof, full::Config as MerkleConfig},
    qmdb::{
        Error, any::value::ValueEncoding, chain, metrics::Metrics, single_operation_root,
        sync::source,
    },
};
use commonware_codec::EncodeShared;
use commonware_cryptography::Hasher;
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use commonware_runtime::Handle;
use std::{num::NonZeroU64, sync::Arc};
use tracing::{debug, warn};

pub mod batch;
mod compact;
pub mod fixed;
mod operation;
pub(crate) mod sync;
pub mod variable;
pub use compact::{
    Config as CompactConfig, Db as CompactDb, MerkleizedBatch as CompactMerkleizedBatch,
    UnmerkleizedBatch as CompactUnmerkleizedBatch,
};
pub use operation::{APPEND_CONTEXT, COMMIT_CONTEXT, Operation};

/// Compute the authenticated root of a newly initialized database without opening storage.
///
/// The initial commit never carries metadata, so this root always represents `Commit(None, 0)`.
pub fn initial_root<F, V, H>() -> H::Digest
where
    F: Family,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, V>: EncodeShared,
{
    single_operation_root::<F, H>(&Operation::<F, V>::Commit(None, Location::new(0)))
}

/// Check that `locs` is strictly increasing and every location is below `end`.
fn check_locations<F: Family>(locs: &[Location<F>], end: Location<F>) -> Result<(), Error<F>> {
    if let Some(pair) = locs.windows(2).find(|pair| pair[0] >= pair[1]) {
        return Err(Error::LocationOutOfOrder(pair[0], pair[1]));
    }
    // Strictly increasing, so only the last location can reach `end`.
    if let Some(&last) = locs.last()
        && last >= end
    {
        return Err(Error::LocationOutOfBounds(last, end));
    }
    Ok(())
}

/// Configuration for a [Keyless] authenticated db.
#[derive(Clone)]
pub struct Config<J, S: Strategy> {
    /// Configuration for the Merkle structure backing the authenticated journal.
    pub merkle: MerkleConfig<S>,

    /// Configuration for the operations log journal.
    pub log: J,
}

/// A keyless authenticated database.
pub struct Keyless<F, E, V, C, H, S>
where
    F: Family,
    E: Context,
    V: ValueEncoding,
    C: Contiguous<Item = Operation<F, V>>,
    H: Hasher,
    S: Strategy,
    Operation<F, V>: EncodeShared,
{
    /// Authenticated journal of operations.
    journal: authenticated::Journal<F, E, C, H, S>,

    /// Cached canonical operations root.
    root: H::Digest,

    /// The inactivity floor declared by the last committed batch. Operations at locations below
    /// this value are considered inactive by the application and may be pruned.
    inactivity_floor_loc: Location<F>,

    /// Metrics for this database.
    metrics: Metrics<E>,
}

impl<F, E, V, C, H, S> std::fmt::Debug for Keyless<F, E, V, C, H, S>
where
    F: Family,
    E: Context,
    V: ValueEncoding,
    C: Mutable<Item = Operation<F, V>>,
    H: Hasher,
    S: Strategy,
    Operation<F, V>: EncodeShared,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Keyless")
            .field("bounds", &self.bounds())
            .field("inactivity_floor_loc", &self.inactivity_floor_loc())
            .finish_non_exhaustive()
    }
}

impl<F, E, V, C, H, S> Keyless<F, E, V, C, H, S>
where
    F: Family,
    E: Context,
    V: ValueEncoding,
    C: Mutable<Item = Operation<F, V>>,
    H: Hasher,
    S: Strategy,
    Operation<F, V>: EncodeShared,
{
    /// Initialize from the latest retained commit, discarding every operation after it.
    /// `Some(max_size)` selects the latest retained commit with at most `max_size` operations.
    /// `None` selects the latest retained state.
    ///
    /// After a bounded initialization past a prune, or a sync of a range that starts above the
    /// floor, the selected commit's inactivity floor can be below the retained start. A batch
    /// read of a location between the two that no batch chain holds then returns
    /// [`crate::journal::Error::ItemPruned`], while a database that retains the location answers
    /// it. Databases at the same commitment therefore answer batch reads alike only when each
    /// retains every operation from its floor onward.
    #[boxed]
    pub async fn init(
        context: E,
        cfg: Config<C::Config, S>,
        max_size: Option<Location<F>>,
    ) -> Result<Self, Error<F>>
    where
        C: authenticated::Backing<E>,
    {
        // Keyless restores commit fields without replaying a keyed snapshot, so its logical floor
        // may precede the retained operation prefix.
        let mut journal = crate::qmdb::init_journal::<F, E, C, H, S>(
            context.child("journal"),
            cfg.merkle,
            cfg.log,
            max_size,
            false,
        )
        .await?;
        let metrics = Metrics::new(context);
        if journal.size() == 0 {
            warn!("no operations found in log, creating initial commit");
            (journal, _) = journal
                .append(&Operation::Commit(None, Location::new(0)))
                .await?;
            journal = journal.sync().await?;
        }

        let (last_commit_loc, inactivity_floor_loc) = {
            let bounds = journal.bounds();
            let last_commit_loc = Location::new(
                bounds
                    .end
                    .checked_sub(1)
                    .expect("at least one commit should exist"),
            );
            let op = journal.read(*last_commit_loc).await?;
            let inactivity_floor_loc = op
                .has_floor()
                .expect("last operation should be a commit with floor");
            (last_commit_loc, inactivity_floor_loc)
        };
        let inactive_peaks = F::inactive_peaks(last_commit_loc + 1, inactivity_floor_loc);
        let root = journal.root(inactive_peaks)?;

        let db = Self {
            journal,
            root,
            inactivity_floor_loc,
            metrics,
        };
        db.update_metrics();
        Ok(db)
    }

    /// Get the value at location `loc` in the database.
    ///
    /// # Errors
    ///
    /// - Returns [`Error::LocationOutOfBounds`] if `loc` >= `self.bounds().end`.
    /// - Returns [`Error::Journal`] with [`crate::journal::Error::ItemPruned`] if `loc` <
    ///   `self.bounds().start`.
    pub async fn get(&self, loc: Location<F>) -> Result<Option<V::Value>, Error<F>> {
        let _timer = self.metrics.get_timer();
        self.metrics.get_calls.inc();
        self.metrics.lookups_requested.inc();
        let op_count = self.journal.bounds().end;
        if loc >= op_count {
            return Err(Error::LocationOutOfBounds(loc, Location::new(op_count)));
        }
        let op = self.journal.read(*loc).await?;

        let result = op.into_value();
        Ok(result)
    }

    /// Batch read values at multiple locations.
    ///
    /// Locations must be strictly increasing.
    /// Returns results in the same order as the input locations.
    ///
    /// # Errors
    ///
    /// - Returns [`Error::LocationOutOfOrder`] if the locations are not strictly increasing.
    /// - Returns [`Error::LocationOutOfBounds`] if any location >= `self.bounds().end`.
    /// - Returns [`Error::Journal`] with [`crate::journal::Error::ItemPruned`] if any location <
    ///   `self.bounds().start`.
    pub async fn get_many(&self, locs: &[Location<F>]) -> Result<Vec<Option<V::Value>>, Error<F>> {
        if locs.is_empty() {
            return Ok(Vec::new());
        }

        let _timer = self.metrics.get_many_timer();
        self.metrics.get_many_calls.inc();
        self.metrics.lookups_requested.inc_by(locs.len() as u64);
        check_locations(locs, Location::new(self.journal.bounds().end))?;
        let positions: Vec<u64> = locs.iter().map(|loc| **loc).collect();
        let ops = self.journal.read_many(&positions).await?;
        let result = ops.into_iter().map(|op| op.into_value()).collect();
        Ok(result)
    }

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

    /// Return the retained operation range `[start, end)`.
    ///
    /// Proof generation also requires the necessary Merkle nodes to be retained.
    pub fn bounds(&self) -> std::ops::Range<Location<F>> {
        let bounds = self.journal.bounds();
        Location::new(bounds.start)..Location::new(bounds.end)
    }

    /// Update state gauges from the current database state.
    fn update_metrics(&self) {
        let bounds = self.journal.bounds();
        self.metrics.update(
            bounds.end,
            bounds.start,
            *self.inactivity_floor_loc,
            bounds.end - 1,
        );
    }

    /// Return the inactivity floor declared by the last committed batch, which is the upper bound
    /// on [`Self::prune`]'s `loc`.
    ///
    /// This logical boundary may precede the retained start in [`Self::bounds`].
    pub const fn sync_boundary(&self) -> Location<F> {
        self.inactivity_floor_loc
    }

    /// Get the metadata associated with the last commit.
    pub async fn get_metadata(&self) -> Result<Option<V::Value>, Error<F>> {
        // The journal always ends with a commit operation.
        let op = self.journal.read(*self.journal.size() - 1).await?;
        let Operation::Commit(metadata, _floor) = op else {
            return Ok(None);
        };

        Ok(metadata)
    }

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

    /// Return a reference to the merkleization strategy.
    pub const fn strategy(&self) -> &S {
        self.journal.strategy()
    }

    /// Generate and return:
    ///  1. a proof of all operations applied to the db in the range starting at (and including)
    ///     location `start_loc`, and ending at the first of either:
    ///     - the last operation performed, or
    ///     - the operation `max_ops` from the start.
    ///  2. the operations corresponding to the leaves in this range.
    ///
    /// # Errors
    ///
    /// - Returns [`Error::Merkle`] with [`crate::merkle::Error::RangeOutOfBounds`] if `start_loc`
    ///   >= the number of operations.
    /// - Returns [`Error::Journal`] with [`crate::journal::Error::ItemPruned`] or [`Error::Merkle`]
    ///   with [`crate::merkle::Error::ElementPruned`] if a required operation or Merkle node has
    ///   been pruned.
    pub async fn proof(
        &self,
        start_loc: Location<F>,
        max_ops: NonZeroU64,
    ) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, V>>), Error<F>> {
        self.historical_proof(self.bounds().end, start_loc, max_ops)
            .await
    }

    /// Analogous to proof, but with respect to the state of the Merkle structure when it had
    /// `op_count` operations.
    ///
    /// `op_count` must be the size of a commit boundary.
    ///
    /// # Errors
    ///
    /// - Returns [`Error::Merkle`] with [`crate::merkle::Error::RangeOutOfBounds`] if `start_loc`
    ///   >= `op_count` or `op_count` > number of operations.
    /// - Returns [`Error::Journal`] with [`crate::journal::Error::ItemPruned`] or [`Error::Merkle`]
    ///   with [`crate::merkle::Error::ElementPruned`] if a required operation or Merkle node has
    ///   been pruned.
    /// - Returns [`Error::HistoricalFloorPruned`] if `op_count - 1` is retained but is not a commit
    ///   op.
    #[allow(clippy::type_complexity)]
    #[tracing::instrument(
        name = "qmdb.keyless.db.historical_proof",
        level = "info",
        skip_all,
        fields(
            op_count = *op_count,
            start_loc = *start_loc,
            max_ops = max_ops.get(),
        ),
    )]
    pub async fn historical_proof(
        &self,
        op_count: Location<F>,
        start_loc: Location<F>,
        max_ops: NonZeroU64,
    ) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, V>>), Error<F>> {
        if op_count > self.journal.size() {
            return Err(crate::merkle::Error::RangeOutOfBounds(op_count).into());
        }

        let inactive_peaks =
            crate::qmdb::inactive_peaks_at::<F, _>(&self.journal, op_count).await?;

        Ok(self
            .journal
            .historical_proof(op_count, start_loc, max_ops, inactive_peaks)
            .await?)
    }

    /// Return the pinned Merkle nodes for a lower operation boundary of `loc`.
    pub async fn pinned_nodes_at(&self, loc: Location<F>) -> Result<Vec<H::Digest>, Error<F>> {
        self.journal
            .merkle
            .pinned_nodes_at(loc)
            .await
            .map_err(Into::into)
    }

    /// Prune historical operations prior to `loc`. This does not affect the db's root.
    ///
    /// The retained start in [`Self::bounds`] can remain below `loc`.
    ///
    /// `prune` requires no prior commit. After a crash, the database remains recoverable;
    /// uncommitted operations are not guaranteed to survive.
    ///
    /// # Errors
    ///
    /// - Returns [`Error::PruneBeyondMinRequired`] if `loc` > the inactivity floor.
    #[tracing::instrument(name = "qmdb.keyless.db.prune", level = "info", skip_all)]
    #[boxed]
    pub async fn prune(mut self, loc: Location<F>) -> Result<Self, Error<F>> {
        let _timer = self.metrics.prune_timer();
        self.metrics.prune_calls.inc();
        if loc > self.inactivity_floor_loc {
            return Err(Error::PruneBeyondMinRequired(
                loc,
                self.inactivity_floor_loc,
            ));
        }
        (self.journal, _) = self.journal.prune(loc).await?;
        self.update_metrics();
        Ok(self)
    }

    /// Sync all database state to disk. While this isn't necessary to ensure durability of
    /// committed operations, periodic invocation may reduce memory usage and the time required to
    /// recover the database on restart.
    #[tracing::instrument(name = "qmdb.keyless.db.sync", level = "info", skip_all)]
    pub async fn sync(mut self) -> Result<Self, Error<F>> {
        let _timer = self.metrics.sync_timer();
        self.metrics.sync_calls.inc();
        self.journal = self.journal.sync().await?;
        Ok(self)
    }

    /// Begin durably persisting the journal state published by prior [`Keyless::apply_batch`]
    /// calls.
    ///
    /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit],
    /// plus a best-effort attempt to bound the recovery needed on startup. 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. A failed data sync also
    /// fails the next durability operation. A failed recovery-watermark sync is not observed by
    /// [Self::commit], and a failed merkle-node sync may not be. Both resurface on the next
    /// [Self::sync].
    #[tracing::instrument(name = "qmdb.keyless.db.start_sync", level = "info", skip_all)]
    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
        self.metrics.start_sync_calls.inc();
        let handle;
        (self.journal, handle) = self.journal.start_sync().await?;
        Ok((self, handle))
    }

    /// Durably commit the journal state published by prior [`Keyless::apply_batch`] calls.
    #[tracing::instrument(name = "qmdb.keyless.db.commit", level = "info", skip_all)]
    pub async fn commit(mut self) -> Result<Self, Error<F>> {
        let _timer = self.metrics.commit_timer();
        self.metrics.commit_calls.inc();
        self.journal = self.journal.commit().await?;
        Ok(self)
    }

    /// Destroy the db, removing all data from disk.
    #[boxed]
    pub async fn destroy(self) -> Result<(), Error<F>> {
        Ok(self.journal.destroy().await?)
    }

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

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

    /// Create an initial [`batch::MerkleizedBatch`] from the DB's current state.
    pub fn to_batch(&self) -> Arc<batch::MerkleizedBatch<F, H::Digest, V, S>> {
        Arc::new(batch::MerkleizedBatch {
            journal_batch: self.journal.to_merkleized_batch(),
            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: &batch::MerkleizedBatch<F, H::Digest, V, S>,
    ) -> Result<(), Error<F>> {
        batch.bounds.validate_apply_to(self.commitment())
    }

    /// Apply a [`batch::MerkleizedBatch`] to the database.
    ///
    /// This publishes the batch to the in-memory database state and appends it to the journal.
    /// Call [`Keyless::commit`] or [`Keyless::sync`], or await the handle returned by
    /// [`Keyless::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.keyless.db.apply_batch", level = "info", skip_all)]
    pub async fn apply_batch(
        mut self,
        batch: Arc<batch::MerkleizedBatch<F, H::Digest, V, S>>,
    ) -> Result<Self, Error<F>> {
        let _timer = self.metrics.apply_batch_timer();
        self.metrics.apply_batch_calls.inc();
        self.validate_batch(&batch)?;
        let start_loc = self.journal.size();

        self.journal = self.journal.apply_batch(&batch.journal_batch).await?;

        self.inactivity_floor_loc = batch.bounds.inactivity_floor;
        self.root = batch.root();
        let end_loc = batch.bounds.tip.size;
        debug!(size = ?end_loc, "applied batch");
        self.update_metrics();
        self.metrics
            .operations_applied
            .inc_by(*end_loc - *start_loc);
        Ok(self)
    }
}

impl<F, E, V, C, H, S> crate::qmdb::sync::Source for Keyless<F, E, V, C, H, S>
where
    F: Family,
    E: Context,
    V: ValueEncoding,
    C: Mutable<Item = Operation<F, V>>,
    H: Hasher,
    S: Strategy,
    Operation<F, V>: EncodeShared,
{
    type Family = F;
    type Digest = H::Digest;
    type Op = Operation<F, V>;
    type Error = Error<F>;

    async fn serve(&self, request: crate::qmdb::sync::Request<F>) -> source::Result<Self> {
        self.journal.serve(request).await
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::qmdb::{verify_proof, verify_proof_and_pinned_nodes};
    use commonware_cryptography::Sha256;
    use commonware_parallel::Strategy;
    use commonware_runtime::{Supervisor as _, deterministic};
    use commonware_utils::NZU64;
    use std::{future::Future, pin::Pin};

    pub(crate) type Reopen<D> =
        Box<dyn Fn(deterministic::Context) -> Pin<Box<dyn Future<Output = D> + Send>>>;

    pub(crate) type BoundedOpen<D, F> = Box<
        dyn Fn(
            deterministic::Context,
            Location<F>,
        ) -> Pin<Box<dyn Future<Output = Result<D, Error<F>>> + Send>>,
    >;

    type TestKeyless<F, V, C, H, S> = Keyless<F, deterministic::Context, V, C, H, S>;

    /// Test value factory: creates distinct values from an index.
    pub(crate) trait TestValue: Clone + PartialEq + std::fmt::Debug + Send + Sync {
        fn make(i: u64) -> Self;
    }

    impl TestValue for Vec<u8> {
        fn make(i: u64) -> Self {
            vec![(i % 255) as u8; ((i % 13) + 7) as usize]
        }
    }

    impl TestValue for commonware_utils::sequence::U64 {
        fn make(i: u64) -> Self {
            Self::new(i * 10 + 1)
        }
    }

    /// Emits the named test against `mmr::Family` and `mmb::Family`.
    macro_rules! keyless_tests {
        ($($name:ident => $scenario:ident, $fixture:ident;)*) => {
            $(
                #[test_traced]
                fn $name() {
                    deterministic::Runner::default().start(|ctx| async move {
                        keyless_tests!(@fixture $fixture, $scenario, mmr, ctx);
                    });
                }
            )*
            paste::paste! {
                $(
                    #[test_traced]
                    fn [<$name _mmb>]() {
                        deterministic::Runner::default().start(|ctx| async move {
                            keyless_tests!(@fixture $fixture, $scenario, mmb, ctx);
                        });
                    }
                )*
            }
        };
        (@fixture bounded, $scenario:ident, $family:ident, $ctx:ident) => {
            let db = open_db::<$family::Family>($ctx.child("db")).await;
            tests::$scenario($ctx, db, reopen::<$family::Family>(), bounded_open::<$family::Family>()).await;
        };
        (@fixture bounded_floor, $scenario:ident, $family:ident, $ctx:ident) => {
            let db = open_db::<$family::Family>($ctx.child("db")).await;
            tests::$scenario($ctx, db, bounded_open::<$family::Family>()).await;
        };
        (@fixture bounded_indexed, $scenario:ident, $family:ident, $ctx:ident) => {
            let db = open_db::<$family::Family>($ctx.child("db").with_attribute("index", 1)).await;
            tests::$scenario($ctx, db, reopen::<$family::Family>(), bounded_open::<$family::Family>()).await;
        };
        (@fixture db, $scenario:ident, $family:ident, $ctx:ident) => {
            let db = open_db::<$family::Family>($ctx.child("db")).await;
            tests::$scenario(db).await;
        };
        (@fixture pair, $scenario:ident, $family:ident, $ctx:ident) => {
            let db = open_db::<$family::Family>($ctx.child("db")).await;
            let foreign = open_db_with_suffix::<$family::Family>("foreign", $ctx.child("foreign")).await;
            tests::$scenario(db, foreign).await;
        };
        (@fixture reopen, $scenario:ident, $family:ident, $ctx:ident) => {
            let db = open_db::<$family::Family>($ctx.child("db")).await;
            tests::$scenario($ctx, db, reopen::<$family::Family>()).await;
        };
        (@fixture reopen_indexed, $scenario:ident, $family:ident, $ctx:ident) => {
            let db =
                open_db::<$family::Family>($ctx.child("db").with_attribute("index", 1)).await;
            tests::$scenario($ctx, db, reopen::<$family::Family>()).await;
        };
    }

    pub(super) use keyless_tests;

    /// Reads stay exact after the ancestor batches are dropped, because the
    /// journal batch retains every ancestor's items.
    #[boxed]
    pub(crate) async fn run_dropped_ancestor_reads<F: Family, V, C, S: Strategy>(
        mut db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        // Commit one value so the chain has committed state below it.
        let seed = db
            .new_batch()
            .append(V::Value::make(0))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        db = db.apply_batch(seed).await.unwrap();
        db = db.commit().await.unwrap();
        let floor = db.inactivity_floor_loc();

        // A three-level unapplied chain, two appends per level. Each level's
        // appends start at its parent's tip.
        let base_a = db.bounds().end;
        let a = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let base_b = a.bounds().tip.size;
        let b = a
            .new_batch::<Sha256>()
            .append(V::Value::make(3))
            .append(V::Value::make(4))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let base_c = b.bounds().tip.size;
        let c = b
            .new_batch::<Sha256>()
            .append(V::Value::make(5))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let child = c.new_batch::<Sha256>();
        drop(a);
        drop(b);

        // Appended values across all three levels resolve from the retained items.
        let locs = [base_a, base_a + 1, base_b, base_b + 1, base_c];
        let expected: Vec<_> = (1..=5u64).map(|i| Some(V::Value::make(i))).collect();
        for (loc, expected) in locs.iter().zip(&expected) {
            assert_eq!(&c.get(*loc, &db).await.unwrap(), expected);
            assert_eq!(&child.get(*loc, &db).await.unwrap(), expected);
        }
        assert_eq!(c.get_many(&locs, &db).await.unwrap(), expected);
        assert_eq!(child.get_many(&locs, &db).await.unwrap(), expected);

        // Merkleizing the child returns `StaleBatch`, because its dropped ancestors are unapplied.
        assert!(matches!(
            child.merkleize(&db, None, floor).await,
            Err(Error::StaleBatch)
        ));

        // Applying the chain lands the same values the reads reported.
        let db = db.apply_batch(c).await.unwrap();
        for (loc, expected) in locs.iter().zip(&expected) {
            assert_eq!(&db.get(*loc).await.unwrap(), expected);
        }
        db.destroy().await.unwrap();
    }

    /// Batch reads below the batch's inactivity floor return `BelowInactivityFloor` however far
    /// the database has pruned, and a merkleized batch returns `LocationOutOfBounds` at or past
    /// its tip, like an unmerkleized one.
    #[boxed]
    pub(crate) async fn run_reads_below_floor_refused<F: Family, V, C, S: Strategy>(
        mut db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        // Locations 1 and 2 hold appends, and the commit at 3 raises the floor to 2.
        let seed = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, Location::new(2))
            .await
            .unwrap();
        db = db.apply_batch(seed).await.unwrap();
        db = db.commit().await.unwrap();

        // A batch built on the database uses its floor.
        let batch = db.new_batch();
        assert!(matches!(
            batch.get(Location::new(1), &db).await,
            Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(1)
        ));
        assert_eq!(
            batch.get(Location::new(2), &db).await.unwrap(),
            Some(V::Value::make(2))
        );
        assert!(matches!(
            batch
                .get_many(&[Location::new(1), Location::new(2)], &db)
                .await,
            Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(1)
        ));

        // Unordered locations are reported before a location below the floor.
        assert!(matches!(
            batch
                .get_many(&[Location::new(1), Location::new(0)], &db)
                .await,
            Err(Error::LocationOutOfOrder(first, second))
                if first == Location::new(1) && second == Location::new(0)
        ));

        // A merkleized batch appending at 4 and raising the floor to 4 returns
        // `BelowInactivityFloor` for location 2.
        let parent = db
            .new_batch()
            .append(V::Value::make(3))
            .merkleize(&db, None, Location::new(4))
            .await
            .unwrap();
        assert!(matches!(
            parent.get(Location::new(2), &db).await,
            Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(2)
        ));
        assert_eq!(
            parent.get(Location::new(4), &db).await.unwrap(),
            Some(V::Value::make(3))
        );
        assert!(matches!(
            parent
                .get_many(&[Location::new(2), Location::new(4)], &db)
                .await,
            Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(2)
        ));
        assert!(matches!(
            parent
                .get_many(&[Location::new(3), Location::new(2)], &db)
                .await,
            Err(Error::LocationOutOfOrder(first, second))
                if first == Location::new(3) && second == Location::new(2)
        ));

        // A merkleized batch bounds reads by its own tip, like an unmerkleized one.
        let tip = parent.bounds().tip.size;
        assert!(matches!(
            parent.get(tip, &db).await,
            Err(Error::LocationOutOfBounds(loc, end)) if loc == tip && end == tip
        ));
        assert!(matches!(
            parent.get_many(&[Location::new(4), tip], &db).await,
            Err(Error::LocationOutOfBounds(loc, end)) if loc == tip && end == tip
        ));

        // Its unmerkleized child builds on the parent's floor.
        let child = parent.new_batch::<Sha256>();
        assert!(matches!(
            child.get(Location::new(3), &db).await,
            Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(3)
        ));
        assert_eq!(
            child.get(Location::new(4), &db).await.unwrap(),
            Some(V::Value::make(3))
        );
        drop(child);

        // A pruned instance returns `BelowInactivityFloor` for the location, not `ItemPruned`.
        db = db.apply_batch(parent).await.unwrap();
        let mut batch = db.new_batch();
        for i in 10..22 {
            batch = batch.append(V::Value::make(i));
        }
        let floor = db.bounds().end + 12;
        let batch = batch.merkleize(&db, None, floor).await.unwrap();
        db = db.apply_batch(batch).await.unwrap();
        db = db.commit().await.unwrap();
        db = db.prune(floor).await.unwrap();
        assert!(db.bounds().start > Location::new(2));
        assert!(matches!(
            db.new_batch().get(Location::new(2), &db).await,
            Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(2)
        ));
        db.destroy().await.unwrap();
    }

    /// A child built before its parent was applied merkleizes to the same root after the
    /// database prunes past the chain's recorded database boundary.
    #[boxed]
    pub(crate) async fn run_merkleize_across_prune<F: Family, V, C, S: Strategy>(
        mut db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        // Seed committed state so the chain's recorded boundary sits above zero.
        let mut seed = db.new_batch();
        for i in 0..80 {
            seed = seed.append(V::Value::make(i));
        }
        let seed = seed
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        db = db.apply_batch(seed).await.unwrap();
        db = db.commit().await.unwrap();

        // The parent declares a floor at its own commit, past the chain's recorded boundary.
        let parent_floor = db.bounds().end + 1;
        let parent = db
            .new_batch()
            .append(V::Value::make(100))
            .merkleize(&db, None, parent_floor)
            .await
            .unwrap();

        // Two identical children built before the apply, appending enough that their merkleize
        // merges deep over the database's Merkle nodes.
        let build = |parent: &Arc<batch::MerkleizedBatch<F, _, V, S>>| {
            let mut child = parent.new_batch::<Sha256>();
            for i in 200..280 {
                child = child.append(V::Value::make(i));
            }
            child
        };
        let child = build(&parent);
        let expected = build(&parent)
            .merkleize(&db, None, parent_floor)
            .await
            .unwrap()
            .root();

        // Apply the parent and prune to its floor.
        let db = db.apply_batch(parent).await.unwrap();
        let db = db.commit().await.unwrap();
        let floor = db.inactivity_floor_loc();
        let db = db.prune(floor).await.unwrap();
        assert!(*db.bounds().start > 0, "the prune must remove history");

        // The child built before the apply still merkleizes to the same root.
        let merkleized = child.merkleize(&db, None, parent_floor).await.unwrap();
        assert_eq!(merkleized.root(), expected);
        db.destroy().await.unwrap();
    }

    /// Once a sibling batch is applied, reads through the other sibling and its child return
    /// `StaleRead`, while merkleizing the child and applying the other sibling return `StaleBatch`.
    #[boxed]
    pub(crate) async fn run_stale_fork_refuses<F: Family, V, C, S: Strategy>(
        mut db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        // Commit one value so the siblings have committed state to read.
        let merkleized = db
            .new_batch()
            .append(V::Value::make(0))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        db = db.apply_batch(merkleized).await.unwrap();
        db = db.commit().await.unwrap();

        let floor = db.inactivity_floor_loc();
        let winner = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let loser = db
            .new_batch()
            .append(V::Value::make(2))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        assert_eq!(winner.bounds().tip.size, loser.bounds().tip.size);
        assert_ne!(winner.root(), loser.root());

        let child = loser.new_batch::<Sha256>().append(V::Value::make(3));
        let direct = db.new_batch().append(V::Value::make(4));
        let committed = Location::new(0);
        let db = db.apply_batch(winner).await.unwrap();

        // A batch built directly on the database before the apply is stale too.
        assert!(matches!(
            direct.get(committed, &db).await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            direct.get_many(&[committed], &db).await,
            Err(Error::StaleRead)
        ));

        assert!(matches!(
            child.get(committed, &db).await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            child.get_many(&[committed], &db).await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            loser.get(committed, &db).await,
            Err(Error::StaleRead)
        ));

        // The compatibility check precedes every other read argument check, so reads of no
        // locations, of a location at the reader's end, and of unordered locations all return
        // `StaleRead`.
        assert!(matches!(
            child.get_many(&[], &db).await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            loser.get_many(&[], &db).await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            child.get(child.size(), &db).await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            loser.get(loser.size(), &db).await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            child
                .get_many(&[Location::new(1), Location::new(0)], &db)
                .await,
            Err(Error::StaleRead)
        ));
        assert!(matches!(
            child.merkleize(&db, None, floor).await,
            Err(Error::StaleBatch)
        ));
        assert!(matches!(db.validate_batch(&loser), Err(Error::StaleBatch)));
        db.destroy().await.unwrap();
    }

    /// Applying a batch's child moves the database to a state outside the parent's chain, so
    /// reads through the parent return `StaleRead` while the child keeps reading.
    #[boxed]
    pub(crate) async fn run_descendant_apply_makes_parent_reads_stale<
        F: Family,
        V,
        C,
        S: Strategy,
    >(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        let floor = db.inactivity_floor_loc();
        let parent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let loc = Location::new(1);
        assert_eq!(parent.get(loc, &db).await.unwrap(), Some(V::Value::make(1)));
        let child = parent
            .new_batch::<Sha256>()
            .append(V::Value::make(2))
            .merkleize(&db, None, floor)
            .await
            .unwrap();

        let db = db.apply_batch(child.clone()).await.unwrap();
        assert_eq!(child.get(loc, &db).await.unwrap(), Some(V::Value::make(1)));
        assert!(matches!(parent.get(loc, &db).await, Err(Error::StaleRead)));
        assert!(matches!(
            parent.get_many(&[loc], &db).await,
            Err(Error::StaleRead)
        ));
        db.destroy().await.unwrap();
    }

    #[boxed]
    pub(crate) async fn run_merkleize_foreign_db<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
        foreign: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let batch = db
            .new_batch()
            .append(V::Value::make(11))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let db = db.apply_batch(batch).await.unwrap();
        let batch = foreign
            .new_batch()
            .append(V::Value::make(99))
            .merkleize(&foreign, None, Location::new(0))
            .await
            .unwrap();
        let foreign = foreign.apply_batch(batch).await.unwrap();
        assert_eq!(db.bounds().end, foreign.bounds().end);
        assert_ne!(db.root(), foreign.root());

        let batch = db.new_batch().append(V::Value::make(22));
        assert!(matches!(
            batch.merkleize(&foreign, None, Location::new(0)).await,
            Err(Error::StaleBatch)
        ));

        // A child must also reject a database outside its ancestor chain.
        let parent = db
            .new_batch()
            .append(V::Value::make(33))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let child = parent.new_batch::<Sha256>().append(V::Value::make(44));
        assert!(matches!(
            child.merkleize(&foreign, None, Location::new(0)).await,
            Err(Error::StaleBatch)
        ));
        db.destroy().await.unwrap();
        foreign.destroy().await.unwrap();
    }

    #[boxed]
    pub(crate) async fn run_merkleize_stale_sibling<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let direct = db.new_batch().append(V::Value::make(11));
        let sibling = db
            .new_batch()
            .append(V::Value::make(22))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let db = db.apply_batch(sibling).await.unwrap();
        let direct_result = direct.merkleize(&db, None, Location::new(0)).await;

        let parent = db
            .new_batch()
            .append(V::Value::make(33))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let child = parent.new_batch::<Sha256>().append(V::Value::make(44));
        let sibling = db
            .new_batch()
            .append(V::Value::make(55))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let db = db.apply_batch(sibling).await.unwrap();
        let child_result = child.merkleize(&db, None, Location::new(0)).await;

        db.destroy().await.unwrap();
        assert!(matches!(direct_result, Err(Error::StaleBatch)));
        assert!(matches!(child_result, Err(Error::StaleBatch)));
    }

    #[boxed]
    pub(crate) async fn run_merkleize_ancestor_states<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let grandparent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let parent = grandparent
            .new_batch::<Sha256>()
            .append(V::Value::make(2))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        let pending = parent
            .new_batch::<Sha256>()
            .append(V::Value::make(3))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();

        // The database may advance to a held intermediate ancestor.
        let db = db.apply_batch(Arc::clone(&grandparent)).await.unwrap();
        let applied = parent
            .new_batch::<Sha256>()
            .append(V::Value::make(3))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        assert_eq!(pending.root(), applied.root());

        // Once that ancestor is dropped, its commitment becomes the effective boundary.
        drop(grandparent);
        let retired = parent
            .new_batch::<Sha256>()
            .append(V::Value::make(3))
            .merkleize(&db, None, Location::new(0))
            .await
            .unwrap();
        assert_eq!(retired.bounds().db, db.commitment());
        assert_eq!(pending.root(), retired.root());

        // A child created before its immediate parent is applied still merkleizes and applies
        // afterward.
        let child = parent.new_batch::<Sha256>().append(V::Value::make(3));
        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 db = db.apply_batch(child).await.unwrap();
        let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap();
        assert!(verify_proof::<Sha256, _, _>(
            &proof,
            Location::new(0),
            &ops,
            &db.root()
        ));
        db.destroy().await.unwrap();
    }

    #[boxed]
    pub(crate) async fn run_empty<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let bounds = db.bounds();
        assert_eq!(bounds.end, 1); // initial commit should exist
        assert_eq!(bounds.start, Location::new(0));
        assert_eq!(db.get_metadata().await.unwrap(), None);
        assert_eq!(db.bounds().end - 1, Location::new(0));

        // Make sure closing/reopening gets us back to the same state, even after adding an uncommitted op.
        let root = db.root();
        {
            db.new_batch().append(V::Value::make(1));

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);

        let db = reopen(context.child("db").with_attribute("index", 2)).await;
        assert_eq!(db.root(), root);
        assert_eq!(db.bounds().end, 1);
        assert_eq!(db.get_metadata().await.unwrap(), None);

        // Test calling commit on an empty db which should make it (durably) non-empty.
        let metadata = V::Value::make(99);
        let merkleized = db
            .new_batch()
            .merkleize(&db, Some(metadata.clone()), db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        assert_eq!(db.bounds().end, 2); // 2 commit ops
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));
        assert_eq!(
            db.get(Location::new(1)).await.unwrap(),
            Some(metadata.clone())
        ); // the commit op
        let root = db.root();

        // Commit op should remain after reopen even without clean shutdown.
        drop(db);
        let db = reopen(context.child("db").with_attribute("index", 3)).await;
        assert_eq!(db.bounds().end, 2); // commit op should remain after re-open.
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
        assert_eq!(db.root(), root);
        assert_eq!(db.bounds().end - 1, Location::new(1));

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

    /// `operations()` must cover exactly the batch's own applied range and match the
    /// operations a post-apply `historical_proof` recovers from the log, for a db-based
    /// batch and for a chained batch applied after its ancestor.
    #[boxed]
    pub(crate) async fn run_operations_match_applied_log<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared + PartialEq + core::fmt::Debug,
    {
        let seed = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let (seed_start, seed_ops) = seed.operations();
        let seed_root = seed.root();
        let seed_proof = seed.proof(&db).unwrap();
        let seed_pins = seed.pinned_nodes(&db).unwrap();
        let applied_start = db.bounds().end;
        let db = db.apply_batch(seed).await.unwrap();
        let seed_range = applied_start..db.bounds().end;
        assert_eq!(seed_start, seed_range.start);
        assert_eq!(*seed_start + seed_ops.len() as u64, *seed_range.end);

        // A chained batch's operations are its own suffix only.
        let parent = db
            .new_batch()
            .append(V::Value::make(3))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let child = parent
            .new_batch::<H>()
            .append(V::Value::make(4))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let (parent_start, parent_ops) = parent.operations();
        let (child_start, child_ops) = child.operations();
        let (parent_root, child_root) = (parent.root(), child.root());
        let (parent_pins, child_pins) = (
            parent.pinned_nodes(&db).unwrap(),
            child.pinned_nodes(&db).unwrap(),
        );
        let (parent_proof, child_proof) = (parent.proof(&db).unwrap(), child.proof(&db).unwrap());
        let applied_start = db.bounds().end;
        let db = db.apply_batch(parent).await.unwrap();
        let parent_range = applied_start..db.bounds().end;

        // With the database at the parent's tip, the child's proof and pinned nodes are
        // unchanged.
        assert_eq!(child.proof(&db).unwrap(), child_proof);
        assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins);
        let applied_start = db.bounds().end;
        let db = db.apply_batch(child).await.unwrap();
        let child_range = applied_start..db.bounds().end;
        assert_eq!(parent_start, parent_range.start);
        assert_eq!(*parent_start + parent_ops.len() as u64, *parent_range.end);
        assert_eq!(child_start, child_range.start);
        assert_eq!(*child_start + child_ops.len() as u64, *child_range.end);

        // A write-free batch still captures its commit-only suffix.
        let empty = db
            .new_batch()
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let (empty_start, empty_ops) = empty.operations();
        let (empty_root, empty_proof) = (empty.root(), empty.proof(&db).unwrap());
        let empty_pins = empty.pinned_nodes(&db).unwrap();
        let applied_start = db.bounds().end;
        let db = db.apply_batch(empty).await.unwrap();
        let empty_range = applied_start..db.bounds().end;
        assert_eq!(empty_start, empty_range.start);
        assert_eq!(*empty_start + empty_ops.len() as u64, *empty_range.end);

        // Every captured delta and proof must match what the log recovers for its
        // range, and verify against the batch's own root with and without the pins.
        for (start, ops, proof, pins, root) in [
            (seed_start, seed_ops, seed_proof, seed_pins, seed_root),
            (
                parent_start,
                parent_ops,
                parent_proof,
                parent_pins,
                parent_root,
            ),
            (child_start, child_ops, child_proof, child_pins, child_root),
            (empty_start, empty_ops, empty_proof, empty_pins, empty_root),
        ] {
            let len = core::num::NonZeroU64::new(ops.len() as u64).unwrap();
            let end = Location::new(*start + ops.len() as u64);
            let (log_proof, log_ops) = db.historical_proof(end, start, len).await.unwrap();
            assert_eq!(log_ops, *ops);
            assert_eq!(log_proof, proof);
            assert!(verify_proof::<H, _, _>(&proof, start, &ops, &root));
            assert!(verify_proof_and_pinned_nodes::<H, _, _>(
                &proof, start, &ops, &pins, &root
            ));
        }

        // Flushing the applied batch prunes the store to its peaks. The late batch's base is
        // mid-mountain, so its artifacts return `ElementPruned` rather than unverifiable data.
        let late = db
            .new_batch()
            .append(V::Value::make(5))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(Arc::clone(&late)).await.unwrap();
        let db = db.commit().await.unwrap();
        assert!(matches!(
            late.proof(&db),
            Err(crate::qmdb::Error::Merkle(
                crate::merkle::Error::ElementPruned(_)
            ))
        ));
        assert!(matches!(
            late.pinned_nodes(&db),
            Err(crate::qmdb::Error::Merkle(
                crate::merkle::Error::ElementPruned(_)
            ))
        ));

        // A batch built on the flushed store reads every node below it from the pinned peaks.
        let flushed = db
            .new_batch()
            .append(V::Value::make(6))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let (flushed_start, flushed_ops) = flushed.operations();
        let flushed_root = flushed.root();
        let flushed_proof = flushed.proof(&db).unwrap();
        let flushed_pins = flushed.pinned_nodes(&db).unwrap();
        assert!(verify_proof_and_pinned_nodes::<H, _, _>(
            &flushed_proof,
            flushed_start,
            &flushed_ops,
            &flushed_pins,
            &flushed_root
        ));
        let applied_start = db.bounds().end;
        let db = db.apply_batch(flushed).await.unwrap();
        let flushed_range = applied_start..db.bounds().end;
        assert_eq!(flushed_start, flushed_range.start);
        assert_eq!(
            *flushed_start + flushed_ops.len() as u64,
            *flushed_range.end
        );

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

    #[boxed]
    pub(crate) async fn run_commit_after_sync_recovery<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let value0 = V::Value::make(10);
        let value1 = V::Value::make(20);

        // Commit and sync the first append so recovery has an older durable boundary.
        let first_loc = Location::new(1);
        let merkleized = db
            .new_batch()
            .append(value0.clone())
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        let db = db.sync().await.unwrap();

        // Commit the next append without syncing; reopen must replay it from the journal.
        let second_loc = db.bounds().end;
        let merkleized = db
            .new_batch()
            .append(value1.clone())
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        let committed_bounds = db.bounds();
        let committed_root = db.root();
        drop(db);

        let db = reopen(context.child("db").with_attribute("index", 2)).await;
        assert_eq!(db.bounds(), committed_bounds);
        assert_eq!(db.root(), committed_root);
        assert_eq!(db.get(first_loc).await.unwrap(), Some(value0));
        assert_eq!(db.get(second_loc).await.unwrap(), Some(value1));

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

    #[boxed]
    pub(crate) async fn run_build_basic<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        mut db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Build a db with 2 values and make sure we can get them back.
        let v1 = V::Value::make(1);
        let v2 = V::Value::make(2);

        {
            let batch = db.new_batch();
            let loc1 = batch.size();
            let batch = batch.append(v1.clone());
            let loc2 = batch.size();
            let batch = batch.append(v2.clone());
            assert_eq!(loc1, Location::new(1));
            assert_eq!(loc2, Location::new(2));
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }

        // Make sure closing/reopening gets us back to the same state.
        assert_eq!(db.bounds().end, 4); // 2 appends, 1 commit + 1 initial commit
        assert_eq!(db.get_metadata().await.unwrap(), None);
        assert_eq!(db.get(Location::new(3)).await.unwrap(), None); // the commit op
        let root = db.root();
        db.sync().await.unwrap();

        let db = reopen(context.child("db").with_attribute("index", 2)).await;
        assert_eq!(db.bounds().end, 4);
        assert_eq!(db.root(), root);
        assert_eq!(db.get(Location::new(1)).await.unwrap().unwrap(), v1);
        assert_eq!(db.get(Location::new(2)).await.unwrap().unwrap(), v2);

        // Make sure commit operation remains after drop/reopen.
        drop(db);
        let db = reopen(context.child("db").with_attribute("index", 3)).await;
        assert_eq!(db.bounds().end, 4);
        assert_eq!(db.root(), root);

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

    #[boxed]
    pub(crate) async fn run_recovery<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let root = db.root();
        const ELEMENTS: u64 = 100;

        // Create uncommitted appends then simulate failure.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i));
            }

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);
        // Should rollback to the previous root.
        let mut db = reopen(context.child("db").with_attribute("index", 2)).await;
        assert_eq!(root, db.root());

        // Apply the updates and commit them this time.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 100));
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        let db = db.commit().await.unwrap();
        let root = db.root();

        // Create uncommitted appends then simulate failure.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 200));
            }

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);
        // Should rollback to the previous root.
        let mut db = reopen(context.child("db").with_attribute("index", 3)).await;
        assert_eq!(root, db.root());

        // Apply the updates and commit them this time.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 300));
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        let db = db.commit().await.unwrap();
        let root = db.root();

        // Make sure we can reopen and get back to the same state.
        drop(db);
        let db = reopen(context.child("db").with_attribute("index", 4)).await;
        assert_eq!(db.bounds().end, 2 * ELEMENTS + 3);
        assert_eq!(db.root(), root);

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

    #[boxed]
    pub(crate) async fn run_proof<F: Family, V, C, S: Strategy>(
        mut db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        const ELEMENTS: u64 = 50;

        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i));
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        let root = db.root();

        let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap();
        assert!(verify_proof::<Sha256, _, _>(
            &proof,
            Location::new(0),
            &ops,
            &root,
        ));
        assert_eq!(ops.len() as u64, 1 + ELEMENTS + 1);

        let (proof, ops) = db.proof(Location::new(10), NZU64!(5)).await.unwrap();
        assert!(verify_proof::<Sha256, _, _>(
            &proof,
            Location::new(10),
            &ops,
            &root,
        ));
        assert_eq!(ops.len(), 5);

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

    #[boxed]
    pub(crate) async fn run_metadata<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let metadata = V::Value::make(99);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, Some(metadata.clone()), db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));

        let merkleized = db
            .new_batch()
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.get_metadata().await.unwrap(), None);

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

    #[boxed]
    pub(crate) async fn run_pruning<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Initial floor is 0, so pruning past 0 should fail.
        assert_eq!(db.inactivity_floor_loc(), Location::new(0));
        let result = db.prune(Location::new(1)).await;
        assert!(
            matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor))
                if prune_loc == Location::new(1) && floor == Location::new(0))
        );

        let db = reopen(context.child("reopen_empty")).await;

        // Add values and commit, advancing the floor to the new commit location.
        let first_commit_loc = Location::<F>::new(3);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, first_commit_loc)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.bounds().end - 1, first_commit_loc);
        assert_eq!(db.inactivity_floor_loc(), first_commit_loc);

        // Append one more, advancing the floor with it.
        let second_commit_loc = Location::<F>::new(5);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(3))
            .merkleize(&db, None, second_commit_loc)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();

        // Valid prune: up to the floor (previous commit location).
        let root = db.root();
        let db = db.prune(first_commit_loc).await.unwrap();
        assert_eq!(db.root(), root);

        // Pruning beyond the current floor fails.
        let new_floor = db.inactivity_floor_loc();
        let beyond = new_floor + 1;
        let result = db.prune(beyond).await;
        assert!(
            matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor))
                if prune_loc == beyond && floor == new_floor)
        );
    }

    #[boxed]
    pub(crate) async fn run_empty_db_recovery<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let root = db.root();
        const ELEMENTS: u64 = 200;

        // Reopen DB without clean shutdown and make sure the state is the same.
        drop(db);
        let db = reopen(context.child("db").with_attribute("index", 2)).await;
        assert_eq!(db.bounds().end, 1); // initial commit should exist
        assert_eq!(db.root(), root);

        // Simulate failure after inserting operations without a commit.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i));
            }

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);
        let db = reopen(context.child("db").with_attribute("index", 3)).await;
        assert_eq!(db.bounds().end, 1); // initial commit should exist
        assert_eq!(db.root(), root);

        // Repeat: simulate failure after inserting operations without a commit.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 500));
            }

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);
        let db = reopen(context.child("db").with_attribute("index", 4)).await;
        assert_eq!(db.bounds().end, 1); // initial commit should exist
        assert_eq!(db.root(), root);

        // One last check: multiple batches of uncommitted appends.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS * 3 {
                batch = batch.append(V::Value::make(i + 1000));
            }

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);
        let mut db = reopen(context.child("db").with_attribute("index", 5)).await;
        assert_eq!(db.bounds().end, 1); // initial commit should exist
        assert_eq!(db.root(), root);
        assert_eq!(db.bounds().end - 1, Location::new(0));

        // Apply the ops one last time but fully commit them this time, then clean up.
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 2000));
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        db.commit().await.unwrap();
        let db = reopen(context.child("db").with_attribute("index", 6)).await;
        assert!(db.bounds().end > 1);
        assert_ne!(db.root(), root);

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

    #[boxed]
    pub(crate) async fn run_replay_with_trailing_appends<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        mut db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Add some initial operations and commit.
        {
            let mut batch = db.new_batch();
            for i in 0..10u64 {
                batch = batch.append(V::Value::make(i));
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        let db = db.commit().await.unwrap();
        let committed_root = db.root();
        let committed_size = db.bounds().end;

        // Add exactly one more append (uncommitted).
        {
            db.new_batch().append(V::Value::make(99));

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);

        // Reopen and verify correct recovery.
        let mut db = reopen(context.child("db").with_attribute("index", 2)).await;
        assert_eq!(
            db.bounds().end,
            committed_size,
            "Should recover to last commit"
        );
        assert_eq!(db.root(), committed_root, "Root should match last commit");
        assert_eq!(
            db.bounds().end - 1,
            committed_size - 1,
            "Last commit location should be correct"
        );

        // Verify we can append and commit new data after recovery.
        let new_value = V::Value::make(77);
        {
            let batch = db.new_batch();
            let loc = batch.size();
            let batch = batch.append(new_value.clone());
            assert_eq!(
                loc, committed_size,
                "New append should get the expected location"
            );
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        let db = db.commit().await.unwrap();

        assert_eq!(db.get(committed_size).await.unwrap(), Some(new_value));

        let new_committed_root = db.root();
        let new_committed_size = db.bounds().end;

        // Add multiple uncommitted appends.
        {
            let mut batch = db.new_batch();
            for i in 0..5u64 {
                batch = batch.append(V::Value::make(200 + i));
            }

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);

        // Reopen and verify correct recovery.
        let db = reopen(context.child("db").with_attribute("index", 3)).await;
        assert_eq!(
            db.bounds().end,
            new_committed_size,
            "Should recover to last commit with multiple trailing appends"
        );
        assert_eq!(
            db.root(),
            new_committed_root,
            "Root should match last commit after multiple appends"
        );
        assert_eq!(
            db.bounds().end - 1,
            new_committed_size - 1,
            "Last commit location should be correct after multiple appends"
        );

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

    /// `get_many` on the DB and on unmerkleized/merkleized batches returns
    /// results consistent with individual `get` calls, and rejects locations that
    /// are not strictly increasing or out of bounds.
    #[boxed]
    pub(crate) async fn run_get_many<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let v1 = V::Value::make(1);
        let v2 = V::Value::make(2);
        let v3 = V::Value::make(3);

        // Commit v1 and v2 to disk.
        let batch = db.new_batch();
        let loc1 = batch.size();
        let batch = batch.append(v1.clone());
        let loc2 = batch.size();
        let batch = batch.append(v2.clone());
        let merkleized = batch
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();

        // DB-level get_many.
        let results = db.get_many(&[loc1, loc2]).await.unwrap();
        assert_eq!(results, vec![Some(v1.clone()), Some(v2.clone())]);

        // Empty input.
        let results = db.get_many(&[]).await.unwrap();
        assert!(results.is_empty());

        // Decreasing and repeated locations are rejected.
        assert!(matches!(
            db.get_many(&[loc2, loc1]).await,
            Err(Error::LocationOutOfOrder(first, second)) if first == loc2 && second == loc1
        ));
        assert!(matches!(
            db.get_many(&[loc1, loc1]).await,
            Err(Error::LocationOutOfOrder(..))
        ));

        // An out-of-bounds location is rejected even when it follows valid ones.
        let end = db.bounds().end;
        assert!(matches!(
            db.get_many(&[loc1, end]).await,
            Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end
        ));

        // Unmerkleized batch: pending appends + DB fallthrough.
        let batch = db.new_batch();
        let loc3 = batch.size();
        let batch = batch.append(v3.clone());
        let results = batch.get_many(&[loc1, loc3], &db).await.unwrap();
        assert_eq!(results, vec![Some(v1.clone()), Some(v3.clone())]);
        assert!(matches!(
            batch.get_many(&[loc3, loc1], &db).await,
            Err(Error::LocationOutOfOrder(..))
        ));
        let end = batch.size();
        assert!(matches!(
            batch.get_many(&[loc1, end], &db).await,
            Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end
        ));

        // Merkleized batch: parent chain + DB fallthrough.
        let parent = db.new_batch();
        let parent_loc = parent.size();
        let parent = parent
            .append(v3.clone())
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let results = parent.get_many(&[loc1, parent_loc], &db).await.unwrap();
        assert_eq!(results, vec![Some(v1.clone()), Some(v3.clone())]);
        assert!(matches!(
            parent.get_many(&[parent_loc, loc1], &db).await,
            Err(Error::LocationOutOfOrder(..))
        ));
        let end = parent.size();
        assert!(matches!(
            parent.get_many(&[loc1, end], &db).await,
            Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end
        ));
        let child = parent.new_batch::<Sha256>().append(V::Value::make(4));
        let results = child.get_many(&[loc1, loc2], &db).await.unwrap();
        assert_eq!(results, vec![Some(v1.clone()), Some(v2.clone())]);
        assert!(matches!(
            child.get_many(&[loc2, loc1], &db).await,
            Err(Error::LocationOutOfOrder(..))
        ));

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

    #[boxed]
    pub(crate) async fn run_batch_chained<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let v1 = V::Value::make(10);
        let v2 = V::Value::make(20);
        let v3 = V::Value::make(30);

        let parent = db.new_batch();
        let loc1 = parent.size();
        let parent = parent.append(v1.clone());
        let parent_m = parent
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        let child = parent_m.new_batch::<Sha256>();
        let loc2 = child.size();
        let child = child.append(v2.clone());
        let loc3 = child.size();
        let child = child.append(v3.clone());
        let child_m = child
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let child_root = child_m.root();

        let db = db.apply_batch(child_m).await.unwrap();
        let db = db.commit().await.unwrap();

        assert_eq!(db.root(), child_root);
        assert_eq!(db.get(loc1).await.unwrap(), Some(v1));
        assert_eq!(db.get(loc2).await.unwrap(), Some(v2));
        assert_eq!(db.get(loc3).await.unwrap(), Some(v3));

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

    #[boxed]
    pub(crate) async fn run_stale_batch<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let batch_a = db
            .new_batch()
            .append(V::Value::make(10))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let batch_b = db
            .new_batch()
            .append(V::Value::make(20))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        let db = db.apply_batch(batch_a).await.unwrap();
        let db = db.commit().await.unwrap();
        let root = db.root();
        let last_commit_loc = db.bounds().end - 1;

        let result = db.apply_batch(batch_b).await;
        assert!(matches!(result, Err(Error::StaleBatch)));

        // The rejection mutated nothing: reopening recovers the committed state.
        let db = reopen(context.child("reopen")).await;
        assert_eq!(db.root(), root);
        assert_eq!(db.bounds().end - 1, last_commit_loc);
        db.destroy().await.unwrap();
    }

    #[boxed]
    pub(crate) async fn run_partial_ancestor_commit<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Chain: DB <- A <- B <- C
        let a = db
            .new_batch()
            .append(V::Value::make(10))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let b = a
            .new_batch::<H>()
            .append(V::Value::make(20))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let c = b
            .new_batch::<H>()
            .append(V::Value::make(30))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        let expected_root = c.root();

        // Apply only A, then apply C directly (B's items applied via ancestor batches).
        let db = db.apply_batch(a).await.unwrap();
        let db = db.apply_batch(c).await.unwrap();

        // Root must match what the full chain produces.
        assert_eq!(db.root(), expected_root);

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

    #[boxed]
    pub(crate) async fn run_delayed_merkleize_after_ancestor_apply<
        F: Family,
        V,
        C,
        H,
        S: Strategy,
    >(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let floor = db.inactivity_floor_loc();
        let a = db
            .new_batch()
            .append(V::Value::make(10))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let b = a
            .new_batch::<H>()
            .append(V::Value::make(20))
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let c = b.new_batch::<H>().append(V::Value::make(30));

        let db = db.apply_batch(a).await.unwrap();
        let c = c.merkleize(&db, None, floor).await.unwrap();
        let expected_root = c.root();
        let db = db.apply_batch(c).await.unwrap();

        assert_eq!(db.root(), expected_root);
        db.destroy().await.unwrap();
    }

    #[boxed]
    pub(crate) async fn run_to_batch<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let batch = db.new_batch();
        let loc1 = batch.size();
        let batch = batch.append(V::Value::make(10));
        let merkleized = batch
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();

        let snapshot = db.to_batch();
        assert_eq!(snapshot.root(), db.root());

        let child_batch = snapshot.new_batch::<Sha256>();
        let loc2 = child_batch.size();
        let child_batch = child_batch.append(V::Value::make(20));
        let merkleized = child_batch
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();

        assert_eq!(db.get(loc1).await.unwrap(), Some(V::Value::make(10)));
        assert_eq!(db.get(loc2).await.unwrap(), Some(V::Value::make(20)));

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

    #[boxed]
    pub(crate) async fn run_non_empty_recovery<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        mut db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Append many values then commit, advancing the floor to the new commit so we can
        // later prune up to it.
        const ELEMENTS: u64 = 200;
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i));
            }
            let new_commit = db.bounds().end + ELEMENTS;
            let merkleized = batch.merkleize(&db, None, new_commit).await.unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        let db = db.commit().await.unwrap();
        let root = db.root();
        let op_count = db.bounds().end;

        // Reopen DB without clean shutdown and make sure the state is the same.
        drop(db);
        let db = reopen(context.child("db").with_attribute("index", 2)).await;
        assert_eq!(db.bounds().end, op_count);
        assert_eq!(db.root(), root);
        assert_eq!(db.bounds().end - 1, op_count - 1);
        drop(db);

        // Insert many operations without commit, then simulate failure.
        let db = reopen(context.child("recovery_a")).await;
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 1000));
            }

            // Don't merkleize/apply, to simulate a failed commit
        }
        drop(db);
        let db = reopen(context.child("recovery_b")).await;
        assert_eq!(db.bounds().end, op_count);
        assert_eq!(db.root(), root);
        drop(db);

        // Repeat after pruning to the last commit.
        let db = reopen(context.child("db").with_attribute("index", 3)).await;
        let last_commit = db.bounds().end - 1;
        let db = db.prune(last_commit).await.unwrap();
        assert_eq!(db.bounds().end, op_count);
        assert_eq!(db.root(), root);
        db.sync().await.unwrap();

        let db = reopen(context.child("recovery_c")).await;
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 2000));
            }
        }
        drop(db);
        let db = reopen(context.child("recovery_d")).await;
        assert_eq!(db.bounds().end, op_count);
        assert_eq!(db.root(), root);
        drop(db);

        // Apply the ops one last time but fully commit them this time, then clean up.
        let mut db = reopen(context.child("db").with_attribute("index", 4)).await;
        {
            let mut batch = db.new_batch();
            for i in 0..ELEMENTS {
                batch = batch.append(V::Value::make(i + 3000));
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        db.commit().await.unwrap();
        let db = reopen(context.child("db").with_attribute("index", 5)).await;
        let bounds = db.bounds();
        assert!(bounds.end > op_count);
        assert_ne!(db.root(), root);

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

    #[boxed]
    pub(crate) async fn run_proof_comprehensive<F: Family, V, C, S: Strategy>(
        mut db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        // Build a db with some values.
        const ELEMENTS: u64 = 100;
        {
            let mut batch = db.new_batch();
            for i in 0u64..ELEMENTS {
                batch = batch.append(V::Value::make(i));
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }

        // Test that historical proof fails with op_count > number of operations.
        assert!(matches!(
            db.historical_proof(db.bounds().end + 1, Location::new(5), NZU64!(10))
                .await,
            Err(Error::<F>::Merkle(crate::merkle::Error::RangeOutOfBounds(
                _
            )))
        ));

        let root = db.root();

        for (start_loc, max_ops) in [
            (0, 10),
            (10, 5),
            (50, 20),
            (90, 15),
            (0, 1),
            (ELEMENTS - 1, 1),
            (ELEMENTS, 1),
        ] {
            let (proof, ops) = db
                .proof(Location::new(start_loc), NZU64!(max_ops))
                .await
                .unwrap();
            assert!(
                verify_proof::<Sha256, _, _>(&proof, Location::new(start_loc), &ops, &root,),
                "Failed to verify proof for range starting at {start_loc} with max {max_ops} ops",
            );
            let expected_ops = std::cmp::min(max_ops, *db.bounds().end - start_loc);
            assert_eq!(ops.len() as u64, expected_ops);

            let wrong_root = Sha256::hash(&[&[0xFF; 32]]);
            assert!(!verify_proof::<Sha256, _, _>(
                &proof,
                Location::new(start_loc),
                &ops,
                &wrong_root,
            ));
            if start_loc > 0 {
                assert!(!verify_proof::<Sha256, _, _>(
                    &proof,
                    Location::new(start_loc - 1),
                    &ops,
                    &root,
                ));
            }
        }

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

    #[boxed]
    pub(crate) async fn run_proof_with_pruning<F: Family, V, C, S: Strategy>(
        context: deterministic::Context,
        mut db: TestKeyless<F, V, C, Sha256, S>,
        reopen: Reopen<TestKeyless<F, V, C, Sha256, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        const ELEMENTS: u64 = 100;
        {
            let mut batch = db.new_batch();
            for i in 0u64..ELEMENTS {
                batch = batch.append(V::Value::make(i));
            }
            let new_commit = db.bounds().end + ELEMENTS;
            let merkleized = batch.merkleize(&db, None, new_commit).await.unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }

        {
            let mut batch = db.new_batch();
            for i in ELEMENTS..ELEMENTS * 2 {
                batch = batch.append(V::Value::make(i));
            }
            let new_commit = db.bounds().end + ELEMENTS;
            let merkleized = batch.merkleize(&db, None, new_commit).await.unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }
        let root = db.root();

        const PRUNE_LOC: u64 = 30;
        let db = db.prune(Location::new(PRUNE_LOC)).await.unwrap();
        let oldest_retained = db.bounds().start;
        assert_eq!(db.root(), root);

        db.sync().await.unwrap();
        let db = reopen(context).await;
        assert_eq!(db.root(), root);

        for (start_loc, max_ops) in [
            (oldest_retained, 10),
            (Location::new(50), 20),
            (Location::new(150), 10),
            (Location::new(190), 15),
        ] {
            if start_loc < oldest_retained {
                continue;
            }
            let (proof, ops) = db.proof(start_loc, NZU64!(max_ops)).await.unwrap();
            assert!(verify_proof::<Sha256, _, _>(&proof, start_loc, &ops, &root,));
        }

        let aggressive_prune: Location<F> = Location::new(150);
        let db = db.prune(aggressive_prune).await.unwrap();

        let new_oldest = db.bounds().start;
        let (proof, ops) = db.proof(new_oldest, NZU64!(20)).await.unwrap();
        assert!(verify_proof::<Sha256, _, _>(
            &proof, new_oldest, &ops, &root,
        ));

        let almost_all = db.bounds().end - 5;
        let db = db.prune(almost_all).await.unwrap();
        let final_oldest = db.bounds().start;
        if final_oldest < db.bounds().end {
            let (final_proof, final_ops) = db.proof(final_oldest, NZU64!(10)).await.unwrap();
            assert!(verify_proof::<Sha256, _, _>(
                &final_proof,
                final_oldest,
                &final_ops,
                &root,
            ));
        }

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

    #[boxed]
    pub(crate) async fn run_get_out_of_bounds<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        assert!(db.get(Location::new(0)).await.unwrap().is_none());

        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();

        assert_eq!(
            db.get(Location::new(1)).await.unwrap(),
            Some(V::Value::make(1))
        );
        assert!(db.get(Location::new(3)).await.unwrap().is_none());
        assert!(matches!(
            db.get(Location::new(4)).await,
            Err(Error::LocationOutOfBounds(loc, size))
                if loc == Location::new(4) && size == Location::new(4)
        ));

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

    #[boxed]
    pub(crate) async fn run_get_pruned<F: Family, V, C, H, S: Strategy>(
        mut db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let mut round = 0u64;
        while db.bounds().start == Location::new(0) {
            round += 1;
            assert!(round <= 64, "failed to prune any history");
            db = commit_appends(db, (0..16).map(|i| V::Value::make(round * 100 + i)), None).await;
            let last_commit = db.bounds().end - 1;
            db = db.prune(last_commit).await.unwrap();
        }

        let start = db.bounds().start;
        let pruned = start - 1;
        assert!(matches!(
            db.get(pruned).await,
            Err(Error::Journal(crate::journal::Error::ItemPruned(loc))) if loc == *pruned
        ));
        assert!(matches!(
            db.get_many(&[pruned, start]).await,
            Err(Error::Journal(crate::journal::Error::ItemPruned(loc))) if loc == *pruned
        ));
        db.get(start).await.unwrap();
        db.get_many(&[start]).await.unwrap();

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

    #[boxed]
    pub(crate) async fn run_batch_get<F: Family, V, C, H, S: Strategy>(
        mut db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let base_vals: Vec<V::Value> = (0..3).map(|i| V::Value::make(10 + i)).collect();
        let mut base_locs = Vec::new();
        {
            let mut batch = db.new_batch();
            for v in &base_vals {
                let loc = batch.size();
                batch = batch.append(v.clone());
                base_locs.push(loc);
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }

        let batch = db.new_batch();
        for (i, loc) in base_locs.iter().enumerate() {
            assert_eq!(
                batch.get(*loc, &db).await.unwrap(),
                Some(base_vals[i].clone()),
            );
        }

        let new_val = V::Value::make(99);
        let new_loc = batch.size();
        let batch = batch.append(new_val.clone());
        assert_eq!(batch.get(new_loc, &db).await.unwrap(), Some(new_val));
        let end = new_loc + 1;
        assert!(matches!(
            batch.get(end, &db).await,
            Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end
        ));
        assert!(matches!(
            batch.get_many(&[base_locs[0], end], &db).await,
            Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end
        ));

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

    #[boxed]
    pub(crate) async fn run_batch_stacked_get<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let v1 = V::Value::make(1);
        let v2 = V::Value::make(2);

        let parent = db.new_batch();
        let loc1 = parent.size();
        let parent = parent.append(v1.clone());
        let parent_m = parent
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        let child = parent_m.new_batch::<Sha256>();
        assert_eq!(child.get(loc1, &db).await.unwrap(), Some(v1));

        let loc2 = child.size();
        let child = child.append(v2.clone());
        assert_eq!(child.get(loc2, &db).await.unwrap(), Some(v2));
        assert!(matches!(
            child.get(Location::new(9999), &db).await,
            Err(Error::LocationOutOfBounds(loc, bound))
                if loc == Location::new(9999) && bound == loc2 + 1
        ));

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

    #[boxed]
    pub(crate) async fn run_batch_speculative_root<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let mut batch = db.new_batch();
        for i in 0u64..10 {
            batch = batch.append(V::Value::make(i));
        }
        let merkleized = batch
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let speculative = merkleized.root();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.root(), speculative);

        let merkleized = db
            .new_batch()
            .append(V::Value::make(100))
            .merkleize(&db, Some(V::Value::make(55)), db.inactivity_floor_loc())
            .await
            .unwrap();
        let speculative = merkleized.root();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.root(), speculative);

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

    #[boxed]
    pub(crate) async fn run_merkleized_batch_get<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let base_val = V::Value::make(10);
        let merkleized = db
            .new_batch()
            .append(base_val.clone())
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();

        let new_val = V::Value::make(20);
        let merkleized = db
            .new_batch()
            .append(new_val.clone())
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        assert_eq!(
            merkleized.get(Location::new(1), &db).await.unwrap(),
            Some(base_val),
        );
        assert_eq!(
            merkleized.get(Location::new(3), &db).await.unwrap(),
            Some(new_val),
        );
        assert_eq!(merkleized.get(Location::new(4), &db).await.unwrap(), None);
        assert!(matches!(
            merkleized.get(Location::new(5), &db).await,
            Err(Error::LocationOutOfBounds(loc, bound))
                if loc == Location::new(5) && bound == Location::new(5)
        ));

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

    #[boxed]
    pub(crate) async fn run_batch_chained_apply_sequential<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let v1 = V::Value::make(1);
        let v2 = V::Value::make(2);

        let parent = db.new_batch();
        let loc1 = parent.size();
        let parent = parent.append(v1.clone());
        let parent_m = parent
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let parent_root = parent_m.root();

        let db = db.apply_batch(parent_m).await.unwrap();
        assert_eq!(db.root(), parent_root);
        assert_eq!(db.get(loc1).await.unwrap(), Some(v1));

        let batch2 = db.new_batch();
        let loc2 = batch2.size();
        let batch2 = batch2.append(v2.clone());
        let batch2_m = batch2
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let batch2_root = batch2_m.root();
        let db = db.apply_batch(batch2_m).await.unwrap();
        assert_eq!(db.root(), batch2_root);
        assert_eq!(db.get(loc2).await.unwrap(), Some(v2));

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

    #[boxed]
    pub(crate) async fn run_batch_many_sequential<F: Family, V, C, S: Strategy>(
        mut db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        const BATCHES: u64 = 20;
        const APPENDS_PER_BATCH: u64 = 5;
        let mut all_values: Vec<V::Value> = Vec::new();
        let mut all_locs: Vec<Location<F>> = Vec::new();

        for batch_idx in 0..BATCHES {
            let mut batch = db.new_batch();
            for j in 0..APPENDS_PER_BATCH {
                let v = V::Value::make(batch_idx * 10 + j);
                let loc = batch.size();
                batch = batch.append(v.clone());
                all_values.push(v);
                all_locs.push(loc);
            }
            let merkleized = batch
                .merkleize(&db, None, db.inactivity_floor_loc())
                .await
                .unwrap();
            db = db.apply_batch(merkleized).await.unwrap();
        }

        for (i, loc) in all_locs.iter().enumerate() {
            assert_eq!(db.get(*loc).await.unwrap(), Some(all_values[i].clone()));
        }

        let root = db.root();
        let (proof, ops) = db.proof(Location::new(0), NZU64!(1000)).await.unwrap();
        assert!(verify_proof::<Sha256, _, _>(
            &proof,
            Location::new(0),
            &ops,
            &root,
        ));
        assert_eq!(db.bounds().end, 1 + BATCHES * (APPENDS_PER_BATCH + 1));

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

    #[boxed]
    pub(crate) async fn run_batch_empty<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let root_before = db.root();
        let size_before = db.bounds().end;

        let merkleized = db
            .new_batch()
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let speculative = merkleized.root();
        let db = db.apply_batch(merkleized).await.unwrap();

        assert_ne!(db.root(), root_before);
        assert_eq!(db.root(), speculative);
        assert_eq!(db.bounds().end, size_before + 1);

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

    #[boxed]
    pub(crate) async fn run_batch_chained_merkleized_get<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let base_val = V::Value::make(10);
        let floor = db.inactivity_floor_loc();
        let merkleized = db
            .new_batch()
            .append(base_val.clone())
            .merkleize(&db, None, floor)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();

        let v1 = V::Value::make(1);
        let parent = db.new_batch();
        let loc1 = parent.size();
        let parent_m = parent
            .append(v1.clone())
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        let v2 = V::Value::make(2);
        let child = parent_m.new_batch::<Sha256>();
        let loc2 = child.size();
        let child_m = child
            .append(v2.clone())
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        assert_eq!(
            child_m.get(Location::new(1), &db).await.unwrap(),
            Some(base_val),
        );
        assert_eq!(child_m.get(loc1, &db).await.unwrap(), Some(v1.clone()));
        assert_eq!(child_m.get(loc2, &db).await.unwrap(), Some(v2.clone()));
        let child_end = child_m.size();
        assert!(matches!(
            child_m.get(child_end, &db).await,
            Err(Error::LocationOutOfBounds(loc, bound)) if loc == child_end && bound == child_end
        ));

        // An ancestor's view ends at its own tip.
        let parent_end = parent_m.size();
        assert!(matches!(
            parent_m.get(loc2, &db).await,
            Err(Error::LocationOutOfBounds(loc, bound)) if loc == loc2 && bound == parent_end
        ));

        // Applying the descendant moves the database off the ancestor's chain states.
        let db = db.apply_batch(Arc::clone(&parent_m)).await.unwrap();
        let db = db.apply_batch(child_m).await.unwrap();
        assert_eq!(db.get(loc2).await.unwrap(), Some(v2));
        assert!(matches!(
            parent_m.get(loc1, &db).await,
            Err(Error::StaleRead)
        ));

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

    #[boxed]
    pub(crate) async fn run_batch_large<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared + std::fmt::Debug,
    {
        const N: u64 = 500;
        let mut values = Vec::new();
        let mut locs = Vec::new();

        let mut batch = db.new_batch();
        for i in 0..N {
            let v = V::Value::make(i);
            locs.push(batch.size());
            batch = batch.append(v.clone());
            values.push(v);
        }
        let merkleized = batch
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();

        for (i, loc) in locs.iter().enumerate() {
            assert_eq!(db.get(*loc).await.unwrap(), Some(values[i].clone()));
        }

        let root = db.root();
        let (proof, ops) = db.proof(Location::new(0), NZU64!(1000)).await.unwrap();
        assert!(verify_proof::<Sha256, _, _>(
            &proof,
            Location::new(0),
            &ops,
            &root,
        ));
        assert_eq!(db.bounds().end, 1 + N + 1);

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

    #[boxed]
    pub(crate) async fn run_stale_batch_chained<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let common_parent = db
            .new_batch()
            .append(V::Value::make(10))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let sibling_a = common_parent
            .new_batch::<Sha256>()
            .append(V::Value::make(11))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let sibling_b = common_parent
            .new_batch::<Sha256>()
            .append(V::Value::make(12))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .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()
            .append(V::Value::make(1))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let parent_b = db
            .new_batch()
            .append(V::Value::make(2))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let child_b = parent_b
            .new_batch::<Sha256>()
            .append(V::Value::make(3))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .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();
    }

    #[boxed]
    pub(crate) async fn run_sequential_commit_parent_then_child<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let parent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let child = parent
            .new_batch::<Sha256>()
            .append(V::Value::make(2))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

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

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

    #[boxed]
    pub(crate) async fn run_stale_batch_child_before_parent<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        let parent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let child = parent
            .new_batch::<Sha256>()
            .append(V::Value::make(2))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

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

    #[boxed]
    pub(crate) async fn run_child_root_matches_pending_and_committed<F: Family, V, C, S: Strategy>(
        db: TestKeyless<F, V, C, Sha256, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        Operation<F, V>: EncodeShared,
    {
        // Build the child while the parent is unapplied.
        let parent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();
        let pending_child = parent
            .new_batch::<Sha256>()
            .append(V::Value::make(2))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        // Commit the parent, then rebuild the same logical child from the
        // committed DB state and compare roots.
        let db = db.apply_batch(parent).await.unwrap();
        let db = db.commit().await.unwrap();

        let committed_child = db
            .new_batch()
            .append(V::Value::make(2))
            .merkleize(&db, None, db.inactivity_floor_loc())
            .await
            .unwrap();

        assert_eq!(pending_child.root(), committed_child.root());

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

    async fn commit_appends<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
        values: impl IntoIterator<Item = V::Value>,
        metadata: Option<V::Value>,
    ) -> TestKeyless<F, V, C, H, S>
    where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Tests that don't specifically exercise floor behavior advance the floor to the new
        // commit location, so pruning up to the last commit works analogously to the pre-floor
        // semantics.
        let base_size = *db.bounds().end;
        let appends_iter: Vec<_> = values.into_iter().collect();
        let new_commit_loc = Location::new(base_size + appends_iter.len() as u64);
        let mut batch = db.new_batch();
        for value in appends_iter {
            batch = batch.append(value);
        }
        let merkleized = batch
            .merkleize(&db, metadata, new_commit_loc)
            .await
            .unwrap();
        let expected_end = merkleized.bounds().tip.size;
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.bounds().end, expected_end);
        db.commit().await.unwrap()
    }

    #[boxed]
    pub(crate) async fn run_bounded_initialization_recovery<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
        bounded: BoundedOpen<TestKeyless<F, V, C, H, S>, F>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let initial_root = db.root();
        let initial_size = db.bounds().end;

        let value_a = V::Value::make(1);
        let value_b = V::Value::make(2);
        let metadata_a = V::Value::make(3);
        let db = commit_appends(
            db,
            [value_a.clone(), value_b.clone()],
            Some(metadata_a.clone()),
        )
        .await;

        let root_before = db.root();
        let size_before = db.bounds().end;
        let commit_before = db.bounds().end - 1;

        let value_c = V::Value::make(4);
        let metadata_b = V::Value::make(5);
        let db = commit_appends(db, [value_c.clone()], Some(metadata_b.clone())).await;
        assert_ne!(db.root(), root_before);
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_b));

        drop(db);
        let db = bounded(context.child("cap"), size_before).await.unwrap();
        assert_eq!(db.root(), root_before);
        assert_eq!(db.bounds().end, size_before);
        assert_eq!(db.bounds().end - 1, commit_before);
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a.clone()));
        assert_eq!(
            db.get(Location::new(1)).await.unwrap(),
            Some(value_a.clone())
        );
        assert_eq!(
            db.get(Location::new(2)).await.unwrap(),
            Some(value_b.clone())
        );
        assert!(
            matches!(
                db.get(Location::new(4)).await,
                Err(Error::LocationOutOfBounds(_, size)) if size == size_before
            ),
            "discarded append should be out of bounds",
        );

        drop(db);
        let db = reopen(context.child("reopen")).await;
        assert_eq!(db.root(), root_before);
        assert_eq!(db.bounds().end, size_before);
        assert_eq!(db.bounds().end - 1, commit_before);
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
        assert_eq!(
            db.get(Location::new(1)).await.unwrap(),
            Some(value_a.clone())
        );
        assert_eq!(
            db.get(Location::new(2)).await.unwrap(),
            Some(value_b.clone())
        );
        assert!(matches!(
            db.get(Location::new(4)).await,
            Err(Error::LocationOutOfBounds(_, size)) if size == size_before
        ));

        drop(db);
        let db = bounded(context.child("cap"), initial_size).await.unwrap();
        assert_eq!(db.root(), initial_root);
        assert_eq!(db.bounds().end, initial_size);
        assert_eq!(db.get_metadata().await.unwrap(), None);
        assert!(matches!(
            db.get(Location::new(1)).await,
            Err(Error::LocationOutOfBounds(_, size)) if size == initial_size
        ));

        drop(db);
        let db = reopen(context.child("reopen_initial_boundary")).await;
        assert_eq!(db.root(), initial_root);
        assert_eq!(db.bounds().end, initial_size);
        assert_eq!(db.get_metadata().await.unwrap(), None);
        assert!(matches!(
            db.get(Location::new(1)).await,
            Err(Error::LocationOutOfBounds(_, size)) if size == initial_size
        ));

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

    #[boxed]
    pub(crate) async fn run_bounded_initialization_pruned_target_errors<
        F: Family,
        V,
        C,
        H,
        S: Strategy,
    >(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
        bounded: BoundedOpen<TestKeyless<F, V, C, H, S>, F>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let first_start = db.bounds().end;
        let mut db = commit_appends(db, (0..16).map(V::Value::make), None).await;

        let mut round = 0u64;
        loop {
            round += 1;
            assert!(
                round <= 64,
                "failed to prune enough history for the initialization test"
            );

            db = commit_appends(db, (0..16).map(|i| V::Value::make(round * 100 + i)), None).await;
            let last_commit = db.bounds().end - 1;
            db = db.prune(last_commit).await.unwrap();

            if db.bounds().start > first_start {
                break;
            }
        }

        let oldest_retained = db.bounds().start;
        drop(db);
        let Err(boundary_err) = bounded(context.child("cap_error"), oldest_retained).await else {
            panic!("expected bounded initialization to fail");
        };
        assert!(
            matches!(
                boundary_err,
                Error::Journal(crate::journal::Error::ItemPruned(_))
            ),
            "unexpected bounded initialization error at retained boundary: {boundary_err:?}"
        );

        let db = reopen(context.child("reopen_boundary")).await;
        drop(db);
        let Err(err) = bounded(context.child("cap_error"), first_start).await else {
            panic!("expected bounded initialization to fail");
        };
        assert!(
            matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
            "unexpected bounded initialization error: {err:?}"
        );
    }

    #[boxed]
    pub(crate) async fn run_floor_tracking<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Freshly created db has floor = 0.
        assert_eq!(db.inactivity_floor_loc(), Location::new(0));

        // Apply a batch with a declared floor; the db's floor should update.
        let floor_a = Location::<F>::new(2);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, floor_a)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        assert_eq!(db.inactivity_floor_loc(), floor_a);

        // Reopen: floor should survive restart (it's part of the last commit operation).
        drop(db);
        let db = reopen(context.child("reopen")).await;
        assert_eq!(db.inactivity_floor_loc(), floor_a);

        // Floor may stay the same across a commit (monotonic non-decreasing).
        let merkleized = db
            .new_batch()
            .append(V::Value::make(3))
            .merkleize(&db, None, floor_a)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.inactivity_floor_loc(), floor_a);

        // Floor may advance further.
        let floor_b = Location::<F>::new(5);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(4))
            .merkleize(&db, None, floor_b)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.inactivity_floor_loc(), floor_b);

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

    #[boxed]
    pub(crate) async fn run_floor_regression_rejected<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Advance floor to 3 and commit so the state is durable.
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, Location::new(3))
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        assert_eq!(db.inactivity_floor_loc(), Location::new(3));

        // Merkleize rejects a commit with a lower floor.
        let Err(err) = db
            .new_batch()
            .append(V::Value::make(3))
            .merkleize(&db, None, Location::new(1))
            .await
        else {
            panic!("merkleize must reject the invalid floor");
        };
        assert!(
            matches!(err, Error::FloorRegressed(new, current) if *new == 1 && *current == 3),
            "unexpected error: {err:?}"
        );

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

    #[boxed]
    pub(crate) async fn run_floor_beyond_commit_loc_rejected<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Batch of 2 appends + 1 commit lands at locations [1..4); commit at 3, total_size = 4.
        // A floor > 3 (the commit location) is invalid. Even floor == 4 (one past the commit)
        // is rejected so a subsequent prune cannot remove the last readable commit.
        let Err(err) = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, Location::new(999))
            .await
        else {
            panic!("merkleize must reject the invalid floor");
        };
        assert!(
            matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 999 && *commit == 3),
            "unexpected error: {err:?}"
        );

        // Boundary: floor == total_size (= commit_loc + 1) is also rejected.
        let Err(err) = db
            .new_batch()
            .append(V::Value::make(3))
            .append(V::Value::make(4))
            .merkleize(&db, None, Location::new(4))
            .await
        else {
            panic!("merkleize must reject the invalid floor");
        };
        assert!(
            matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 4 && *commit == 3),
            "unexpected error: {err:?}"
        );
    }

    #[boxed]
    pub(crate) async fn run_bounded_initialization_restores_floor<F: Family, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        bounded: BoundedOpen<TestKeyless<F, V, C, H, S>, F>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // First commit: floor advances to 3 (= commit location).
        let floor_a = Location::<F>::new(3);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, floor_a)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        let initialization_bound = db.bounds().end;

        // Second commit: floor advances to 6.
        let floor_b = Location::<F>::new(6);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(3))
            .append(V::Value::make(4))
            .merkleize(&db, None, floor_b)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        assert_eq!(db.inactivity_floor_loc(), floor_b);

        // Reopen at the first commit. The floor should restore to floor_a.
        drop(db);
        let db = bounded(context.child("cap"), initialization_bound)
            .await
            .unwrap();
        assert_eq!(db.inactivity_floor_loc(), floor_a);

        // Prune is now gated at floor_a: pruning up to the floor works.
        let db = db.prune(floor_a).await.unwrap();

        // Pruning past the floor fails.
        let beyond = floor_a + 1;
        let Err(err) = db.prune(beyond).await else {
            panic!("expected prune to fail");
        };
        assert!(matches!(err, Error::PruneBeyondMinRequired(_, _)));
    }

    /// Floor is embedded in the Commit operation and therefore in the Merkle root: two databases
    /// with identical appends but different floors must produce different roots.
    #[boxed]
    pub(crate) async fn run_floor_changes_root<F: Family, V, C, H, S: Strategy>(
        db_a: TestKeyless<F, V, C, H, S>,
        db_b: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        let appends = [V::Value::make(1), V::Value::make(2)];

        // db_a commits with floor=0.
        let mut batch_a = db_a.new_batch();
        for v in appends.iter() {
            batch_a = batch_a.append(v.clone());
        }
        let merkleized = batch_a
            .merkleize(&db_a, None, Location::new(0))
            .await
            .unwrap();
        let db_a = db_a.apply_batch(merkleized).await.unwrap();

        // db_b commits the same appends but with floor=3 (= commit location).
        let mut batch_b = db_b.new_batch();
        for v in appends.iter() {
            batch_b = batch_b.append(v.clone());
        }
        let merkleized = batch_b
            .merkleize(&db_b, None, Location::new(3))
            .await
            .unwrap();
        let db_b = db_b.apply_batch(merkleized).await.unwrap();

        assert_ne!(db_a.root(), db_b.root());

        db_a.destroy().await.unwrap();
        db_b.destroy().await.unwrap();
    }

    /// A floor equal to the commit operation's location is on the tight boundary of acceptance.
    #[boxed]
    pub(crate) async fn run_floor_at_commit_loc_accepted<F: Family, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // 2 appends + 1 commit on top of the initial commit: commit lands at location 3.
        // floor == 3 (= commit_loc) is the maximum accepted value under the per-commit bound.
        let commit_loc = Location::<F>::new(3);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, commit_loc)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        assert_eq!(db.inactivity_floor_loc(), commit_loc);

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

    /// Commit, reinitialize at an earlier bound, then verify the restored floor survives another
    /// reopen.
    #[boxed]
    pub(crate) async fn run_bounded_initialization_after_reopen_with_floor<
        F: Family,
        V,
        C,
        H,
        S: Strategy,
    >(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
        bounded: BoundedOpen<TestKeyless<F, V, C, H, S>, F>,
    ) where
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // First commit: 2 appends + commit, floor advances to 3.
        let floor_a = Location::<F>::new(3);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .merkleize(&db, None, floor_a)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        let initialization_bound = db.bounds().end;

        // Second commit: 2 appends + commit, floor advances to 6.
        let floor_b = Location::<F>::new(6);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(3))
            .append(V::Value::make(4))
            .merkleize(&db, None, floor_b)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        db.commit().await.unwrap();

        // Reopen to simulate a crash after both commits were durable.
        let db = reopen(context.child("reopen")).await;
        assert_eq!(db.inactivity_floor_loc(), floor_b);

        // Reopen at the first commit. The floor should restore to floor_a.
        drop(db);
        let db = bounded(context.child("cap"), initialization_bound)
            .await
            .unwrap();
        assert_eq!(db.inactivity_floor_loc(), floor_a);
        assert_eq!(db.bounds().end - 1, Location::new(3));

        // Bounded initialization is durable. Reopen and confirm the floor again.
        drop(db);
        let db = reopen(context.child("reopen").with_attribute("index", 2)).await;
        assert_eq!(db.inactivity_floor_loc(), floor_a);

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

    /// Merkleize returns `FloorRegressed` for a chained batch whose floor is below its parent's
    /// floor, even though that floor is still at or above the database's floor.
    #[boxed]
    pub(crate) async fn run_chained_floor_regression_rejected<F, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        F: Family,
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // parent: 1 append + commit at loc 2 with floor=2 (the parent's commit_loc).
        let parent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, Location::new(2))
            .await
            .unwrap();

        // child: 1 append + commit at loc 4 with floor=1 (regressed from parent's floor=2).
        let Err(err) = parent
            .new_batch::<H>()
            .append(V::Value::make(2))
            .merkleize(&db, None, Location::new(1))
            .await
        else {
            panic!("merkleize must reject the invalid floor");
        };
        assert!(
            matches!(err, Error::FloorRegressed(new, prev) if *new == 1 && *prev == 2),
            "unexpected error: {err:?}"
        );

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

    /// Merkleize returns `FloorBeyondSize` for a chained batch whose floor exceeds its own commit
    /// location. The commit location includes the parent's operations.
    #[boxed]
    pub(crate) async fn run_chained_floor_beyond_commit_rejected<F, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        F: Family,
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // parent: 1 append + commit at loc 2 with floor=2.
        let parent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, Location::new(2))
            .await
            .unwrap();

        // child: 1 append + commit at loc 4, declaring floor=5 (one past its commit). The error
        // names commit location 4, so the bound counts the parent's operations.
        let Err(err) = parent
            .new_batch::<H>()
            .append(V::Value::make(2))
            .merkleize(&db, None, Location::new(5))
            .await
        else {
            panic!("merkleize must reject the invalid floor");
        };
        assert!(
            matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 5 && *commit == 4),
            "unexpected error: {err:?}"
        );
    }

    /// After committing with `floor = commit_loc` and pruning down to it, exactly one operation,
    /// the commit itself, is at or above the floor. This is the smallest nonempty such set the
    /// per-commit bound allows. The DB must remain fully usable: the commit is readable, the root
    /// is preserved, reopen recovers `inactivity_floor_loc` from the sole remaining op, and a
    /// follow-on batch applies cleanly on top.
    #[boxed]
    pub(crate) async fn run_single_commit_live_set<F, V, C, H, S: Strategy>(
        context: deterministic::Context,
        db: TestKeyless<F, V, C, H, S>,
        reopen: Reopen<TestKeyless<F, V, C, H, S>>,
    ) where
        F: Family,
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // Initial commit is at loc 0. 3 appends + commit -> commit lands at loc 4.
        // Declare floor = 4 (= commit_loc), the tight maximum.
        let metadata = V::Value::make(42);
        let commit_loc = Location::<F>::new(4);
        let merkleized = db
            .new_batch()
            .append(V::Value::make(1))
            .append(V::Value::make(2))
            .append(V::Value::make(3))
            .merkleize(&db, Some(metadata.clone()), commit_loc)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        assert_eq!(db.bounds().end - 1, commit_loc);
        assert_eq!(db.inactivity_floor_loc(), commit_loc);
        let root_after_commit = db.root();

        // Prune at the floor, the maximum prune allowed under the invariant. Pruning is
        // blob-aligned, so `bounds.start` may not physically advance all the way to `commit_loc`.
        // What matters semantically is that the floor has authorized pruning of everything below
        // the commit and that any further prune is rejected.
        let db = db.prune(commit_loc).await.unwrap();
        let bounds = db.bounds();
        assert!(
            bounds.start <= commit_loc,
            "prune must not advance bounds.start past the floor"
        );
        assert_eq!(bounds.end, commit_loc + 1);

        // The commit op remains readable; its metadata is intact.
        assert_eq!(db.get(commit_loc).await.unwrap(), Some(metadata.clone()));
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));
        assert_eq!(db.bounds().end - 1, commit_loc);
        assert_eq!(db.inactivity_floor_loc(), commit_loc);
        // Prune does not affect the root (documented invariant on `prune`).
        assert_eq!(db.root(), root_after_commit);

        // Persist the prune, then verify pruning one past the floor is rejected. The
        // floor is the hard ceiling.
        let db = db.sync().await.unwrap();
        let Err(err) = db.prune(commit_loc + 1).await else {
            panic!("expected prune to fail");
        };
        assert!(matches!(err, Error::PruneBeyondMinRequired(p, f)
                if *p == *commit_loc + 1 && *f == *commit_loc));

        // Reopening restores the inactivity floor from the last commit.
        let db = reopen(context.child("reopened")).await;
        let reopened_bounds = db.bounds();
        assert_eq!(reopened_bounds.end, commit_loc + 1);
        assert_eq!(db.bounds().end - 1, commit_loc);
        assert_eq!(db.inactivity_floor_loc(), commit_loc);
        assert_eq!(db.root(), root_after_commit);
        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));

        // A follow-on batch applies on top. Monotonicity requires the new floor to be at
        // least `commit_loc` (= 4); advancing to the new tight max (= 7) exercises the
        // ancestor-to-tip floor transition from a single-commit starting point.
        let next_commit_loc = Location::<F>::new(7);
        let v5 = V::Value::make(5);
        let v6 = V::Value::make(6);
        let merkleized = db
            .new_batch()
            .append(v5.clone())
            .append(v6.clone())
            .merkleize(&db, None, next_commit_loc)
            .await
            .unwrap();
        let db = db.apply_batch(merkleized).await.unwrap();
        let db = db.commit().await.unwrap();
        assert_eq!(db.bounds().end - 1, next_commit_loc);
        assert_eq!(db.inactivity_floor_loc(), next_commit_loc);

        // New appends are readable, and the journal still holds the original commit op (it was
        // not pruned again), so reading it still returns its metadata.
        assert_eq!(db.get(Location::new(5)).await.unwrap(), Some(v5));
        assert_eq!(db.get(Location::new(6)).await.unwrap(), Some(v6));
        assert_eq!(db.get(commit_loc).await.unwrap(), Some(metadata));

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

    /// A multi-level chain with strictly-monotonic, within-bounds floors applies cleanly.
    #[boxed]
    pub(crate) async fn run_chained_apply_with_valid_floors_succeeds<F, V, C, H, S: Strategy>(
        db: TestKeyless<F, V, C, H, S>,
    ) where
        F: Family,
        V: ValueEncoding<Value: TestValue>,
        C: Mutable<Item = Operation<F, V>>,
        H: Hasher,
        Operation<F, V>: EncodeShared,
    {
        // parent:     1 append, commit at loc 2, floor = 2.
        // child:      1 append, commit at loc 4, floor = 3.
        // grandchild: 1 append, commit at loc 6, floor = 5.
        let parent = db
            .new_batch()
            .append(V::Value::make(1))
            .merkleize(&db, None, Location::new(2))
            .await
            .unwrap();
        let child = parent
            .new_batch::<H>()
            .append(V::Value::make(2))
            .merkleize(&db, None, Location::new(3))
            .await
            .unwrap();
        let grandchild = child
            .new_batch::<H>()
            .append(V::Value::make(3))
            .merkleize(&db, None, Location::new(5))
            .await
            .unwrap();

        let db = db.apply_batch(grandchild).await.unwrap();

        // Grandchild's commit is the last op, and the tip's floor is the database's floor.
        assert_eq!(db.bounds().end - 1, Location::new(6));
        assert_eq!(db.inactivity_floor_loc(), Location::new(5));

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