//! A mutable key-value database that supports variable-sized values, but without authentication.
//!
//! # Example
//!
//! ```rust
//! use commonware_storage::{
//!     journal::contiguous::variable::Config as JournalConfig,
//!     qmdb::store::db::{Config, Db},
//!     translator::TwoCap,
//! };
//! use commonware_utils::{NZUsize, NZU16, NZU64};
//! use commonware_cryptography::{blake3::Digest, Digest as _};
//! use commonware_math::algebra::Random;
//! use commonware_runtime::{
//!     buffer::paged::CacheRef, deterministic::Runner, Metrics, Runner as _, Supervisor as _,
//! };
//!
//! use std::num::NonZeroU16;
//! const PAGE_SIZE: NonZeroU16 = NZU16!(8192);
//! const PAGE_CACHE_SIZE: usize = 100;
//!
//! let executor = Runner::default();
//! executor.start(|mut ctx| async move {
//!     let config = Config {
//!         log: JournalConfig {
//!             partition: "test-partition".into(),
//!             write_buffer: NZUsize!(64 * 1024),
//!             replay_buffer: NZUsize!(64 * 1024),
//!             compression: None,
//!             codec_config: ((), ()),
//!             items_per_section: NZU64!(4),
//!             page_cache: CacheRef::from_pooler(&ctx, PAGE_SIZE, NZUsize!(PAGE_CACHE_SIZE)),
//!         },
//!         translator: TwoCap,
//!         init_cache_size: Some(NZUsize!(1 << 16)),
//!         init_buffer: NZUsize!(1 << 21),
//!     };
//!     let db =
//!         Db::<_, Digest, Digest, TwoCap>::init(ctx.child("store"), config)
//!             .await
//!             .unwrap();
//!
//!     // Insert a key-value pair
//!     let k = Digest::random(&mut ctx);
//!     let v = Digest::random(&mut ctx);
//!     let metadata = Some(Digest::random(&mut ctx));
//!     let batch = db.new_batch().update(k, v).finalize(metadata);
//!     let (db, _) = db.apply_batch(batch).await.unwrap();
//!     let db = db.commit().await.unwrap();
//!
//!     // Fetch the value
//!     let fetched_value = db.get(&k).await.unwrap();
//!     assert_eq!(fetched_value.unwrap(), v);
//!
//!     // Delete the key's value
//!     let batch = db.new_batch().delete(k).finalize(None);
//!     let (db, _) = db.apply_batch(batch).await.unwrap();
//!     let db = db.commit().await.unwrap();
//!
//!     // Fetch the value
//!     let fetched_value = db.get(&k).await.unwrap();
//!     assert!(fetched_value.is_none());
//!
//!     // Destroy the store
//!     db.destroy().await.unwrap();
//! });
//! ```
//!
//! ```ignore
//! // Apply a batch and commit it, then build a child batch from the newly published state
//! // and apply it. Each mutation takes the database and returns it, so committing and
//! // building run in sequence on the threaded handle.
//! let batch = db.new_batch().update(key_a, value_a).finalize(None);
//! let (db, _) = db.apply_batch(batch).await?;
//! let db = db.commit().await?;
//!
//! let child = db.new_batch().update(key_b, value_b).finalize(None);
//! let (db, _) = db.apply_batch(child).await?;
//! let db = db.commit().await?;
//! ```

use crate::{
    Context,
    index::{Unordered as _, unordered::Index},
    journal::contiguous::{
        Contiguous, Mutable as _,
        variable::{Config as JournalConfig, Journal},
    },
    merkle::mmr::Location,
    qmdb::{
        FloorHelper,
        any::{
            VariableValue,
            unordered::{Update, variable::Operation},
        },
        build_snapshot_from_log, delete_key,
        operation::{Committable as _, Floored as _, Key},
        update_key,
    },
    translator::Translator,
};
use commonware_codec::{CodecShared, Read};
use commonware_macros::boxed;
use commonware_runtime::Handle;
use core::{num::NonZeroUsize, ops::Range};
use std::collections::BTreeMap;
use tracing::{debug, warn};

type Error = crate::qmdb::Error<crate::mmr::Family>;

/// Configuration for initializing a [Db].
#[derive(Clone)]
pub struct Config<T: Translator, C> {
    /// Configuration for the variable-size operations log journal.
    pub log: JournalConfig<C>,

    /// The [Translator] used by the [Index].
    pub translator: T,

    /// Capacity (in entries) of the `(location -> key)` cache used during init to resolve snapshot
    /// collisions without re-reading the log; `None` disables it.
    pub init_cache_size: Option<NonZeroUsize>,

    /// Size (in bytes) of the read buffer used to replay the log during init.
    pub init_buffer: NonZeroUsize,
}

/// A finalized batch of writes and deletes ready to be applied to the store.
pub struct Changeset<K: Key, V: CodecShared + Clone> {
    diff: BTreeMap<K, Option<V>>,
    metadata: Option<V>,
}

impl<K: Key, V: CodecShared + Clone> Changeset<K, V> {
    fn into_parts(self) -> (BTreeMap<K, Option<V>>, Option<V>) {
        (self.diff, self.metadata)
    }
}

impl<K: Key, V: CodecShared + Clone> FromIterator<(K, Option<V>)> for Changeset<K, V> {
    fn from_iter<TIter: IntoIterator<Item = (K, Option<V>)>>(iter: TIter) -> Self {
        Self {
            diff: iter.into_iter().collect(),
            metadata: None,
        }
    }
}

impl<K: Key, V: CodecShared + Clone, const N: usize> From<[(K, Option<V>); N]> for Changeset<K, V> {
    fn from(items: [(K, Option<V>); N]) -> Self {
        items.into_iter().collect()
    }
}

/// A mutable batch of writes and deletes staged against the current store state.
pub struct Batch<'a, E, K, V, T>
where
    E: Context,
    K: Key,
    V: VariableValue,
    T: Translator,
{
    db: &'a Db<E, K, V, T>,
    diff: BTreeMap<K, Option<V>>,
}

impl<'a, E, K, V, T> Batch<'a, E, K, V, T>
where
    E: Context,
    K: Key,
    V: VariableValue,
    T: Translator,
{
    const fn new(db: &'a Db<E, K, V, T>) -> Self {
        Self {
            db,
            diff: BTreeMap::new(),
        }
    }

    /// Finalize the batch into a changeset that can be applied to the store.
    pub fn finalize(self, metadata: Option<V>) -> Changeset<K, V> {
        Changeset {
            diff: self.diff,
            metadata,
        }
    }

    /// Get the value of `key` in the batch, or the value in the store if it has
    /// not been modified by the batch.
    pub async fn get(&self, key: &K) -> Result<Option<V>, Error> {
        if let Some(value) = self.diff.get(key) {
            return Ok(value.clone());
        }
        self.db.get(key).await
    }

    /// Update the value of `key` in the batch.
    pub fn update(mut self, key: K, value: V) -> Self {
        self.diff.insert(key, Some(value));
        self
    }

    /// Delete the value of `key` in the batch.
    pub fn delete(mut self, key: K) -> Self {
        self.diff.insert(key, None);
        self
    }
}

/// An unauthenticated key-value database based off of an append-only [Journal] of operations.
pub struct Db<E, K, V, T>
where
    E: Context,
    K: Key,
    V: VariableValue,
    T: Translator,
{
    /// A log of all [Operation]s that have been applied to the store.
    ///
    /// # Invariants
    ///
    /// - There is always at least one commit operation in the log.
    /// - The log is never pruned beyond the inactivity floor.
    log: Journal<E, Operation<crate::mmr::Family, K, V>>,

    /// A snapshot of all currently active operations in the form of a map from each key to the
    /// location containing its most recent update.
    ///
    /// # Invariant
    ///
    /// Only references operations of type [Operation::Update].
    snapshot: Index<T, Location>,

    /// The number of active keys in the store.
    active_keys: usize,

    /// A location before which all operations are "inactive" (that is, operations before this point
    /// are over keys that have been updated by some operation at or after this point).
    pub inactivity_floor_loc: Location,

    /// The location of the last commit operation.
    pub last_commit_loc: Location,

    /// The number of _steps_ to raise the inactivity floor. Each step involves moving exactly one
    /// active operation to tip.
    pub steps: u64,
}

impl<E, K, V, T> std::fmt::Debug for Db<E, K, V, T>
where
    E: Context,
    K: Key,
    V: VariableValue,
    T: Translator,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Db")
            .field("bounds", &self.bounds())
            .field("inactivity_floor_loc", &self.inactivity_floor_loc())
            .finish_non_exhaustive()
    }
}

impl<E, K, V, T> Db<E, K, V, T>
where
    E: Context,
    K: Key,
    V: VariableValue,
    T: Translator,
{
    /// Get the value of `key` in the db, or None if it has no value.
    pub async fn get(&self, key: &K) -> Result<Option<V>, Error> {
        for &loc in self.snapshot.get(key) {
            let Operation::Update(Update(k, v)) = self.get_op(loc).await? else {
                unreachable!("location ({loc}) does not reference update operation");
            };

            if &k == key {
                return Ok(Some(v));
            }
        }

        Ok(None)
    }

    /// Returns a new empty batch of changes.
    pub const fn new_batch(&self) -> Batch<'_, E, K, V, T> {
        Batch::new(self)
    }

    /// Whether the db currently has no active keys.
    pub const fn is_empty(&self) -> bool {
        self.active_keys == 0
    }

    /// Gets a [Operation] from the log at the given location. Returns [Error::OperationPruned]
    /// if the location precedes the oldest retained location. The location is otherwise assumed
    /// valid.
    async fn get_op(&self, loc: Location) -> Result<Operation<crate::mmr::Family, K, V>, Error> {
        assert!(*loc < self.log.bounds().end);
        self.log.read(*loc).await.map_err(|e| match e {
            crate::journal::Error::ItemPruned(_) => Error::OperationPruned(loc),
            e => Error::Journal(e),
        })
    }

    /// Return [start, end) where `start` and `end - 1` are the Locations of the oldest and newest
    /// retained operations respectively.
    pub fn bounds(&self) -> std::ops::Range<Location> {
        let bounds = self.log.bounds();
        Location::new(bounds.start)..Location::new(bounds.end)
    }

    /// Return the Location of the next operation appended to this db.
    pub fn size(&self) -> Location {
        Location::new(self.log.size())
    }

    /// Return the inactivity floor location. This is the location before which all operations are
    /// known to be inactive. Operations before this point can be safely pruned.
    pub const fn inactivity_floor_loc(&self) -> Location {
        self.inactivity_floor_loc
    }

    /// Get the metadata associated with the last commit.
    pub async fn get_metadata(&self) -> Result<Option<V>, Error> {
        let Operation::CommitFloor(metadata, _) = self.log.read(*self.last_commit_loc).await?
        else {
            unreachable!("last commit should be a commit floor operation");
        };

        Ok(metadata)
    }

    /// Prune historical operations prior to `prune_loc`. This does not affect the db's root
    /// or current snapshot.
    ///
    /// `prune` requires no prior commit. After a crash, the database remains recoverable;
    /// uncommitted operations are not guaranteed to survive.
    #[boxed]
    pub async fn prune(mut self, prune_loc: Location) -> Result<Self, Error> {
        if prune_loc > self.inactivity_floor_loc {
            return Err(Error::PruneBeyondMinRequired(
                prune_loc,
                self.inactivity_floor_loc,
            ));
        }

        // The floor justifying the boundary may exist only in buffered operations (it
        // advances before its batch is durable), and pruning does not guarantee buffered
        // appends are durable. Commit so the justification survives the prune.
        self.log = self.log.commit().await?;

        // Prune the log. The log will prune at section boundaries, so the actual oldest retained
        // location may be less than requested.
        let pruned;
        (self.log, pruned) = self.log.prune(*prune_loc).await?;
        if !pruned {
            return Ok(self);
        }

        let bounds = self.log.bounds();
        let log_size = Location::new(bounds.end);
        let oldest_retained_loc = Location::new(bounds.start);
        debug!(
            ?log_size,
            ?oldest_retained_loc,
            ?prune_loc,
            "pruned inactive ops"
        );

        Ok(self)
    }

    /// Initializes a new [Db] with the given configuration.
    pub async fn init(
        context: E,
        cfg: Config<T, <Operation<crate::mmr::Family, K, V> as Read>::Cfg>,
    ) -> Result<Self, Error> {
        let log =
            Journal::<E, Operation<crate::mmr::Family, K, V>>::init(context.child("log"), cfg.log)
                .await?;

        // Rewind log to remove uncommitted operations.
        let (mut log, size) = log.rewind_to(|op| op.is_commit()).await?;
        if size == 0 {
            warn!("Log is empty, initializing new db");
            (log, _) = log
                .append(&Operation::CommitFloor(None, Location::new(0)))
                .await?;
        }

        // Sync the log to avoid having to repeat any recovery that may have been performed on next
        // startup.
        let log = log.sync().await?;

        let last_commit_loc =
            Location::new(log.size().checked_sub(1).expect("commit should exist"));

        // Build the snapshot.
        let cache_size = cfg.init_cache_size;
        let init_buffer = cfg.init_buffer;
        let mut snapshot = Index::new(context.child("snapshot"), cfg.translator);
        let (inactivity_floor_loc, active_keys) = {
            let op = log.read(*last_commit_loc).await?;
            let inactivity_floor_loc = op.has_floor().expect("last op should be a commit");
            if inactivity_floor_loc > last_commit_loc {
                return Err(crate::qmdb::Error::DataCorrupted(
                    "inactivity floor exceeds last commit",
                ));
            }
            let active_keys = build_snapshot_from_log(
                inactivity_floor_loc,
                &log,
                &mut snapshot,
                init_buffer,
                cache_size,
                |_, _| {},
            )
            .await?;
            (inactivity_floor_loc, active_keys)
        };

        Ok(Self {
            log,
            snapshot,
            active_keys,
            inactivity_floor_loc,
            last_commit_loc,
            steps: 0,
        })
    }

    /// 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.
    #[boxed]
    pub async fn sync(mut self) -> Result<Self, Error> {
        self.log = self.log.sync().await?;
        Ok(self)
    }

    /// Destroy the db, removing all data from disk.
    #[boxed]
    pub async fn destroy(self) -> Result<(), Error> {
        self.log.destroy().await.map_err(Into::into)
    }

    /// Applies a finalized batch to the in-memory database state and appends its operations to the
    /// journal, returning the range of written locations.
    ///
    /// This publishes the batch to the in-memory database state and appends it to the journal.
    /// Call [`Db::commit`] or [`Db::sync`], or await the handle returned by [`Db::start_sync`], to
    /// make the applied state durable.
    #[boxed]
    pub async fn apply_batch(
        mut self,
        batch: Changeset<K, V>,
    ) -> Result<(Self, Range<Location>), Error> {
        let start_loc = self.last_commit_loc + 1;
        let (diff, metadata) = batch.into_parts();

        for (key, value) in diff {
            if let Some(value) = value {
                let updated = {
                    let new_loc = self.log.bounds().end;
                    update_key::<crate::mmr::Family, _, _>(
                        &mut self.snapshot,
                        &self.log,
                        &key,
                        Location::new(new_loc),
                        None,
                    )
                    .await?
                };
                if updated.is_some() {
                    self.steps += 1;
                } else {
                    self.active_keys += 1;
                }
                (self.log, _) = self
                    .log
                    .append(&Operation::Update(Update(key, value)))
                    .await?;
            } else {
                let deleted = delete_key::<crate::mmr::Family, _, _>(
                    &mut self.snapshot,
                    &self.log,
                    &key,
                    None,
                )
                .await?;
                if deleted.is_some() {
                    (self.log, _) = self.log.append(&Operation::Delete(key)).await?;
                    self.steps += 1;
                    self.active_keys -= 1;
                }
            }
        }

        // Raise the inactivity floor by `self.steps` steps, plus 1 to account for the previous
        // commit becoming inactive.
        if self.is_empty() {
            self.inactivity_floor_loc = self.size();
            debug!(tip = ?self.inactivity_floor_loc, "db is empty, raising floor to tip");
        } else {
            let steps_to_take = self.steps + 1;
            let mut helper = FloorHelper {
                snapshot: &mut self.snapshot,
                log: self.log,
            };
            let mut inactivity_floor_loc = self.inactivity_floor_loc;
            for _ in 0..steps_to_take {
                (helper, inactivity_floor_loc) = helper.raise_floor(inactivity_floor_loc).await?;
            }
            self.log = helper.log;
            self.inactivity_floor_loc = inactivity_floor_loc;
        }

        // Append the commit operation with the new inactivity floor.
        let commit_loc;
        (self.log, commit_loc) = self
            .log
            .append(&Operation::CommitFloor(metadata, self.inactivity_floor_loc))
            .await?;
        self.last_commit_loc = Location::new(commit_loc);

        self.steps = 0;

        let end_loc = self.size();
        Ok((self, start_loc..end_loc))
    }

    /// Begin durably persisting the journal state published by prior [`Db::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 offsets or recovery-watermark sync is not
    /// observed by [Self::commit] and resurfaces on the next [Self::sync].
    #[boxed]
    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error> {
        let handle;
        (self.log, handle) = self.log.start_sync().await?;
        Ok((self, handle))
    }

    /// Durably commit the journal state published by prior [`Db::apply_batch`] calls.
    #[boxed]
    pub async fn commit(mut self) -> Result<Self, Error> {
        self.log = self.log.commit().await?;
        Ok(self)
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::translator::TwoCap;
    use commonware_cryptography::{
        Hasher as _,
        blake3::{Blake3, Digest},
    };
    use commonware_macros::test_traced;
    use commonware_math::algebra::Random;
    use commonware_runtime::{
        Runner, Spawner as _, Supervisor as _,
        buffer::paged::CacheRef,
        deterministic,
        mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs},
        reschedule,
    };
    use commonware_utils::{NZU16, NZU64, NZUsize};
    use core::future::Future;
    use futures::FutureExt as _;
    use std::num::{NonZeroU16, NonZeroUsize};

    const PAGE_SIZE: NonZeroU16 = NZU16!(77);
    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9);

    /// The type of the store used in tests.
    type TestStore = Db<deterministic::Context, Digest, Vec<u8>, TwoCap>;

    async fn create_test_store(context: deterministic::Context) -> TestStore {
        let cfg = Config {
            log: JournalConfig {
                partition: "journal".into(),
                write_buffer: NZUsize!(64 * 1024),
                replay_buffer: NZUsize!(64 * 1024),
                compression: None,
                codec_config: ((), ((0..=10000).into(), ())),
                items_per_section: NZU64!(7),
                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
            },
            translator: TwoCap,
            init_cache_size: Some(NZUsize!(1024)),
            init_buffer: NZUsize!(1 << 21),
        };
        TestStore::init(context, cfg).await.unwrap()
    }

    async fn apply_entries(
        db: TestStore,
        iter: impl IntoIterator<Item = (Digest, Option<Vec<u8>>)> + Send,
    ) -> (TestStore, Range<Location>) {
        db.apply_batch(iter.into_iter().collect()).await.unwrap()
    }

    /// A store over a delayed-sync storage backend.
    type DelayedStore = Db<DelayedSyncContext<deterministic::Context>, Digest, Vec<u8>, TwoCap>;

    /// Open a [DelayedStore] whose blob syncs park on `pending`.
    ///
    /// Init durably persists the recovered database, so while syncs park the returned future
    /// must be driven with [drive_pending_syncs] (or the mock unblocked first). The journal
    /// uses large pages and sections: an apply that fills the write buffer or rolls the blob
    /// over waits for the in-flight sync, so mid-sync applies must stay clear of both.
    fn open_delayed_store(
        context: &deterministic::Context,
        label: &'static str,
        suffix: &str,
        pending: &PendingSyncs,
    ) -> impl Future<Output = Result<DelayedStore, Error>> {
        let cfg = Config {
            log: JournalConfig {
                partition: format!("journal-{suffix}"),
                write_buffer: NZUsize!(64 * 1024),
                replay_buffer: NZUsize!(64 * 1024),
                compression: None,
                codec_config: ((), ((0..=10000).into(), ())),
                items_per_section: NZU64!(1000),
                page_cache: CacheRef::from_pooler(context, NZU16!(1024), NZUsize!(8)),
            },
            translator: TwoCap,
            init_cache_size: Some(NZUsize!(1024)),
            init_buffer: NZUsize!(1 << 21),
        };
        DelayedStore::init(
            DelayedSyncContext {
                inner: context.child(label),
                pending: pending.clone(),
            },
            cfg,
        )
    }

    /// Apply a single-key batch writing `key -> value`.
    async fn apply_write(db: DelayedStore, key: Digest, value: Vec<u8>) -> DelayedStore {
        let (db, _) = db.apply_batch([(key, Some(value))].into()).await.unwrap();
        db
    }

    /// A sync handle must not block database use while the backend sync is pending.
    #[test_traced]
    fn test_store_start_sync_overlaps_work() {
        deterministic::Runner::default().start(|ctx| async move {
            let pending = PendingSyncs::default();
            let open = open_delayed_store(&ctx, "delayed", "start-sync-overlap", &pending);
            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
            let key0 = Blake3::hash(&[&0u64.to_be_bytes()]);
            let value0 = vec![1u8; 8];
            db = apply_write(db, key0, value0.clone()).await;

            let starts_before = pending.starts();
            let entered_before = pending.entered();
            let completions_before = pending.completions();
            let handle;
            (db, handle) = db.start_sync().await.unwrap();
            assert!(pending.starts() > starts_before);
            assert_eq!(pending.completions(), completions_before);

            // Observe the sync while the database keeps working.
            let waiter = ctx
                .child("await_sync")
                .spawn(|_| async move { handle.await.unwrap() });
            while pending.entered() == entered_before {
                reschedule().await;
            }

            // Reads and applies complete before the sync does.
            assert_eq!(db.get(&key0).await.unwrap(), Some(value0));
            let key1 = Blake3::hash(&[&1u64.to_be_bytes()]);
            let value1 = vec![2u8; 8];
            db = apply_write(db, key1, value1.clone()).await;
            assert_eq!(
                pending.completions(),
                completions_before,
                "the database made progress while the sync was still in flight"
            );

            pending.unblock();
            waiter.await.unwrap();

            // The mid-sync batch is durable after the next start_sync completes.
            let handle;
            (db, handle) = db.start_sync().await.unwrap();
            handle.await.unwrap();
            let size = db.size();
            drop(db);

            let db = open_delayed_store(&ctx, "reopen", "start-sync-overlap", &pending)
                .await
                .unwrap();
            assert_eq!(db.size(), size);
            assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
            db.destroy().await.unwrap();
        });
    }

    /// A sync begun by `start_sync` that fails in flight surfaces the error through both the
    /// returned handle and the next durability operation.
    #[test_traced]
    fn test_store_start_sync_failure_propagates() {
        deterministic::Runner::default().start(|ctx| async move {
            // Pass syncs through so opening the database doesn't park.
            let pending = PendingSyncs::default();
            pending.unblock();
            let mut db = open_delayed_store(&ctx, "delayed", "start-sync-fail", &pending)
                .await
                .unwrap();
            db = apply_write(db, Blake3::hash(&[&0u64.to_be_bytes()]), vec![1u8; 8]).await;

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

            let handle;
            (db, handle) = db.start_sync().await.unwrap();
            assert!(
                handle.await.is_err(),
                "the sync handle surfaces the failure"
            );
            let starts_before = pending.starts();
            // A failed mutable method consumes the database per the failures-are-fatal contract.
            assert!(
                db.commit().await.is_err(),
                "the next durability op surfaces the failed in-flight sync"
            );
            assert_eq!(
                pending.starts(),
                starts_before,
                "the surfaced error is the retained failure, not a fresh sync's"
            );
        });
    }

    /// State persisted via an awaited start_sync handle is recovered on reopen.
    #[test_traced]
    fn test_store_start_sync_recovery() {
        deterministic::Runner::default().start(|ctx| async move {
            let pending = PendingSyncs::default();
            pending.unblock();
            let mut db = open_delayed_store(&ctx, "delayed", "start-sync-recovery", &pending)
                .await
                .unwrap();
            let key = Blake3::hash(&[&0u64.to_be_bytes()]);
            let value = vec![1u8; 8];
            db = apply_write(db, key, value.clone()).await;

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

            let db = open_delayed_store(&ctx, "reopen", "start-sync-recovery", &pending)
                .await
                .unwrap();
            assert_eq!(db.size(), size);
            assert_eq!(db.get(&key).await.unwrap(), Some(value));
            db.destroy().await.unwrap();
        });
    }

    /// Pruning drains the in-flight sync before mutating storage.
    #[test_traced]
    fn test_store_start_sync_prune_waits() {
        deterministic::Runner::default().start(|ctx| async move {
            let pending = PendingSyncs::default();
            let open = open_delayed_store(&ctx, "delayed", "start-sync-prune", &pending);
            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
            // Two batches so floor-raising steps leave a non-trivial prune target.
            db = apply_write(db, Blake3::hash(&[&0u64.to_be_bytes()]), vec![1u8; 8]).await;
            db = apply_write(db, Blake3::hash(&[&1u64.to_be_bytes()]), vec![2u8; 8]).await;

            let starts_before = pending.starts();
            let handle;
            (db, handle) = db.start_sync().await.unwrap();
            assert!(pending.starts() > starts_before);

            let floor = db.inactivity_floor_loc();
            assert!(*floor > 0);
            let db = {
                let mut prune = std::pin::pin!(db.prune(floor));
                assert!(
                    prune.as_mut().now_or_never().is_none(),
                    "prune proceeded while the started sync was pending"
                );
                pending.unblock();
                prune.await.unwrap()
            };
            handle.await.unwrap();
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("DEBUG")]
    pub fn test_store_construct_empty() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let db = create_test_store(context.child("store").with_attribute("index", 0)).await;
            assert_eq!(db.bounds().end, 1);
            assert_eq!(db.log.bounds().start, 0);
            assert_eq!(db.inactivity_floor_loc(), 0);
            assert!(db.get_metadata().await.unwrap().is_none());
            let floor = db.inactivity_floor_loc();
            let db = db.prune(floor).await.unwrap();
            assert!(matches!(
                db.prune(Location::new(1)).await,
                Err(Error::PruneBeyondMinRequired(_, _))
            ));

            let db = create_test_store(context.child("store").with_attribute("index", 3)).await;

            // Make sure closing/reopening gets us back to the same state, even after adding an uncommitted op.
            let d1 = Digest::random(&mut context);
            let v1 = vec![1, 2, 3];
            let (db, _) = apply_entries(db, [(d1, Some(v1))]).await;
            drop(db);

            let db = create_test_store(context.child("store").with_attribute("index", 1)).await;
            assert_eq!(db.bounds().end, 1);

            // Test calling commit on an empty db which should make it (durably) non-empty.
            let metadata = vec![1, 2, 3];
            let batch = db.new_batch().finalize(Some(metadata.clone()));
            let (db, range) = db.apply_batch(batch).await.unwrap();
            assert_eq!(range.start, 1);
            assert_eq!(range.end, 2);
            let db = db.commit().await.unwrap();
            assert_eq!(db.bounds().end, 2);
            let floor = db.inactivity_floor_loc();
            let db = db.prune(floor).await.unwrap();
            assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));

            let db = create_test_store(context.child("store").with_attribute("index", 2)).await;
            assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));

            // Confirm the inactivity floor doesn't fall endlessly behind with multiple commits on a
            // non-empty db.
            let (db, _) =
                apply_entries(db, [(Digest::random(&mut context), Some(vec![1, 2, 3]))]).await;
            let mut db = db.commit().await.unwrap();
            for _ in 1..100 {
                let merkleized = db.new_batch().finalize(None);
                (db, _) = db.apply_batch(merkleized).await.unwrap();
                db = db.commit().await.unwrap();
                // Distance should equal 3 after the second commit, with inactivity_floor
                // referencing the previous commit operation.
                assert!(db.bounds().end - db.inactivity_floor_loc <= 3);
                assert!(db.get_metadata().await.unwrap().is_none());
            }

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

    #[test_traced("DEBUG")]
    fn test_store_construct_basic() {
        let executor = deterministic::Runner::default();

        executor.start(|mut ctx| async move {
            let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await;

            // Ensure the store is empty
            assert_eq!(db.bounds().end, 1);
            assert_eq!(db.inactivity_floor_loc, 0);

            let key = Digest::random(&mut ctx);
            let value = vec![2, 3, 4, 5];

            // Attempt to get a key that does not exist
            let result = db.get(&key).await;
            assert!(result.unwrap().is_none());

            // Insert a key-value pair. apply_batch writes the Update, a floor-raise move, and a
            // CommitFloor: 3 new ops on top of the initial commit.
            let (db, _) = apply_entries(db, [(key, Some(value.clone()))]).await;

            assert_eq!(*db.bounds().end, 4);
            assert_eq!(*db.inactivity_floor_loc, 2);

            // Fetch the value
            let fetched_value = db.get(&key).await.unwrap();
            assert_eq!(fetched_value.unwrap(), value);

            // Simulate commit failure: drop without commit.
            drop(db);

            // Re-open the store
            let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await;

            // Ensure the re-opened store removed the uncommitted operations
            assert_eq!(*db.bounds().end, 1);
            assert_eq!(*db.inactivity_floor_loc, 0);
            assert!(db.get_metadata().await.unwrap().is_none());

            // Insert a key-value pair and persist with metadata.
            let metadata = vec![99, 100];
            let batch = db
                .new_batch()
                .update(key, value.clone())
                .finalize(Some(metadata.clone()));
            let (db, range) = db.apply_batch(batch).await.unwrap();
            assert_eq!(*range.start, 1);
            assert_eq!(*range.end, 4);
            let db = db.commit().await.unwrap();
            assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));

            assert_eq!(*db.bounds().end, 4);
            assert_eq!(*db.inactivity_floor_loc, 2);

            // Re-open the store
            let db = create_test_store(ctx.child("store").with_attribute("index", 2)).await;

            // Ensure the re-opened store retained the committed operations
            assert_eq!(*db.bounds().end, 4);
            assert_eq!(*db.inactivity_floor_loc, 2);

            // Fetch the value, ensuring it is still present
            let fetched_value = db.get(&key).await.unwrap();
            assert_eq!(fetched_value.unwrap(), value);

            // Insert two new k/v pairs to force pruning of the first section.
            let (k1, v1) = (Digest::random(&mut ctx), vec![2, 3, 4, 5, 6]);
            let (k2, v2) = (Digest::random(&mut ctx), vec![6, 7, 8]);
            let (db, _) = apply_entries(db, [(k1, Some(v1.clone()))]).await;
            let (db, _) = apply_entries(db, [(k2, Some(v2.clone()))]).await;

            assert_eq!(*db.bounds().end, 10);
            assert_eq!(*db.inactivity_floor_loc, 5);

            // Each apply_entries writes a CommitFloor with None metadata, replacing
            // the previously committed metadata.
            assert_eq!(db.get_metadata().await.unwrap(), None);

            let db = db.commit().await.unwrap();
            assert_eq!(db.get_metadata().await.unwrap(), None);

            // commit() is just an fsync now, so bounds and floor are unchanged.
            assert_eq!(*db.bounds().end, 10);
            assert_eq!(*db.inactivity_floor_loc, 5);

            // Ensure all keys can be accessed, despite the first section being pruned.
            assert_eq!(db.get(&key).await.unwrap().unwrap(), value);
            assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
            assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);

            // Update existing key with modified value.
            let mut v1_updated = db.get(&k1).await.unwrap().unwrap();
            v1_updated.push(7);
            let (db, _) = apply_entries(db, [(k1, Some(v1_updated))]).await;
            let db = db.commit().await.unwrap();
            assert_eq!(db.get(&k1).await.unwrap().unwrap(), vec![2, 3, 4, 5, 6, 7]);

            // Create new key.
            let k3 = Digest::random(&mut ctx);
            let (db, _) = apply_entries(db, [(k3, Some(vec![8]))]).await;
            let db = db.commit().await.unwrap();
            assert_eq!(db.get(&k3).await.unwrap().unwrap(), vec![8]);

            // Destroy the store
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("DEBUG")]
    fn test_store_log_replay() {
        let executor = deterministic::Runner::default();

        executor.start(|mut ctx| async move {
            let mut db = create_test_store(ctx.child("store").with_attribute("index", 0)).await;

            // Update the same key many times.
            const UPDATES: u64 = 100;
            let k = Digest::random(&mut ctx);
            for _ in 0..UPDATES {
                let v = vec![1, 2, 3, 4, 5];
                (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await;
            }

            let iter = db.snapshot.get(&k);
            assert_eq!(iter.count(), 1);

            let db = db.commit().await.unwrap();
            db.sync().await.unwrap();

            // Re-open the store, prune it, then ensure it replays the log correctly.
            let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await;
            let floor = db.inactivity_floor_loc();
            let db = db.prune(floor).await.unwrap();

            let iter = db.snapshot.get(&k);
            assert_eq!(iter.count(), 1);

            // First apply_entries: Update + 1 move + CommitFloor = 3 ops. Subsequent 99: Update + 2
            // moves + CommitFloor = 4 ops each. Total: 1 (init) + 3 + 99*4 = 400.
            assert_eq!(*db.bounds().end, 400);
            // Only the last Update and CommitFloor are active → floor = 398.
            assert_eq!(*db.inactivity_floor_loc, 398);
            let floor = db.inactivity_floor_loc;

            // All blobs prior to the inactivity floor are pruned, so the oldest retained location
            // is the first in the last retained blob.
            assert_eq!(db.log.bounds().start, *floor - *floor % 7);

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

    #[test_traced("DEBUG")]
    fn test_store_build_snapshot_keys_with_shared_prefix() {
        let executor = deterministic::Runner::default();

        executor.start(|mut ctx| async move {
            let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await;

            let (k1, v1) = (Digest::random(&mut ctx), vec![1, 2, 3, 4, 5]);
            let (mut k2, v2) = (Digest::random(&mut ctx), vec![6, 7, 8, 9, 10]);

            // Ensure k2 shares 2 bytes with k1 (test DB uses `TwoCap` translator.)
            k2.0[0..2].copy_from_slice(&k1.0[0..2]);

            let (db, _) = apply_entries(db, [(k1, Some(v1.clone()))]).await;
            let (db, _) = apply_entries(db, [(k2, Some(v2.clone()))]).await;

            assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
            assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);

            let db = db.commit().await.unwrap();
            db.sync().await.unwrap();

            // Re-open the store to ensure it builds the snapshot for the conflicting
            // keys correctly.
            let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await;

            assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
            assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);

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

    #[test_traced("DEBUG")]
    fn test_store_delete() {
        let executor = deterministic::Runner::default();

        executor.start(|mut ctx| async move {
            let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await;

            // Insert a key-value pair
            let k = Digest::random(&mut ctx);
            let v = vec![1, 2, 3, 4, 5];
            let (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await;
            let db = db.commit().await.unwrap();

            // Fetch the value
            let fetched_value = db.get(&k).await.unwrap();
            assert_eq!(fetched_value.unwrap(), v);

            // Delete the key
            assert!(db.get(&k).await.unwrap().is_some());
            let (db, _) = apply_entries(db, [(k, None)]).await;

            // Ensure the key is no longer present
            let fetched_value = db.get(&k).await.unwrap();
            assert!(fetched_value.is_none());
            assert!(db.get(&k).await.unwrap().is_none());

            // Commit the changes
            db.commit().await.unwrap();

            // Re-open the store and ensure the key is still deleted
            let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await;
            let fetched_value = db.get(&k).await.unwrap();
            assert!(fetched_value.is_none());

            // Re-insert the key
            let (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await;
            let fetched_value = db.get(&k).await.unwrap();
            assert_eq!(fetched_value.unwrap(), v);

            // Commit the changes
            db.commit().await.unwrap();

            // Re-open the store and ensure the snapshot restores the key, after processing
            // the delete and the subsequent set.
            let db = create_test_store(ctx.child("store").with_attribute("index", 2)).await;
            let fetched_value = db.get(&k).await.unwrap();
            assert_eq!(fetched_value.unwrap(), v);

            // Delete a non-existent key (no-op)
            let k_n = Digest::random(&mut ctx);
            let (db, range) = apply_entries(db, [(k_n, None)]).await;
            assert_eq!(range.start, 9);
            assert_eq!(range.end, 11);
            let db = db.commit().await.unwrap();

            assert!(db.get(&k_n).await.unwrap().is_none());
            // Make sure k is still there
            assert!(db.get(&k).await.unwrap().is_some());

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

    /// Tests the pruning example in the module documentation.
    #[test_traced("DEBUG")]
    fn test_store_pruning() {
        let executor = deterministic::Runner::default();

        executor.start(|mut ctx| async move {
            let db = create_test_store(ctx.child("store")).await;

            let k_a = Digest::random(&mut ctx);
            let k_b = Digest::random(&mut ctx);

            let v_a = vec![1];
            let v_b = vec![];
            let v_c = vec![4, 5, 6];

            let (db, _) = apply_entries(db, [(k_a, Some(v_a.clone()))]).await;
            let (db, _) = apply_entries(db, [(k_b, Some(v_b.clone()))]).await;

            let db = db.commit().await.unwrap();
            assert_eq!(*db.bounds().end, 7);
            assert_eq!(*db.inactivity_floor_loc, 3);
            assert_eq!(db.get(&k_a).await.unwrap().unwrap(), v_a);

            let (db, _) = apply_entries(db, [(k_b, Some(v_a.clone()))]).await;
            let (db, _) = apply_entries(db, [(k_a, Some(v_c.clone()))]).await;

            let db = db.commit().await.unwrap();
            assert_eq!(*db.bounds().end, 15);
            assert_eq!(*db.inactivity_floor_loc, 12);
            assert_eq!(db.get(&k_a).await.unwrap().unwrap(), v_c);
            assert_eq!(db.get(&k_b).await.unwrap().unwrap(), v_a);

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

    /// Pruning to a floor advanced by applied-but-uncommitted entries must not durably outrun
    /// the last durable commit: after a crash, the recovered floor would lie below the pruned
    /// boundary and the store could never reopen.
    #[test_traced("WARN")]
    pub fn test_store_db_prune_after_unsynced_floor_recovery() {
        let executor = deterministic::Runner::default();
        const ELEMENTS: u64 = 1000;
        executor.start(|context| async move {
            let mut db = create_test_store(context.child("store").with_attribute("index", 0)).await;

            // Establish a durable state whose last commit declares an early inactivity floor.
            for i in 0u64..ELEMENTS {
                let k = Blake3::hash(&[&i.to_be_bytes()]);
                let v = vec![(i % 255) as u8; ((i % 13) + 7) as usize];
                (db, _) = apply_entries(db, [(k, Some(v))]).await;
            }
            let mut db = db.commit().await.unwrap();
            let durable_floor = db.inactivity_floor_loc;

            // Apply (but do not commit) entries that advance the in-memory floor past the
            // durable commit's floor.
            for i in 0u64..ELEMENTS {
                let k = Blake3::hash(&[&i.to_be_bytes()]);
                let v = vec![((i + 1) % 255) as u8; ((i % 13) + 8) as usize];
                (db, _) = apply_entries(db, [(k, Some(v))]).await;
            }
            let unsynced_floor = db.inactivity_floor_loc;
            assert!(unsynced_floor > durable_floor);

            // Prune to the in-memory floor, then crash before any further commit.
            let db = db.prune(unsynced_floor).await.unwrap();
            let op_count = db.bounds().end;
            drop(db);

            // Reopening must succeed: prune committed the buffered operations first, so the
            // replayed log reproduces the advanced floor.
            let db = create_test_store(context.child("store").with_attribute("index", 1)).await;
            assert_eq!(db.bounds().end, op_count);
            assert_eq!(db.inactivity_floor_loc, unsynced_floor);
            db.destroy().await.unwrap();
        });
    }

    #[test_traced("WARN")]
    pub fn test_store_db_recovery() {
        let executor = deterministic::Runner::default();
        // Build a db with 1000 keys, some of which we update and some of which we delete.
        const ELEMENTS: u64 = 1000;
        executor.start(|context| async move {
            let db = create_test_store(context.child("store").with_attribute("index", 0)).await;

            // Simulate building batches but not applying them (data is not persisted).
            {
                let mut batch = db.new_batch();
                for i in 0u64..ELEMENTS {
                    let k = Blake3::hash(&[&i.to_be_bytes()]);
                    let v = vec![(i % 255) as u8; ((i % 13) + 7) as usize];
                    batch = batch.update(k, v);
                }
                // Drop the batch without applying -- simulates a failure before apply.
            }
            drop(db);
            let mut db = create_test_store(context.child("store").with_attribute("index", 1)).await;
            assert_eq!(*db.bounds().end, 1);

            // Apply the updates and commit them.
            for i in 0u64..ELEMENTS {
                let k = Blake3::hash(&[&i.to_be_bytes()]);
                let v = vec![(i % 255) as u8; ((i % 13) + 7) as usize];
                (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await;
            }
            let mut db = db.commit().await.unwrap();

            // Update every 3rd key and commit.
            for i in 0u64..ELEMENTS {
                if i % 3 != 0 {
                    continue;
                }
                let k = Blake3::hash(&[&i.to_be_bytes()]);
                let v = vec![((i + 1) % 255) as u8; ((i % 13) + 8) as usize];
                (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await;
            }
            let mut db = db.commit().await.unwrap();
            assert_eq!(db.snapshot.items(), 1000);

            // Delete every 7th key and commit.
            for i in 0u64..ELEMENTS {
                if i % 7 != 1 {
                    continue;
                }
                let k = Blake3::hash(&[&i.to_be_bytes()]);
                (db, _) = apply_entries(db, [(k, None)]).await;
            }
            let db = db.commit().await.unwrap();
            let final_count = db.bounds().end;
            let final_floor = db.inactivity_floor_loc;

            // Sync and reopen the store to ensure the state is preserved.
            db.sync().await.unwrap();
            let db = create_test_store(context.child("store").with_attribute("index", 2)).await;
            assert_eq!(db.bounds().end, final_count);
            assert_eq!(db.inactivity_floor_loc, final_floor);

            let floor = db.inactivity_floor_loc();
            let db = db.prune(floor).await.unwrap();
            assert_eq!(db.log.bounds().start, *final_floor - *final_floor % 7);
            assert_eq!(db.snapshot.items(), 857);

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

    #[test_traced("WARN")]
    pub fn test_store_commit_after_sync_recovers_without_second_sync() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = create_test_store(context.child("store").with_attribute("index", 0)).await;
            let key0 = Blake3::hash(&[&0u64.to_be_bytes()]);
            let key1 = Blake3::hash(&[&1u64.to_be_bytes()]);
            let value0 = vec![0, 1, 2];
            let value1 = vec![3, 4, 5, 6];

            // Commit and sync an initial update so restart recovery has an older watermark.
            let (db, _) = apply_entries(db, [(key0, Some(value0.clone()))]).await;
            let db = db.commit().await.unwrap();
            let db = db.sync().await.unwrap();

            // Persist a later commit without syncing; recovery must replay it after reopen.
            let (db, _) = apply_entries(db, [(key1, Some(value1.clone()))]).await;
            let db = db.commit().await.unwrap();
            let committed_end = db.bounds().end;
            let committed_floor = db.inactivity_floor_loc();
            drop(db);

            let db = create_test_store(context.child("store").with_attribute("index", 1)).await;
            assert_eq!(db.bounds().end, committed_end);
            assert_eq!(db.inactivity_floor_loc(), committed_floor);
            assert_eq!(db.get(&key0).await.unwrap(), Some(value0));
            assert_eq!(db.get(&key1).await.unwrap(), Some(value1));

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

    #[test_traced("DEBUG")]
    fn test_store_batch() {
        let executor = deterministic::Runner::default();

        executor.start(|mut ctx| async move {
            let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await;

            // Ensure the store is empty
            assert_eq!(db.bounds().end, 1);
            assert_eq!(db.inactivity_floor_loc, 0);

            let key = Digest::random(&mut ctx);
            let value = vec![2, 3, 4, 5];

            let batch = db.new_batch();

            // Attempt to get a key that does not exist
            let result = batch.get(&key).await;
            assert!(result.unwrap().is_none());

            // Insert a key-value pair
            let batch = batch.update(key, value.clone());

            assert_eq!(db.bounds().end, 1); // The batch is not applied yet
            assert_eq!(db.inactivity_floor_loc, 0);

            // Fetch the value
            let fetched_value = batch.get(&key).await.unwrap();
            assert_eq!(fetched_value.unwrap(), value);
            let changeset = batch.finalize(None);
            db.apply_batch(changeset).await.unwrap();

            // Re-open the store
            let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await;

            // Ensure the batch was not applied since we didn't commit.
            assert_eq!(db.bounds().end, 1);
            assert_eq!(db.inactivity_floor_loc, 0);
            assert!(db.get_metadata().await.unwrap().is_none());

            // Insert a key-value pair and persist the change.
            let metadata = vec![99, 100];
            let batch = db
                .new_batch()
                .update(key, value.clone())
                .finalize(Some(metadata.clone()));
            let (db, range) = db.apply_batch(batch).await.unwrap();
            assert_eq!(range.start, 1);
            assert_eq!(range.end, 4);
            let db = db.commit().await.unwrap();
            assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));
            drop(db);

            // Re-open the store
            let db = create_test_store(ctx.child("store").with_attribute("index", 2)).await;

            // Ensure the re-opened store retained the committed operations
            assert_eq!(db.bounds().end, 4);
            assert_eq!(db.inactivity_floor_loc, 2);

            // Fetch the value, ensuring it is still present
            let fetched_value = db.get(&key).await.unwrap();
            assert_eq!(fetched_value.unwrap(), value);

            // Destroy the store
            db.destroy().await.unwrap();
        });
    }

    /// A [Db] keyed by variable-length byte keys.
    type VecKeyStore = Db<deterministic::Context, Vec<u8>, Vec<u8>, TwoCap>;

    #[test_traced("DEBUG")]
    fn test_store_variable_length_keys() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Configure the operation codec for variable-length keys.
            let cfg = Config {
                log: JournalConfig {
                    partition: "journal".into(),
                    write_buffer: NZUsize!(64 * 1024),
                    replay_buffer: NZUsize!(64 * 1024),
                    compression: None,
                    codec_config: (((0..=64).into(), ()), ((0..=10000).into(), ())),
                    items_per_section: NZU64!(7),
                    page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
                },
                translator: TwoCap,
                init_cache_size: Some(NZUsize!(1024)),
                init_buffer: NZUsize!(1 << 21),
            };

            // Commit two keys of different lengths that share a translated prefix.
            let db = VecKeyStore::init(
                context.child("store").with_attribute("index", 0),
                cfg.clone(),
            )
            .await
            .unwrap();
            let short = b"key".to_vec();
            let long = b"key-extended".to_vec();
            let batch = db
                .new_batch()
                .update(short.clone(), vec![1])
                .update(long.clone(), vec![2])
                .finalize(None);
            let (db, _) = db.apply_batch(batch).await.unwrap();
            let db = db.commit().await.unwrap();
            assert_eq!(db.get(&short).await.unwrap(), Some(vec![1]));
            assert_eq!(db.get(&long).await.unwrap(), Some(vec![2]));
            drop(db);

            // Reopen the store and verify both committed values.
            let db = VecKeyStore::init(context.child("store").with_attribute("index", 1), cfg)
                .await
                .unwrap();
            assert_eq!(db.get(&short).await.unwrap(), Some(vec![1]));
            assert_eq!(db.get(&long).await.unwrap(), Some(vec![2]));
            db.destroy().await.unwrap();
        });
    }

    fn is_send<T: Send>(_: T) {}

    #[allow(dead_code)]
    fn assert_read_futures_are_send(db: TestStore, key: Digest, loc: Location) {
        is_send(db.get(&key));
        is_send(db.get_metadata());
        is_send(db.prune(loc));
    }

    #[allow(dead_code)]
    fn assert_sync_is_send(db: TestStore) {
        is_send(db.sync());
    }

    #[allow(dead_code)]
    fn assert_write_futures_are_send(
        db: Db<deterministic::Context, Digest, Vec<u8>, TwoCap>,
        key: Digest,
        value: Vec<u8>,
    ) {
        is_send(db.get(&key));
        let batch = db.new_batch();
        is_send(batch.get(&key));
        is_send(db.apply_batch(Changeset::from([(key, Some(value))])));
    }

    #[allow(dead_code)]
    fn assert_commit_is_send(db: Db<deterministic::Context, Digest, Vec<u8>, TwoCap>) {
        is_send(db.commit());
    }
}