commonware-storage 2026.10.0

//! An authenticated database that provides succinct proofs of _any_ value ever associated
//! with a key, maintains a next-key ordering for each active key, and allows values to have
//! variable sizes.
//!
//! _If the values you wish to store all have the same size, use [crate::qmdb::any::ordered::fixed]
//! instead for better performance._

use crate::{
    Context,
    index::ordered::Index,
    journal::contiguous::variable::Journal,
    merkle::{Family, Location},
    qmdb::{
        Error,
        any::{VariableConfig, VariableValue, ordered, value::VariableEncoding},
        operation::Key,
    },
    translator::OrderedTranslator,
};
use commonware_codec::Read;
use commonware_cryptography::Hasher;
use commonware_parallel::Strategy;
use commonware_runtime::Spawner;

pub type Update<K, V> = ordered::Update<K, VariableEncoding<V>>;
pub type Operation<F, K, V> = ordered::Operation<F, K, VariableEncoding<V>>;

/// A key-value QMDB based on an authenticated log of operations, supporting authentication of any
/// value ever associated with a key.
pub type Db<F, E, K, V, H, T, S> = super::Db<
    F,
    E,
    Journal<E, Operation<F, K, V>>,
    Index<T, Location<F>>,
    H,
    Update<K, V>,
    { crate::qmdb::any::BITMAP_CHUNK_BYTES },
    S,
>;

impl<
    F: Family,
    E: Context + Spawner,
    K: Key,
    V: VariableValue,
    H: Hasher,
    T: OrderedTranslator,
    S: Strategy,
> Db<F, E, K, V, H, T, S>
{
    /// Returns a [Db] QMDB initialized from `cfg`. Any uncommitted log operations will be
    /// discarded and the state of the db will be as of the last committed operation.
    /// `Some(max_size)` selects the latest retained commit with at most `max_size` operations.
    /// `None` selects the latest retained state.
    pub async fn init(
        context: E,
        cfg: VariableConfig<T, <Operation<F, K, V> as Read>::Cfg, S>,
        max_size: Option<Location<F>>,
    ) -> Result<Self, Error<F>> {
        crate::qmdb::any::init(context, cfg, max_size).await
    }
}

/// Partitioned index variants that divide the key space into `2^(P*8)` partitions.
///
/// See [partitioned::Db] for the generic type, or use the convenience aliases:
/// - [partitioned::p256::Db] for 256 partitions (P=1)
/// - [partitioned::p64k::Db] for 65,536 partitions (P=2)
///
/// `p256` suits smaller datasets, but its partitions spill to a `BTreeMap` once the index holds more
/// than 130,816 entries (see [`crate::index::partitioned::ordered`]); prefer `p64k` to keep ordered
/// access mostly inline at larger scale.
pub mod partitioned {
    pub use super::{Operation, Update};
    use crate::{
        Context,
        index::partitioned::ordered::Index,
        journal::contiguous::variable::Journal,
        merkle::{Family, Location},
        qmdb::{
            Error,
            any::{VariableConfig, VariableValue},
            operation::Key,
        },
        translator::OrderedTranslator,
    };
    use commonware_codec::Read;
    use commonware_cryptography::Hasher;
    use commonware_parallel::Strategy;
    use commonware_runtime::Spawner;

    /// An ordered key-value QMDB with a partitioned snapshot index and variable-size values.
    ///
    /// This is the partitioned variant of [super::Db]. The const generic `P` specifies
    /// the number of prefix bytes used for partitioning:
    /// - `P = 1`: 256 partitions
    /// - `P = 2`: 65,536 partitions
    ///
    /// Use partitioned indices when you have a large number of keys (>> 2^(P*8)) and memory
    /// efficiency is important. Keys should be uniformly distributed across the prefix space.
    pub type Db<F, E, K, V, H, T, const P: usize, S> = crate::qmdb::any::ordered::Db<
        F,
        E,
        Journal<E, Operation<F, K, V>>,
        Index<T, Location<F>, P>,
        H,
        Update<K, V>,
        { crate::qmdb::any::BITMAP_CHUNK_BYTES },
        S,
    >;

    impl<
        F: Family,
        E: Context + Spawner,
        K: Key,
        V: VariableValue,
        H: Hasher,
        T: OrderedTranslator,
        const P: usize,
        S: Strategy,
    > Db<F, E, K, V, H, T, P, S>
    {
        /// Returns a [Db] QMDB initialized from `cfg`. Uncommitted log operations will be
        /// discarded and the state of the db will be as of the last committed operation.
        /// `Some(max_size)` selects the latest retained commit with at most `max_size` operations.
        /// `None` selects the latest retained state.
        pub async fn init(
            context: E,
            cfg: VariableConfig<T, <Operation<F, K, V> as Read>::Cfg, S, core::num::NonZeroUsize>,
            max_size: Option<Location<F>>,
        ) -> Result<Self, Error<F>> {
            crate::qmdb::any::init(context, cfg, max_size).await
        }
    }

    /// Convenience type aliases for 256 partitions (P=1).
    pub mod p256 {
        /// Variable-value DB with 256 partitions.
        pub type Db<F, E, K, V, H, T, S> = super::Db<F, E, K, V, H, T, 1, S>;
    }

    /// Convenience type aliases for 65,536 partitions (P=2).
    pub mod p64k {
        /// Variable-value DB with 65,536 partitions.
        pub type Db<F, E, K, V, H, T, S> = super::Db<F, E, K, V, H, T, 2, S>;
    }
}

#[cfg(test)]
pub(crate) mod test {
    use super::*;
    use crate::{
        mmr,
        qmdb::{
            any::{
                batch::{DiffEntry, Slot, StagedLoc, lookup_sorted},
                ordered::test::{
                    test_ordered_any_db_basic, test_ordered_any_db_empty,
                    test_ordered_any_update_collision_edge_case,
                },
                test::variable_db_config,
            },
            floor::{Bounded, Proportional},
        },
        translator::TwoCap,
    };
    use commonware_cryptography::{Sha256, sha256::Digest};
    use commonware_macros::test_traced;
    use commonware_math::algebra::Random;
    use commonware_parallel::Sequential;
    use commonware_runtime::{
        Runner as _, Supervisor as _,
        buffer::paged::CacheRef,
        deterministic::{self, Context},
    };
    use commonware_utils::{NZU16, NZU64, NZUsize, TestRng, sequence::FixedBytes};
    use rand::Rng;
    // Janky page & cache sizes to exercise boundary conditions.
    const PAGE_SIZE: u16 = 103;
    const PAGE_CACHE_SIZE: usize = 13;

    pub(crate) type VarConfig<B = ()> =
        VariableConfig<TwoCap, ((), (commonware_codec::RangeCfg<usize>, ())), Sequential, B>;

    /// Type alias for the concrete [Db] type used in these unit tests.
    pub(crate) type AnyTest =
        Db<mmr::Family, deterministic::Context, Digest, Vec<u8>, Sha256, TwoCap, Sequential>;

    pub(crate) fn create_test_config<B>(seed: u64, init_concurrency: B) -> VarConfig<B> {
        let page_cache = CacheRef::new(NZU16!(PAGE_SIZE), NZUsize!(PAGE_CACHE_SIZE));
        VariableConfig {
            merkle_config: crate::mmr::full::Config {
                journal_partition: format!("mmr-journal-{seed}"),
                metadata_partition: format!("mmr-metadata-{seed}"),
                items_per_blob: NZU64!(12), // intentionally small and janky size
                write_buffer: NZUsize!(64),
                replay_buffer: NZUsize!(64),
                strategy: Sequential,
                page_cache: page_cache.clone(),
            },
            journal_config: crate::journal::contiguous::variable::Config {
                partition: format!("log-journal-{seed}"),
                items_per_section: NZU64!(14), // intentionally small and janky size
                write_buffer: NZUsize!(64),
                replay_buffer: NZUsize!(64),
                compression: None,
                codec_config: ((), ((0..=10000).into(), ())),
                page_cache,
            },
            translator: TwoCap,
            init_cache: Some(NZUsize!(1024)),
            init_buffer: NZUsize!(1 << 21),
            init_concurrency,
        }
    }

    /// Create a test database with unique partition names
    pub(crate) async fn create_test_db(mut context: Context) -> AnyTest {
        let seed = context.next_u64();
        let config = create_test_config(seed, ());
        AnyTest::init(context, config, None).await.unwrap()
    }

    #[test_traced]
    fn test_foreign_db_merkleize_rejected() {
        deterministic::Runner::default().start(|context| async move {
            let db_a = create_test_db(context.child("a")).await;
            let db_b = create_test_db(context.child("b")).await;

            let seed_a = db_a
                .new_batch()
                .write(Sha256::hash(&[b"a"]), Some(vec![1]))
                .merkleize(&db_a, None, &mut Proportional)
                .await
                .unwrap();
            let db_a = db_a.apply_batch(seed_a).await.unwrap();
            let seed_b = db_b
                .new_batch()
                .write(Sha256::hash(&[b"b"]), Some(vec![2]))
                .merkleize(&db_b, None, &mut Proportional)
                .await
                .unwrap();
            let db_b = db_b.apply_batch(seed_b).await.unwrap();

            assert_eq!(db_a.bounds().end, db_b.bounds().end);
            assert_ne!(db_a.root(), db_b.root());

            let batch = db_a.new_batch().write(Sha256::hash(&[b"a"]), Some(vec![3]));
            assert!(matches!(
                batch.merkleize(&db_b, None, &mut Proportional).await,
                Err(Error::StaleBatch)
            ));
        });
    }

    /// Serial-vs-parallel init equivalence for the variable-value partitioned db. The parallel
    /// build streams variable-size ops through the shared log, which the fixed-value equivalence
    /// tests cannot cover.
    #[test_traced("WARN")]
    fn test_ordered_partitioned_variable_parallel_init_equivalence() {
        deterministic::Runner::default().start(|context| async move {
            type PartDb<S> = partitioned::Db<
                mmr::Family,
                deterministic::Context,
                Digest,
                Vec<u8>,
                Sha256,
                TwoCap,
                1,
                S,
            >;

            /// The value each key holds after the two commits below.
            fn expected_value(i: u64) -> Option<Vec<u8>> {
                if i % 7 == 1 {
                    None
                } else if i.is_multiple_of(3) {
                    Some(vec![0xAB; (i % 17 + 1) as usize])
                } else {
                    Some(vec![(i % 251) as u8; (i % 40 + 1) as usize])
                }
            }

            // Commit 1: insert every key.
            let cfg = create_test_config(77, NZUsize!(1));
            let db = PartDb::<Sequential>::init(context.child("populate"), cfg, None)
                .await
                .unwrap();
            let mut batch = db.new_batch();
            for i in 0u64..500 {
                let k = Sha256::hash(&[&i.to_be_bytes()]);
                let v = vec![(i % 251) as u8; (i % 40 + 1) as usize];
                batch = batch.write(k, Some(v));
            }
            let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();
            let db = db.commit().await.unwrap();

            // Commit 2: update a third and delete a seventh so the replay carries churn.
            let mut batch = db.new_batch();
            for i in (0u64..500).step_by(3) {
                let k = Sha256::hash(&[&i.to_be_bytes()]);
                batch = batch.write(k, Some(vec![0xAB; (i % 17 + 1) as usize]));
            }
            for i in (1u64..500).step_by(7) {
                let k = Sha256::hash(&[&i.to_be_bytes()]);
                batch = batch.write(k, None);
            }
            let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();
            let db = db.commit().await.unwrap();
            let db = db.sync().await.unwrap();
            let root = db.root();
            drop(db);

            for concurrency in [1usize, 4] {
                let cfg =
                    create_test_config(77, core::num::NonZeroUsize::new(concurrency).unwrap());
                let ctx = context
                    .child("reopen")
                    .with_attribute("concurrency", concurrency);
                let db = PartDb::<Sequential>::init(ctx, cfg, None).await.unwrap();
                assert_eq!(
                    db.root(),
                    root,
                    "root mismatch at concurrency={concurrency}"
                );
                for i in 0u64..500 {
                    let k = Sha256::hash(&[&i.to_be_bytes()]);
                    assert_eq!(
                        db.get(&k).await.unwrap(),
                        expected_value(i),
                        "value mismatch for key {i}"
                    );
                }
                drop(db);
            }
        });
    }

    /// Deterministic byte vector generator for variable-value tests.
    fn to_bytes(i: u64) -> Vec<u8> {
        let len = ((i % 13) + 7) as usize;
        vec![(i % 255) as u8; len]
    }

    /// Create n random operations using the default seed (0). Some portion of
    /// the updates are deletes. create_test_ops(n) is a prefix of
    /// create_test_ops(n') for n < n'.
    pub(crate) fn create_test_ops(n: usize) -> Vec<Operation<mmr::Family, Digest, Vec<u8>>> {
        create_test_ops_seeded(n, 0)
    }

    /// Create n random operations using a specific seed. Use different seeds
    /// when you need non-overlapping keys in the same test.
    pub(crate) fn create_test_ops_seeded(
        n: usize,
        seed: u64,
    ) -> Vec<Operation<mmr::Family, Digest, Vec<u8>>> {
        let mut rng = TestRng::new(seed);
        let mut prev_key = Digest::random(&mut rng);
        let mut ops = Vec::new();
        for i in 0..n {
            if i % 10 == 0 && i > 0 {
                ops.push(Operation::Delete(prev_key));
            } else {
                let key = Digest::random(&mut rng);
                let next_key = Digest::random(&mut rng);
                let value = to_bytes(rng.next_u64());
                ops.push(Operation::Update(ordered::Update {
                    key,
                    value,
                    next_key,
                }));
                prev_key = key;
            }
        }
        ops
    }

    /// Applies the given operations to the database.
    pub(crate) async fn apply_ops(
        db: AnyTest,
        ops: Vec<Operation<mmr::Family, Digest, Vec<u8>>>,
    ) -> AnyTest {
        let mut batch = db.new_batch();
        for op in ops {
            match op {
                Operation::Update(data) => {
                    batch = batch.write(data.key, Some(data.value));
                }
                Operation::Delete(key) => {
                    batch = batch.write(key, None);
                }
                Operation::CommitFloor(_, _) => {
                    // CommitFloor consumes self, so this helper does not support it.
                    // Test data from create_test_ops never includes CommitFloor.
                    panic!("CommitFloor not supported in apply_ops");
                }
            }
        }
        let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
        db.apply_batch(merkleized).await.unwrap()
    }

    // Tests using FixedBytes<4> keys (for edge cases that require specific key patterns)

    /// Type alias for a variable db with FixedBytes<4> keys.
    type VariableDb = Db<mmr::Family, Context, FixedBytes<4>, Digest, Sha256, TwoCap, Sequential>;

    /// Return a variable db with FixedBytes<4> keys.
    async fn open_variable_db(context: Context) -> VariableDb {
        let cfg = variable_db_config("fixed-bytes-var-partition");
        VariableDb::init(context, cfg, None).await.unwrap()
    }

    #[test_traced("WARN")]
    fn test_ordered_any_variable_db_empty() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = open_variable_db(context.child("initial")).await;
            test_ordered_any_db_empty(context, db, |ctx| Box::pin(open_variable_db(ctx))).await;
        });
    }

    #[test_traced("WARN")]
    fn test_ordered_any_variable_db_basic() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = open_variable_db(context.child("initial")).await;
            test_ordered_any_db_basic(context, db, |ctx| Box::pin(open_variable_db(ctx))).await;
        });
    }

    #[test_traced("WARN")]
    fn test_ordered_any_update_collision_edge_case_variable() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = open_variable_db(context.child("storage")).await;
            test_ordered_any_update_collision_edge_case(db).await;
        });
    }

    /// Builds a db with two colliding keys, and creates a new one between them using a batch
    /// update.
    #[test_traced("WARN")]
    fn test_ordered_any_update_batch_create_between_collisions() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = open_variable_db(context.child("storage")).await;

            // This DB uses a TwoCap so we use equivalent two byte prefixes for each key to ensure
            // collisions.
            let key1 = FixedBytes::from([0xFFu8, 0xFFu8, 5u8, 5u8]);
            let key2 = FixedBytes::from([0xFFu8, 0xFFu8, 6u8, 6u8]);
            let key3 = FixedBytes::from([0xFFu8, 0xFFu8, 7u8, 0u8]);
            let val = Sha256::fill(1u8);

            let merkleized = db
                .new_batch()
                .write(key1.clone(), Some(val))
                .write(key3.clone(), Some(val))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();

            assert_eq!(db.get(&key1).await.unwrap().unwrap(), val);
            assert!(db.get(&key2).await.unwrap().is_none());
            assert_eq!(db.get(&key3).await.unwrap().unwrap(), val);

            // Batch-insert the middle key.
            let merkleized = db
                .new_batch()
                .write(key2.clone(), Some(val))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();

            assert_eq!(db.get(&key1).await.unwrap().unwrap(), val);
            assert_eq!(db.get(&key2).await.unwrap().unwrap(), val);
            assert_eq!(db.get(&key3).await.unwrap().unwrap(), val);

            let span1 = db.get_span(&key1).await.unwrap().unwrap();
            assert_eq!(span1.1.next_key, key2);
            let span2 = db.get_span(&key2).await.unwrap().unwrap();
            assert_eq!(span2.1.next_key, key3);
            let span3 = db.get_span(&key3).await.unwrap().unwrap();
            assert_eq!(span3.1.next_key, key1);

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

    /// Batch create/delete cases where the deleted key is the previous key of a newly created key,
    /// and vice-versa.
    #[test_traced("WARN")]
    fn test_ordered_any_batch_create_delete_prev_links() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let key1 = FixedBytes::from([0x10u8, 0x00, 0x00, 0x00]);
            let key2 = FixedBytes::from([0x20u8, 0x00, 0x00, 0x00]);
            let key3 = FixedBytes::from([0x30u8, 0x00, 0x00, 0x00]);
            let val1 = Sha256::fill(1u8);
            let val2 = Sha256::fill(2u8);
            let val3 = Sha256::fill(3u8);

            // Delete the previous key of a newly created key.
            let db = open_variable_db(context.child("first")).await;
            let merkleized = db
                .new_batch()
                .write(key1.clone(), Some(val1))
                .write(key3.clone(), Some(val3))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();

            let merkleized = db
                .new_batch()
                .write(key1.clone(), None)
                .write(key2.clone(), Some(val2))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();

            assert!(db.get(&key1).await.unwrap().is_none());
            assert_eq!(db.get(&key2).await.unwrap(), Some(val2));
            assert_eq!(db.get(&key3).await.unwrap(), Some(val3));
            let span2 = db.get_span(&key2).await.unwrap().unwrap();
            assert_eq!(span2.1.next_key, key3);
            let span3 = db.get_span(&key3).await.unwrap().unwrap();
            assert_eq!(span3.1.next_key, key2);
            db.destroy().await.unwrap();

            // Create a key that becomes the previous key of a concurrently deleted key.
            let db = open_variable_db(context.child("second")).await;
            let merkleized = db
                .new_batch()
                .write(key1.clone(), Some(val1))
                .write(key3.clone(), Some(val3))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();

            let merkleized = db
                .new_batch()
                .write(key2.clone(), Some(val2))
                .write(key3.clone(), None)
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();
            let db = db.apply_batch(merkleized).await.unwrap();

            assert_eq!(db.get(&key1).await.unwrap(), Some(val1));
            assert_eq!(db.get(&key2).await.unwrap(), Some(val2));
            assert!(db.get(&key3).await.unwrap().is_none());
            let span1 = db.get_span(&key1).await.unwrap().unwrap();
            assert_eq!(span1.1.next_key, key2);
            let span2 = db.get_span(&key2).await.unwrap().unwrap();
            assert_eq!(span2.1.next_key, key1);
            db.destroy().await.unwrap();
        });
    }

    /// A staged read that resolved in a grandparent's diff must stay correct when that grandparent
    /// is applied and dropped before `Staged::merkleize`. The resolved location and the cached
    /// successor both come from the grandparent's update, and applying the grandparent makes that
    /// update the key's database update. Merkleize must therefore supersede the resolved location
    /// and link each rewritten key to its cached successor (see `StagedLoc`).
    #[test_traced("WARN")]
    fn ordered_variable_staged_ancestor_applied_before_merkleize() {
        deterministic::Runner::default().start(|context| async move {
            let db = create_test_db(context.child("staged_ancestor")).await;

            let key = |i: u64| Sha256::hash(&[&i.to_be_bytes()]);

            // Committed base state, so the grandparent's write of key(0) supersedes a
            // committed location. Its create of key(100) supersedes none.
            let mut seed = db.new_batch();
            for i in 0..8u64 {
                seed = seed.write(key(i), Some(to_bytes(i)));
            }
            let seed = seed.merkleize(&db, None, &mut Proportional).await.unwrap();
            let db = db.apply_batch(seed).await.unwrap().commit().await.unwrap();

            // Grandparent -> parent chain. The parent touches neither staged key, so the
            // staged reads resolve in the grandparent's diff.
            let grandparent = db
                .new_batch()
                .write(key(0), Some(to_bytes(1_000)))
                .write(key(100), Some(to_bytes(1_001)))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();
            let parent = grandparent
                .new_batch::<Sha256>()
                .write(key(1), Some(to_bytes(1_002)))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();

            // Stage both keys through the parent. Each must resolve in the grandparent's diff.
            let read_keys = [key(0), key(100)];
            let keys: Vec<&Digest> = read_keys.iter().collect();
            let (values, staged) = parent
                .new_batch::<Sha256>()
                .stage(&keys, &db)
                .await
                .unwrap();
            assert_eq!(values, vec![Some(to_bytes(1_000)), Some(to_bytes(1_001))]);
            for (resolved, key) in staged.resolved_locs().into_iter().zip(&read_keys) {
                let written =
                    lookup_sorted(grandparent.diff.as_slice(), key).and_then(DiffEntry::loc);
                assert!(
                    matches!(
                        resolved,
                        Some(StagedLoc::Ancestor { loc, .. }) if Some(loc) == written
                    ),
                    "a staged read must resolve in the grandparent's diff"
                );
            }

            // Apply and drop the grandparent, so the staged resolutions' locations migrate
            // into the database region, retiring their recorded bases.
            let db = db.apply_batch(grandparent).await.unwrap();

            // Merkleize the staged writes over the applied state.
            let slots = vec![Slot::Put(to_bytes(2_000)), Slot::Put(to_bytes(2_001))];
            let staged = staged
                .merkleize(slots, Vec::new(), None, &db, &mut Proportional)
                .await
                .unwrap();

            // The explicit path over the same applied state must agree.
            let explicit_root = parent
                .new_batch::<Sha256>()
                .write(key(0), Some(to_bytes(2_000)))
                .write(key(100), Some(to_bytes(2_001)))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap()
                .root();
            assert_eq!(staged.root(), explicit_root);

            // Apply the rest of the chain and read the writes back.
            let db = db.apply_batch(parent).await.unwrap();
            let db = db.apply_batch(staged).await.unwrap();
            let db = db.commit().await.unwrap();

            assert_eq!(db.get(&key(0)).await.unwrap(), Some(to_bytes(2_000)));
            assert_eq!(db.get(&key(100)).await.unwrap(), Some(to_bytes(2_001)));
            assert_eq!(db.get(&key(1)).await.unwrap(), Some(to_bytes(1_002)));

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

    /// A staged read that resolved in an unapplied ancestor's diff keeps its recorded base while
    /// that ancestor is reachable at `Staged::merkleize` (see `StagedLoc`). The full lifecycle must
    /// hold under both apply shapes: one `apply_batch` call applying the unapplied ancestors with
    /// the staged batch (the recorded bases are used directly), and the ancestors applied first by
    /// their own calls (the staged batch's apply re-resolves each base in the applied ancestors'
    /// diffs).
    ///
    /// Covers grandparent-updated, grandparent-created, and parent-updated keys plus a staged
    /// delete, so both `Some` and `None` recorded bases flow through each path. The delete also
    /// rewrites its predecessor's link.
    #[test_traced("WARN")]
    fn ordered_variable_staged_ancestor_alive_through_apply() {
        deterministic::Runner::default().start(|context| async move {
            for apply_ancestors_first in [false, true] {
                let label = if apply_ancestors_first {
                    "staged_ancestor_alive_separate"
                } else {
                    "staged_ancestor_alive_combined"
                };
                let db = create_test_db(context.child(label)).await;

                let key = |i: u64| Sha256::hash(&[&i.to_be_bytes()]);

                // Seed a committed base state. The grandparent's write of key(0) and the parent's
                // write of key(1) supersede committed locations. key(100) is created by the
                // grandparent, so its recorded base is `None`. Enough keys are seeded that key(3)
                // stays in the database.
                let mut seed = db.new_batch();
                for i in 0..64u64 {
                    seed = seed.write(key(i), Some(to_bytes(i)));
                }
                let seed = seed.merkleize(&db, None, &mut Proportional).await.unwrap();
                let db = db.apply_batch(seed).await.unwrap().commit().await.unwrap();

                let grandparent = db
                    .new_batch()
                    .write(key(0), Some(to_bytes(1_000)))
                    .write(key(100), Some(to_bytes(1_001)))
                    .merkleize(&db, None, &mut Proportional)
                    .await
                    .unwrap();
                let parent = grandparent
                    .new_batch::<Sha256>()
                    .write(key(1), Some(to_bytes(1_002)))
                    .merkleize(&db, None, &mut Proportional)
                    .await
                    .unwrap();

                // key(3) is untouched by the ancestors (their floor raises and relinks move
                // other keys), so its staged read stays committed-resolved.
                let read_keys = [key(0), key(100), key(1), key(3)];
                let keys: Vec<&Digest> = read_keys.iter().collect();
                let (values, staged) = parent
                    .new_batch::<Sha256>()
                    .stage(&keys, &db)
                    .await
                    .unwrap();
                assert_eq!(
                    values,
                    vec![
                        Some(to_bytes(1_000)),
                        Some(to_bytes(1_001)),
                        Some(to_bytes(1_002)),
                        Some(to_bytes(3)),
                    ]
                );

                // Each ancestor-written key must resolve in the diff that wrote it, and key(3)
                // in the database.
                let resolved = staged.resolved_locs();
                for (slot, written) in [
                    lookup_sorted(grandparent.diff.as_slice(), &key(0)),
                    lookup_sorted(grandparent.diff.as_slice(), &key(100)),
                    lookup_sorted(parent.diff.as_slice(), &key(1)),
                ]
                .into_iter()
                .enumerate()
                {
                    let written = written.and_then(DiffEntry::loc);
                    assert!(
                        matches!(
                            resolved[slot],
                            Some(StagedLoc::Ancestor { loc, .. }) if Some(loc) == written
                        ),
                        "slot {slot} must resolve in the diff of the ancestor that wrote it"
                    );
                }
                assert!(
                    matches!(resolved[3], Some(StagedLoc::Committed(_))),
                    "a key no ancestor wrote must resolve in the database"
                );

                // Merkleize with every ancestor still held, so the ancestor-resolved slots
                // trust their recorded bases. Slot 2 is a staged delete.
                let slots = vec![
                    Slot::Put(to_bytes(2_000)),
                    Slot::Put(to_bytes(2_001)),
                    Slot::Delete,
                    Slot::Put(to_bytes(2_002)),
                ];
                let staged = staged
                    .merkleize(slots, Vec::new(), None, &db, &mut Proportional)
                    .await
                    .unwrap();

                // The explicit path over the same chain must agree.
                let explicit_root = parent
                    .new_batch::<Sha256>()
                    .write(key(0), Some(to_bytes(2_000)))
                    .write(key(100), Some(to_bytes(2_001)))
                    .write(key(1), None)
                    .write(key(3), Some(to_bytes(2_002)))
                    .merkleize(&db, None, &mut Proportional)
                    .await
                    .unwrap()
                    .root();
                assert_eq!(staged.root(), explicit_root);

                // Apply under the chosen shape and read the writes back.
                let db = if apply_ancestors_first {
                    let db = db.apply_batch(grandparent).await.unwrap();
                    db.apply_batch(parent).await.unwrap()
                } else {
                    db
                };
                let db = db.apply_batch(staged).await.unwrap();
                let db = db.commit().await.unwrap();

                assert_eq!(db.get(&key(0)).await.unwrap(), Some(to_bytes(2_000)));
                assert_eq!(db.get(&key(100)).await.unwrap(), Some(to_bytes(2_001)));
                assert_eq!(db.get(&key(1)).await.unwrap(), None);
                assert_eq!(db.get(&key(3)).await.unwrap(), Some(to_bytes(2_002)));

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

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

    #[allow(dead_code)]
    fn assert_non_trait_futures_are_send(db: &mut AnyTest, key: Digest) {
        is_send(db.get_all(&key));
        is_send(db.get_with_loc(&key));
        is_send(db.get_span(&key));
        let mut policy = Bounded {
            entries: 1,
            skips: 1,
        };
        is_send(db.new_batch().merkleize(db, None, &mut policy));
        is_send(async move {
            let (_, staged) = db.new_batch().stage(&[&key], db).await?;
            staged
                .merkleize(vec![Slot::Keep], Vec::new(), None, db, &mut policy)
                .await
        });
    }

    /// Parent inserts a key, child inserts another; commit parent then
    /// apply child sequentially. Verifies next-key pointers
    /// are correct after both commits.
    #[test_traced("WARN")]
    fn test_ordered_sequential_commit_basic() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = create_test_db(context).await;

            // Seed with initial data so the ordered index is non-trivial.
            let db = apply_ops(db, create_test_ops(10)).await;
            let db = db.commit().await.unwrap();

            let base = db.to_batch();

            // Parent batch: insert key_a.
            let key_a = Digest::random(TestRng::new(800));
            let val_a = vec![1u8; 10];
            let parent_batch = base
                .new_batch::<Sha256>()
                .write(key_a, Some(val_a.clone()))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();

            // Child batch: insert key_b.
            let key_b = Digest::random(TestRng::new(801));
            let val_b = vec![2u8; 10];
            let child_batch = parent_batch
                .new_batch::<Sha256>()
                .write(key_b, Some(val_b.clone()))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();

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

            // Commit child.
            let db = db.apply_batch(child_batch).await.unwrap();
            let db = db.commit().await.unwrap();

            // Both keys should be readable.
            assert_eq!(db.get(&key_a).await.unwrap().unwrap(), val_a);
            assert_eq!(db.get(&key_b).await.unwrap().unwrap(), val_b);

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

    /// Parent inserts key_x, child deletes key_x. After committing parent
    /// then child sequentially, key_x should be gone and the
    /// next-key ring should exclude it.
    #[test_traced("WARN")]
    fn test_ordered_sequential_commit_delete_after_insert() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = create_test_db(context).await;

            let db = apply_ops(db, create_test_ops(5)).await;
            let db = db.commit().await.unwrap();

            let base = db.to_batch();

            let key_x = Digest::random(TestRng::new(810));
            let val_x = vec![10u8; 8];
            let parent_batch = base
                .new_batch::<Sha256>()
                .write(key_x, Some(val_x.clone()))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();

            let child_batch = parent_batch
                .new_batch::<Sha256>()
                .write(key_x, None)
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();

            let db = db.apply_batch(parent_batch).await.unwrap();
            let db = db.commit().await.unwrap();
            assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_x);

            // Commit child.
            let db = db.apply_batch(child_batch).await.unwrap();
            let db = db.commit().await.unwrap();

            // key_x should be deleted.
            assert!(db.get(&key_x).await.unwrap().is_none());

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

    /// Parent and child both modify the same key. After committing parent
    /// then child sequentially, the child's value wins.
    #[test_traced("WARN")]
    fn test_ordered_sequential_commit_overlapping_keys() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let db = create_test_db(context).await;

            let db = apply_ops(db, create_test_ops(5)).await;
            let db = db.commit().await.unwrap();

            let base = db.to_batch();

            let key_x = Digest::random(TestRng::new(820));
            let val_a = vec![10u8; 8];
            let parent_batch = base
                .new_batch::<Sha256>()
                .write(key_x, Some(val_a.clone()))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();

            let val_b = vec![20u8; 8];
            let child_batch = parent_batch
                .new_batch::<Sha256>()
                .write(key_x, Some(val_b.clone()))
                .merkleize(&db, None, &mut Proportional)
                .await
                .unwrap();

            let db = db.apply_batch(parent_batch).await.unwrap();
            let db = db.commit().await.unwrap();
            assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_a);

            // Commit child.
            let db = db.apply_batch(child_batch).await.unwrap();
            let db = db.commit().await.unwrap();

            assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_b);

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

    // FromSyncTestable implementation for from_sync_result tests
    mod from_sync_testable {
        use super::*;
        use crate::{
            merkle::{Family, Location, full::Merkle},
            qmdb::any::sync::tests::FromSyncTestable,
        };
        use futures::future::join_all;

        impl<F: Family> FromSyncTestable
            for Db<F, deterministic::Context, Digest, Vec<u8>, Sha256, TwoCap, Sequential>
        {
            type Merkle = Merkle<F, deterministic::Context, Digest, Sequential>;

            fn into_log_components(self) -> (Self::Merkle, Self::Journal) {
                (self.log.merkle, self.log.journal)
            }

            async fn pinned_nodes_at(&self, loc: Location<F>) -> Vec<Digest> {
                join_all(F::nodes_to_pin(loc).map(|p| self.log.merkle.get_node(p)))
                    .await
                    .into_iter()
                    .map(|n| n.unwrap().unwrap())
                    .collect()
            }
        }
    }
}