//! Sync tests for [`crate::qmdb::current`] databases. //! //! The harness contract and the shared sync tests live in [`crate::qmdb::sync::harness`]. This //! module implements the harness for `current` databases, runs a subset of the `any`-specific //! tests from [`crate::qmdb::any::sync::tests`], and adds `current`-specific tests. Sync targets //! the ops root, while the harness's canonical root is the root returned by `Db::root()`. use crate::qmdb::{ current::tests::{fixed_config, variable_config}, floor::Proportional, sync::{ Database as SyncDatabase, harness::{ConfigOf, SyncTestHarness}, }, }; use commonware_cryptography::{Sha256, sha256::Digest}; use commonware_macros::test_traced; use commonware_parallel::Sequential; use commonware_runtime::{Runner as _, Supervisor as _, deterministic, deterministic::Context}; use commonware_utils::{NZU64, NZUsize, non_empty_range}; use rand::Rng as _; mod harnesses { use super::*; use crate::merkle::{self, mmb, mmr}; use commonware_math::algebra::Random; use commonware_utils::TestRng; type OrderedFixedDb = crate::qmdb::current::ordered::fixed::Db< F, Context, Digest, Digest, Sha256, crate::translator::OneCap, 32, Sequential, >; type OrderedVariableDb = crate::qmdb::current::ordered::variable::Db< F, Context, Digest, Digest, Sha256, crate::translator::OneCap, 32, Sequential, >; type UnorderedFixedDb = crate::qmdb::current::unordered::fixed::Db< F, Context, Digest, Digest, Sha256, crate::translator::TwoCap, 32, Sequential, >; type UnorderedVariableDb = crate::qmdb::current::unordered::variable::Db< F, Context, Digest, Digest, Sha256, crate::translator::TwoCap, 32, Sequential, >; fn create_unordered_fixed_ops( n: usize, seed: u64, ) -> Vec> { use crate::qmdb::any::operation::{Operation, update::Unordered as Update}; 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(Operation::Delete(prev_key)); } else { let value = Digest::random(&mut rng); ops.push(Operation::Update(Update(key, value))); prev_key = key; } } ops } fn create_unordered_variable_ops( n: usize, seed: u64, ) -> Vec> { use crate::qmdb::any::operation::{Operation, update::Unordered as Update}; 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(Operation::Delete(prev_key)); } else { let value = Digest::random(&mut rng); ops.push(Operation::Update(Update(key, value))); prev_key = key; } } ops } fn create_ordered_fixed_ops( n: usize, seed: u64, ) -> Vec> { use crate::qmdb::any::operation::{Operation, update::Ordered as Update}; let mut rng = TestRng::new(seed); let mut ops = Vec::new(); for i in 0..n { if i % 10 == 0 && i > 0 { let key = Digest::random(&mut rng); ops.push(Operation::Delete(key)); } else { let key = Digest::random(&mut rng); let value = Digest::random(&mut rng); let next_key = Digest::random(&mut rng); ops.push(Operation::Update(Update { key, value, next_key, })); } } ops } fn create_ordered_variable_ops( n: usize, seed: u64, ) -> Vec> { use crate::qmdb::any::operation::{Operation, update::Ordered as Update}; let mut rng = TestRng::new(seed); let mut ops = Vec::new(); for i in 0..n { let key = Digest::random(&mut rng); if i % 10 == 0 && i > 0 { ops.push(Operation::Delete(key)); } else { let value = Digest::random(&mut rng); let next_key = Digest::random(&mut rng); ops.push(Operation::Update(Update { key, value, next_key, })); } } ops } async fn apply_unordered_fixed_ops( db: UnorderedFixedDb, ops: Vec>, metadata: Option, ) -> UnorderedFixedDb { use crate::qmdb::any::operation::{Operation, update::Unordered as Update}; let merkleized = { let mut batch = db.new_batch(); for op in ops { match op { Operation::Update(Update(key, value)) => { batch = batch.write(key, Some(value)); } Operation::Delete(key) => { batch = batch.write(key, None); } Operation::CommitFloor(_, _) => {} } } batch .merkleize(&db, metadata, &mut Proportional) .await .unwrap() }; let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap() } async fn apply_unordered_variable_ops( db: UnorderedVariableDb, ops: Vec>, metadata: Option, ) -> UnorderedVariableDb { use crate::qmdb::any::operation::{Operation, update::Unordered as Update}; let merkleized = { let mut batch = db.new_batch(); for op in ops { match op { Operation::Update(Update(key, value)) => { batch = batch.write(key, Some(value)); } Operation::Delete(key) => { batch = batch.write(key, None); } Operation::CommitFloor(_, _) => {} } } batch .merkleize(&db, metadata, &mut Proportional) .await .unwrap() }; let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap() } async fn apply_ordered_fixed_ops( db: OrderedFixedDb, ops: Vec>, metadata: Option, ) -> OrderedFixedDb { use crate::qmdb::any::operation::{Operation, update::Ordered as Update}; let merkleized = { let mut batch = db.new_batch(); for op in ops { match op { Operation::Update(Update { key, value, .. }) => { batch = batch.write(key, Some(value)); } Operation::Delete(key) => { batch = batch.write(key, None); } Operation::CommitFloor(_, _) => {} } } batch .merkleize(&db, metadata, &mut Proportional) .await .unwrap() }; let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap() } async fn apply_ordered_variable_ops( db: OrderedVariableDb, ops: Vec>, metadata: Option, ) -> OrderedVariableDb { use crate::qmdb::any::operation::{Operation, update::Ordered as Update}; let merkleized = { let mut batch = db.new_batch(); for op in ops { match op { Operation::Update(Update { key, value, .. }) => { batch = batch.write(key, Some(value)); } Operation::Delete(key) => { batch = batch.write(key, None); } Operation::CommitFloor(_, _) => {} } } batch .merkleize(&db, metadata, &mut Proportional) .await .unwrap() }; let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap() } pub struct UnorderedFixedHarness(std::marker::PhantomData); impl SyncTestHarness for UnorderedFixedHarness { type Family = F; type Db = UnorderedFixedDb; type Metadata = Digest; fn config(suffix: &str) -> ConfigOf { fixed_config::(suffix) } fn create_ops( n: usize, ) -> Vec> { create_unordered_fixed_ops::(n, 0) } fn create_ops_seeded( n: usize, seed: u64, ) -> Vec> { create_unordered_fixed_ops::(n, seed) } fn sample_metadata() -> Digest { Digest::from([1; 32]) } async fn init_db(ctx: Context) -> Self::Db { let cfg = fixed_config::("default"); Self::Db::init(ctx, cfg, None).await.unwrap() } async fn apply_ops( db: Self::Db, ops: Vec>, metadata: Option, ) -> Self::Db { apply_unordered_fixed_ops(db, ops, metadata).await } crate::qmdb::any::sync::tests::db_any_harness_methods!(); } pub type UnorderedFixedMmrHarness = UnorderedFixedHarness; pub type UnorderedFixedMmbHarness = UnorderedFixedHarness; pub struct UnorderedVariableHarness(std::marker::PhantomData); impl SyncTestHarness for UnorderedVariableHarness { type Family = F; type Db = UnorderedVariableDb; type Metadata = Digest; fn config(suffix: &str) -> ConfigOf { variable_config::(suffix) } fn create_ops( n: usize, ) -> Vec> { create_unordered_variable_ops::(n, 0) } fn create_ops_seeded( n: usize, seed: u64, ) -> Vec> { create_unordered_variable_ops::(n, seed) } fn sample_metadata() -> Digest { Digest::from([1; 32]) } async fn init_db(ctx: Context) -> Self::Db { let cfg = variable_config::("default"); Self::Db::init(ctx, cfg, None).await.unwrap() } async fn apply_ops( db: Self::Db, ops: Vec>, metadata: Option, ) -> Self::Db { apply_unordered_variable_ops(db, ops, metadata).await } crate::qmdb::any::sync::tests::db_any_harness_methods!(); } pub type UnorderedVariableMmrHarness = UnorderedVariableHarness; pub type UnorderedVariableMmbHarness = UnorderedVariableHarness; pub struct OrderedFixedHarness(std::marker::PhantomData); impl SyncTestHarness for OrderedFixedHarness { type Family = F; type Db = OrderedFixedDb; type Metadata = Digest; fn config(suffix: &str) -> ConfigOf { fixed_config::(suffix) } fn create_ops( n: usize, ) -> Vec> { create_ordered_fixed_ops::(n, 0) } fn create_ops_seeded( n: usize, seed: u64, ) -> Vec> { create_ordered_fixed_ops::(n, seed) } fn sample_metadata() -> Digest { Digest::from([1; 32]) } async fn init_db(ctx: Context) -> Self::Db { let cfg = fixed_config::("default"); Self::Db::init(ctx, cfg, None).await.unwrap() } async fn apply_ops( db: Self::Db, ops: Vec>, metadata: Option, ) -> Self::Db { apply_ordered_fixed_ops(db, ops, metadata).await } crate::qmdb::any::sync::tests::db_any_harness_methods!(); } pub type OrderedFixedMmrHarness = OrderedFixedHarness; pub type OrderedFixedMmbHarness = OrderedFixedHarness; pub struct OrderedVariableHarness(std::marker::PhantomData); impl SyncTestHarness for OrderedVariableHarness { type Family = F; type Db = OrderedVariableDb; type Metadata = Digest; fn config(suffix: &str) -> ConfigOf { variable_config::(suffix) } fn create_ops( n: usize, ) -> Vec> { create_ordered_variable_ops::(n, 0) } fn create_ops_seeded( n: usize, seed: u64, ) -> Vec> { create_ordered_variable_ops::(n, seed) } fn sample_metadata() -> Digest { Digest::from([1; 32]) } async fn init_db(ctx: Context) -> Self::Db { let cfg = variable_config::("default"); Self::Db::init(ctx, cfg, None).await.unwrap() } async fn apply_ops( db: Self::Db, ops: Vec>, metadata: Option, ) -> Self::Db { apply_ordered_variable_ops(db, ops, metadata).await } crate::qmdb::any::sync::tests::db_any_harness_methods!(); } pub type OrderedVariableMmrHarness = OrderedVariableHarness; pub type OrderedVariableMmbHarness = OrderedVariableHarness; } /// Regression test: sync a pruned MMB-backed current DB and verify the synced DB has the /// same canonical root, reopens cleanly, and returns the expected value. /// /// The target DB commits the same key 100 times, forcing the inactivity floor past a full /// 256-bit chunk boundary. The receiver derives `pruned_chunks` from `range.start` and must /// source the grafted pinned nodes for that region from the ops tree (zero-chunk identity). /// If those nodes do not match the sender's, the canonical roots diverge. #[test_traced("INFO")] fn test_current_mmb_sync_with_pruned_full_chunk_reopens() { let executor = deterministic::Runner::default(); executor.start(|mut context: Context| async move { type Db = crate::qmdb::current::unordered::variable::Db< crate::merkle::mmb::Family, Context, Digest, Digest, Sha256, crate::translator::TwoCap, 32, Sequential, >; const COMMITS: u64 = 100; let target_suffix = context.next_u64().to_string(); let target_context = context.child("target"); let mut target_db: Db = Db::init( target_context.child("target"), variable_config::(&target_suffix), None, ) .await .unwrap(); let key = Digest::from([7u8; 32]); let mut expected = None; for round in 0..COMMITS { expected = Some(Digest::from([round as u8; 32])); let merkleized = target_db .new_batch() .write(key, expected) .merkleize(&target_db, None, &mut Proportional) .await .unwrap(); target_db = target_db.apply_batch(merkleized).await.unwrap(); target_db = target_db.commit().await.unwrap(); } assert!( *target_db.inactivity_floor_loc() >= 256, "expected inactivity floor past chunk 0" ); let boundary = target_db.sync_boundary(); let target_db = target_db.prune(boundary).await.unwrap(); let sync_root = SyncDatabase::root(&target_db); let verification_root = target_db.root(); let lower_bound = target_db.sync_boundary(); let upper_bound = target_db.bounds().end; let client_suffix = context.next_u64().to_string(); let client_config = variable_config::(&client_suffix); let target_db = std::sync::Arc::new(target_db); // Sync targets the ops root, which is what the engine verifies. `build_db` reconstructs // the canonical root without authenticating it, so the assertions below compare it // against `verification_root`. let synced_db: Db = crate::qmdb::sync::sync(crate::qmdb::sync::engine::Config { context: context.child("client"), db_config: client_config.clone(), fetch_batch_size: commonware_utils::NZU64!(64), target: crate::qmdb::sync::Target { root: sync_root, range: commonware_utils::non_empty_range!(lower_bound, upper_bound), }, source: target_db.clone(), apply_batch_size: NZU64!(1024), max_outstanding_requests: NZUsize!(4), update_rx: None, finish_rx: None, reached_target_tx: None, }) .await .unwrap(); assert_eq!(SyncDatabase::root(&synced_db), sync_root); assert_eq!(synced_db.root(), verification_root); assert_eq!(synced_db.sync_boundary(), lower_bound); assert_eq!(synced_db.get(&key).await.unwrap(), expected); drop(synced_db); let reopened: Db = Db::init(context.child("reopened"), client_config, None) .await .unwrap(); assert_eq!(SyncDatabase::root(&reopened), sync_root); assert_eq!(reopened.root(), verification_root); assert_eq!(reopened.sync_boundary(), lower_bound); assert_eq!(reopened.get(&key).await.unwrap(), expected); reopened.destroy().await.unwrap(); std::sync::Arc::try_unwrap(target_db) .unwrap_or_else(|_| panic!("failed to unwrap Arc")) .destroy() .await .unwrap(); }); } #[test_traced] fn test_current_local_pinned_nodes_rejects_target_before_local_lower_bound() { type Db = crate::qmdb::current::unordered::variable::Db< crate::merkle::mmr::Family, Context, Digest, Digest, Sha256, crate::translator::TwoCap, 32, Sequential, >; let executor = deterministic::Runner::default(); executor.start(|mut context: Context| async move { let suffix = context.next_u64().to_string(); let config = variable_config::(&suffix); let mut db: Db = Db::init(context.child("db"), config.clone(), None) .await .unwrap(); let key = Digest::from([9u8; 32]); for round in 0..300u64 { let merkleized = db .new_batch() .write(key, Some(Digest::from([round as u8; 32]))) .merkleize(&db, None, &mut Proportional) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); db = db.commit().await.unwrap(); } let prune_loc = crate::merkle::Location::new(256); assert!(db.sync_boundary() >= prune_loc); let db = db.prune(prune_loc).await.unwrap(); let bounds = db.bounds(); let local_start = bounds.start; let local_end = bounds.end; let sync_root = SyncDatabase::root(&db); assert!(local_start > crate::merkle::Location::new(0)); // Reopen the operation journal independently to probe the persisted Merkle boundary. drop(db); let journal = <::Journal as crate::qmdb::sync::Journal< crate::merkle::mmr::Family, >>::new( context.child("journal"), crate::qmdb::sync::DatabaseConfig::journal_config(&config), non_empty_range!(local_start, local_end), ) .await .unwrap(); let stale_target = crate::qmdb::sync::Target { root: sync_root, range: non_empty_range!(local_start.checked_sub(1).unwrap(), local_end), }; assert!( ::local_pinned_nodes( context.child("probe_stale"), &config, &stale_target, &journal, ) .await .unwrap() .is_none() ); let matching_target = crate::qmdb::sync::Target { root: sync_root, range: non_empty_range!(local_start, local_end), }; assert!( ::local_pinned_nodes( context.child("probe_matching"), &config, &matching_target, &journal, ) .await .unwrap() .is_some() ); drop(journal); }); } /// Emits the `any`-specific sync tests that also run against `current` databases for `$harness`. macro_rules! current_sync_tests { ($harness:ty) => { #[test_traced("WARN")] fn test_sync_waits_for_explicit_finish() { crate::qmdb::any::sync::tests::test_sync_waits_for_explicit_finish::<$harness>(); } #[test_traced("WARN")] fn test_sync_handles_early_finish_signal() { crate::qmdb::any::sync::tests::test_sync_handles_early_finish_signal::<$harness>(); } #[test_traced("WARN")] fn test_sync_fails_when_finish_sender_dropped() { crate::qmdb::any::sync::tests::test_sync_fails_when_finish_sender_dropped::<$harness>(); } #[test_traced("WARN")] fn test_sync_allows_dropped_reached_target_receiver() { crate::qmdb::any::sync::tests::test_sync_allows_dropped_reached_target_receiver::< $harness, >(); } #[test_traced("WARN")] fn test_sync_post_sync_usability() { crate::qmdb::any::sync::tests::test_sync_post_sync_usability::<$harness>(); } #[test_traced("WARN")] fn test_local_pinned_nodes_below_floor() { crate::qmdb::any::sync::tests::test_local_pinned_nodes_below_floor::<$harness>(); } }; } crate::qmdb::sync::harness::sync_tests!( harnesses::UnorderedFixedMmrHarness, unordered_fixed_mmr, current_sync_tests ); crate::qmdb::sync::harness::sync_tests!( harnesses::UnorderedFixedMmbHarness, unordered_fixed_mmb, current_sync_tests ); crate::qmdb::sync::harness::sync_tests!( harnesses::UnorderedVariableMmrHarness, unordered_variable_mmr, current_sync_tests ); crate::qmdb::sync::harness::sync_tests!( harnesses::UnorderedVariableMmbHarness, unordered_variable_mmb, current_sync_tests ); crate::qmdb::sync::harness::sync_tests!( harnesses::OrderedFixedMmrHarness, ordered_fixed_mmr, current_sync_tests ); crate::qmdb::sync::harness::sync_tests!( harnesses::OrderedFixedMmbHarness, ordered_fixed_mmb, current_sync_tests ); crate::qmdb::sync::harness::sync_tests!( harnesses::OrderedVariableMmrHarness, ordered_variable_mmr, current_sync_tests ); crate::qmdb::sync::harness::sync_tests!( harnesses::OrderedVariableMmbHarness, ordered_variable_mmb, current_sync_tests );