//! An authenticated database that provides succinct proofs of _any_ value ever associated //! with a key, where values can have varying sizes. //! //! _If the values you wish to store all have the same size, use [crate::qmdb::any::unordered::fixed] //! instead for better performance._ use crate::{ index::unordered::Index, journal::contiguous::variable::Journal, merkle::{Family, Location}, qmdb::{ any::{unordered, value::VariableEncoding, VariableConfig, VariableValue}, operation::Key, Error, }, translator::Translator, Context, }; use commonware_codec::{Codec, Read}; use commonware_cryptography::Hasher; use commonware_parallel::Strategy; pub type Update = unordered::Update>; pub type Operation = unordered::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 Db where Operation: Codec, { /// 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. pub async fn init( context: E, cfg: VariableConfig as Read>::Cfg, S>, ) -> Result> { crate::qmdb::any::init(context, cfg).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) pub mod partitioned { pub use super::{Operation, Update}; use crate::{ index::partitioned::unordered::Index, journal::contiguous::variable::Journal, merkle::{Family, Location}, qmdb::{ any::{VariableConfig, VariableValue}, operation::Key, Error, }, translator::Translator, Context, }; use commonware_codec::{Codec, Read}; use commonware_cryptography::Hasher; use commonware_parallel::Strategy; /// A 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::unordered::Db< F, E, Journal>, Index, P>, H, Update, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, S, >; impl< F: Family, E: Context, K: Key, V: VariableValue, H: Hasher, T: Translator, const P: usize, S: Strategy, > Db where Operation: Codec, { /// 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. pub async fn init( context: E, cfg: VariableConfig as Read>::Cfg, S>, ) -> Result> { crate::qmdb::any::init(context, cfg).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::{index::Unordered as _, mmr, translator::TwoCap}; use commonware_cryptography::{sha256::Digest, Sha256}; use commonware_macros::test_traced; use commonware_math::algebra::Random; use commonware_parallel::Sequential; use commonware_runtime::{ buffer::paged::CacheRef, deterministic::{self, Context}, BufferPooler, Runner as _, Supervisor as _, }; use commonware_utils::{NZUsize, TestRng, NZU16, NZU64}; use rand::Rng; use std::{ num::{NonZeroU16, NonZeroUsize}, sync::Arc, }; const PAGE_SIZE: NonZeroU16 = NZU16!(77); const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9); pub(crate) fn create_test_config(seed: u64, pooler: &impl BufferPooler) -> VarConfig { let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE); VariableConfig { merkle_config: crate::mmr::full::Config { journal_partition: format!("journal-{seed}"), metadata_partition: format!("metadata-{seed}"), items_per_blob: NZU64!(13), write_buffer: NZUsize!(1024), strategy: Sequential, page_cache: page_cache.clone(), }, journal_config: crate::journal::contiguous::variable::Config { partition: format!("log-journal-{seed}"), items_per_section: NZU64!(7), write_buffer: NZUsize!(1024), compression: None, codec_config: ((), ((0..=10000).into(), ())), page_cache, }, translator: TwoCap, init_cache_size: Some(NZUsize!(1024)), } } pub(crate) type VarConfig = VariableConfig, ())), Sequential>; /// A type alias for the concrete [Db] type used in these unit tests. pub(crate) type AnyTest = Db, Sha256, TwoCap, Sequential>; /// 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, &context); AnyTest::init(context, config).await.unwrap() } /// 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 { let key = Digest::random(&mut rng); if i % 10 == 0 && i > 0 { ops.push(unordered::Operation::Delete(prev_key)); } else { let value = to_bytes(rng.next_u64()); ops.push(unordered::Operation::Update(unordered::Update(key, value))); prev_key = key; } } ops } /// Applies the given operations to the database. pub(crate) async fn apply_ops( db: &mut AnyTest, ops: Vec>>>, ) { let mut batch = db.new_batch(); for op in ops { match op { unordered::Operation::Update(unordered::Update(key, value)) => { batch = batch.write(key, Some(value)); } unordered::Operation::Delete(key) => { batch = batch.write(key, None); } unordered::Operation::CommitFloor(_, _) => { panic!("CommitFloor not supported in apply_ops"); } } } let merkleized = batch.merkleize(db, None).await.unwrap(); db.apply_batch(merkleized).await.unwrap(); } /// The staged path (`stage` + `Staged::merkleize`) must produce the same values and root as an /// explicit `get_many` + `write` + `merkleize` for variable-encoded values, across updates, /// a delete, upserts, a duplicate read slot, and a missing key. Guards the staged /// cached-location reuse against a fixed-vs-variable op-encoding divergence. #[test_traced("WARN")] fn unordered_variable_staged_matches_explicit_writes() { deterministic::Runner::default().start(|context| async move { let mut db = create_test_db(context.child("staged")).await; let key = |i: u64| Sha256::hash(&i.to_be_bytes()); let mut seed = db.new_batch(); for i in 0..200u64 { seed = seed.write(key(i), Some(to_bytes(i))); } let seed = seed.merkleize(&db, None).await.unwrap(); db.apply_batch(seed).await.unwrap(); db.commit().await.unwrap(); // Read set: key(5) duplicated at slots 0/3, read-only key(6), missing key(9000), // key(20) deleted via index at slot 4. let read_keys = [key(5), key(6), key(9000), key(5), key(20)]; let keys: Vec<&Digest> = read_keys.iter().collect(); let indexed_updates = vec![ (0, Some(to_bytes(5_000))), (2, Some(to_bytes(5_001))), (3, Some(to_bytes(5_002))), (4, None), ]; let upserts = vec![ (key(7000), Some(to_bytes(6_000))), (key(30), Some(to_bytes(6_001))), (key(31), None), ]; let mut explicit = db.new_batch(); let explicit_values = explicit.get_many(&keys, &db).await.unwrap(); for (slot, value) in &indexed_updates { explicit = explicit.write(read_keys[*slot], value.clone()); } for (k, v) in &upserts { explicit = explicit.write(*k, v.clone()); } let explicit_root = explicit.merkleize(&db, None).await.unwrap().root(); let (staged_values, staged) = db.new_batch().stage(&keys, &db).await.unwrap(); let staged_root = staged .merkleize(indexed_updates.clone(), upserts.clone(), None, &db) .await .unwrap() .root(); assert_eq!(explicit_values, staged_values); assert_eq!(explicit_root, staged_root); db.destroy().await.unwrap(); }); } /// A staged read that resolved in a grandparent's diff must survive that grandparent /// committing and being freed before `Staged::merkleize`. The recorded base is one /// transition older than the resolved location (it is the committed location the /// grandparent's own write superseded), and the grandparent's apply performs that /// transition, making the resolved location the key's committed one. Merkleize must /// therefore supersede the resolved location itself (see `StagedLoc`). Trusting the /// recorded base instead fails at apply: a rewritten key's snapshot location no longer /// matches the stale base (a panic), and a created key's `None` base emits a second /// create that leaks the migrated location's active bit. #[test_traced("WARN")] fn unordered_variable_staged_ancestor_commit_before_merkleize() { deterministic::Runner::default().start(|context| async move { let mut 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).await.unwrap(); db.apply_batch(seed).await.unwrap(); db.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) .await .unwrap(); let parent = grandparent .new_batch::() .write(key(1), Some(to_bytes(1_002))) .merkleize(&db, None) .await .unwrap(); 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))]); // Commit and free the grandparent: the staged resolutions' locations migrate // into the committed region, retiring their recorded bases. db.apply_batch(grandparent).await.unwrap(); let updates = vec![(0, Some(to_bytes(2_000))), (1, Some(to_bytes(2_001)))]; let staged = staged .merkleize(updates, Vec::new(), None, &db) .await .unwrap(); // The explicit path over the same post-commit 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) .await .unwrap() .root(); assert_eq!(staged.root(), explicit_root); db.apply_batch(parent).await.unwrap(); db.apply_batch(staged).await.unwrap(); 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 uncommitted ancestor's diff keeps its recorded /// base while that ancestor is alive at `Staged::merkleize` (see `StagedLoc`), and the /// full lifecycle must hold under both apply shapes: one `apply_batch` call committing /// the still-pending ancestors together 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. #[test_traced("WARN")] fn unordered_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 mut db = create_test_db(context.child(label)).await; let key = |i: u64| Sha256::hash(&i.to_be_bytes()); // 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`. 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).await.unwrap(); db.apply_batch(seed).await.unwrap(); db.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) .await .unwrap(); let parent = grandparent .new_batch::() .write(key(1), Some(to_bytes(1_002))) .merkleize(&db, None) .await .unwrap(); // key(3) is untouched by the ancestors (their floor raises 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)), ] ); // Merkleize with every ancestor still alive: the ancestor-resolved slots // trust their recorded bases. Slot 2 is a staged delete. let updates = vec![ (0, Some(to_bytes(2_000))), (1, Some(to_bytes(2_001))), (2, None), (3, Some(to_bytes(2_002))), ]; let staged = staged .merkleize(updates, Vec::new(), None, &db) .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) .await .unwrap() .root(); assert_eq!(staged.root(), explicit_root); if apply_ancestors_first { db.apply_batch(grandparent).await.unwrap(); db.apply_batch(parent).await.unwrap(); } db.apply_batch(staged).await.unwrap(); 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(); } }); } /// Return an `Any` database initialized with a fixed config. async fn open_db(context: deterministic::Context) -> AnyTest { let cfg = create_test_config(0, &context); AnyTest::init(context, cfg).await.unwrap() } #[test_traced("WARN")] pub fn test_any_variable_db_build_and_authenticate() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let db = open_db(context.child("storage")).await; crate::qmdb::any::test::test_any_db_build_and_authenticate( context, db, |ctx| Box::pin(open_db(ctx)), to_bytes, ) .await; }); } #[test_traced("WARN")] pub fn test_any_variable_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 = open_db(context.child("open").with_attribute("index", 1)).await; let root = db.root(); // Build a batch but don't apply it (simulate failure before commit). { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.write( Sha256::hash(&i.to_be_bytes()), Some(vec![(i % 255) as u8; ((i % 13) + 7) as usize]), ); } let _ = batch.merkleize(&db, None).await.unwrap(); } // Simulate a failure and test that we rollback to the previous root. drop(db); let mut db = open_db(context.child("open").with_attribute("index", 2)).await; assert_eq!(root, db.root()); // Re-apply the updates and commit them this time. let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&i.to_be_bytes()); let v = vec![(i % 255) as u8; ((i % 13) + 7) as usize]; batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None).await.unwrap(); db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap(); let root = db.root(); // Update every 3rd key but don't apply (simulate failure). { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { if i % 3 != 0 { continue; } let k = Sha256::hash(&i.to_be_bytes()); let v = vec![((i + 1) % 255) as u8; ((i % 13) + 8) as usize]; batch = batch.write(k, Some(v)); } let _ = batch.merkleize(&db, None).await.unwrap(); } // Simulate a failure and test that we rollback to the previous root. drop(db); let mut db = open_db(context.child("open").with_attribute("index", 3)).await; assert_eq!(root, db.root()); // Re-apply updates for every 3rd key and commit them this time. let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { if i % 3 != 0 { continue; } let k = Sha256::hash(&i.to_be_bytes()); let v = vec![((i + 1) % 255) as u8; ((i % 13) + 8) as usize]; batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None).await.unwrap(); db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap(); let root = db.root(); // Delete every 7th key but don't apply (simulate failure). { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { if i % 7 != 1 { continue; } let k = Sha256::hash(&i.to_be_bytes()); batch = batch.write(k, None); } let _ = batch.merkleize(&db, None).await.unwrap(); } // Simulate a failure and test that we rollback to the previous root. drop(db); let mut db = open_db(context.child("open").with_attribute("index", 4)).await; assert_eq!(root, db.root()); // Re-delete every 7th key and commit this time. let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { if i % 7 != 1 { continue; } let k = Sha256::hash(&i.to_be_bytes()); batch = batch.write(k, None); } let merkleized = batch.merkleize(&db, None).await.unwrap(); db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap(); let root = db.root(); let inactivity_floor = db.inactivity_floor_loc(); db.sync().await.unwrap(); // test pruning boundary after sync w/ prune db.prune(inactivity_floor).await.unwrap(); let bounds = db.bounds(); let snapshot_items = db.snapshot.items(); db.sync().await.unwrap(); drop(db); // Confirm state is preserved after reopen. let db = open_db(context.child("open").with_attribute("index", 5)).await; assert_eq!(root, db.root()); assert_eq!(db.bounds(), bounds); assert_eq!(db.inactivity_floor_loc(), inactivity_floor); assert_eq!(db.snapshot.items(), snapshot_items); db.destroy().await.unwrap(); }); } #[test_traced] fn test_any_variable_db_prune_beyond_inactivity_floor() { let executor = deterministic::Runner::default(); executor.start(|mut context| async move { let mut db = open_db(context.child("storage")).await; // Add some operations let key1 = Digest::random(&mut context); let key2 = Digest::random(&mut context); let key3 = Digest::random(&mut context); let merkleized = db .new_batch() .write(key1, Some(vec![10])) .write(key2, Some(vec![20])) .write(key3, Some(vec![30])) .merkleize(&db, None) .await .unwrap(); db.apply_batch(merkleized).await.unwrap(); // inactivity_floor should be at some location < op_count let inactivity_floor = db.inactivity_floor_loc(); let beyond_floor = Location::new(*inactivity_floor + 1); // Try to prune beyond the inactivity floor let result = db.prune(beyond_floor).await; assert!( matches!(result, Err(Error::PruneBeyondMinRequired(loc, floor)) if loc == beyond_floor && floor == inactivity_floor) ); db.destroy().await.unwrap(); }); } #[test_traced] fn test_stale_batch_rejected() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = open_db(context.child("storage")).await; let key1 = Sha256::hash(&[1]); let key2 = Sha256::hash(&[2]); // Create two batches from the same DB state. let batch_a = db .new_batch() .write(key1, Some(vec![10])) .merkleize(&db, None) .await .unwrap(); let batch_b = db .new_batch() .write(key2, Some(vec![20])) .merkleize(&db, None) .await .unwrap(); // Apply the first -- should succeed. db.apply_batch(batch_a).await.unwrap(); let expected_root = db.root(); let expected_bounds = db.bounds(); assert_eq!(db.get(&key1).await.unwrap(), Some(vec![10])); assert_eq!(db.get(&key2).await.unwrap(), None); // Apply the second -- should fail because the DB was modified. let result = db.apply_batch(batch_b).await; assert!( matches!(result, Err(Error::StaleBatch { .. })), "expected StaleBatch error, got {result:?}" ); assert_eq!(db.root(), expected_root); assert_eq!(db.bounds(), expected_bounds); assert_eq!(db.get(&key1).await.unwrap(), Some(vec![10])); assert_eq!(db.get(&key2).await.unwrap(), None); db.destroy().await.unwrap(); }); } /// Sibling batches with different operation counts are still detected /// as stale. #[test_traced] fn test_stale_batch_rejected_different_sizes() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = open_db(context.child("storage")).await; // A writes 1 key, B writes 5 keys -- different total_size. let batch_a = db .new_batch() .write(Sha256::hash(&[1]), Some(vec![10])) .merkleize(&db, None) .await .unwrap(); let batch_b = db .new_batch() .write(Sha256::hash(&[2]), Some(vec![20])) .write(Sha256::hash(&[3]), Some(vec![30])) .write(Sha256::hash(&[4]), Some(vec![40])) .write(Sha256::hash(&[5]), Some(vec![50])) .write(Sha256::hash(&[6]), Some(vec![60])) .merkleize(&db, None) .await .unwrap(); // B has more ops than A. assert!(batch_b.bounds.total_size > batch_a.bounds.total_size); // Apply A, then B must be stale. db.apply_batch(batch_a).await.unwrap(); let result = db.apply_batch(batch_b).await; assert!( matches!(result, Err(Error::StaleBatch { .. })), "expected StaleBatch for asymmetric sibling, got {result:?}" ); db.destroy().await.unwrap(); }); } /// Applying C (grandchild of A) after only A is committed must /// apply B's data + C's data. Uncommitted ancestor B's snapshot /// entries are applied via ancestor_diffs with committed_locs override. #[test_traced] fn test_partial_ancestor_commit() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = open_db(context.child("storage")).await; let key1 = Sha256::hash(&[1]); let key2 = Sha256::hash(&[2]); let key3 = Sha256::hash(&[3]); // Chain: DB <- A <- B <- C let a = db .new_batch() .write(key1, Some(vec![10])) .merkleize(&db, None) .await .unwrap(); let b = a .new_batch::() .write(key2, Some(vec![20])) .merkleize(&db, None) .await .unwrap(); let c = b .new_batch::() .write(key3, Some(vec![30])) .merkleize(&db, None) .await .unwrap(); let expected_root = c.root(); // Apply only A, then apply C directly (B uncommitted). db.apply_batch(a).await.unwrap(); db.apply_batch(c).await.unwrap(); assert_eq!(db.root(), expected_root); assert_eq!(db.get(&key1).await.unwrap(), Some(vec![10])); assert_eq!(db.get(&key2).await.unwrap(), Some(vec![20])); assert_eq!(db.get(&key3).await.unwrap(), Some(vec![30])); db.destroy().await.unwrap(); }); } #[test_traced] fn test_stale_batch_chained() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = open_db(context.child("storage")).await; let key1 = Sha256::hash(&[1]); let key2 = Sha256::hash(&[2]); let key3 = Sha256::hash(&[3]); // Commit initial state. let merkleized = db .new_batch() .write(key1, Some(vec![10])) .merkleize(&db, None) .await .unwrap(); db.apply_batch(merkleized).await.unwrap(); // Create a parent batch, then fork two children. let parent = db .new_batch() .write(key2, Some(vec![20])) .merkleize(&db, None) .await .unwrap(); let child_a = parent .new_batch::() .write(key3, Some(vec![30])) .merkleize(&db, None) .await .unwrap(); let child_b = parent .new_batch::() .write(key3, Some(vec![40])) .merkleize(&db, None) .await .unwrap(); // Apply child_a, then child_b should be stale. db.apply_batch(child_a).await.unwrap(); let result = db.apply_batch(child_b).await; assert!( matches!(result, Err(Error::StaleBatch { .. })), "expected StaleBatch error for sibling, got {result:?}" ); db.destroy().await.unwrap(); }); } /// Apply parent then child -- this is the sequential commit pattern /// and must succeed. `apply_batch` detects that the child's ancestors /// were committed and applies only the child's own operations. #[test_traced] fn test_sequential_commit_parent_then_child() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = open_db(context.child("storage")).await; let key1 = Sha256::hash(&[1]); let key2 = Sha256::hash(&[2]); // Create parent, then child. let parent = db .new_batch() .write(key1, Some(vec![10])) .merkleize(&db, None) .await .unwrap(); let child = parent .new_batch::() .write(key2, Some(vec![20])) .merkleize(&db, None) .await .unwrap(); // Apply parent first, then child -- sequential commit. db.apply_batch(parent).await.unwrap(); db.apply_batch(child).await.unwrap(); assert_eq!(db.get(&key1).await.unwrap(), Some(vec![10])); assert_eq!(db.get(&key2).await.unwrap(), Some(vec![20])); db.destroy().await.unwrap(); }); } #[test_traced] fn test_stale_batch_child_applied_before_parent() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = open_db(context.child("storage")).await; let key1 = Sha256::hash(&[1]); let key2 = Sha256::hash(&[2]); // Create parent, then child. let parent = db .new_batch() .write(key1, Some(vec![10])) .merkleize(&db, None) .await .unwrap(); let child = parent .new_batch::() .write(key2, Some(vec![20])) .merkleize(&db, None) .await .unwrap(); // Apply child first -- parent should now be stale. db.apply_batch(child).await.unwrap(); let result = db.apply_batch(parent).await; assert!( matches!(result, Err(Error::StaleBatch { .. })), "expected StaleBatch for parent after child applied, got {result:?}" ); db.destroy().await.unwrap(); }); } // FromSyncTestable implementation for from_sync_result tests mod from_sync_testable { use super::*; use crate::{ merkle::mmr::{self, full::Mmr}, qmdb::any::sync::tests::FromSyncTestable, }; use futures::future::join_all; type TestMmr = Mmr; impl FromSyncTestable for AnyTest { type Merkle = TestMmr; fn into_log_components(self) -> (Self::Merkle, Self::Journal) { (self.log.merkle, self.log.journal) } async fn pinned_nodes_at(&self, loc: mmr::Location) -> Vec { join_all(mmr::Family::nodes_to_pin(loc).map(|p| self.log.merkle.get_node(p))) .await .into_iter() .map(|n| n.unwrap().unwrap()) .collect() } } } /// Regression test for https://github.com/commonwarexyz/monorepo/issues/2787 #[allow(dead_code, clippy::manual_async_fn)] fn issue_2787_regression( db: &crate::qmdb::immutable::variable::Db< mmr::Family, deterministic::Context, Digest, Vec, Sha256, TwoCap, Sequential, >, key: Digest, ) -> impl std::future::Future + Send + use<'_> { async move { let _ = db.get(&key).await; } } fn is_send(_: T) {} #[allow(dead_code)] fn assert_non_trait_futures_are_send(db: &AnyTest, key: Digest, value: Vec) { let batch = db.new_batch().write(key, Some(value)); is_send(batch.merkleize(db, None)); is_send(db.get_with_loc(&key)); } /// Owned batch root matches the borrow-based batch root. #[test_traced("WARN")] fn test_owned_batch_root_matches() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; // Apply some initial data. apply_ops(&mut db, create_test_ops(20)).await; db.commit().await.unwrap(); // Build an owned batch from committed state. let base = db.to_batch(); // Create a child batch via owned and via borrow-based API. Same ops. let ops = create_test_ops_seeded(10, 99); // Borrow-based path. let mut batch = db.new_batch(); for op in &ops { match op { unordered::Operation::Update(unordered::Update(k, v)) => { batch = batch.write(*k, Some(v.clone())); } unordered::Operation::Delete(k) => { batch = batch.write(*k, None); } _ => unreachable!(), } } let borrow_root = batch.merkleize(&db, None).await.unwrap().root(); // Owned batch path. let mut batch = base.new_batch::(); for op in &ops { match op { unordered::Operation::Update(unordered::Update(k, v)) => { batch = batch.write(*k, Some(v.clone())); } unordered::Operation::Delete(k) => { batch = batch.write(*k, None); } _ => unreachable!(), } } let batch_root = batch.merkleize(&db, None).await.unwrap().root(); assert_eq!(borrow_root, batch_root); db.destroy().await.unwrap(); }); } /// Owned batch can be merkleized and applied to the database. #[test_traced("WARN")] fn test_owned_batch_apply() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; // Apply initial data. apply_ops(&mut db, create_test_ops(20)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Build a child batch via owned API, merkleize, and apply. let key = Digest::random(commonware_utils::TestRng::new(200)); let value = vec![42u8; 16]; let child_batch = base .new_batch::() .write(key, Some(value.clone())) .merkleize(&db, None) .await .unwrap(); // Apply the batch. db.apply_batch(child_batch).await.unwrap(); db.commit().await.unwrap(); // Verify the key was written. let fetched = db.get(&key).await.unwrap(); assert_eq!(fetched.unwrap(), value); db.destroy().await.unwrap(); }); } /// Batch chains: parent batch committed, child applied sequentially. #[test_traced("WARN")] fn test_owned_batch_chain_commit_parent_first() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; // Build initial data. apply_ops(&mut db, create_test_ops(10)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Parent batch (via owned API). let key_a = Digest::random(commonware_utils::TestRng::new(300)); let val_a = vec![1u8; 10]; let parent_batch = base .new_batch::() .write(key_a, Some(val_a.clone())) .merkleize(&db, None) .await .unwrap(); // Child batch (built on parent batch). let key_b = Digest::random(commonware_utils::TestRng::new(301)); let val_b = vec![2u8; 10]; let child_batch = parent_batch .new_batch::() .write(key_b, Some(val_b.clone())) .merkleize(&db, None) .await .unwrap(); db.apply_batch(parent_batch).await.unwrap(); db.commit().await.unwrap(); // Commit child. db.apply_batch(child_batch).await.unwrap(); 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(); }); } /// Multiple forks from the same batch. #[test_traced("WARN")] fn test_owned_batch_multiple_forks() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; apply_ops(&mut db, create_test_ops(10)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Fork A. let key_a = Digest::random(commonware_utils::TestRng::new(400)); let fork_a = base .new_batch::() .write(key_a, Some(vec![10u8; 8])) .merkleize(&db, None) .await .unwrap(); // Fork B (different key, same parent). let key_b = Digest::random(commonware_utils::TestRng::new(401)); let fork_b = base .new_batch::() .write(key_b, Some(vec![20u8; 8])) .merkleize(&db, None) .await .unwrap(); // Roots differ. assert_ne!(fork_a.root(), fork_b.root()); // Apply fork A. db.apply_batch(fork_a).await.unwrap(); db.commit().await.unwrap(); assert_eq!(db.get(&key_a).await.unwrap().unwrap(), vec![10u8; 8]); assert!(db.get(&key_b).await.unwrap().is_none()); db.destroy().await.unwrap(); }); } /// Batches can be stored in a homogeneous collection. #[test_traced("WARN")] fn test_owned_batch_homogeneous_collection() { use crate::qmdb::any::batch::MerkleizedBatch; use commonware_cryptography::sha256; use std::collections::HashMap; type Snap = MerkleizedBatch< mmr::Family, sha256::Digest, super::Update>, Sequential, >; let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; apply_ops(&mut db, create_test_ops(10)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Build several batches at different depths and store them by root. let mut collection: HashMap> = HashMap::new(); // Depth 1. let key = Digest::random(commonware_utils::TestRng::new(500)); let batch1 = base .new_batch::() .write(key, Some(vec![1u8; 8])) .merkleize(&db, None) .await .unwrap(); collection.insert(batch1.root(), batch1); // Depth 2 (retrieve batch1 from collection, build child). let batch1_root = *collection.keys().next().unwrap(); let batch1_ref = collection.get(&batch1_root).unwrap(); let key = Digest::random(commonware_utils::TestRng::new(501)); let batch2 = batch1_ref .new_batch::() .write(key, Some(vec![2u8; 8])) .merkleize(&db, None) .await .unwrap(); collection.insert(batch2.root(), batch2); // All batches in the same HashMap -- type erasure works. assert_eq!(collection.len(), 2); db.destroy().await.unwrap(); }); } /// Batch chains: parent inserts key, child deletes it. #[test_traced("WARN")] fn test_owned_batch_chain_delete_after_ancestor_insert() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; apply_ops(&mut db, create_test_ops(5)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Parent batch: insert key_x. let key_x = Digest::random(commonware_utils::TestRng::new(700)); let val_a = vec![10u8; 8]; let parent_batch = base .new_batch::() .write(key_x, Some(val_a.clone())) .merkleize(&db, None) .await .unwrap(); // Child batch: delete key_x. let child_batch = parent_batch .new_batch::() .write(key_x, None) .merkleize(&db, None) .await .unwrap(); db.apply_batch(parent_batch).await.unwrap(); db.commit().await.unwrap(); assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_a); // Commit child. db.apply_batch(child_batch).await.unwrap(); db.commit().await.unwrap(); // key_x should be deleted. assert!(db.get(&key_x).await.unwrap().is_none()); db.destroy().await.unwrap(); }); } /// Batch chains: parent and child both modify the same key. #[test_traced("WARN")] fn test_owned_batch_chain_overlapping_keys() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; // Build initial data. apply_ops(&mut db, create_test_ops(5)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Parent batch: insert key_x with value_a. let key_x = Digest::random(commonware_utils::TestRng::new(600)); let val_a = vec![10u8; 8]; let parent_batch = base .new_batch::() .write(key_x, Some(val_a.clone())) .merkleize(&db, None) .await .unwrap(); // Child batch: update key_x to value_b (overlapping key). let val_b = vec![20u8; 8]; let child_batch = parent_batch .new_batch::() .write(key_x, Some(val_b.clone())) .merkleize(&db, None) .await .unwrap(); db.apply_batch(parent_batch).await.unwrap(); db.commit().await.unwrap(); // key_x should have parent's value. assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_a); // Commit child. db.apply_batch(child_batch).await.unwrap(); db.commit().await.unwrap(); // key_x should now have child's value. assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_b); db.destroy().await.unwrap(); }); } /// Three-deep batch chain: grandparent -> parent -> child. /// Commit each layer sequentially. #[test_traced("WARN")] fn test_owned_batch_chain_three_deep() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; apply_ops(&mut db, create_test_ops(10)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Grandparent: insert key_a. let key_a = Digest::random(commonware_utils::TestRng::new(900)); let val_a = vec![1u8; 10]; let grandparent_batch = base .new_batch::() .write(key_a, Some(val_a.clone())) .merkleize(&db, None) .await .unwrap(); // Parent: insert key_b. let key_b = Digest::random(commonware_utils::TestRng::new(901)); let val_b = vec![2u8; 10]; let parent_batch = grandparent_batch .new_batch::() .write(key_b, Some(val_b.clone())) .merkleize(&db, None) .await .unwrap(); // Child: insert key_c. let key_c = Digest::random(commonware_utils::TestRng::new(902)); let val_c = vec![3u8; 10]; let child_batch = parent_batch .new_batch::() .write(key_c, Some(val_c.clone())) .merkleize(&db, None) .await .unwrap(); db.apply_batch(grandparent_batch).await.unwrap(); db.commit().await.unwrap(); assert_eq!(db.get(&key_a).await.unwrap().unwrap(), val_a); // Commit parent. db.apply_batch(parent_batch).await.unwrap(); db.commit().await.unwrap(); assert_eq!(db.get(&key_b).await.unwrap().unwrap(), val_b); // Commit child. db.apply_batch(child_batch).await.unwrap(); db.commit().await.unwrap(); assert_eq!(db.get(&key_c).await.unwrap().unwrap(), val_c); // All three keys readable. assert_eq!(db.get(&key_a).await.unwrap().unwrap(), val_a); assert_eq!(db.get(&key_b).await.unwrap().unwrap(), val_b); assert_eq!(db.get(&key_c).await.unwrap().unwrap(), val_c); db.destroy().await.unwrap(); }); } /// Three-deep chain where each layer touches the same key. #[test_traced("WARN")] fn test_owned_batch_chain_three_deep_overlapping_key() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; apply_ops(&mut db, create_test_ops(5)).await; db.commit().await.unwrap(); let base = db.to_batch(); let key_x = Digest::random(commonware_utils::TestRng::new(910)); // Grandparent: insert key_x = val_a. let val_a = vec![10u8; 8]; let grandparent_batch = base .new_batch::() .write(key_x, Some(val_a.clone())) .merkleize(&db, None) .await .unwrap(); // Parent: update key_x = val_b. let val_b = vec![20u8; 8]; let parent_batch = grandparent_batch .new_batch::() .write(key_x, Some(val_b.clone())) .merkleize(&db, None) .await .unwrap(); // Child: delete key_x. let child_batch = parent_batch .new_batch::() .write(key_x, None) .merkleize(&db, None) .await .unwrap(); db.apply_batch(grandparent_batch).await.unwrap(); db.commit().await.unwrap(); assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_a); // Commit parent. db.apply_batch(parent_batch).await.unwrap(); db.commit().await.unwrap(); assert_eq!(db.get(&key_x).await.unwrap().unwrap(), val_b); // Commit child. db.apply_batch(child_batch).await.unwrap(); db.commit().await.unwrap(); assert!(db.get(&key_x).await.unwrap().is_none()); db.destroy().await.unwrap(); }); } /// After committing and dropping an ancestor, building a new child /// from a surviving descendant must not panic or return wrong data. /// Regression test: the Merkleizer's `read_op` fell into the /// "ancestor chain" region for operations that belonged to the freed /// ancestor, causing wrong indexing. #[test_traced("WARN")] fn test_new_child_after_ancestor_committed_and_dropped() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; apply_ops(&mut db, create_test_ops(5)).await; db.commit().await.unwrap(); // Chain: DB <-- a <-- b let key_a = Digest::random(commonware_utils::TestRng::new(800)); let val_a = vec![10u8; 8]; let a = db .new_batch() .write(key_a, Some(val_a.clone())) .merkleize(&db, None) .await .unwrap(); let key_b = Digest::random(commonware_utils::TestRng::new(801)); let val_b = vec![20u8; 8]; let b = a .new_batch::() .write(key_b, Some(val_b.clone())) .merkleize(&db, None) .await .unwrap(); // Commit a and drop it. b's Weak becomes invalid. db.apply_batch(a).await.unwrap(); db.commit().await.unwrap(); // Build c from b. This must not panic despite a being freed. let key_c = Digest::random(commonware_utils::TestRng::new(802)); let val_c = vec![30u8; 8]; let c = b .new_batch::() .write(key_c, Some(val_c.clone())) .merkleize(&db, None) .await .unwrap(); // Commit b (skip_ancestors path since a is committed). db.apply_batch(b).await.unwrap(); db.commit().await.unwrap(); // Commit c. db.apply_batch(c).await.unwrap(); db.commit().await.unwrap(); // All three keys present with correct values. assert_eq!(db.get(&key_a).await.unwrap().unwrap(), val_a); assert_eq!(db.get(&key_b).await.unwrap().unwrap(), val_b); assert_eq!(db.get(&key_c).await.unwrap().unwrap(), val_c); db.destroy().await.unwrap(); }); } /// Regression: applying a batch after its ancestor Arc is dropped (without /// committing) must still apply the ancestor's snapshot diffs. Before the /// fix, the Weak parent chain was dead and ancestor diffs were silently /// lost, causing the journal and snapshot to diverge. #[test_traced("WARN")] fn test_apply_batch_after_ancestor_dropped_without_commit() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = create_test_db(context).await; apply_ops(&mut db, create_test_ops(5)).await; db.commit().await.unwrap(); let base = db.to_batch(); // Chain: base <-- a <-- b <-- c let key_a = Digest::random(commonware_utils::TestRng::new(700)); let val_a = vec![1u8; 10]; let a = base .new_batch::() .write(key_a, Some(val_a.clone())) .merkleize(&db, None) .await .unwrap(); let key_b = Digest::random(commonware_utils::TestRng::new(701)); let val_b = vec![2u8; 10]; let b = a .new_batch::() .write(key_b, Some(val_b.clone())) .merkleize(&db, None) .await .unwrap(); let key_c = Digest::random(commonware_utils::TestRng::new(702)); let val_c = vec![3u8; 10]; let c = b .new_batch::() .write(key_c, Some(val_c.clone())) .merkleize(&db, None) .await .unwrap(); // Drop a and b without committing. Their Weak refs in c are now dead. drop(a); drop(b); // Apply only the tip. This is !skip_ancestors (db hasn't changed). // Before the fix, a's and b's snapshot diffs would be silently lost. db.apply_batch(c).await.unwrap(); db.commit().await.unwrap(); // All three keys must be in the snapshot. assert_eq!(db.get(&key_a).await.unwrap().unwrap(), val_a); assert_eq!(db.get(&key_b).await.unwrap().unwrap(), val_b); assert_eq!(db.get(&key_c).await.unwrap().unwrap(), val_c); db.destroy().await.unwrap(); }); } }