//! 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 = ordered::Update>; pub type Operation = ordered::Operation>; /// A key-value QMDB based on an authenticated log of operations, supporting authentication of any /// value ever associated with a key. pub type Db = super::Db< F, E, Journal>, Index>, H, Update, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, S, >; impl< F: Family, E: Context + Spawner, K: Key, V: VariableValue, H: Hasher, T: OrderedTranslator, S: Strategy, > Db { /// 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 as Read>::Cfg, S>, max_size: Option>, ) -> Result> { 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 = crate::qmdb::any::ordered::Db< F, E, Journal>, Index, P>, H, Update, { 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 { /// 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 as Read>::Cfg, S, core::num::NonZeroUsize>, max_size: Option>, ) -> Result> { 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 = super::Db; } /// Convenience type aliases for 65,536 partitions (P=2). pub mod p64k { /// Variable-value DB with 65,536 partitions. pub type Db = super::Db; } } #[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 = VariableConfig, ())), Sequential, B>; /// Type alias for the concrete [Db] type used in these unit tests. pub(crate) type AnyTest = Db, Sha256, TwoCap, Sequential>; pub(crate) fn create_test_config(seed: u64, init_concurrency: B) -> VarConfig { 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 = partitioned::Db< mmr::Family, deterministic::Context, Digest, Vec, Sha256, TwoCap, 1, S, >; /// The value each key holds after the two commits below. fn expected_value(i: u64) -> Option> { 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::::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::::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 { 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>> { 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>> { 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>>, ) -> 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, 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::() .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::() .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::() .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::() .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::() .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::() .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) {} #[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::() .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::() .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::() .write(key_x, Some(val_x.clone())) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let child_batch = parent_batch .new_batch::() .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::() .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::() .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 FromSyncTestable for Db, Sha256, TwoCap, Sequential> { type Merkle = Merkle; fn into_log_components(self) -> (Self::Merkle, Self::Journal) { (self.log.merkle, self.log.journal) } async fn pinned_nodes_at(&self, loc: Location) -> Vec { join_all(F::nodes_to_pin(loc).map(|p| self.log.merkle.get_node(p))) .await .into_iter() .map(|n| n.unwrap().unwrap()) .collect() } } } }