//! An _Any_ authenticated database provides succinct proofs of any value ever associated with a //! key. //! //! The specific variants provided within this module include: //! - Unordered: The database does not maintain or require any ordering over the key space. //! - Fixed-size values //! - Variable-size values //! - Ordered: The database maintains a total order over active keys. //! - Fixed-size values //! - Variable-size values //! //! # Examples //! //! ```ignore //! // 1. Create a batch and apply it. //! let batch = db.new_batch() //! .write(key, Some(value)) // upsert //! .write(other_key, None) // delete //! .merkleize(&db, None, &mut Proportional).await?; //! let root = batch.root(); // speculative root //! let db = db.apply_batch(batch).await?.commit().await?; // flush to disk //! ``` //! //! ```ignore //! // 2. Fork two batches from the same parent. Apply one; the other is stale. //! let parent = db.new_batch().write(k1, Some(v1)) //! .merkleize(&db, None, &mut Proportional).await?; //! let fork_a = parent.new_batch::().write(k2, Some(v2)) //! .merkleize(&db, None, &mut Proportional).await?; //! let fork_b = parent.new_batch::().write(k3, Some(v3)) //! .merkleize(&db, None, &mut Proportional).await?; //! //! let db = db.apply_batch(fork_a).await?; // OK: includes parent //! assert!(db.validate_batch(&fork_b).is_err()); // StaleBatch; applying would consume the db //! ``` //! //! ```ignore //! // 3. Chain two batches. Apply parent first, then child. //! let parent = db.new_batch().write(k1, Some(v1)) //! .merkleize(&db, None, &mut Proportional).await?; //! let child = parent.new_batch::().write(k2, Some(v2)) //! .merkleize(&db, None, &mut Proportional).await?; //! //! let db = db.apply_batch(parent).await?; // apply parent //! let db = db.apply_batch(child).await?; // ancestors skipped automatically //! let db = db.commit().await?; //! ``` //! //! ```ignore //! // 4. Chain two batches. Apply child directly (includes parent's changes). //! let parent = db.new_batch().write(k1, Some(v1)) //! .merkleize(&db, None, &mut Proportional).await?; //! let child = parent.new_batch::().write(k2, Some(v2)) //! .merkleize(&db, None, &mut Proportional).await?; //! //! let db = db.apply_batch(child).await?; // OK: includes parent //! assert!(db.validate_batch(&parent).is_err()); // StaleBatch //! ``` //! //! ```ignore //! // 5. Two independent chains. Apply the tail of one; the other chain is stale. //! let a1 = db.new_batch().write(k1, Some(v1)) //! .merkleize(&db, None, &mut Proportional).await?; //! let a2 = a1.new_batch::().write(k2, Some(v2)) //! .merkleize(&db, None, &mut Proportional).await?; //! //! let b1 = db.new_batch().write(k3, Some(v3)) //! .merkleize(&db, None, &mut Proportional).await?; //! let b2 = b1.new_batch::().write(k4, Some(v4)) //! .merkleize(&db, None, &mut Proportional).await?; //! //! let db = db.apply_batch(a2).await?; // OK: includes a1 //! assert!(db.validate_batch(&b2).is_err()); // StaleBatch //! ``` //! //! ```ignore //! // 6. Advance the floor with a policy. `Key` and `Value` are the caller's types, `Value` has an //! // `expiry` height, and `now` is the caller's current height. This policy evicts expired //! // updates at the floor, records each evicted key and value, and stops at the first unexpired //! // update. Stopping at the first unexpired update leaves any later expired updates to a later //! // walk. It decides at most 8 updates and passes at most 16 inactive locations. //! struct Expire { //! now: u64, //! expired: Vec<(Key, Value)>, //! } //! impl Policy for Expire { //! fn evicts(&self) -> bool { //! true //! } //! fn limits(&self, _inactive: usize) -> Limits { //! Limits { entries: 8, skips: 16 } //! } //! fn decide<'a>(&mut self, entry: Entry<'a, F, Key, Value>) -> Decision<'a, Value> { //! if entry.value().expiry > self.now { //! return entry.stop(); //! } //! let key = entry.key().clone(); //! let (decision, value) = entry.evict(); //! self.expired.push((key, value)); //! decision //! } //! } //! let mut policy = Expire { now, expired: Vec::new() }; //! let batch = db.new_batch().write(k1, Some(v1)); //! let merkleized = batch.merkleize(&db, None, &mut policy).await?; //! let db = db.apply_batch(merkleized).await?; //! //! // Hold the floor at its inherited location. //! let batch = db.new_batch().write(k2, None); //! let merkleized = batch.merkleize(&db, None, &mut Hold).await?; //! let db = db.apply_batch(merkleized).await?; //! //! // Keep at most 4 updates and pass at most 8 inactive locations. //! let mut policy = Bounded { entries: 4, skips: 8 }; //! let batch = db.new_batch().write(k3, Some(v3)); //! let merkleized = batch.merkleize(&db, None, &mut policy).await?; //! let db = db.apply_batch(merkleized).await?; //! ``` use crate::{ Context, index::Factory as IndexFactory, journal::{ authenticated, contiguous::{fixed::Config as FConfig, variable::Config as VConfig}, }, merkle::{Family, Location, full::Config as MerkleConfig}, qmdb::{ Error as QmdbError, any::operation::{Operation, Update}, bitmap::Shared, metrics::Metrics, single_operation_root, }, translator::Translator, }; use commonware_codec::Codec; use commonware_cryptography::Hasher; use commonware_macros::boxed; use commonware_parallel::Strategy; use commonware_runtime::Spawner; use core::num::NonZeroUsize; use std::sync::Arc; use tracing::warn; pub mod batch; pub mod db; pub mod operation; #[cfg(any(test, feature = "test-traits"))] pub mod traits; pub mod value; pub use value::{FixedValue, ValueEncoding, VariableValue}; pub mod ordered; pub(crate) mod sync; pub mod unordered; /// Compute the authenticated root of a newly initialized database without opening storage. /// /// The initial commit never carries metadata, so this root always represents /// `CommitFloor(None, 0)`. pub fn initial_root() -> H::Digest where F: Family, H: Hasher, U: Update, Operation: Codec, { single_operation_root::(&Operation::::CommitFloor(None, Location::new(0))) } pub(crate) const BITMAP_CHUNK_BYTES: usize = 64; /// Configuration for an `Any` authenticated db. #[derive(Clone)] pub struct Config { /// Configuration for the Merkle structure backing the authenticated journal. pub merkle_config: MerkleConfig, /// Configuration for the operations log journal. pub journal_config: J, /// The translator used by the compressed index. Ordered databases require an /// [OrderedTranslator](crate::translator::OrderedTranslator). pub translator: T, /// Maximum number of entries in the `(location -> key)` cache used during init to resolve /// snapshot collisions without re-reading the log; `None` disables it. pub init_cache: Option, /// Size (in bytes) of the read buffer used to replay the log during init. pub init_buffer: NonZeroUsize, /// The index's snapshot-build concurrency (see [crate::qmdb::SnapshotBuild::Concurrency]): `()` /// for index types that build serially, and the number of build tasks for index types that /// build in parallel. A value of `1` builds the index entirely on the init task. Values of /// `2` and `3` decode on the init task and insert on one or two workers. Larger values split /// between spawned decode and insert tasks while the init task merely forwards batches. pub init_concurrency: B, } /// Configuration for an `Any` authenticated db with fixed-size values. pub type FixedConfig = Config; /// Configuration for an `Any` authenticated db with variable-sized values. pub type VariableConfig = Config, S, B>; /// Initialize an `Any` authenticated db from the given config. /// `Some(max_size)` selects the latest retained commit with at most `max_size` operations. /// `None` selects the latest retained state. /// /// The selected state is durable before return. Zero is invalid because the initial commit /// requires one operation. Subsequent appends may exceed this initialization bound. pub async fn init( context: E, cfg: Config>::Concurrency>, max_size: Option>, ) -> Result, QmdbError> where F: Family, E: Context + Spawner, U: Update, H: Hasher, I: IndexFactory> + crate::qmdb::SnapshotBuild, J: authenticated::Backing> + 'static, S: Strategy, Operation: Codec, { init_with_bitmap::(context, cfg, None, max_size, None) .await } /// Like [`init`] but accepts a pre-allocated bitmap (used by `current::Db`, which sizes pruned /// chunks from grafted metadata). `bitmap = None` allocates internally. #[boxed] pub(crate) async fn init_with_bitmap( context: E, cfg: Config>::Concurrency>, bitmap: Option>>, max_size: Option>, pair_absorption_threshold: Option, ) -> Result, QmdbError> where F: Family, E: Context + Spawner, U: Update, H: Hasher, I: IndexFactory> + crate::qmdb::SnapshotBuild, J: authenticated::Backing> + 'static, S: Strategy, Operation: Codec, { // Keep the selected commit unpublished until every variant-owned reconstruction constraint has // been checked against the same retained prefix. let pending = crate::qmdb::prepare_initialization::( context.child("log"), cfg.merkle_config, cfg.journal_config, max_size, ) .await?; // Snapshot replay requires both the selected commit and its floor to remain above the bitmap // boundary. Current also requires every absorbed chunk pair to exist at the selected size. let size = pending.bounds().end; if bitmap .as_ref() .is_some_and(|bitmap| size < bitmap.pruned_bits()) { return Err(QmdbError::HistoricalFloorPruned(Location::new(size))); } let floor = crate::qmdb::validate_initialization(&pending, true).await?; if let (Some(bitmap), Some(floor)) = (&bitmap, floor) && *floor < bitmap.pruned_bits() { return Err(QmdbError::HistoricalFloorPruned(Location::new(size))); } if pair_absorption_threshold.is_some_and(|minimum| size < minimum) { return Err(QmdbError::HistoricalFloorPruned(Location::new(size))); } // Publish the selected journal prefix only after its in-memory state is reconstructible. let mut log = pending.finish().await?; if log.size() == 0 { warn!("Authenticated log is empty, initializing new db"); let commit_floor = Operation::CommitFloor(None, Location::new(0)); (log, _) = log.append(&commit_floor).await?; log = log.sync().await?; } // Rebuild the volatile snapshot and bitmap from the selected commit's retained floor. let index = I::new(context.child("index"), cfg.translator); let snapshot_context = context.child("snapshot"); let metrics = Metrics::new(context); db::Db::init_from_log( snapshot_context, index, log, bitmap, cfg.init_concurrency, cfg.init_buffer, cfg.init_cache, metrics, ) .await } #[cfg(test)] // pub(crate) so qmdb/current can use the generic tests. pub(crate) mod test { use super::*; use crate::{ index::Unordered as UnorderedIndex, journal::contiguous::{ Contiguous as _, Mutable, fixed::Config as FConfig, variable::Config as VConfig, }, merkle::Location as GenericLocation, qmdb::{ any::{FixedConfig, MerkleConfig, VariableConfig, db::Db}, chain::Bounds, }, translator::OneCap, }; use commonware_codec::{Codec, CodecShared, Encode as _}; use commonware_cryptography::{Hasher, Sha256, sha256::Digest}; use commonware_runtime::{Supervisor as _, buffer::paged::CacheRef, deterministic::Context}; use commonware_utils::{ NZU16, NZU64, NZUsize, Widen, bitmap::{Prunable, Readable as _}, }; use core::{fmt::Debug, future::Future, pin::Pin}; use std::{ collections::{BTreeMap, BTreeSet, HashMap}, num::{NonZeroU16, NonZeroUsize}, sync::Weak, }; pub(crate) fn colliding_digest(prefix: u8, suffix: u64) -> Digest { let mut bytes = [0u8; 32]; bytes[0] = prefix; bytes[24..].copy_from_slice(&suffix.to_be_bytes()); Digest::from(bytes) } // Janky page & cache sizes to exercise boundary conditions. const PAGE_SIZE: NonZeroU16 = NZU16!(101); const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(11); pub(crate) fn fixed_db_config( suffix: &str, ) -> FixedConfig { fixed_db_config_with_strategy(suffix, Sequential) } pub(crate) fn fixed_db_config_with_strategy< T: Translator + Default, S: commonware_parallel::Strategy, >( suffix: &str, strategy: S, ) -> FixedConfig { fixed_db_config_full(suffix, strategy, ()) } /// Shared config construction for every fixed-value flavor, generic over the strategy and /// the index's snapshot-build concurrency. pub(crate) fn fixed_db_config_full< T: Translator + Default, S: commonware_parallel::Strategy, B, >( suffix: &str, strategy: S, init_concurrency: B, ) -> FixedConfig { let page_cache = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE); FixedConfig { merkle_config: MerkleConfig { journal_partition: format!("journal-{suffix}"), metadata_partition: format!("metadata-{suffix}"), items_per_blob: NZU64!(11), write_buffer: NZUsize!(1024), replay_buffer: NZUsize!(1024), strategy, page_cache: page_cache.clone(), }, journal_config: FConfig { partition: format!("log-journal-{suffix}"), items_per_blob: NZU64!(7), page_cache, write_buffer: NZUsize!(1024), replay_buffer: NZUsize!(1024), }, translator: T::default(), init_cache: Some(NZUsize!(1024)), init_buffer: NZUsize!(1 << 21), init_concurrency, } } /// Like [fixed_db_config], typed for a partitioned index at the serial concurrency. pub(crate) fn fixed_db_config_partitioned( suffix: &str, ) -> FixedConfig { fixed_db_config_full(suffix, Sequential, NZUsize!(1)) } /// Like [variable_db_config], typed for a partitioned index at the serial concurrency. pub(crate) fn variable_db_config_partitioned( suffix: &str, ) -> VariableConfig { variable_db_config_full(suffix, NZUsize!(1)) } pub(crate) fn variable_db_config( suffix: &str, ) -> VariableConfig { variable_db_config_full(suffix, ()) } /// Shared config construction for every variable-value flavor, generic over the index's /// snapshot-build concurrency. pub(crate) fn variable_db_config_full( suffix: &str, init_concurrency: B, ) -> VariableConfig { let page_cache = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE); VariableConfig { merkle_config: MerkleConfig { journal_partition: format!("journal-{suffix}"), metadata_partition: format!("metadata-{suffix}"), items_per_blob: NZU64!(11), write_buffer: NZUsize!(1024), replay_buffer: NZUsize!(1024), strategy: Sequential, page_cache: page_cache.clone(), }, journal_config: VConfig { partition: format!("log-journal-{suffix}"), items_per_section: NZU64!(7), compression: None, codec_config: ((), ()), page_cache, write_buffer: NZUsize!(1024), replay_buffer: NZUsize!(1024), }, translator: T::default(), init_cache: Some(NZUsize!(1024)), init_buffer: NZUsize!(1 << 21), init_concurrency, } } use crate::{ merkle::mmr, qmdb::any::traits::{DbAny, Provable, UnmerkleizedBatch as _}, }; type Error = crate::qmdb::Error; type Location = mmr::Location; /// Test recovery on non-empty db. pub(crate) async fn test_any_db_non_empty_recovery( context: Context, mut db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, { const ELEMENTS: u64 = 1000; // Commit initial batch. { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value(i * 1000); batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let db = db.commit().await.unwrap(); let boundary = db.sync_boundary(); let db = db.prune(boundary).await.unwrap(); let root = db.root(); let op_count = db.size(); let inactivity_floor_loc = db.inactivity_floor_loc(); drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 1)).await; assert_eq!(db.size(), op_count); assert_eq!(db.inactivity_floor_loc(), inactivity_floor_loc); assert_eq!(db.root(), root); // Write without applying (unapplied batch should be lost on reopen). { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let _merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); } drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 2)).await; assert_eq!(db.size(), op_count); assert_eq!(db.inactivity_floor_loc(), inactivity_floor_loc); assert_eq!(db.root(), root); // Write without applying again. { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let _merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); } drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 3)).await; assert_eq!(db.size(), op_count); assert_eq!(db.root(), root); // Three rounds of unapplied batches. for _ in 0..3 { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let _merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); } drop(db); let mut db = reopen_db(context.child("reopen").with_attribute("index", 4)).await; assert_eq!(db.size(), op_count); assert_eq!(db.root(), root); // Now actually commit a batch. { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } db.commit().await.unwrap(); let db = reopen_db(context.child("reopen").with_attribute("index", 5)).await; assert!(db.size() > op_count); assert_ne!(db.inactivity_floor_loc(), inactivity_floor_loc); assert_ne!(db.root(), root); db.destroy().await.unwrap(); } /// Test recovery on empty db. pub(crate) async fn test_any_db_empty_recovery( context: Context, db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, { let root = db.root(); drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 1)).await; assert_eq!(db.size(), 1); assert_eq!(db.root(), root); // Write without applying (unapplied batch should be lost on reopen). { let mut batch = db.new_batch(); for i in 0u64..1000 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let _merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); } drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 2)).await; assert_eq!(db.size(), 1); assert_eq!(db.root(), root); // Write without applying again. { let mut batch = db.new_batch(); for i in 0u64..1000 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let _merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); } drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 3)).await; assert_eq!(db.size(), 1); assert_eq!(db.root(), root); // Three rounds of unapplied batches. for _ in 0..3 { let mut batch = db.new_batch(); for i in 0u64..1000 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let _merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); } drop(db); let mut db = reopen_db(context.child("reopen").with_attribute("index", 4)).await; assert_eq!(db.size(), 1); assert_eq!(db.root(), root); // Now actually commit a batch. { let mut batch = db.new_batch(); for i in 0u64..1000 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } db.commit().await.unwrap(); let db = reopen_db(context.child("reopen").with_attribute("index", 5)).await; assert!(db.size() > 1); assert_ne!(db.root(), root); db.destroy().await.unwrap(); } /// Test that a commit after an older sync boundary is recovered without another sync. pub(crate) async fn test_any_db_commit_after_sync_recovery( context: Context, db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, V: Clone + CodecShared + Eq + std::fmt::Debug, { let key0 = Sha256::hash(&[&0u64.to_be_bytes()]); let key1 = Sha256::hash(&[&1u64.to_be_bytes()]); let value0 = make_value(100); let value1 = make_value(200); // Establish a synced baseline so recovery starts before the later commit. let merkleized = db .new_batch() .write(key0, Some(value0.clone())) .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(); // Commit a second batch without syncing; reopen must replay it from the journal. let merkleized = db .new_batch() .write(key1, Some(value1.clone())) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let committed_root = db.root(); let committed_size = db.size(); drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 1)).await; assert_eq!(db.root(), committed_root); assert_eq!(db.size(), committed_size); assert_eq!(db.get(&key0).await.unwrap(), Some(value0)); assert_eq!(db.get(&key1).await.unwrap(), Some(value1)); db.destroy().await.unwrap(); } /// Test that state committed via an awaited start_sync handle is recovered on reopen. pub(crate) async fn test_any_db_start_sync_recovery( context: Context, db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, V: Clone + CodecShared + Eq + std::fmt::Debug, { let key0 = Sha256::hash(&[&0u64.to_be_bytes()]); let value0 = make_value(100); // Apply a batch and begin committing it, awaiting the handle for durability. let merkleized = db .new_batch() .write(key0, Some(value0.clone())) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let (db, handle) = db.start_sync().await.unwrap(); handle.await.unwrap(); let committed_root = db.root(); let committed_size = db.size(); drop(db); let db = reopen_db(context.child("reopen").with_attribute("index", 1)).await; assert_eq!(db.root(), committed_root); assert_eq!(db.size(), committed_size); assert_eq!(db.get(&key0).await.unwrap(), Some(value0)); db.destroy().await.unwrap(); } /// Pruning to a floor advanced by an applied-but-uncommitted batch must not durably outrun /// the last durable commit: after a crash, the recovered commit's floor would lie below the /// pruned boundary and the database could never reopen. pub(crate) async fn test_any_db_prune_after_unsynced_floor_recovery< F: Family, D, V: Clone + CodecShared, >( context: Context, db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, { const ELEMENTS: u64 = 1000; // Establish a durable state whose last commit declares an early inactivity floor. let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); batch = batch.write(k, Some(make_value(i))); } 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 durable_floor = db.inactivity_floor_loc(); // Apply (but do not commit) a batch that advances the in-memory floor well past the // durable commit's floor. let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); batch = batch.write(k, Some(make_value(i + 1))); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let unsynced_floor = db.inactivity_floor_loc(); assert!(unsynced_floor > durable_floor); // Prune to the in-memory floor, then crash before any further commit. let boundary = db.sync_boundary(); let db = db.prune(boundary).await.unwrap(); let root = db.root(); let op_count = db.size(); drop(db); // Reopening must succeed: pruning made the floor-declaring commit durable before the // journal durably advanced its boundary past positions that commit still needs. let db = reopen_db(context.child("reopen").with_attribute("index", 1)).await; assert_eq!(db.size(), op_count); assert_eq!(db.inactivity_floor_loc(), unsynced_floor); assert_eq!(db.root(), root); db.destroy().await.unwrap(); } /// Open a prior committed state at a bound and recover it on ordinary reopen. pub(crate) async fn test_any_db_bounded_initialization_recovery( context: Context, db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, bounded_db: impl Fn(Context, Location) -> Pin> + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, V: Clone + CodecShared + Eq + std::fmt::Debug, { let key0 = Sha256::hash(&[&0u64.to_be_bytes()]); let key1 = Sha256::hash(&[&1u64.to_be_bytes()]); let key2 = Sha256::hash(&[&2u64.to_be_bytes()]); let initial_root = db.root(); let initial_size = db.size(); let initial_floor = db.inactivity_floor_loc(); // Selecting the earlier empty commit must rebuild an empty snapshot. let merkleized = db .new_batch() .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); drop(db); let db = bounded_db(context.child("cap"), initial_size) .await .unwrap(); assert_eq!(db.root(), initial_root); assert_eq!(db.size(), initial_size); assert_eq!(db.inactivity_floor_loc(), initial_floor); assert_eq!(db.get_metadata().await.unwrap(), None); let value0_a = make_value(10); let value1_a = make_value(11); let metadata_a = make_value(12); let merkleized = db .new_batch() .write(key0, Some(value0_a.clone())) .write(key1, Some(value1_a.clone())) .merkleize(&db, Some(metadata_a.clone()), &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let root_a = db.root(); let size_a = db.size(); let floor_a = db.inactivity_floor_loc(); let value0_b = make_value(20); let value2_b = make_value(21); let metadata_b = make_value(22); let merkleized = db .new_batch() .write(key0, Some(value0_b)) .write(key1, None) .write(key2, Some(value2_b)) .merkleize(&db, Some(metadata_b), &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_ne!(db.root(), root_a); let value0_c = make_value(30); let value1_c = make_value(31); let metadata_c = make_value(32); let merkleized = db .new_batch() .write(key0, Some(value0_c)) .write(key1, Some(value1_c)) .write(key2, None) .merkleize(&db, Some(metadata_c), &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); // Select a commit before a tail where // - the same key (`key0`) was updated multiple times // - `key1` was deleted then recreated, with opposite membership changes to `key2` // keeping each batch's active-key delta at zero drop(db); let db = bounded_db(context.child("cap"), size_a).await.unwrap(); assert_eq!(db.root(), root_a); assert_eq!(db.size(), size_a); assert_eq!(db.inactivity_floor_loc(), floor_a); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a.clone())); assert_eq!(db.get(&key0).await.unwrap(), Some(value0_a)); assert_eq!(db.get(&key1).await.unwrap(), Some(value1_a)); assert_eq!(db.get(&key2).await.unwrap(), None); db.commit().await.unwrap(); let db = reopen_db(context.child("reopen_after_rewind")).await; assert_eq!(db.root(), root_a); assert_eq!(db.size(), size_a); assert_eq!(db.inactivity_floor_loc(), floor_a); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a)); assert_eq!(db.get(&key0).await.unwrap(), Some(make_value(10))); assert_eq!(db.get(&key1).await.unwrap(), Some(make_value(11))); assert_eq!(db.get(&key2).await.unwrap(), None); // Fresh writes from the selected tip should produce a correct new chain and persist // across reopen. let value2_d = make_value(40); let metadata_d = make_value(41); let merkleized = db .new_batch() .write(key2, Some(value2_d.clone())) .merkleize(&db, Some(metadata_d.clone()), &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_d.clone())); assert_eq!(db.get(&key0).await.unwrap(), Some(make_value(10))); assert_eq!(db.get(&key1).await.unwrap(), Some(make_value(11))); assert_eq!(db.get(&key2).await.unwrap(), Some(value2_d.clone())); drop(db); let db = reopen_db(context.child("reopen_after_rewind_new_writes")).await; assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_d)); assert_eq!(db.get(&key0).await.unwrap(), Some(make_value(10))); assert_eq!(db.get(&key1).await.unwrap(), Some(make_value(11))); assert_eq!(db.get(&key2).await.unwrap(), Some(value2_d)); // Reopen at the initial commit boundary (`first_commit_loc + 1`). drop(db); let db = bounded_db(context.child("cap"), initial_size) .await .unwrap(); assert_eq!(db.root(), initial_root); assert_eq!(db.size(), initial_size); assert_eq!(db.inactivity_floor_loc(), initial_floor); assert_eq!(db.get_metadata().await.unwrap(), None); assert_eq!(db.get(&key0).await.unwrap(), None); assert_eq!(db.get(&key1).await.unwrap(), None); assert_eq!(db.get(&key2).await.unwrap(), None); db.commit().await.unwrap(); let db = reopen_db(context.child("reopen_initial_boundary")).await; assert_eq!(db.root(), initial_root); assert_eq!(db.size(), initial_size); assert_eq!(db.inactivity_floor_loc(), initial_floor); assert_eq!(db.get_metadata().await.unwrap(), None); assert_eq!(db.get(&key0).await.unwrap(), None); assert_eq!(db.get(&key1).await.unwrap(), None); assert_eq!(db.get(&key2).await.unwrap(), None); db.destroy().await.unwrap(); } /// Test that a large mixed workload can be authenticated and replayed correctly. #[boxed] pub(crate) async fn test_any_db_build_and_authenticate( context: Context, mut db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny + Provable, V: CodecShared + Clone + Eq + std::hash::Hash + std::fmt::Debug, >::Operation: Codec, { use crate::qmdb::verify_proof; const ELEMENTS: u64 = 1000; let mut map = HashMap::::default(); { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value(i * 1000); batch = batch.write(k, Some(v.clone())); map.insert(k, v); } // Update every 3rd key. for i in 0u64..ELEMENTS { if i % 3 != 0 { continue; } let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value((i + 1) * 10000); batch = batch.write(k, Some(v.clone())); map.insert(k, v); } // Delete every 7th key. for i in 0u64..ELEMENTS { if i % 7 != 1 { continue; } let k = Sha256::hash(&[&i.to_be_bytes()]); batch = batch.write(k, None); map.remove(&k); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } // Commit + sync with pruning raises inactivity floor. let db = db.sync().await.unwrap(); let boundary = db.sync_boundary(); let db = db.prune(boundary).await.unwrap(); // Drop & reopen and ensure state matches. let root = db.root(); db.sync().await.unwrap(); let db = reopen_db(context.child("reopened")).await; assert_eq!(root, db.root()); // State matches reference map. for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); if let Some(map_value) = map.get(&k) { let Some(db_value) = db.get(&k).await.unwrap() else { panic!("key not found in db: {k}"); }; assert_eq!(*map_value, db_value); } else { assert!(db.get(&k).await.unwrap().is_none()); } } let bounds = db.bounds(); let inactivity_floor = db.inactivity_floor_loc(); for loc in *inactivity_floor..*bounds.end { let loc = Location::new(loc); let (proof, ops) = db.proof(loc, NZU64!(10)).await.unwrap(); assert!(verify_proof::(&proof, loc, &ops, &root)); } db.destroy().await.unwrap(); } /// Test that replaying multiple updates of the same key on startup preserves correct state. pub(crate) async fn test_any_db_log_replay< F: Family, D, V: Clone + CodecShared + PartialEq + std::fmt::Debug, >( context: Context, mut db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, { // Update the same key many times within a single batch. const UPDATES: u64 = 100; let k = Sha256::hash(&[&UPDATES.to_be_bytes()]); let mut last_value = None; { let mut batch = db.new_batch(); for i in 0u64..UPDATES { let v = make_value(i * 1000); last_value = Some(v.clone()); batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let db = db.commit().await.unwrap(); let root = db.root(); // Reopen and verify the state is preserved correctly. drop(db); let db = reopen_db(context.child("reopened")).await; assert_eq!(db.root(), root); assert_eq!(db.get(&k).await.unwrap(), last_value); db.destroy().await.unwrap(); } /// Test that historical_proof returns correct proofs for past database states. pub(crate) async fn test_any_db_historical_proof_basic( _context: Context, mut db: D, make_value: impl Fn(u64) -> V, ) where D: DbAny + Provable, >::Operation: Codec + PartialEq + std::fmt::Debug, { use crate::qmdb::verify_proof; use commonware_utils::NZU64; // Add some operations const OPS: u64 = 20; { let mut batch = db.new_batch(); for i in 0u64..OPS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value(i * 1000); batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let root_hash = db.root(); let original_op_count = db.size(); // Historical proof should match "regular" proof when historical size == current database size let max_ops = NZU64!(10); let start_loc = Location::new(5); let (historical_proof, historical_ops) = db .historical_proof(original_op_count, start_loc, max_ops) .await .unwrap(); let (regular_proof, regular_ops) = db.proof(start_loc, max_ops).await.unwrap(); assert_eq!(historical_proof.leaves, regular_proof.leaves); assert_eq!(historical_proof.digests, regular_proof.digests); assert_eq!(historical_ops, regular_ops); assert!(verify_proof::( &historical_proof, start_loc, &historical_ops, &root_hash, )); // Add more operations to the database { let mut batch = db.new_batch(); for i in OPS..(OPS + 5) { let k = Sha256::hash(&[&(i + 1000).to_be_bytes()]); // different keys let v = make_value(i * 1000); batch = batch.write(k, Some(v)); } let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } // Historical proof should remain the same even though database has grown let (historical_proof2, historical_ops2) = db .historical_proof(original_op_count, start_loc, max_ops) .await .unwrap(); assert_eq!(historical_proof2.leaves, original_op_count); assert_eq!(historical_proof2.digests, regular_proof.digests); assert_eq!(historical_ops2, regular_ops); assert!(verify_proof::( &historical_proof2, start_loc, &historical_ops2, &root_hash, )); db.destroy().await.unwrap(); } /// Test that tampering with historical proofs causes verification to fail. pub(crate) async fn test_any_db_historical_proof_invalid( _context: Context, mut db: D, make_value: impl Fn(u64) -> V, ) where D: DbAny + Provable, >::Operation: Codec + PartialEq + std::fmt::Debug + Clone, { use crate::qmdb::verify_proof; use commonware_utils::NZU64; // Apply two single-write batches and capture the commit-boundary size after the // first batch. `historical_proof` requires the historical size to land on a commit // boundary when the db commits to an inactive peak boundary. let mut historical_op_count = Location::new(0); for i in 0u64..2 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value(i * 1000); let merkleized = db .new_batch() .write(k, Some(v)) .merkleize(&db, None, &mut Proportional) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); if i == 0 { historical_op_count = db.bounds().end; } } let expected_ops_len = (*historical_op_count - 1) as usize; let (proof, ops) = db .historical_proof(historical_op_count, Location::new(1), NZU64!(10)) .await .unwrap(); assert_eq!(proof.leaves, historical_op_count); assert_eq!(ops.len(), expected_ops_len); // Changing the proof digests should cause verification to fail { let mut tampered_proof = proof.clone(); tampered_proof.digests[0] = Sha256::hash(&[b"invalid"]); let root_hash = db.root(); assert!(!verify_proof::( &tampered_proof, Location::new(1), &ops, &root_hash, )); } // Appending an extra digest should cause verification to fail { let mut tampered_proof = proof.clone(); tampered_proof.digests.push(Sha256::hash(&[b"invalid"])); let root_hash = db.root(); assert!(!verify_proof::( &tampered_proof, Location::new(1), &ops, &root_hash, )); } // Changing the ops should cause verification to fail { let root_hash = db.root(); let mut tampered_ops = ops.clone(); // Swap first two ops if we have at least 2 if tampered_ops.len() >= 2 { tampered_ops.swap(0, 1); assert!(!verify_proof::( &proof, Location::new(1), &tampered_ops, &root_hash, )); } } // Appending an extra (duplicate) op should cause verification to fail { let root_hash = db.root(); let mut tampered_ops = ops.clone(); tampered_ops.push(tampered_ops[0].clone()); assert!(!verify_proof::( &proof, Location::new(1), &tampered_ops, &root_hash, )); } // Changing the start location should cause verification to fail { let root_hash = db.root(); assert!(!verify_proof::( &proof, Location::new(2), &ops, &root_hash, )); } // Changing the root digest should cause verification to fail { let invalid_root = Sha256::hash(&[b"invalid"]); assert!(!verify_proof::( &proof, Location::new(1), &ops, &invalid_root, )); } // Changing the proof leaves count should cause verification to fail { let mut tampered_proof = proof.clone(); tampered_proof.leaves = Location::new(100); let root_hash = db.root(); assert!(!verify_proof::( &tampered_proof, Location::new(1), &ops, &root_hash, )); } db.destroy().await.unwrap(); } /// Test historical_proof edge cases: singleton db, limited ops, min position. pub(crate) async fn test_any_db_historical_proof_edge_cases( _context: Context, mut db: D, make_value: impl Fn(u64) -> V, ) where D: DbAny + Provable, >::Operation: Codec + PartialEq + std::fmt::Debug, { use commonware_utils::NZU64; // Apply a sequence of single-write batches and record the commit-boundary size // reached after each. `historical_proof` requires the historical size to be a // commit boundary when the db commits to an inactive peak boundary, so we anchor each test on // one of the boundaries we recorded here rather than hardcoding sizes that depend // on internal floor-raising behavior. let initial_size = db.bounds().end; let mut boundaries = vec![initial_size]; for i in 0u64..5 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = make_value(i * 1000); let merkleized = db .new_batch() .write(k, Some(v)) .merkleize(&db, None, &mut Proportional) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); boundaries.push(db.bounds().end); } // Singleton historical state: only the initial CommitFloor is visible. let singleton_size = boundaries[0]; let (single_proof, single_ops) = db .historical_proof(singleton_size, Location::new(0), NZU64!(1)) .await .unwrap(); assert_eq!(single_proof.leaves, singleton_size); assert_eq!(single_ops.len(), 1); // max_ops exceeds the ops remaining at this historical size, so the returned count // is capped at `historical_size - start_loc`. Anchor at the earliest post-batch // boundary that has at least 3 ops past `boundaries[1]`. let limited_size = boundaries[2]; let limited_start = boundaries[1]; let expected_limited = (*limited_size - *limited_start) as usize; assert!(expected_limited > 0); let (_limited_proof, limited_ops) = db .historical_proof(limited_size, limited_start, NZU64!(20)) .await .unwrap(); assert_eq!(limited_ops.len(), expected_limited); // Standard historical proof anchored at an early commit boundary, requesting a // bounded number of ops within the historical range. let min_size = boundaries[2]; let max_ops = NZU64!(3); let expected_min = core::cmp::min(max_ops.get(), *min_size - 1) as usize; let (min_proof, min_ops) = db .historical_proof(min_size, Location::new(1), max_ops) .await .unwrap(); assert_eq!(min_proof.leaves, min_size); assert_eq!(min_ops.len(), expected_min); db.destroy().await.unwrap(); } /// Test making multiple commits, one of which deletes a key from a previous commit. pub(crate) async fn test_any_db_multiple_commits_delete_replayed( context: Context, mut db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> V, ) where D: DbAny, V: Clone + CodecShared + Eq + std::fmt::Debug, { let mut map = HashMap::::default(); const ELEMENTS: u64 = 10; let metadata_value = make_value(42); let key_at = |j: u64, i: u64| Sha256::hash(&[&(j * 1000 + i).to_be_bytes()]); for j in 0u64..ELEMENTS { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = key_at(j, i); let v = make_value(i * 1000); batch = batch.write(k, Some(v.clone())); map.insert(k, v); } let merkleized = batch .merkleize(&db, Some(metadata_value.clone()), &mut Proportional) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); db = db.commit().await.unwrap(); } assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_value)); let k = key_at(ELEMENTS - 1, ELEMENTS - 1); let merkleized = db .new_batch() .write(k, None) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), None); assert!(db.get(&k).await.unwrap().is_none()); let root = db.root(); drop(db); let db = reopen_db(context.child("reopened")).await; assert_eq!(root, db.root()); assert_eq!(db.get_metadata().await.unwrap(), None); assert!(db.get(&k).await.unwrap().is_none()); db.destroy().await.unwrap(); } /// State of an `any` db that a rebuild from the log must reproduce. struct Observed { /// Number of active keys. active_keys: usize, /// Sorted snapshot locations in the translated bucket of each observed key. locs: Vec>>, /// Active locations of the bitmap in `[pruned_bits, len)`. bits: Vec, /// Length of the bitmap. len: u64, /// Inactivity floor. floor: GenericLocation, /// Root. root: D, /// Value of each observed key. values: Vec>, } /// Observe the state of `db` for `keys`. async fn observe( db: &Db, keys: &[U::Key], ) -> Observed where F: Family, C: Mutable>, I: UnorderedIndex>, H: Hasher, U: Update, S: Strategy, Operation: Codec, { let mut values = Vec::with_capacity(keys.len()); for key in keys { values.push(db.get(key).await.unwrap()); } let bitmap = &db.bitmap; Observed { active_keys: db.active_keys, locs: keys .iter() .map(|key| { let mut locs: Vec<_> = db.snapshot.get(key).copied().collect(); locs.sort(); locs }) .collect(), bits: (bitmap.pruned_bits()..bitmap.len()) .filter(|loc| bitmap.get_bit(*loc)) .collect(), len: bitmap.len(), floor: db.inactivity_floor_loc(), root: db.root(), values, } } /// Commit and drop `db`, await `reopen` to rebuild it from the log, and assert the rebuilt /// snapshot locations, bitmap, inactivity floor, root, and values of `keys` equal the live /// ones. The live snapshot must also hold as many entries as active keys. Returns the rebuilt /// db. /// /// The rebuilt pruned prefix is the retained log start rounded down to a chunk boundary, so it /// lies below the live prefix after a prune whose chunk floor exceeds the retained start. #[boxed] pub(crate) async fn assert_rebuild_matches( db: Db, reopen: impl Future>, keys: &[U::Key], ) -> Db where F: Family, C: Mutable>, I: UnorderedIndex>, H: Hasher, U: Update, S: Strategy, Operation: Codec, { assert_eq!( db.snapshot.items(), db.active_keys, "live snapshot entries diverged from active keys", ); assert_exact(&db).await; // Capture the live state, then commit, drop, and rebuild from the log. let live = observe(&db, keys).await; let pruned = db.bitmap.pruned_bits(); let start = *db.bounds().start; db.commit().await.unwrap(); let db = reopen.await; // The rebuilt pruned prefix derives from the retained start. let chunk_bits = Prunable::::CHUNK_SIZE_BITS; assert_eq!( db.bitmap.pruned_bits(), start / chunk_bits * chunk_bits, "pruned_bits diverged from the retained start on reopen", ); assert!( db.bitmap.pruned_bits() <= pruned, "pruned_bits exceeds the live prefix on reopen", ); // Every captured value must survive the rebuild. let rebuilt = observe(&db, keys).await; assert_eq!( rebuilt.active_keys, live.active_keys, "active keys diverged on reopen", ); assert_eq!( rebuilt.locs, live.locs, "snapshot locations diverged on reopen", ); assert_eq!(rebuilt.len, live.len, "bitmap len diverged on reopen"); assert_eq!( rebuilt.bits, live.bits, "active locations diverged on reopen", ); assert_eq!( rebuilt.floor, live.floor, "inactivity floor diverged on reopen", ); assert_eq!(rebuilt.root, live.root, "root diverged on reopen"); assert_eq!(rebuilt.values, live.values, "values diverged on reopen"); assert_exact(&db).await; db } /// Record in `live` the location of each key's last update in `ops` and remove each key /// `ops` last deletes. The first operation in `ops` lies at `start`. pub(crate) fn replay( live: &mut BTreeMap>, start: GenericLocation, ops: &[Operation], ) { for (i, op) in ops.iter().enumerate() { match op { Operation::Update(update) => { live.insert(update.key().clone(), start + Widen::widen(i)); } Operation::Delete(key) => { live.remove(key); } Operation::CommitFloor(..) => {} } } } /// Replay the retained log of `db` and return the location of every live key's update. pub(crate) async fn live( db: &Db, ) -> BTreeMap> where F: Family, C: Mutable>, I: UnorderedIndex>, H: Hasher, U: Update, S: Strategy, Operation: Codec, { let bounds = db.bounds(); let positions: Vec = (*bounds.start..*bounds.end).collect(); let ops = db.log.read_many(&positions).await.unwrap(); let mut live = BTreeMap::new(); replay(&mut live, bounds.start, &ops); live } /// Assert that the activity bitmap of `db` is exact against a replay of its retained log. /// The bitmap covers the log. Every unpruned bit is set if and only if its location holds a /// live update or the last commit, and every live update lies at or above the inactivity /// floor. The active key count matches the live keys, and the snapshot maps each live key to /// its update and holds no other entry. pub(crate) async fn assert_exact( db: &Db, ) where F: Family, C: Mutable>, I: UnorderedIndex>, H: Hasher, U: Update, S: Strategy, Operation: Codec, { let live = live(db).await; let size = *db.bounds().end; let floor = db.inactivity_floor_loc(); assert!( live.values().all(|loc| *loc >= floor), "a live update lies below the inactivity floor", ); assert_eq!(db.bitmap.len(), size, "bitmap length diverged from the log"); assert_bits(db.bitmap.as_ref(), &live); assert_eq!( db.active_keys, live.len(), "active keys diverged from the log" ); assert_eq!( db.snapshot.items(), live.len(), "snapshot entries diverged from the log", ); for (key, loc) in &live { assert!( db.snapshot.get(key).any(|entry| entry == loc), "snapshot misses the live update at {loc}", ); } } /// Assert that the unpruned bits of `bitmap` are set exactly at the locations in `live` and /// at the last location, which holds the commit. pub(crate) fn assert_bits( bitmap: &impl commonware_utils::bitmap::Readable, live: &BTreeMap>, ) { let len = bitmap.len(); let active: BTreeSet = live.values().map(|loc| **loc).chain([len - 1]).collect(); for loc in bitmap.pruned_bits()..len { assert_eq!( bitmap.get_bit(loc), active.contains(&loc), "bit {loc} diverged from the live updates", ); } } /// Siblings seeded by each chained-rebuild shape. const SIBLINGS: u64 = 48; /// Colliding sibling `i`. Every key the chained-rebuild shapes write shares one translated /// bucket. fn sibling(i: u64) -> Digest { colliding_digest(0xA0, i) } /// Expected value of every key written by applied batches. #[derive(Clone, Default)] struct Model { /// Value of each active key. values: BTreeMap, /// Every key ever written. keys: BTreeSet, } impl Model { /// Record `writes` as applied. fn apply(&mut self, writes: &[(Digest, Option)]) { for &(key, value) in writes { self.keys.insert(key); match value { Some(value) => self.values.insert(key, value), None => self.values.remove(&key), }; } } } /// Merkleize `writes` on top of `batch`. pub(crate) async fn build( db: &D, batch: D::Batch, writes: &[(Digest, Option)], ) -> D::Merkleized where D: DbAny, { writes .iter() .fold(batch, |batch, &(key, value)| batch.write(key, value)) .merkleize(db, None, &mut Proportional) .await .unwrap() } /// Apply and commit `writes` as one batch, recording them in `model`. async fn write_all( db: D, model: &mut Model, writes: &[(Digest, Option)], ) -> D where D: DbAny, { let batch = build(&db, db.new_batch(), writes).await; let db = db.apply_batch(batch).await.unwrap(); model.apply(writes); db.commit().await.unwrap() } /// Commit siblings `0..SIBLINGS`. async fn seed_siblings( db: D, model: &mut Model, make_value: &impl Fn(u64) -> Digest, ) -> D where D: DbAny, { let writes: Vec<_> = (0..SIBLINGS) .map(|i| (sibling(i), Some(make_value(i)))) .collect(); write_all(db, model, &writes).await } /// Apply the chain A <- B <- C. A is applied and dropped before C is merkleized on B, so C /// resolves the keys A created through the base locations of the dropped prefix. B and C /// touch only keys A created, so a lost base location duplicates them silently. #[boxed] async fn apply_dropped_chain( db: D, model: &mut Model, make_value: &impl Fn(u64) -> Digest, ) -> D where U: Update, S: Strategy, Operation: Codec, D: DbAny< F, Key = Digest, Value = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, { // A updates and deletes siblings and creates new ones. let a = vec![ (sibling(5), Some(make_value(100))), (sibling(6), Some(make_value(101))), (sibling(7), None), (sibling(1000), Some(make_value(102))), (sibling(1001), Some(make_value(103))), ]; // B and C update the keys A created. let b = vec![ (sibling(1000), Some(make_value(104))), (sibling(1001), Some(make_value(105))), ]; let c = vec![ (sibling(1000), Some(make_value(106))), (sibling(1001), Some(make_value(107))), ]; // Build A <- B, apply A and drop it, then merkleize C on B and apply it. let a_batch = build(&db, db.new_batch(), &a).await; let b_batch = build(&db, a_batch.new_batch::(), &b).await; let db = db.apply_batch(a_batch).await.unwrap(); let db = db.commit().await.unwrap(); let c_batch = build(&db, b_batch.new_batch::(), &c).await; let db = db.apply_batch(c_batch).await.unwrap(); for writes in [&a, &b, &c] { model.apply(writes); } db } /// Apply the chain A <- B <- C by applying A, then C over pending B, so the applied and /// pending ancestors are both non-empty. On a db without active keys, every key in A's diff /// is created by A, so a key misresolved through either partition is duplicated silently. #[boxed] async fn apply_mixed_chain( db: D, model: &mut Model, make_value: &impl Fn(u64) -> Digest, ) -> D where U: Update, S: Strategy, Operation: Codec, D: DbAny< F, Key = Digest, Value = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, { // A recreates deleted siblings and creates new ones. let a = vec![ (sibling(0), Some(make_value(200))), (sibling(1), Some(make_value(201))), (sibling(1000), Some(make_value(202))), (sibling(1001), Some(make_value(203))), ]; // B updates a key A created and creates a sibling. C updates keys A and B created. let b = vec![ (sibling(1001), Some(make_value(204))), (sibling(2000), Some(make_value(205))), ]; let c = vec![ (sibling(1000), Some(make_value(206))), (sibling(2000), Some(make_value(207))), ]; // Build A <- B <- C, then apply A and C. let a_batch = build(&db, db.new_batch(), &a).await; let b_batch = build(&db, a_batch.new_batch::(), &b).await; let c_batch = build(&db, b_batch.new_batch::(), &c).await; let db = db.apply_batch(a_batch).await.unwrap(); let db = db.apply_batch(c_batch).await.unwrap(); for writes in [&a, &b, &c] { model.apply(writes); } db } /// Apply the chain A <- B <- C <- D. A is applied and dropped before D is merkleized, and B /// is applied before D. D then resolves B's keys through the applied ancestor and the keys A /// shares with C through the base locations of the dropped prefix. #[boxed] async fn apply_crossed_chain( db: D, model: &mut Model, make_value: &impl Fn(u64) -> Digest, ) -> D where U: Update, S: Strategy, Operation: Codec, D: DbAny< F, Key = Digest, Value = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, { // A updates and deletes siblings and creates new ones. let a = vec![ (sibling(5), Some(make_value(300))), (sibling(6), Some(make_value(301))), (sibling(7), None), (sibling(1000), Some(make_value(302))), (sibling(1001), Some(make_value(303))), (sibling(1002), Some(make_value(304))), ]; // B updates keys A updated and created, and creates a sibling. let b = vec![ (sibling(5), Some(make_value(305))), (sibling(1000), Some(make_value(306))), (sibling(2000), Some(make_value(307))), ]; // C updates keys A updated and created and recreates the key A deleted. B touches none // of them. let c = vec![ (sibling(6), Some(make_value(308))), (sibling(7), Some(make_value(309))), (sibling(1001), Some(make_value(310))), ]; // D updates keys written by A, B, and C. let d = vec![ (sibling(5), Some(make_value(311))), (sibling(1001), Some(make_value(312))), (sibling(1002), Some(make_value(313))), (sibling(2000), Some(make_value(314))), ]; // Build A <- B <- C, then apply A and drop it before D is merkleized on C. let a_batch = build(&db, db.new_batch(), &a).await; let b_batch = build(&db, a_batch.new_batch::(), &b).await; let c_batch = build(&db, b_batch.new_batch::(), &c).await; let db = db.apply_batch(a_batch).await.unwrap(); let db = db.commit().await.unwrap(); let d_batch = build(&db, c_batch.new_batch::(), &d).await; // Apply B, then D over pending C. let db = db.apply_batch(b_batch).await.unwrap(); let db = db.apply_batch(d_batch).await.unwrap(); for writes in [&a, &b, &c, &d] { model.apply(writes); } db } /// Assert `db` serves `model` for every written key, then that a rebuild from the log matches /// the live state. Returns the rebuilt db. #[boxed] async fn assert_model_rebuild( db: Db, model: &Model, reopen: impl Future>, ) -> Db where F: Family, C: Mutable>, I: UnorderedIndex>, U: Update, S: Strategy, Operation: Codec, { assert_eq!( db.active_keys, model.values.len(), "active keys diverged from the model", ); let keys: Vec<_> = model.keys.iter().copied().collect(); for key in &keys { assert_eq!( db.get(key).await.unwrap(), model.values.get(key).copied(), "value of {key} diverged from the model", ); } assert_rebuild_matches(db, reopen, &keys).await } /// Batches applied across dropped, applied, and pending ancestors must leave a live snapshot, /// bitmap, floor, root, and values equal to a rebuild from the log. A superseded location /// misresolved to `None` leaves a duplicate snapshot entry and a stale bitmap bit. pub(crate) async fn test_any_db_chained_rebuild( context: Context, db: Db, reopen: impl Fn(Context) -> Fut, make_value: impl Fn(u64) -> Digest, ) where F: Family, C: Mutable>, I: UnorderedIndex>, U: Update, S: Strategy, Operation: Codec, Db: DbAny< F, Key = Digest, Value = Digest, Digest = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, Fut: Future>, { // Shape 1: A is applied and dropped before C is merkleized on B. let mut model = Model::default(); let db = seed_siblings(db, &mut model, &make_value).await; let db = apply_dropped_chain(db, &mut model, &make_value).await; let db = assert_model_rebuild( db, &model, reopen(context.child("rebuild").with_attribute("index", 1)), ) .await; db.destroy().await.unwrap(); // Shape 2: A is applied, then C is applied over pending B. The seed is deleted first, so // the chain starts without active keys. let db = reopen(context.child("shape").with_attribute("index", 2)).await; let mut model = Model::default(); let db = seed_siblings(db, &mut model, &make_value).await; let clear: Vec<_> = (0..SIBLINGS).map(|i| (sibling(i), None)).collect(); let db = write_all(db, &mut model, &clear).await; let db = apply_mixed_chain(db, &mut model, &make_value).await; let db = assert_model_rebuild( db, &model, reopen(context.child("rebuild").with_attribute("index", 2)), ) .await; db.destroy().await.unwrap(); // Shape 3: the applied ancestor and the dropped prefix both resolve keys in one apply. let db = reopen(context.child("shape").with_attribute("index", 3)).await; let mut model = Model::default(); let db = seed_siblings(db, &mut model, &make_value).await; let db = apply_crossed_chain(db, &mut model, &make_value).await; let db = assert_model_rebuild( db, &model, reopen(context.child("rebuild").with_attribute("index", 3)), ) .await; db.destroy().await.unwrap(); // Shape 4: rewrite the seed eight times so the floor passes a bitmap chunk, then prune to // the sync boundary and reopen. Shape 3 then runs over a rebuilt snapshot and a pruned // bitmap. let db = reopen(context.child("shape").with_attribute("index", 4)).await; let mut model = Model::default(); let mut db = seed_siblings(db, &mut model, &make_value).await; for round in 1..=8 { let writes: Vec<_> = (0..SIBLINGS) .map(|i| (sibling(i), Some(make_value(round * 1000 + i)))) .collect(); db = write_all(db, &mut model, &writes).await; } let boundary = db.sync_boundary(); let db = db.prune(boundary).await.unwrap(); db.commit().await.unwrap(); let db = reopen(context.child("pruned")).await; assert!( db.bitmap.pruned_bits() > 0, "setup must prune a bitmap chunk", ); let db = apply_crossed_chain(db, &mut model, &make_value).await; let db = assert_model_rebuild( db, &model, reopen(context.child("rebuild").with_attribute("index", 4)), ) .await; db.destroy().await.unwrap(); } use crate::qmdb::{ any::{ batch::Slot, ordered::{fixed::Db as OrderedFixedDb, variable::Db as OrderedVariableDb}, traits::MerkleizedBatch as MerkleizedTrait, unordered::{fixed::Db as UnorderedFixedDb, variable::Db as UnorderedVariableDb}, }, floor::{Bounded, Decision, Entry, Hold, Limits, Policy, Proportional}, }; use commonware_macros::{test_group, test_traced}; use commonware_parallel::Sequential; use commonware_runtime::{ Clock as _, Metrics as _, Runner as _, deterministic, telemetry::metrics::metric_samples, }; use core::time::Duration; use futures::{pin_mut, poll}; use rand::RngExt as _; // Type aliases for all 12 MMR variants (all use OneCap for collision coverage). type UnorderedFixed = UnorderedFixedDb; type UnorderedVariable = UnorderedVariableDb; type OrderedFixed = OrderedFixedDb; type OrderedVariable = OrderedVariableDb; type UnorderedFixedP1 = unordered::fixed::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 1, Sequential, >; type UnorderedVariableP1 = unordered::variable::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 1, Sequential, >; type OrderedFixedP1 = ordered::fixed::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 1, Sequential, >; type OrderedVariableP1 = ordered::variable::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 1, Sequential, >; type UnorderedFixedP2 = unordered::fixed::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 2, Sequential, >; type UnorderedVariableP2 = unordered::variable::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 2, Sequential, >; type OrderedFixedP2 = ordered::fixed::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 2, Sequential, >; type OrderedVariableP2 = ordered::variable::partitioned::Db< mmr::Family, Context, Digest, Digest, Sha256, OneCap, 2, Sequential, >; // MMB type aliases for with_all_variants. mod mmb_types { use super::*; use crate::{ index::{ordered::Index as OrderedIndex, unordered::Index as UnorderedIndex}, journal::contiguous::{fixed::Journal as FJournal, variable::Journal as VJournal}, merkle::{Location, mmb}, qmdb::any::{ operation::{Operation, update}, value::{FixedEncoding, VariableEncoding}, }, }; type MmbLocation = Location; pub type MmbUnorderedFixed = super::super::db::Db< mmb::Family, Context, FJournal< Context, Operation>>, >, UnorderedIndex, Sha256, update::Unordered>, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, Sequential, >; pub type MmbUnorderedVariable = super::super::db::Db< mmb::Family, Context, VJournal< Context, Operation>>, >, UnorderedIndex, Sha256, update::Unordered>, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, Sequential, >; pub type MmbOrderedFixed = super::super::db::Db< mmb::Family, Context, FJournal< Context, Operation>>, >, OrderedIndex, Sha256, update::Ordered>, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, Sequential, >; pub type MmbOrderedVariable = super::super::db::Db< mmb::Family, Context, VJournal< Context, Operation>>, >, OrderedIndex, Sha256, update::Ordered>, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, Sequential, >; } use mmb_types::*; #[inline] fn to_digest(i: u64) -> Digest { Sha256::hash(&[&i.to_be_bytes()]) } // Defines MMR-only variants (for tests that require mmr::Family, e.g. proof verification). macro_rules! with_mmr_variants { ($cb:ident!($($args:tt)*)) => { $cb!($($args)*, uf, UnorderedFixed, mmr::Family, fixed_db_config); $cb!($($args)*, uv, UnorderedVariable, mmr::Family, variable_db_config); $cb!($($args)*, of, OrderedFixed, mmr::Family, fixed_db_config); $cb!($($args)*, ov, OrderedVariable, mmr::Family, variable_db_config); $cb!($($args)*, ufp1, UnorderedFixedP1, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, uvp1, UnorderedVariableP1, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, ofp1, OrderedFixedP1, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, ovp1, OrderedVariableP1, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, ufp2, UnorderedFixedP2, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, uvp2, UnorderedVariableP2, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, ofp2, OrderedFixedP2, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, ovp2, OrderedVariableP2, mmr::Family, variable_db_config_partitioned); }; } // Defines all variants (MMR + MMB). Calls $cb!($($args)*, $label, $type, $family, $config) for each. macro_rules! with_all_variants { ($cb:ident!($($args:tt)*)) => { $cb!($($args)*, uf, UnorderedFixed, mmr::Family, fixed_db_config); $cb!($($args)*, uv, UnorderedVariable, mmr::Family, variable_db_config); $cb!($($args)*, of, OrderedFixed, mmr::Family, fixed_db_config); $cb!($($args)*, ov, OrderedVariable, mmr::Family, variable_db_config); $cb!($($args)*, ufp1, UnorderedFixedP1, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, uvp1, UnorderedVariableP1, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, ofp1, OrderedFixedP1, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, ovp1, OrderedVariableP1, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, ufp2, UnorderedFixedP2, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, uvp2, UnorderedVariableP2, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, ofp2, OrderedFixedP2, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, ovp2, OrderedVariableP2, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, uf_mmb, MmbUnorderedFixed, mmb::Family, fixed_db_config); $cb!($($args)*, uv_mmb, MmbUnorderedVariable, mmb::Family, variable_db_config); $cb!($($args)*, of_mmb, MmbOrderedFixed, mmb::Family, fixed_db_config); $cb!($($args)*, ov_mmb, MmbOrderedVariable, mmb::Family, variable_db_config); }; } // Defines the ordered variants (MMR + MMB). macro_rules! with_ordered_variants { ($cb:ident!($($args:tt)*)) => { $cb!($($args)*, of, OrderedFixed, mmr::Family, fixed_db_config); $cb!($($args)*, ov, OrderedVariable, mmr::Family, variable_db_config); $cb!($($args)*, ofp1, OrderedFixedP1, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, ovp1, OrderedVariableP1, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, ofp2, OrderedFixedP2, mmr::Family, fixed_db_config_partitioned); $cb!($($args)*, ovp2, OrderedVariableP2, mmr::Family, variable_db_config_partitioned); $cb!($($args)*, of_mmb, MmbOrderedFixed, mmb::Family, fixed_db_config); $cb!($($args)*, ov_mmb, MmbOrderedVariable, mmb::Family, variable_db_config); }; } // Emit one `#[test_group("slow")] #[test_traced]` test per variant, named // `_`. `with_reopen` hands the test a db plus a reopen // closure, `with_make_value` hands it just the db. macro_rules! test_for_variant { (with_cap: $f:ident, $traced:literal, $l:ident, $db:ty, $family:ty, $cfg:ident) => { paste::paste! { #[test_group("slow")] #[test_traced($traced)] fn [<$f _ $l>]() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child(stringify!($l)); let db = <$db>::init(ctx.child("storage"), $cfg::("db"), None) .await .unwrap(); $f( ctx, db, |ctx| { Box::pin(async move { <$db>::init(ctx.child("storage"), $cfg::("db"), None) .await .unwrap() }) }, |ctx, cap| { Box::pin(async move { <$db>::init( ctx.child("storage"), $cfg::("db"), Some(cap), ) .await }) }, to_digest, ) .await; }); } } }; (with_reopen: $f:ident, $traced:literal, $l:ident, $db:ty, $family:ty, $cfg:ident) => { paste::paste! { #[test_group("slow")] #[test_traced($traced)] fn [<$f _ $l>]() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child(stringify!($l)); let db = <$db>::init(ctx.child("storage"), $cfg::("db"), None) .await .unwrap(); $f( ctx, db, |ctx| { Box::pin(async move { <$db>::init(ctx.child("storage"), $cfg::("db"), None) .await .unwrap() }) }, to_digest, ) .await; }); } } }; (with_make_value: $f:ident, $traced:literal, $l:ident, $db:ty, $family:ty, $cfg:ident) => { paste::paste! { #[test_group("slow")] #[test_traced($traced)] fn [<$f _ $l>]() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child(stringify!($l)); let db = <$db>::init(ctx.child("storage"), $cfg::("db"), None) .await .unwrap(); $f(ctx, db, to_digest).await; }); } } }; (with_seeds: $f:ident, $traced:literal, $l:ident, $db:ty, $family:ty, $cfg:ident) => { paste::paste! { #[rstest::rstest] #[test_traced($traced)] fn [<$f _ $l>](#[values(0, 1, 7, 0x5133)] seed: u64) { let executor = deterministic::Runner::seeded(seed); executor.start(|context| async move { let ctx = context.child(stringify!($l)); let db = <$db>::init(ctx.child("storage"), $cfg::("db"), None) .await .unwrap(); $f(ctx, db, to_digest).await; }); } } }; } // Generate one slow test per variant across all variants (MMR + MMB). macro_rules! test_for_all_variants { (with_reopen: $f:ident, $traced:literal) => { with_all_variants!(test_for_variant!(with_reopen: $f, $traced)); }; (with_make_value: $f:ident, $traced:literal) => { with_all_variants!(test_for_variant!(with_make_value: $f, $traced)); }; } // Generate one slow test per variant across the MMR-only variants (for // tests that use mmr::Family-specific features like Location::new or // verify_proof). macro_rules! test_for_mmr_variants { (with_reopen: $f:ident, $traced:literal) => { with_mmr_variants!(test_for_variant!(with_reopen: $f, $traced)); }; (with_make_value: $f:ident, $traced:literal) => { with_mmr_variants!(test_for_variant!(with_make_value: $f, $traced)); }; } test_for_all_variants!(with_reopen: test_any_db_log_replay, "WARN"); test_for_mmr_variants!(with_reopen: test_any_db_build_and_authenticate, "WARN"); test_for_mmr_variants!(with_make_value: test_any_db_historical_proof_basic, "WARN"); test_for_mmr_variants!(with_make_value: test_any_db_historical_proof_invalid, "WARN"); test_for_mmr_variants!(with_make_value: test_any_db_historical_proof_edge_cases, "WARN"); test_for_all_variants!(with_reopen: test_any_db_multiple_commits_delete_replayed, "WARN"); test_for_all_variants!(with_reopen: test_any_db_non_empty_recovery, "WARN"); test_for_all_variants!(with_reopen: test_any_db_empty_recovery, "WARN"); test_for_all_variants!(with_reopen: test_any_db_commit_after_sync_recovery, "WARN"); test_for_all_variants!(with_reopen: test_any_db_start_sync_recovery, "WARN"); test_for_all_variants!(with_reopen: test_any_db_prune_after_unsynced_floor_recovery, "WARN"); test_for_all_variants!(with_reopen: test_any_db_chained_rebuild, "WARN"); test_for_all_variants!(with_make_value: test_any_policy_hold, "WARN"); test_for_all_variants!(with_reopen: test_any_policy_keep_evict_and_recover, "WARN"); test_for_all_variants!(with_make_value: test_any_policy_limits_after_colliding_writes, "WARN"); test_for_all_variants!(with_make_value: test_any_policy_matches_proportional, "WARN"); test_for_all_variants!(with_make_value: test_any_policy_decisions_match_writes, "WARN"); test_for_all_variants!(with_make_value: test_any_policy_after_staged_writes, "WARN"); test_for_all_variants!(with_reopen: test_any_policy_after_ancestor_applied, "WARN"); test_for_all_variants!(with_reopen: test_any_activity_depths, "WARN"); test_for_all_variants!(with_make_value: test_any_proportional_bound, "WARN"); test_for_all_variants!(with_make_value: test_any_policy_limits_with_writes, "WARN"); test_for_all_variants!(with_reopen: test_any_policy_own_writes, "WARN"); test_for_all_variants!(with_reopen: test_any_policy_evicts_parent_created_key, "WARN"); test_for_all_variants!(with_make_value: test_any_policy_ancestor_twins, "WARN"); test_for_all_variants!(with_make_value: test_any_proportional_one_batch, "WARN"); test_for_variant!( with_seeds: test_any_proportional_randomized_bound, "WARN", uf, UnorderedFixed, mmr::Family, fixed_db_config ); test_for_variant!( with_seeds: test_any_proportional_randomized_bound, "WARN", of, OrderedFixed, mmr::Family, fixed_db_config ); test_for_variant!( with_seeds: test_any_proportional_pending_bound, "WARN", uf, UnorderedFixed, mmr::Family, fixed_db_config ); test_for_variant!( with_seeds: test_any_proportional_pending_bound, "WARN", of, OrderedFixed, mmr::Family, fixed_db_config ); with_ordered_variants!( test_for_variant!(with_reopen: test_any_ordered_policy_eviction_matrix, "WARN") ); with_ordered_variants!( test_for_variant!(with_reopen: test_any_ordered_policy_repair_across_ancestors, "WARN") ); test_for_variant!( with_reopen: test_any_policy_reads_in_one_read, "WARN", uf, UnorderedFixed, mmr::Family, fixed_db_config ); test_for_variant!( with_reopen: test_any_policy_reads_in_one_read, "WARN", of, OrderedFixed, mmr::Family, fixed_db_config ); test_for_variant!( with_make_value: test_any_policy_unordered_reads_nothing, "WARN", uf, UnorderedFixed, mmr::Family, fixed_db_config ); with_mmr_variants!( test_for_variant!(with_cap: test_any_db_bounded_initialization_recovery, "WARN") ); /// A test policy that decides each update from its key and records it. pub(crate) struct Script { /// The fixed limits the policy returns, or `None` for proportional limits. limits: Option, /// Decides an update from its key. decide: D, /// The location, key, and value of each decided update, in order. pub(crate) visited: Vec<(GenericLocation, Digest, Digest)>, } /// A scripted decision for an update. #[derive(Clone, Copy, Debug)] pub(crate) enum Choice { /// Move the update to the tip. Keep, /// Delete the key. Evict, /// Write the value for the key at the tip. Replace(Digest), /// End the walk without deciding the update. Stop, } impl Choice> Script { /// Return a policy with fixed limits that decides each update with `decide`. pub(crate) const fn new(entries: usize, skips: u64, decide: D) -> Self { Self { limits: Some(Limits { entries, skips }), decide, visited: Vec::new(), } } /// Return a policy with proportional limits whose `decide` must keep every entry. pub(crate) const fn proportional(decide: D) -> Self { Self { limits: None, decide, visited: Vec::new(), } } /// Return the location of each decided update, in order. pub(crate) fn locations(&self) -> Vec> { self.visited.iter().map(|(loc, _, _)| *loc).collect() } } impl Choice> Policy for Script { fn evicts(&self) -> bool { self.limits.is_some() } fn limits(&self, made_inactive: usize) -> Limits { self.limits.unwrap_or_else(|| { Policy::::limits(&Proportional, made_inactive) }) } fn decide<'a>(&mut self, entry: Entry<'a, F, Digest, Digest>) -> Decision<'a, Digest> { self.visited .push((entry.location(), *entry.key(), *entry.value())); match (self.decide)(entry.key()) { Choice::Keep => entry.keep(), Choice::Evict => entry.evict().0, Choice::Replace(value) => entry.replace(value), Choice::Stop => entry.stop(), } } } pub(crate) const fn keep(_: &Digest) -> Choice { Choice::Keep } /// Merkleize `writes` on `batch` with [`Hold`]. pub(crate) async fn hold_batch( db: &D, batch: D::Batch, writes: &[(Digest, Option)], ) -> D::Merkleized where D: DbAny, { writes .iter() .fold(batch, |batch, &(key, value)| batch.write(key, value)) .merkleize(db, None, &mut Hold) .await .unwrap() } /// Apply `writes` as one batch with [`Hold`]. pub(crate) async fn hold(db: D, writes: &[(Digest, Option)]) -> D where D: DbAny, { let merkleized = hold_batch(&db, db.new_batch(), writes).await; db.apply_batch(merkleized).await.unwrap() } /// Merkleize `batch` with `policy` and without metadata through [`DbAny`], for callers whose /// generic database type implements the batch trait only through its [`DbAny::Batch`]. async fn merkleize( db: &D, batch: D::Batch, policy: &mut (impl Policy + Send), ) -> Result> where D: DbAny, { batch.merkleize(db, None, policy).await } /// Merkleize `batch` with a policy whose entries and skips are unbounded and that decides each /// update with `choose`. Returns the merkleized batch and the decided locations. async fn decide( db: &D, batch: D::Batch, choose: impl FnMut(&Digest) -> Choice + Send, ) -> (D::Merkleized, Vec>) where D: DbAny, { let mut policy = Script::new(usize::MAX, u64::MAX, choose); let merkleized = batch.merkleize(db, None, &mut policy).await.unwrap(); (merkleized, policy.locations()) } /// Replay the log of `db`, then each of the pending `batches` (oldest first). Return the /// location of every live key's update. async fn replayed>( db: &D, batches: &[&D::Merkleized], ) -> BTreeMap> { let mut live = db.live().await; for batch in batches { let (start, ops) = D::ops(batch); replay(&mut live, start, &ops); } live } /// Return the ascending locations of the live updates after the log of `db` and the pending /// `batches` (oldest first). pub(crate) async fn active>( db: &D, batches: &[&D::Merkleized], ) -> Vec> { let mut active: Vec<_> = replayed(db, batches).await.into_values().collect(); active.sort(); active } /// The state a floor walk sees: the state after a batch's writes. pub(crate) struct Written { /// The location of every live key's update. pub(crate) live: BTreeMap>, /// The tip the writes reach, where the walk ends. pub(crate) tip: GenericLocation, /// The operations the writes make inactive: one for each update of a key live before the /// batch, whose update it supersedes, and two for each delete, which supersedes an update /// and is itself inactive. pub(crate) inactive: usize, /// The encoded operations the writes append. ops: Vec>, } impl Written { /// Return the ascending locations of the live updates. pub(crate) fn active(&self) -> Vec> { let mut active: Vec<_> = self.live.values().copied().collect(); active.sort(); active } } /// Return the state after `writes` on `batch`, a child of the pending `ancestors` (oldest /// first) of `db`. /// /// Every policy sees the same state: the writes resolve before the floor walk, so every batch /// with these writes appends the same operations below the tip. [`Hold`] decides nothing, so /// its batch appends only those operations and its commit. pub(crate) async fn written>( db: &D, ancestors: &[&D::Merkleized], batch: D::Batch, writes: &[(Digest, Option)], ) -> Written where Operation: Codec, { let held = hold_batch(db, batch, writes).await; let before = replayed(db, ancestors).await; let batches: Vec<_> = ancestors.iter().copied().chain([&held]).collect(); let (_, ops) = D::ops(&held); let ops = &ops[..ops.len() - 1]; let inactive = ops .iter() .map(|op| match op { Operation::Update(update) => usize::from(before.contains_key(update.key())), Operation::Delete(_) => 2, Operation::CommitFloor(..) => panic!("the writes append a commit"), }) .sum(); Written { live: replayed(db, &batches).await, tip: D::span(&held).tip.size - 1, inactive, ops: ops.iter().map(|op| op.encode().to_vec()).collect(), } } /// Return what the floor walk of `batch` appends: its operations from the tip of its writes up /// to its commit, an update as its key and value and a delete as its key and `None`. Asserts /// that its operations below the tip are the writes of `written`. pub(crate) fn appended>( batch: &D::Merkleized, written: &Written, ) -> Vec<(Digest, Option)> where Operation: Codec, { let (start, ops) = D::ops(batch); let below = (*written.tip - *start) as usize; let prefix: Vec<_> = ops[..below].iter().map(|op| op.encode().to_vec()).collect(); assert_eq!( prefix, written.ops, "the writes diverged from the held batch" ); ops[below..ops.len() - 1] .iter() .map(|op| match op { Operation::Update(update) => (*update.key(), Some(*update.value())), Operation::Delete(key) => (*key, None), Operation::CommitFloor(..) => panic!("a commit precedes the last operation"), }) .collect() } /// Sum the samples of the counter `name` across every database instance. pub(crate) fn counter(context: &Context, name: &str) -> u64 { metric_samples(&context.encode(), name) .map(|(_, value)| value.parse::().unwrap()) .sum() } /// Model a fixed-limit walk from `floor` over the ascending `active` locations below /// `tip`, the tip of the batch's writes, where the policy stops at the `stops` locations and /// decides every other one. Returns the floor the walk reaches and the locations it visits, /// a stopped one included. /// /// Reaching an active update spends the inactive gap before it in skips; when the gap exceeds /// the remaining skips, the floor advances by them and the walk ends. Deciding an update spends /// an entry and moves the floor past it, and stopping leaves the floor at it. Once no active /// update remains, the floor moves to `tip` if the remaining skips reach it, and by them /// otherwise. With no entries left the walk ends. pub(crate) fn walk_model( active: &[u64], mut floor: u64, tip: u64, mut entries: usize, mut skips: u64, stops: &[u64], ) -> (u64, Vec) { let mut visited = Vec::new(); while entries > 0 { let next = active .iter() .copied() .find(|loc| *loc >= floor && *loc < tip) .unwrap_or(tip); let gap = next - floor; if gap > skips { floor += skips; break; } skips -= gap; floor = next; if next == tip { break; } visited.push(next); if stops.contains(&next) { break; } floor = next + 1; entries -= 1; } (floor, visited) } /// [`Hold`] keeps the inherited floor while the batch's writes apply, and moves it to the new /// commit location when the batch empties the database. pub(crate) async fn test_any_policy_hold( _context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where D: DbAny, { let writes: Vec<_> = (0..4) .map(|i| (to_digest(i), Some(make_value(i)))) .collect(); let db = hold(db, &writes).await; let floor = db.inactivity_floor_loc(); // Hold the floor for a batch that updates, deletes, and creates keys. let merkleized = db .new_batch() .write(to_digest(0), Some(make_value(100))) .write(to_digest(1), None) .write(to_digest(4), Some(make_value(4))) .merkleize(&db, None, &mut Hold) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // The writes apply and the untouched keys keep their values. assert_eq!(db.inactivity_floor_loc(), floor); for (key, expected) in [ (0, Some(make_value(100))), (1, None), (2, Some(make_value(2))), (3, Some(make_value(3))), (4, Some(make_value(4))), ] { assert_eq!(db.get(&to_digest(key)).await.unwrap(), expected); } // Deleting every key moves the floor to the new commit location. let deletes: Vec<_> = [0, 2, 3, 4].map(|i| (to_digest(i), None)).into(); let db = hold(db, &deletes).await; assert_eq!(db.inactivity_floor_loc(), db.size() - 1); db.destroy().await.unwrap(); } /// A policy in a batch with writes passes the locations the writes make inactive, keeps and /// evicts applied updates, and replaces and evicts the batch's own writes. The result /// survives commit, reopen, and prune. pub(crate) async fn test_any_policy_keep_evict_and_recover( context: Context, db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> Digest, ) where D: Inspect, Operation: Codec, { // Seed four keys in key order at 1..5, with the initial commit at 0 and the seed commit // at 5. let keys = [0x10, 0x20, 0x30, 0x40, 0x50].map(|prefix| colliding_digest(prefix, 0)); let seed: Vec<_> = (0..4) .map(|i| (keys[i], Some(make_value(i as u64)))) .collect(); let db = hold(db, &seed).await; assert_eq!(*db.size(), 6); // Delete the smallest key, update the largest, and create a larger one. Both orderings // append the delete at 6, the update at 7, and the create at 8: an ordered batch rewrites // no predecessor, since the updated key precedes both the deleted key (by wrapping) and // the created one. The writes reach the tip 9. let with = |batch: D::Batch| { batch .write(keys[0], None) .write(keys[3], Some(make_value(103))) .write(keys[4], Some(make_value(4))) }; let replacement = make_value(203); let choose = |key: &Digest| match key { key if *key == keys[1] => Choice::Keep, key if *key == keys[3] => Choice::Replace(replacement), _ => Choice::Evict, }; // The active updates are the second and third keys at 2 and 3 and the batch's own writes // at 7 and 8. Reaching them passes 0 and 1, then 4 (superseded), 5 (the seed commit), and // 6 (the batch's delete): five skips. With four, the walk ends at 4 + 2 = 6. let mut policy = Script::new(usize::MAX, 4, choose); let merkleized = with(db.new_batch()) .merkleize(&db, None, &mut policy) .await .unwrap(); assert_eq!(policy.locations(), [2, 3].map(GenericLocation::::new)); assert_eq!(*D::span(&merkleized).inactivity_floor, 6); drop(merkleized); // With five skips, the policy keeps the second key, evicts the third, replaces the // batch's update of the largest key, and evicts the key the batch creates. Each decided // update sees the value the writes left, and the floor reaches the tip. let mut policy = Script::new(usize::MAX, 5, choose); let merkleized = with(db.new_batch()) .merkleize(&db, None, &mut policy) .await .unwrap(); let visited = [ (2, keys[1], make_value(1)), (3, keys[2], make_value(2)), (7, keys[3], make_value(103)), (8, keys[4], make_value(4)), ] .map(|(loc, key, value)| (GenericLocation::::new(loc), key, value)); assert_eq!(policy.visited, visited); assert_eq!(*D::span(&merkleized).inactivity_floor, 9); // The decisions follow the writes in walk order. An ordered batch folds each evicted // key's link into its kept or replaced predecessor, so it appends no other rewrite. let (_, ops) = D::ops(&merkleized); let decided: Vec<_> = ops[3..ops.len() - 1] .iter() .map(|op| match op { Operation::Update(update) => (*update.key(), Some(*update.value())), Operation::Delete(key) => (*key, None), Operation::CommitFloor(..) => unreachable!(), }) .collect(); assert_eq!( decided, [ (keys[1], Some(make_value(1))), (keys[2], None), (keys[3], Some(make_value(203))), (keys[4], None), ] ); // The batch commits the tip as the floor. The created key is absent, so two keys remain. let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(*db.inactivity_floor_loc(), 9); assert_eq!(db.live().await.len(), 2); db.assert_exact().await; let expected = [ (keys[0], None), (keys[1], Some(make_value(1))), (keys[2], None), (keys[3], Some(make_value(203))), (keys[4], None), ]; for (key, value) in expected { assert_eq!(db.get(&key).await.unwrap(), value); } // The state survives commit and reopen. let db = db.commit().await.unwrap(); let root = db.root(); let size = db.size(); let floor = db.inactivity_floor_loc(); drop(db); let db = reopen_db(context.child("committed")).await; assert_eq!(db.root(), root); assert_eq!(db.size(), size); assert_eq!(db.inactivity_floor_loc(), floor); for (key, value) in expected { assert_eq!(db.get(&key).await.unwrap(), value); } // The state survives prune and reopen. let db = db.sync().await.unwrap(); let boundary = db.sync_boundary(); let db = db.prune(boundary).await.unwrap(); let root = db.root(); let size = db.size(); let floor = db.inactivity_floor_loc(); drop(db); let db = reopen_db(context.child("pruned")).await; assert_eq!(db.root(), root); assert_eq!(db.size(), size); assert_eq!(db.inactivity_floor_loc(), floor); for (key, value) in expected { assert_eq!(db.get(&key).await.unwrap(), value); } db.destroy().await.unwrap(); } /// Writes to keys that share a translated-key bucket with active updates leave the floor where /// [`walk_model`] predicts from the state after the writes under every limit pair, and the /// walk reads nothing past its window. pub(crate) async fn test_any_policy_limits_after_colliding_writes>( context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where Operation: Codec, { // Lay out the initial commit at 0, six colliding updates at 1..7, and the seed commit // at 7. let keys: Vec<_> = (0..7).map(|i| colliding_digest(0xAA, i)).collect(); let seed: Vec<_> = keys[..6] .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let db = hold(db, &seed).await; let floor = db.inactivity_floor_loc(); let tip = db.size(); assert_eq!((*floor, *tip), (0, 8)); // Updating the last seeded key supersedes its update at 6, and creating a seventh key // supersedes nothing. Every active update shares the written keys' bucket. let writes = [ (keys[5], Some(make_value(105))), (keys[6], Some(make_value(6))), ]; let with = |batch: D::Batch| { writes .iter() .fold(batch, |batch, &(key, value)| batch.write(key, value)) }; // Both orderings append the update at 8 and the create at 9: an ordered batch rewrites no // predecessor, since the created key's predecessor is the updated one. The walk sees the // five untouched updates and both writes, up to the tip 10. let state = written(&db, &[], db.new_batch(), &writes).await; let active: Vec = state.active().into_iter().map(|loc| *loc).collect(); assert_eq!(active, [1, 2, 3, 4, 5, 8, 9]); assert_eq!(*state.tip, 10); // Resolving the writes reads their bucket. [`Hold`] decides nothing, so its batch reads // only what the resolution reads. let items = || counter(&context, "log_journal_items_read_total"); let before = items(); drop(hold_batch(&db, db.new_batch(), &writes).await); let resolution = items() - before; assert!( resolution >= 6, "the resolution reads the six colliding updates" ); // A window of one skip and one entry ends at 2, so the walk reads only the update at 1, // after the resolution's reads and before deciding it. let before = items(); let mut reads = Vec::new(); let mut policy = Script::new(1, 1, |_: &Digest| { reads.push(items()); Choice::Keep }); let merkleized = with(db.new_batch()) .merkleize(&db, None, &mut policy) .await .unwrap(); assert_eq!( policy.visited, [(GenericLocation::new(1), keys[0], make_value(0))] ); assert_eq!(*D::span(&merkleized).inactivity_floor, 2); assert_eq!(reads, [before + resolution + 1]); assert_eq!(items(), before + resolution + 1); drop(merkleized); // Every limit pair matches the model. Entries beyond the five updates below the base decide // the batch's own writes. for entries in 0..=8 { for skips in 0..=8 { let (expected, decisions) = walk_model(&active, *floor, *state.tip, entries, skips, &[]); let mut policy = Script::new(entries, skips, keep); let merkleized = with(db.new_batch()) .merkleize(&db, None, &mut policy) .await .unwrap(); let reached = *D::span(&merkleized).inactivity_floor; let decided: Vec = policy.locations().into_iter().map(|loc| *loc).collect(); assert_eq!(decided, decisions, "entries={entries} skips={skips}"); assert_eq!(reached, expected, "entries={entries} skips={skips}"); } } db.destroy().await.unwrap(); } /// Under every pair of limits in a grid, a batch with no writes and a batch that updates, /// deletes, and creates keys walk the state after their writes, their own writes included, up /// to the tip those writes reach. /// /// A policy that keeps every update, and policies that evict, replace, and stop at chosen /// keys, decide the updates and reach the floor that [`walk_model`] predicts from that state. /// Each of their batches serves the values its decisions leave. /// /// Each batch then applies under a window that ends one skip short of the tip of its writes, /// commits the modeled floor, and serves the model. The batch with writes walks the state the /// applied batch with no writes leaves. /// /// The grid includes zero entries, zero skips, skips that exactly reach the tip, and windows /// that end inside the batch's writes. pub(crate) async fn test_any_policy_limits_with_writes>( _context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where Operation: Codec, { // Lay out the initial commit at 0, superseded updates at 1..4, active updates at 4..7, // the seed commit at 7, active updates at 8..11, and the last commit at 11. let keys = [0x10, 0x20, 0x30, 0x40, 0x50, 0x60, 0x70].map(|prefix| colliding_digest(prefix, 0)); let seed: Vec<_> = keys[..6] .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let mut db = hold(db, &seed).await; let updates: Vec<_> = keys[..3] .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64 + 100)))) .collect(); db = hold(db, &updates).await; let mut model = Model::default(); model.apply(&seed); model.apply(&updates); // The second batch updates an applied key, deletes an updated one, and creates one key // between two others and one past the largest. let writes = [ (keys[4], Some(make_value(204))), (keys[1], None), (colliding_digest(0x35, 0), Some(make_value(235))), (keys[6], Some(make_value(206))), ]; // One script keeps every update. Each other script evicts, replaces, and stops at keys // live in both batches, and keeps the rest. let replacement = make_value(300); let scripts = [ None, Some((keys[3], keys[0], keys[5])), Some((keys[5], keys[3], keys[2])), ]; let choose = |script: Option<(Digest, Digest, Digest)>, key: &Digest| match script { Some((evicted, _, _)) if *key == evicted => Choice::Evict, Some((_, replaced, _)) if *key == replaced => Choice::Replace(replacement), Some((_, _, stopped)) if *key == stopped => Choice::Stop, _ => Choice::Keep, }; for writes in [&[][..], &writes[..]] { let with = |batch: D::Batch| { writes .iter() .fold(batch, |batch, &(key, value)| batch.write(key, value)) }; let mut written_model = model.clone(); written_model.apply(writes); let floor = *db.inactivity_floor_loc(); let base = *db.size(); // The walk sees the update of each live key: the batch's update, both creates, and // each predecessor an ordered batch rewrites lie at or above the base, and the // untouched updates below it. let state = written(&db, &[], db.new_batch(), writes).await; let active: Vec = state.active().into_iter().map(|loc| *loc).collect(); let tip = *state.tip; if writes.is_empty() { assert_eq!( (floor, base, tip, active.as_slice()), (0, 12, 12, [4, 5, 6, 8, 9, 10].as_slice()) ); } else { assert_eq!(active.len(), 7); assert!(active.iter().filter(|loc| **loc >= base).count() >= 3); } let inactive = tip - floor - active.len() as u64; // The model covers the limit regimes the grid exercises: zero entries pass nothing, // zero skips cannot pass the initial commit, and with an entry for every live update // and one more, the walk reaches the tip with exactly as many skips as inactive // locations but not with one fewer. let model_of = |entries, skips| walk_model(&active, floor, tip, entries, skips, &[]); let entries = active.len() + 1; assert_eq!(model_of(0, u64::MAX), (floor, Vec::new())); assert_eq!(model_of(usize::MAX, 0), (floor, Vec::new())); assert_eq!(model_of(entries, inactive), (tip, active.clone())); assert!(model_of(entries, inactive - 1).0 < tip); // Every limit pair matches the model with each script's stops, and the batch serves // each decided key's evicted, replaced, or kept value. With writes, some windows end // inside them after deciding one of them. let mut inside = false; for script in scripts { let stops: Vec = script .iter() .map(|(_, _, stopped)| *state.live[stopped]) .collect(); for entries in 0..=active.len() + 1 { for skips in 0..=inactive + 1 { let (expected, decisions) = walk_model(&active, floor, tip, entries, skips, &stops); let mut policy = Script::new(entries, skips, |key: &Digest| choose(script, key)); let merkleized = with(db.new_batch()) .merkleize(&db, None, &mut policy) .await .unwrap(); let reached = *D::span(&merkleized).inactivity_floor; let decided: Vec = policy.locations().into_iter().map(|loc| *loc).collect(); assert_eq!(decided, decisions, "entries={entries} skips={skips}"); assert_eq!(reached, expected, "entries={entries} skips={skips}"); inside |= (base..tip).contains(&expected) && decided.iter().any(|loc| *loc >= base); let mut decided_model = written_model.clone(); for (_, key, _) in &policy.visited { match choose(script, key) { Choice::Evict => decided_model.apply(&[(*key, None)]), Choice::Replace(value) => { decided_model.apply(&[(*key, Some(value))]) } Choice::Keep | Choice::Stop => {} } } assert_serves(&db, &merkleized, &decided_model).await; } } } assert_eq!(inside, !writes.is_empty()); // The applied batch commits the floor of a window that ends one skip short of the tip. let (expected, _) = model_of(entries, inactive - 1); let mut policy = Script::new(entries, inactive - 1, keep); let merkleized = with(db.new_batch()) .merkleize(&db, None, &mut policy) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(*db.inactivity_floor_loc(), expected); db.assert_exact().await; model = written_model; assert_values(&db, &model).await; } db.destroy().await.unwrap(); } /// Evicting a key a pending parent created leaves it absent once the child applies over the /// still-pending parent: the child's diff deletes the key with no base location, the live /// keys return to the base's, and the applied bitmap never marks the parent's create. #[boxed] pub(crate) async fn test_any_policy_evicts_parent_created_key( context: Context, db: D, reopen: impl Fn(Context) -> Fut, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Changes, Operation: Codec, Fut: Future, { // Seed three keys with a held floor, and create a fourth in a pending parent. let [k1, k2, k3, k4] = [0x10, 0x20, 0x30, 0x40].map(|prefix| colliding_digest(prefix, 0)); let seed: Vec<_> = [k1, k2, k3] .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let db = hold(db, &seed).await; let live = db.live().await.len(); let parent = hold_batch(&db, db.new_batch(), &[(k4, Some(make_value(4)))]).await; let (start, ops) = D::ops(&parent); let created = ops .iter() .position(|op| matches!(op, Operation::Update(update) if *update.key() == k4)) .expect("the parent creates the key"); let created = start + created as u64; // The child evicts the created key and keeps every other update it reaches. let mut policy = Script::new(usize::MAX, u64::MAX, move |key: &Digest| { if *key == k4 { Choice::Evict } else { Choice::Keep } }); let child = D::child(&parent) .merkleize(&db, None, &mut policy) .await .unwrap(); assert!(policy.visited.contains(&(created, k4, make_value(4)))); assert_eq!(D::changes(&child, &k4), [(None, None)]); // Applying the child over the pending parent leaves the base's keys alone and the created // key absent, with an exact bitmap, through commit and reopen. let db = db.apply_batch(child).await.unwrap(); drop(parent); assert_eq!(db.get(&k4).await.unwrap(), None); assert_eq!(db.live().await.len(), live); db.assert_exact().await; let mut model = Model::default(); model.apply(&seed); assert_values(&db, &model).await; let db = db.commit().await.unwrap(); let (root, floor) = (db.root(), db.inactivity_floor_loc()); drop(db); let db = reopen(context.child("reopened")).await; assert_eq!((db.root(), db.inactivity_floor_loc()), (root, floor)); assert_eq!(db.get(&k4).await.unwrap(), None); db.assert_exact().await; assert_values(&db, &model).await; db.destroy().await.unwrap(); } /// A policy decides the batch's own writes like any other active update: it keeps and replaces /// the batch's updates and creates, and evicting a key the batch creates leaves it absent with /// no net change in live keys. A policy that stops at the batch's own write commits the floor /// inside the batch. Both batches survive reopen with an exact activity bitmap. #[boxed] pub(crate) async fn test_any_policy_own_writes( context: Context, db: D, reopen: impl Fn(Context) -> Fut, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, Fut: Future, { // Seed six keys in key order at 1..7, with the seed commit at 7. The batch creates two // more keys between them. let keys = [0x10, 0x20, 0x25, 0x30, 0x40, 0x45, 0x50, 0x60] .map(|prefix| colliding_digest(prefix, 0)); let [k1, k2, k25, k3, k4, k45, k5, k6] = keys; let seed: Vec<_> = [k1, k2, k3, k4, k5, k6] .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let db = hold(db, &seed).await; let mut model = Model::default(); model.apply(&seed); // Update two keys, delete one, and create two whose predecessors are the updated keys, so // an ordered batch rewrites no predecessor. The writes append five operations at 8..13. let writes = [ (k2, Some(make_value(102))), (k4, Some(make_value(104))), (k5, None), (k25, Some(make_value(125))), (k45, Some(make_value(145))), ]; let state = written(&db, &[], db.new_batch(), &writes).await; assert_eq!(*state.tip, 13); let with = |batch: D::Batch| { writes .iter() .fold(batch, |batch, &(key, value)| batch.write(key, value)) }; // The policy keeps, evicts, and replaces the untouched applied updates, keeps and // replaces the two updates the batch writes, keeps one created key, and evicts the other. let choices = [ (k1, Choice::Keep), (k3, Choice::Evict), (k6, Choice::Replace(make_value(606))), (k2, Choice::Keep), (k4, Choice::Replace(make_value(404))), (k25, Choice::Keep), (k45, Choice::Evict), ]; let choose = |key: &Digest| choices.iter().find(|(k, _)| k == key).unwrap().1; // The walk visits the applied updates at 1, 3, and 6, then the four live writes, each // with the value the writes left. Both orderings append the updates before the creates and // keep key order within each. let mut policy = Script::new(usize::MAX, u64::MAX, choose); let merkleized = with(db.new_batch()) .merkleize(&db, None, &mut policy) .await .unwrap(); let visited: Vec<_> = [ (k1, 0), (k3, 2), (k6, 5), (k2, 102), (k4, 104), (k25, 125), (k45, 145), ] .into_iter() .map(|(key, value)| (state.live[&key], key, make_value(value))) .collect(); assert_eq!(policy.visited, visited); assert_eq!( policy.locations()[..3], [1, 3, 6].map(GenericLocation::::new) ); assert!(policy.locations()[3..].iter().all(|loc| **loc >= 8)); // The floor reaches the tip of the writes, and the walk appends its decisions in order. // An ordered batch folds the link past each evicted key into its decided predecessor: the // kept created key precedes the third key, and the replaced fourth key the evicted create. assert_eq!(D::span(&merkleized).inactivity_floor, state.tip); assert_eq!( appended::(&merkleized, &state), [ (k1, Some(make_value(0))), (k3, None), (k6, Some(make_value(606))), (k2, Some(make_value(102))), (k4, Some(make_value(404))), (k25, Some(make_value(125))), (k45, None), ] ); assert_eq!(db.read(&merkleized, &k45).await, None); // The created and evicted key leaves no trace: five keys remain, the six seeded less the // deleted and evicted ones plus the kept create. let db = db.apply_batch(merkleized).await.unwrap(); model.apply(&writes); model.apply(&[ (k3, None), (k6, Some(make_value(606))), (k4, Some(make_value(404))), (k45, None), ]); assert_eq!(db.inactivity_floor_loc(), state.tip); assert!(db.live().await.keys().eq(model.values.keys())); assert_eq!(model.values.len(), 5); assert_values(&db, &model).await; db.assert_exact().await; // The state survives reopen. let db = db.commit().await.unwrap(); let (root, floor) = (db.root(), db.inactivity_floor_loc()); drop(db); let db = reopen(context.child("decided")).await; assert_eq!((db.root(), db.inactivity_floor_loc()), (root, floor)); assert_values(&db, &model).await; db.assert_exact().await; // A batch updates the first key at its first location. Its policy keeps every other live // update, which the previous batch moved, and stops at that write, so the floor stays at // the batch's first location. let start = db.size(); let update = [(k1, Some(make_value(201)))]; let state = written(&db, &[], db.new_batch(), &update).await; assert_eq!(state.live[&k1], start); let mut policy = Script::new(usize::MAX, u64::MAX, |key: &Digest| { if *key == k1 { Choice::Stop } else { Choice::Keep } }); let merkleized = db .new_batch() .write(k1, update[0].1) .merkleize(&db, None, &mut policy) .await .unwrap(); assert_eq!(policy.locations(), state.active()); assert_eq!(policy.visited.last(), Some(&(start, k1, make_value(201)))); assert_eq!(D::span(&merkleized).inactivity_floor, start); let kept: Vec<_> = policy.visited[..4] .iter() .map(|(_, key, value)| (*key, Some(*value))) .collect(); assert_eq!(appended::(&merkleized, &state), kept); // The applied floor survives reopen, and the next batch's policy decides the stopped write // without a skip. let db = db.apply_batch(merkleized).await.unwrap(); model.apply(&update); assert_eq!(db.inactivity_floor_loc(), start); db.assert_exact().await; let db = db.commit().await.unwrap(); let root = db.root(); drop(db); let db = reopen(context.child("stopped")).await; assert_eq!((db.root(), db.inactivity_floor_loc()), (root, start)); assert_values(&db, &model).await; db.assert_exact().await; let mut policy = Script::new(1, 0, keep); drop( db.new_batch() .merkleize(&db, None, &mut policy) .await .unwrap(), ); assert_eq!(policy.visited, [(start, k1, make_value(201))]); db.destroy().await.unwrap(); } /// Merkleize `writes` on batches from `start`, children of the pending `ancestors` (oldest /// first) of `db`, under [`Proportional`], under proportional limits with a policy that decides /// nothing, and under [`Bounded`] with unlimited skips and one entry for the previous commit /// and for each operation the writes make inactive. /// /// Asserts that all three append the same operations under the same floor and root. /// /// Asserts that they move exactly the live updates a [modeled walk](walk_model) with that many /// entries reaches from the inherited floor, each with its value. /// /// Returns the state after the writes, the floor, and the moved locations. async fn assert_proportional( db: &D, ancestors: &[&D::Merkleized], start: impl Fn() -> D::Batch, writes: &[(Digest, Option)], ) -> (Written, u64, Vec) where F: Family, D: Inspect, Operation: Codec, { let with = || { writes .iter() .fold(start(), |batch, &(key, value)| batch.write(key, value)) }; let floor = ancestors.last().map_or_else( || db.inactivity_floor_loc(), |parent| D::span(parent).inactivity_floor, ); let state = written(db, ancestors, start(), writes).await; let entries = state.inactive + 1; let active: Vec = state.active().iter().map(|loc| **loc).collect(); let (expected, moved) = walk_model(&active, *floor, *state.tip, entries, u64::MAX, &[]); // The three policies append the same operations under the same floor and root. let built = build(db, start(), writes).await; let mut policy = Script::proportional(|_: &Digest| Choice::Keep); let proportional = with().merkleize(db, None, &mut policy).await.unwrap(); assert_same(db, &built, &proportional); let mut policy = Bounded { entries, skips: u64::MAX, }; let compacted = with().merkleize(db, None, &mut policy).await.unwrap(); assert_same(db, &built, &compacted); // They move the modeled updates and commit the modeled floor. let at: BTreeMap = state.live.iter().map(|(key, loc)| (**loc, *key)).collect(); let mut kept = Vec::new(); for loc in &moved { let key = at[loc]; kept.push((key, db.read(&built, &key).await)); } assert_eq!(appended::(&built, &state), kept); assert_eq!(*D::span(&built).inactivity_floor, expected); (state, expected, moved) } /// A [`Bounded`] policy with unlimited skips and one entry for the previous commit and for /// each operation the batch's writes make inactive (an update of a key live before the batch, /// an ordered predecessor rewrite included, and two for a delete) reproduces the operations, /// floor, and root of a [`Proportional`] batch, as does a policy with proportional limits that /// decides nothing. Each moves the live updates a walk with that allowance reaches. /// /// From the database and from a pending parent that moved its own floor, the allowance runs /// out below the base tip, runs out among the batch's own writes after moving some, and /// outlasts the live updates so the floor reaches the tip of the writes. pub(crate) async fn test_any_policy_matches_proportional( _context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, { // Seed 32 updates with a held floor, then supersede every third one. let seed: Vec<_> = (0..32) .map(|i| (to_digest(i), Some(make_value(i)))) .collect(); let churn: Vec<_> = (0..32) .step_by(3) .map(|i| (to_digest(i), Some(make_value(i + 100)))) .collect(); let db = hold(db, &seed).await; let db = hold(db, &churn).await; // A small batch updates, deletes, and creates a key. Deleting 16 keys while creating 40 // leaves more live updates than the allowance. Deleting 24 keys and updating 4 leaves // fewer. let small = [ (to_digest(1), Some(make_value(201))), (to_digest(4), None), (to_digest(40), Some(make_value(40))), ]; let exhausted: Vec<_> = (0..16) .map(|i| (to_digest(i), None)) .chain((100..140).map(|i| (to_digest(i), Some(make_value(i))))) .collect(); let outlasted: Vec<_> = (0..24) .map(|i| (to_digest(i), None)) .chain((24..28).map(|i| (to_digest(i), Some(make_value(i + 300))))) .collect(); // Run each batch from the database, then from a pending parent. let parent = build( &db, db.new_batch(), &[ (to_digest(2), Some(make_value(202))), (to_digest(41), Some(make_value(41))), ], ) .await; let starts: [(Vec<&D::Merkleized>, GenericLocation); 2] = [ (Vec::new(), db.bounds().end), (vec![&parent], D::span(&parent).tip.size), ]; for (ancestors, base) in starts { let start = || { ancestors .last() .map_or_else(|| db.new_batch(), |p| D::child(p)) }; let (_, floor, _) = assert_proportional(&db, &ancestors, start, &small).await; assert!(floor < *base, "the allowance outlasts the base region"); let (state, floor, moved) = assert_proportional(&db, &ancestors, start, &exhausted).await; assert!( *base <= floor && floor < *state.tip, "the allowance runs out outside the batch's writes", ); assert!(moved.iter().any(|loc| *loc >= *base), "no own write moved"); let (state, floor, moved) = assert_proportional(&db, &ancestors, start, &outlasted).await; assert_eq!(floor, *state.tip, "the allowance runs out before the tip"); assert!(moved.iter().any(|loc| *loc >= *base), "no own write moved"); } drop(parent); db.destroy().await.unwrap(); } /// Assert that `decided` and `written`, batches of `db`, serve the same value for every key in /// `keys`, that `decided` serves `value` for `target`, and that each commits the tip of its /// writes (`tips`, in order) as its floor. async fn assert_same_state>( db: &D, decided: &D::Merkleized, written: &D::Merkleized, keys: &[Digest], (target, value): (Digest, Option), tips: [GenericLocation; 2], ) { for key in keys { assert_eq!( db.read(decided, key).await, db.read(written, key).await, "{key} diverged", ); } assert_eq!(db.read(decided, &target).await, value); assert_eq!( [decided, written].map(|batch| D::span(batch).inactivity_floor), tips ); } /// Evicting or replacing an update leaves the same state as deleting or writing its key in the /// batch. The update may resolve in the applied database, in a live parent, or in a /// grandparent applied and freed before merkleize. Every key shares one translated-key bucket. /// /// The two batches append different operations: a decision follows the writes, while a write /// resolves among them, and the walk of the writing batch also decides its own write. Each /// policy keeps every other update, so each walk ends at the tip of its batch's writes. pub(crate) async fn test_any_policy_decisions_match_writes( _context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, { // Seed colliding keys with a held floor. let keys: Vec<_> = (0..8).map(|i| colliding_digest(0xAA, i)).collect(); let seed: Vec<_> = keys .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let mut db = hold(db, &seed).await; let target = keys[2]; let decisions = [ (Choice::Evict, None), (Choice::Replace(make_value(500)), Some(make_value(500))), ]; let only = |decision: &Choice| { let decision = *decision; move |key: &Digest| { if *key == target { decision } else { Choice::Keep } } }; // The decided update resolves in the applied database. for (decision, value) in &decisions { let tips = [ written(&db, &[], db.new_batch(), &[]).await.tip, written(&db, &[], db.new_batch(), &[(target, *value)]) .await .tip, ]; let (decided, _) = decide(&db, db.new_batch(), only(decision)).await; let (writing, _) = decide(&db, db.new_batch().write(target, *value), keep).await; assert_same_state(&db, &decided, &writing, &keys, (target, *value), tips).await; } // The decided update resolves in a live parent. let parent = hold_batch( &db, db.new_batch(), &[ (target, Some(make_value(300))), (keys[5], Some(make_value(305))), ], ) .await; for (decision, value) in &decisions { let tips = [ written(&db, &[&parent], D::child(&parent), &[]).await.tip, written(&db, &[&parent], D::child(&parent), &[(target, *value)]) .await .tip, ]; let (decided, _) = decide(&db, D::child(&parent), only(decision)).await; let writing = D::child(&parent).write(target, *value); let (writing, _) = decide(&db, writing, keep).await; assert_same_state(&db, &decided, &writing, &keys, (target, *value), tips).await; } drop(parent); // The decided update resolves in a grandparent that is applied and freed after both // batches start and before merkleize. The decided batch then applies with an exact // activity bitmap, and the next grandparent writes the target again. for (decision, value) in &decisions { let grandparent = hold_batch(&db, db.new_batch(), &[(target, Some(make_value(400)))]).await; let parent = hold_batch( &db, D::child(&grandparent), &[(keys[6], Some(make_value(406)))], ) .await; let tips = [ written(&db, &[&grandparent, &parent], D::child(&parent), &[]) .await .tip, written( &db, &[&grandparent, &parent], D::child(&parent), &[(target, *value)], ) .await .tip, ]; let decided = D::child(&parent); let writing = D::child(&parent).write(target, *value); drop(parent); db = db.apply_batch(grandparent).await.unwrap(); let (decided, _) = decide(&db, decided, only(decision)).await; let (writing, _) = decide(&db, writing, keep).await; assert_same_state(&db, &decided, &writing, &keys, (target, *value), tips).await; drop(writing); db = db.apply_batch(decided).await.unwrap(); assert_eq!(db.get(&target).await.unwrap(), *value); assert_eq!(db.get(&keys[6]).await.unwrap(), Some(make_value(406))); db.assert_exact().await; } db.destroy().await.unwrap(); } /// Staged merkleization with a policy over either update kind. pub(crate) trait StagedPolicy: Sized { /// The merkleized batch. type Merkleized; /// Merkleize with `slots`, `unstaged`, and `policy`, and without metadata. async fn merkleize_staged + Send>( self, slots: Vec>, unstaged: Vec<(Digest, Option)>, db: &D, policy: &mut P, ) -> Self::Merkleized; } impl StagedPolicy, N, S>, F> for batch::Staged, S> where F: Family, C: Mutable>>, I: UnorderedIndex>, V: ValueEncoding, S: Strategy, Operation>: Codec, { type Merkleized = Arc, S>>; async fn merkleize_staged + Send>( self, slots: Vec>, unstaged: Vec<(Digest, Option)>, db: &Db, N, S>, policy: &mut P, ) -> Self::Merkleized { self.merkleize(slots, unstaged, None, db, policy) .await .unwrap() } } impl StagedPolicy, N, S>, F> for batch::Staged, S> where F: Family, C: Mutable>>, I: crate::index::Ordered>, V: ValueEncoding, S: Strategy, Operation>: Codec, { type Merkleized = Arc, S>>; async fn merkleize_staged + Send>( self, slots: Vec>, unstaged: Vec<(Digest, Option)>, db: &Db, N, S>, policy: &mut P, ) -> Self::Merkleized { self.merkleize(slots, unstaged, None, db, policy) .await .unwrap() } } /// Writes staged in the batch supersede their keys' updates, so the policy passes those updates /// as inactive and decides the staged writes themselves. It evicts a staged update that won by /// last-write-wins and replaces an unstaged write. The batch matches an unstaged batch with the /// same writes and policy. A staged key resolves in the applied snapshot or in a live parent. pub(crate) async fn test_any_policy_after_staged_writes( _context: Context, db: Db, make_value: impl Fn(u64) -> Digest, ) where F: Family, C: Mutable>, I: UnorderedIndex> + 'static, U: Update, S: Strategy, Operation: Codec, Db: DbAny< F, Key = Digest, Value = Digest, Digest = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, batch::Staged: StagedPolicy< Db, F, Merkleized = Arc>, >, { // Seed five updates in key order at locations 1..6 with a held floor. let mut keys: Vec<_> = (0..5).map(to_digest).collect(); keys.sort(); let seed: Vec<_> = keys .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let db = hold(db, &seed).await; let writes = [ (keys[1], Some(make_value(101))), (keys[2], None), (keys[3], Some(make_value(103))), ]; // The staged keys resolve in the applied snapshot. let decided = staged_matches_writes(&db, &[], || db.new_batch(), &writes, &make_value).await; assert_eq!(decided[..2], [1, 5].map(GenericLocation::::new)); // The staged update resolves in a live parent that supersedes its applied update. let parent = hold_batch(&db, db.new_batch(), &[(keys[1], Some(make_value(201)))]).await; let decided = staged_matches_writes( &db, &[&parent], || parent.new_batch::(), &writes, &make_value, ) .await; assert_eq!(decided[..2], [1, 5].map(GenericLocation::::new)); drop(parent); db.destroy().await.unwrap(); } /// Stage the first two `writes` through reads of their keys and write the third unstaged, then /// merkleize a batch from `make`, a child of the pending `ancestors`, under unbounded limits /// with a policy that evicts the first written key, replaces the third, and keeps every other /// update. The staged updates write a stale value to the first key before its final one, and to /// the second key before an unstaged write of its final value wins. /// /// Asserts that the policy decides every live update after the writes in location order, the /// two surviving writes included, that the floor reaches the tip of the writes, that the batch /// serves the decided values, and that it matches an unstaged batch with the same writes and /// policy. Returns the decided locations. async fn staged_matches_writes( db: &Db, ancestors: &[&Arc>], make: impl Fn() -> batch::UnmerkleizedBatch, writes: &[(Digest, Option); 3], make_value: &impl Fn(u64) -> Digest, ) -> Vec> where F: Family, C: Mutable>, I: UnorderedIndex> + 'static, U: Update, S: Strategy, Operation: Codec, Db: DbAny< F, Key = Digest, Value = Digest, Digest = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, batch::Staged: StagedPolicy< Db, F, Merkleized = Arc>, >, { // The walk decides every live update after the writes, up to their tip. Two of them are // the batch's surviving writes, at or above the batch's base. let state = written(db, ancestors, make(), writes).await; let expected = state.active(); let base = written(db, ancestors, make(), &[]).await.tip; assert_eq!(expected.iter().filter(|loc| **loc >= base).count(), 2); let replacement = make_value(303); let choose = |key: &Digest| match key { key if *key == writes[0].0 => Choice::Evict, key if *key == writes[2].0 => Choice::Replace(replacement), _ => Choice::Keep, }; // Merkleize the staged writes. The first key's later slot and the unstaged writes carry the // final values in `writes`. let read = [&writes[0].0, &writes[1].0, &writes[0].0]; let (_, staged) = make().stage(&read, db).await.unwrap(); let mut policy = Script::new(usize::MAX, u64::MAX, choose); let staged = staged .merkleize_staged( vec![ Slot::Put(make_value(901)), Slot::Put(make_value(902)), writes[0].1.map_or(Slot::Delete, Slot::Put), ], vec![writes[2], writes[1]], db, &mut policy, ) .await; assert_eq!(policy.locations(), expected); assert_eq!(staged.bounds().inactivity_floor, state.tip); for (key, value) in [ (writes[0].0, None), (writes[1].0, None), (writes[2].0, Some(replacement)), ] { assert_eq!(staged.get(&key, db).await.unwrap(), value); } // An unstaged batch with the same writes and policy decides the same updates and produces // the same batch. let unstaged = writes .iter() .fold(make(), |batch, &(key, value)| batch.write(key, value)); let mut twin = Script::new(usize::MAX, u64::MAX, choose); let unstaged = merkleize(db, unstaged, &mut twin).await.unwrap(); assert_eq!(twin.visited, policy.visited); assert_same(db, &staged, &unstaged); expected } /// A batch whose parent is applied before merkleize decides the same updates, reads only the /// decided updates below the parent's operations, and produces the same batch as a twin over /// the pending parent. A batch whose chain a fork replaced returns `StaleBatch` before /// deciding any update. pub(crate) async fn test_any_policy_after_ancestor_applied( context: Context, db: Db, reopen: impl Fn(Context) -> Fut, make_value: impl Fn(u64) -> Digest, ) where F: Family, C: Mutable>, I: UnorderedIndex> + 'static, U: Update, S: Strategy, Operation: Codec, Db: DbAny< F, Key = Digest, Value = Digest, Digest = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, Fut: Future>, { // Seed eight updates in key order at locations 1..9 with a held floor. let mut keys: Vec<_> = (0..8).map(to_digest).collect(); keys.sort(); let seed: Vec<_> = keys .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let db = hold(db, &seed).await; let superseded = GenericLocation::::new(2); // A pending parent supersedes the second update. let parent = hold_batch(&db, db.new_batch(), &[(keys[1], Some(make_value(101)))]).await; // A twin over the pending parent keeps every update. let (twin, expected) = decide(&db, parent.new_batch::(), keep).await; assert!(!expected.contains(&superseded)); // Apply the parent, then merkleize a batch started before the apply. The parent's // update resolves in memory. let batch = parent.new_batch::(); let base = parent.bounds().base.size; let db = db.apply_batch(parent).await.unwrap(); let before = counter(&context, "log_journal_items_read_total"); let (merkleized, decided) = decide(&db, batch, keep).await; assert_eq!(decided, expected); let below_base = expected.iter().filter(|loc| **loc < base).count() as u64; assert!(below_base < expected.len() as u64); assert_eq!( counter(&context, "log_journal_items_read_total"), before + below_base ); assert_same(&db, &twin, &merkleized); drop(twin); // The applied batch serves the parent's write and survives reopen. let root = merkleized.root(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.root(), root); assert_eq!(db.get(&keys[1]).await.unwrap(), Some(make_value(101))); let db = db.commit().await.unwrap(); drop(db); let db = reopen(context.child("reopen")).await; assert_eq!(db.root(), root); assert_eq!(db.get(&keys[1]).await.unwrap(), Some(make_value(101))); // A fork replaces the chain of a pending batch. The batch returns `StaleBatch` before // deciding any update. let parent = hold_batch(&db, db.new_batch(), &[(keys[3], Some(make_value(103)))]).await; let fork = hold_batch(&db, db.new_batch(), &[(keys[4], Some(make_value(104)))]).await; let batch = parent.new_batch::(); let db = db.apply_batch(fork).await.unwrap(); let mut policy = Script::new(usize::MAX, u64::MAX, keep); assert!(matches!( merkleize(&db, batch, &mut policy).await, Err(crate::qmdb::Error::StaleBatch) )); assert!(policy.visited.is_empty()); drop(parent); db.destroy().await.unwrap(); } /// Database access the policy tests need beyond [`DbAny`]. pub(crate) trait Inspect: DbAny { /// The update kind. type Update: Update; /// Assert that the activity bitmap is [exact](assert_exact). async fn assert_exact(&self); /// Return the location of every live key's update in a [replay](live) of the retained log. async fn live(&self) -> BTreeMap>; /// Start a child of `batch`. fn child(batch: &Self::Merkleized) -> Self::Batch; /// Return the bounds of `batch`. fn span(batch: &Self::Merkleized) -> &Bounds; /// Return the operations `batch` appends and the location of the first. #[allow(clippy::type_complexity)] fn ops( batch: &Self::Merkleized, ) -> (GenericLocation, Arc>>); /// Read `key` through `batch`. async fn read(&self, batch: &Self::Merkleized, key: &Digest) -> Option; } impl Inspect for Db where F: Family, C: Mutable>, I: UnorderedIndex> + 'static, U: Update, S: Strategy, Operation: Codec, Self: DbAny< F, Key = Digest, Value = Digest, Digest = Digest, Merkleized = Arc>, Batch = batch::UnmerkleizedBatch, >, { type Update = U; async fn assert_exact(&self) { assert_exact(self).await; } async fn live(&self) -> BTreeMap> { live(self).await } fn child(batch: &Self::Merkleized) -> Self::Batch { batch.new_batch::() } fn span(batch: &Self::Merkleized) -> &Bounds { batch.bounds() } fn ops(batch: &Self::Merkleized) -> (GenericLocation, Arc>>) { batch.operations() } async fn read(&self, batch: &Self::Merkleized, key: &Digest) -> Option { batch.get(key, self).await.unwrap() } } /// Ordered database access the link tests need beyond [`Inspect`]. pub(crate) trait Links: Inspect { /// Assert that `key` holds `value` and links to `next`. async fn assert_link(&self, key: Digest, value: Digest, next: Digest); /// Assert that `key` is absent. async fn assert_absent(&self, key: Digest); } impl Links for Db, N, S> where F: Family, C: Mutable>>, I: crate::index::Ordered> + 'static, V: ValueEncoding, S: Strategy, Operation>: Codec, Self: Inspect, { async fn assert_link(&self, key: Digest, value: Digest, next: Digest) { assert_eq!( self.get_all(&key).await.unwrap(), Some((value, next)), "{key} diverged from its value and link", ); } async fn assert_absent(&self, key: Digest) { assert_eq!(self.get_all(&key).await.unwrap(), None, "{key} is live"); } } /// Assert that two merkleized batches of `db` append the same operations under the same floor /// and root. pub(crate) fn assert_same(_: &D, a: &D::Merkleized, b: &D::Merkleized) where F: Family, D: Inspect, Operation: Codec, { let encoded = |merkleized: &D::Merkleized| { let (start, ops) = D::ops(merkleized); let ops: Vec> = ops.iter().map(|op| op.encode().to_vec()).collect(); (start, ops) }; assert_eq!(encoded(a), encoded(b)); assert_eq!(D::span(a).inactivity_floor, D::span(b).inactivity_floor); assert_eq!(MerkleizedTrait::root(a), MerkleizedTrait::root(b)); } /// Assert that `db` serves the value `model` holds for every written key. async fn assert_values(db: &D, model: &Model) where F: Family, D: DbAny, { for key in &model.keys { assert_eq!( db.get(key).await.unwrap(), model.values.get(key).copied(), "value of {key} diverged from the model", ); } } /// Assert that `batch` serves the value `model` holds for every written key. async fn assert_serves>(db: &D, batch: &D::Merkleized, model: &Model) { for key in &model.keys { assert_eq!( db.read(batch, key).await, model.values.get(key).copied(), "value of {key} diverged from the model", ); } } /// Merkleize `batch` with a policy and apply the result to `db`. `batch` is a child of the /// pending `ancestors` (oldest first). The policy writes the value `decisions` holds for a key /// (`None` evicts) and keeps every other update. It decides the live update of every key in /// location order. The applied state serves `model` with the decisions recorded and keeps an /// exact activity bitmap. async fn apply_decided>( db: D, ancestors: &[&D::Merkleized], batch: D::Batch, decisions: &[(Digest, Option)], model: &mut Model, ) -> D where Operation: Codec, { // Replay the log, then each ancestor. The policy decides every live update. let expected = active(&db, ancestors).await; // Decide each update, then apply over the pending ancestors. let choose = |key: &Digest| match decisions.iter().find(|(k, _)| k == key) { Some((_, Some(value))) => Choice::Replace(*value), Some((_, None)) => Choice::Evict, None => Choice::Keep, }; let (merkleized, decided) = decide(&db, batch, choose).await; assert_eq!(decided, expected); model.apply(decisions); let db = db.apply_batch(merkleized).await.unwrap(); assert_values(&db, model).await; db.assert_exact().await; db } /// Over two pending ancestors, one, and none, and for batches started before their ancestors /// are applied, the [`Proportional`] walk produces the same batch, and so does a policy that /// keeps every update. Policies that keep, evict, and replace each ancestor's updates then /// apply over two pending ancestors and over one. Every policy decides exactly the live /// updates after its batch's writes, the batch's own included, every batch serves the model's /// values, and every applied state keeps an exact activity bitmap. Every key shares one /// translated-key bucket. #[boxed] pub(crate) async fn test_any_activity_depths( context: Context, db: D, reopen: impl Fn(Context) -> Fut, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, Fut: Future, { // Seed eight keys with a held floor. let key = |i: u64| colliding_digest(0xB0, i); let seed: Vec<_> = (0..8).map(|i| (key(i), Some(make_value(i)))).collect(); let mut model = Model::default(); let db = hold(db, &seed).await; model.apply(&seed); db.assert_exact().await; // The grandparent updates and deletes seeded keys and creates two keys. The parent updates // and deletes seeded keys and the grandparent's creates, and creates a key. let grand = [ (key(1), Some(make_value(101))), (key(2), None), (key(100), Some(make_value(102))), (key(101), Some(make_value(103))), ]; let middle = [ (key(3), Some(make_value(201))), (key(4), None), (key(100), Some(make_value(202))), (key(101), None), (key(102), Some(make_value(203))), ]; let grandparent = hold_batch(&db, db.new_batch(), &grand).await; let parent = hold_batch(&db, D::child(&grandparent), &middle).await; model.apply(&grand); model.apply(&middle); // The batch deletes and updates seeded keys, updates the grandparent's update, and updates // the parent's create. let writes = [ (key(0), None), (key(1), Some(make_value(301))), (key(5), Some(make_value(302))), (key(102), Some(make_value(303))), ]; let with = |batch: D::Batch| { writes .iter() .fold(batch, |batch, &(k, v)| batch.write(k, v)) }; // Replay the log, the grandparent, the parent, and the writes. A policy decides the live // update of every key in location order, including the three the batch writes. let state = written(&db, &[&grandparent, &parent], D::child(&parent), &writes).await; let expected = state.active(); let base = D::span(&parent).tip.size; assert_eq!(expected.iter().filter(|loc| **loc >= base).count(), 3); model.apply(&writes); // Depth 2: the proportional walk passes the grandparent's operations, and a policy decides // the expected updates. let proportional = build(&db, D::child(&parent), &writes).await; assert!( D::span(&proportional).inactivity_floor > D::span(&parent).base.size, "the walk passes the grandparent's operations", ); assert_serves(&db, &proportional, &model).await; let (kept, decided) = decide(&db, with(D::child(&parent)), keep).await; assert_eq!(decided, expected); assert_serves(&db, &kept, &model).await; // A proportional batch and two policy batches start before the grandparent is applied. let early = with(D::child(&parent)); let first = with(D::child(&parent)); let last = with(D::child(&parent)); // Depth 1: apply and free the grandparent. The proportional and policy batches started // after the apply, and the proportional and first policy batches started before it, match // depth 2. The first policy batch has one more entry than there are updates to decide, so // its walk reaches the tip in read rounds sized to its remaining entries. let db = db.apply_batch(grandparent).await.unwrap(); db.assert_exact().await; let proportional1 = build(&db, D::child(&parent), &writes).await; assert_same(&db, &proportional, &proportional1); let early = early.merkleize(&db, None, &mut Proportional).await.unwrap(); assert_same(&db, &proportional, &early); let (kept1, decided1) = decide(&db, with(D::child(&parent)), keep).await; assert_eq!(decided1, expected); assert_same(&db, &kept, &kept1); let mut policy = Script::new(expected.len() + 1, u64::MAX, keep); let first = first.merkleize(&db, None, &mut policy).await.unwrap(); assert_eq!(policy.locations(), expected); assert_same(&db, &kept, &first); // Depth 0: apply the parent. The proportional and policy batches started after the apply, // and the last policy batch started before the grandparent's apply, match depth 2. let db = db.apply_batch(parent).await.unwrap(); db.assert_exact().await; let (last, decided) = decide(&db, last, keep).await; assert_eq!(decided, expected); assert_same(&db, &kept, &last); let proportional0 = build(&db, db.new_batch(), &writes).await; assert_same(&db, &proportional, &proportional0); let (kept0, decided0) = decide(&db, with(db.new_batch()), keep).await; assert_eq!(decided0, expected); assert_same(&db, &kept, &kept0); // Apply the proportional batch. The state serves the model and keeps an exact bitmap. drop(( proportional, proportional1, early, kept, kept1, first, last, kept0, )); let db = db.apply_batch(proportional0).await.unwrap(); assert_values(&db, &model).await; db.assert_exact().await; // Apply a policy that keeps every live update. The state serves the model and keeps an // exact bitmap. let expected = active(&db, &[]).await; let (kept, decided) = decide(&db, db.new_batch(), keep).await; assert_eq!(decided, expected); let db = db.apply_batch(kept).await.unwrap(); assert_values(&db, &model).await; db.assert_exact().await; // Depth 2 apply: the grandparent and the parent each update two existing keys and create a // key. The policy keeps one update of each ancestor, evicts the other, and replaces the // create. let grand = [ (key(1), Some(make_value(401))), (key(6), Some(make_value(402))), (key(103), Some(make_value(403))), ]; let middle = [ (key(3), Some(make_value(404))), (key(102), Some(make_value(405))), (key(104), Some(make_value(406))), ]; let grandparent = hold_batch(&db, db.new_batch(), &grand).await; let parent = hold_batch(&db, D::child(&grandparent), &middle).await; model.apply(&grand); model.apply(&middle); let decisions = [ (key(6), None), (key(103), Some(make_value(407))), (key(102), None), (key(104), Some(make_value(408))), ]; let ancestors = [&grandparent, &parent]; let db = apply_decided(db, &ancestors, D::child(&parent), &decisions, &mut model).await; drop((grandparent, parent)); // Depth 1 apply: the parent updates three existing keys and creates a key. The policy keeps // one update, evicts another and the create, and replaces the third. let middle = [ (key(1), Some(make_value(501))), (key(5), Some(make_value(502))), (key(104), Some(make_value(503))), (key(105), Some(make_value(504))), ]; let parent = hold_batch(&db, db.new_batch(), &middle).await; model.apply(&middle); let decisions = [ (key(5), None), (key(104), Some(make_value(505))), (key(105), None), ]; let db = apply_decided(db, &[&parent], D::child(&parent), &decisions, &mut model).await; drop(parent); // Prune to the sync boundary. The bitmap stays exact. let db = db.commit().await.unwrap(); let boundary = db.sync_boundary(); let db = db.prune(boundary).await.unwrap(); db.assert_exact().await; // Reopen. The rebuilt state matches and keeps an exact bitmap. let root = db.root(); let db = db.commit().await.unwrap(); drop(db); let db = reopen(context.child("reopen")).await; assert_eq!(db.root(), root); assert_values(&db, &model).await; db.assert_exact().await; db.destroy().await.unwrap(); } /// Return the keys of `live` ordered by the location of their updates, oldest first. pub(crate) fn age(live: &BTreeMap) -> Vec { let mut entries: Vec<_> = live.iter().map(|(key, loc)| (*loc, *key)).collect(); entries.sort_unstable(); entries.into_iter().map(|(_, key)| key).collect() } /// Return the keys live after applying `writes`, in order, to the keys of `live`. pub(crate) fn keys_after( live: &BTreeMap, writes: &[(K, Option)], ) -> BTreeSet { let mut keys: BTreeSet<_> = live.keys().copied().collect(); for (key, value) in writes { if value.is_some() { keys.insert(*key); } else { keys.remove(key); } } keys } /// Assert the [`Proportional`] bound at a batch boundary with `floor`, exclusive tip `tip`, /// and the location of every live key's update in `live`. /// /// Every live update lies at or above the floor and below the commit at `tip - 1`. /// /// When every batch since the initial commit is [`Proportional`], the floor trails the tip by /// at most `3 * n + 1` operations for `n` live keys. pub(crate) fn assert_bound( floor: GenericLocation, tip: GenericLocation, live: &BTreeMap>, ) { let n = live.len() as u64; assert!(floor < tip, "floor={floor:?}, tip={tip:?}"); assert!( live.values().all(|loc| *loc >= floor && **loc < *tip - 1), "a live update lies outside the floor and the commit: floor={floor:?}, tip={tip:?}", ); assert!( *tip - *floor <= 3 * n + 1, "tip={tip:?}, floor={floor:?}, n={n}, gap={}, bound={}", *tip - *floor, 3 * n + 1, ); } /// Return deletes of the `n` live keys with the highest locations. fn newest( live: &BTreeMap, n: usize, ) -> Vec<(K, Option)> { age(live) .into_iter() .rev() .take(n) .map(|key| (key, None)) .collect() } /// Apply a large batch of creates, hot-key updates, delete and recreate churn, shrinking by /// deleting the newest keys, bursts of creates followed by deletes of the newest keys, and /// large mixed batches to `db`. Keys come from `key` and values from `value`. `apply` applies /// one batch and records the location of each live key in its map. pub(crate) async fn churn( mut db: D, key: impl Fn(u64) -> K, value: impl Fn(u64) -> V, mut apply: impl AsyncFnMut(D, &mut BTreeMap, &[(K, Option)]) -> D, ) -> D where K: Copy + Ord, L: Copy + Ord, { let mut live = BTreeMap::new(); // Create 256 keys in one batch. let writes: Vec<_> = (0..256).map(|i| (key(i), Some(value(i)))).collect(); db = apply(db, &mut live, &writes).await; // Update four hot keys in each of 64 batches. for round in 0..64 { let writes: Vec<_> = (0..4) .map(|i| (key(i), Some(value(1000 + round)))) .collect(); db = apply(db, &mut live, &writes).await; } // Delete eight keys, then delete the next eight in each of 64 batches while recreating // the eight the previous batch deleted. let window = |round: u64| (8 * round..8 * round + 8).map(|i| key(i % 256)); let writes: Vec<_> = window(0).map(|key| (key, None)).collect(); db = apply(db, &mut live, &writes).await; for round in 0..64 { let writes: Vec<_> = window(round) .map(|key| (key, Some(value(2000 + round)))) .chain(window(round + 1).map(|key| (key, None))) .collect(); db = apply(db, &mut live, &writes).await; } // Delete the eight newest keys in each batch until eight keys remain. while live.len() > 8 { let writes = newest(&live, 8); db = apply(db, &mut live, &writes).await; } // Four times, create a burst of 64 keys and then delete the eight newest keys in each of // eight batches. for burst in 0..4 { let start = 1000 + 64 * burst; let writes: Vec<_> = (start..start + 64) .map(|i| (key(i), Some(value(i)))) .collect(); db = apply(db, &mut live, &writes).await; for _ in 0..8 { let writes = newest(&live, 8); db = apply(db, &mut live, &writes).await; } } // Create 512 keys in one batch, then delete 64 keys, update 128, and create 32 in each of // four batches. let writes: Vec<_> = (2000..2512).map(|i| (key(i), Some(value(i)))).collect(); db = apply(db, &mut live, &writes).await; for round in 0..4 { let keys: Vec<_> = live.keys().copied().collect(); let start = 3000 + 32 * round; let deletes = keys[..64].iter().map(|&key| (key, None)); let updates = keys[64..192].iter().map(|&key| (key, Some(value(round)))); let creates = (start..start + 32).map(|i| (key(i), Some(value(i)))); let writes: Vec<_> = deletes.chain(updates).chain(creates).collect(); db = apply(db, &mut live, &writes).await; } db } /// Apply a fixed then a randomized schedule of batches to `db`, drawing choices from /// `context`. Keys come from `key` and values from `value`. `apply` applies one batch and /// records the location of each live key in its map. /// /// The fixed prefix creates 20 keys, deletes all but the oldest in one batch, and empties the /// database. It then creates 19 keys, deletes the three newest in each batch until one /// remains, and empties the database again. /// /// Each of two epochs then creates 96 keys, rewrites hot keys, and deletes the newest (first /// epoch) or the oldest (second epoch) keys in batches until one remains. It applies batches /// of random writes and empty batches, and ends by emptying the database. pub(crate) async fn randomized_churn( context: &mut Context, mut db: D, key: impl Fn(u64) -> K, value: impl Fn(u64) -> V, mut apply: impl AsyncFnMut(D, &mut BTreeMap, &[(K, Option)]) -> D, ) -> D where K: Copy + Ord, L: Copy + Ord, { let mut live = BTreeMap::new(); // Delete all but the oldest of 20 keys in one batch, then empty the database. let writes: Vec<_> = (0..20).map(|i| (key(i), Some(value(i)))).collect(); db = apply(db, &mut live, &writes).await; let writes: Vec<_> = age(&live).into_iter().skip(1).map(|k| (k, None)).collect(); db = apply(db, &mut live, &writes).await; assert_eq!(live.len(), 1); let writes: Vec<_> = live.keys().map(|k| (*k, None)).collect(); db = apply(db, &mut live, &writes).await; db = apply(db, &mut live, &[]).await; // Create 19 keys, delete the three newest in each batch until one remains, then empty the // database. let writes: Vec<_> = (0..19).map(|i| (key(i), Some(value(i)))).collect(); db = apply(db, &mut live, &writes).await; while live.len() > 1 { let writes = newest(&live, 3.min(live.len() - 1)); db = apply(db, &mut live, &writes).await; } let writes: Vec<_> = live.keys().map(|k| (*k, None)).collect(); db = apply(db, &mut live, &writes).await; for epoch in 0..2u64 { let writes: Vec<_> = (0..96) .map(|i| (key(i), Some(value(epoch * 1000 + i)))) .collect(); db = apply(db, &mut live, &writes).await; // Rewrite up to four of eight hot keys in each of 32 batches. let hot: Vec<_> = live.keys().copied().take(8).collect(); for round in 0..32u64 { let count = context.random_range(0..=4u64); let writes: Vec<_> = (0..count) .map(|i| { let k = hot[context.random_range(0..hot.len())]; (k, Some(value(epoch * 1000 + round * 8 + i))) }) .collect(); db = apply(db, &mut live, &writes).await; } // Delete up to eight of the newest (first epoch) or oldest (second epoch) keys in each // batch until one remains. while live.len() > 1 { let count = context.random_range(1..=8.min(live.len() - 1)); let mut keys = age(&live); if epoch == 0 { keys.reverse(); } let writes: Vec<_> = keys.into_iter().take(count).map(|k| (k, None)).collect(); db = apply(db, &mut live, &writes).await; } // Apply 48 batches of up to 12 random writes over 160 keys, each a delete with // probability 0.45. They likely create, update, and delete live keys, delete absent // keys, write a key twice, and recreate deleted keys. An empty batch follows each // round divisible by eight. for round in 0..48 { let count = context.random_range(0..=12usize); let writes: Vec<_> = (0..count) .map(|_| { let k = key(context.random_range(0..160u64)); let v = context .random_bool(0.55) .then(|| value(context.random::())); (k, v) }) .collect(); db = apply(db, &mut live, &writes).await; if round % 8 == 0 { db = apply(db, &mut live, &[]).await; } } // Delete all but the oldest key, apply an empty batch, then empty the database. if live.is_empty() { db = apply(db, &mut live, &[(key(0), Some(value(epoch)))]).await; } let writes: Vec<_> = age(&live).into_iter().skip(1).map(|k| (k, None)).collect(); db = apply(db, &mut live, &writes).await; assert_eq!(live.len(), 1); db = apply(db, &mut live, &[]).await; let writes: Vec<_> = live.keys().map(|k| (*k, None)).collect(); db = apply(db, &mut live, &writes).await; for _ in 0..3 { db = apply(db, &mut live, &[]).await; assert!(live.is_empty()); } } db } /// Merkleize and apply `writes` as one [`Proportional`] batch, replaying its operations into /// `live`. Asserts that the live keys match an independent key-set model and that both the /// merkleized and the applied state satisfy [`assert_bound`], so every batch since the initial /// commit must be [`Proportional`]. async fn bounded>( db: D, live: &mut BTreeMap>, writes: &[(Digest, Option)], ) -> D where Operation: Codec, { let expected = keys_after(live, writes); let merkleized = build(&db, db.new_batch(), writes).await; let (start, ops) = D::ops(&merkleized); replay(live, start, &ops); assert_eq!(live.keys().copied().collect::>(), expected); let span = D::span(&merkleized); let (floor, tip) = (span.inactivity_floor, span.tip.size); assert_bound(floor, tip, live); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!((db.inactivity_floor_loc(), db.size()), (floor, tip)); db } /// Merkleize `writes` as one [`Proportional`] batch on the newest of the `pending` batches, /// or on the database when none is pending, and assert [`assert_bound`] on its bounds and the /// live keys after it and its ancestors. Applies the newest batch, and with it its ancestors, /// once three are pending. async fn pending_bounded>( (mut db, mut pending): (D, Vec), live: &mut BTreeMap>, writes: &[(Digest, Option)], ) -> (D, Vec) where Operation: Codec, { let expected = keys_after(live, writes); let batch = pending.last().map_or_else(|| db.new_batch(), D::child); let merkleized = build(&db, batch, writes).await; let (start, ops) = D::ops(&merkleized); replay(live, start, &ops); assert_eq!(live.keys().copied().collect::>(), expected); let span = D::span(&merkleized); assert_bound(span.inactivity_floor, span.tip.size, live); pending.push(merkleized); if pending.len() == 3 { let tip = pending.pop().unwrap(); db = db.apply_batch(tip).await.unwrap(); pending.clear(); assert_bound(db.inactivity_floor_loc(), db.size(), live); } (db, pending) } /// Every [`Proportional`] batch of [`randomized_churn`] satisfies [`assert_bound`] before /// and after it applies. Every batch since the initial commit is [`Proportional`], as the /// bound requires. /// /// The final state's activity bitmap is exact. pub(crate) async fn test_any_proportional_randomized_bound( mut context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, { let db = randomized_churn(&mut context, db, to_digest, make_value, bounded::).await; db.assert_exact().await; db.destroy().await.unwrap(); } /// Every [`Proportional`] batch of [`randomized_churn`], merkleized on a chain of up to two /// pending ancestors, satisfies [`assert_bound`] over the chain's live keys, and so does each /// applied state. Every batch since the initial commit is [`Proportional`], as the bound /// requires. pub(crate) async fn test_any_proportional_pending_bound( mut context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, { let (mut db, mut pending) = randomized_churn( &mut context, (db, Vec::new()), to_digest, make_value, pending_bounded::, ) .await; if let Some(tip) = pending.pop() { db = db.apply_batch(tip).await.unwrap(); } drop(pending); db.assert_exact().await; db.destroy().await.unwrap(); } /// Every batch of [`churn`] satisfies [`assert_bound`]. Every batch since the initial commit /// is [`Proportional`], as the bound requires. pub(crate) async fn test_any_proportional_bound( _context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, { let db = churn(db, to_digest, make_value, bounded::).await; db.destroy().await.unwrap(); } /// A [`Proportional`] batch moves one active update for each operation it makes inactive: two /// for each delete (the delete and the update it supersedes), one for each update it /// supersedes, and one for its previous commit. /// /// Its walk ends at the tip of its writes, so deleting 19 of 20 keys in one batch leaves the /// floor two operations below the tip, within `3 * n + 1` for the one remaining key. /// Every batch since the initial commit is [`Proportional`], as the bound requires. pub(crate) async fn test_any_proportional_one_batch( _context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, Operation: Codec, { // Seed 20 keys in key order at 1..21. The seed makes only its previous commit inactive, so // it moves one update, the smallest key's from 1 to 21, and the floor passes it. let mut keys: Vec<_> = (0..20).map(to_digest).collect(); keys.sort(); let seed: Vec<_> = keys .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let merkleized = build(&db, db.new_batch(), &seed).await; let db = db.apply_batch(merkleized).await.unwrap(); let live = db.live().await; assert_eq!((*db.inactivity_floor_loc(), *db.size()), (2, 23)); assert_eq!(*live[&keys[0]], 21); // Deleting the five newest keys (the smallest and the four largest) makes ten operations // inactive, plus one for each predecessor an ordered batch rewrites (an update of a live // key, which supersedes it), plus the previous commit. The live updates from 2 on are // contiguous, so the walk moves that many of them and the floor passes them. let deletes = newest(&live, 5); let state = written(&db, &[], db.new_batch(), &deletes).await; let rewrites = state.live.values().filter(|loc| **loc >= db.size()).count(); let entries = 2 * deletes.len() + rewrites + 1; let active = state.active(); assert!( active[..entries] .iter() .zip(2..) .all(|(loc, at)| **loc == at) ); let at: BTreeMap<_, _> = state.live.iter().map(|(key, loc)| (*loc, *key)).collect(); let moved: Vec<_> = active[..entries] .iter() .map(|loc| { let key = at[loc]; let i = keys.binary_search(&key).unwrap(); (key, Some(make_value(i as u64))) }) .collect(); let merkleized = build(&db, db.new_batch(), &deletes).await; assert_eq!(appended::(&merkleized, &state), moved); assert_eq!(*D::span(&merkleized).inactivity_floor, 2 + entries as u64); drop(merkleized); // Delete every key but the one at 2. The only live update after the writes is its own: // at 2, or where an ordered batch rewrites it as the deleted keys' predecessor. The walk // moves it and reaches the tip of the writes. let deletes: Vec<_> = keys .iter() .filter(|key| **key != keys[1]) .map(|key| (*key, None)) .collect(); let state = written(&db, &[], db.new_batch(), &deletes).await; assert!(state.live.keys().eq([&keys[1]])); let merkleized = build(&db, db.new_batch(), &deletes).await; assert_eq!( appended::(&merkleized, &state), [(keys[1], Some(make_value(1)))] ); assert_eq!(D::span(&merkleized).inactivity_floor, state.tip); // The commit follows the move, so the floor trails the tip by two. let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(*db.size() - *db.inactivity_floor_loc(), 2); assert_eq!(db.get(&keys[1]).await.unwrap(), Some(make_value(1))); db.assert_exact().await; db.destroy().await.unwrap(); } /// The same batch, a child of a pending parent of a pending grandparent, makes the same /// decisions and appends the same operations under the same floor and root whether it /// merkleizes with both ancestors pending, after the grandparent is applied while still /// referenced and after its reference is dropped, and after the parent is applied while still /// referenced and after its reference is dropped. Its policy evicts and replaces updates of /// the applied state, of each ancestor, and of the batch itself, and stops at the batch's /// own create. The applied batch serves the model's values with an exact activity bitmap. #[boxed] pub(crate) async fn test_any_policy_ancestor_twins( _context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Inspect, D::Merkleized: Clone, Operation: Codec, { // Seed eight keys and a sibling of the third in its translated-key bucket. let key = |prefix: u8| colliding_digest(prefix, 0); let sibling = colliding_digest(0x30, 1); let seed: Vec<_> = [0x10, 0x20, 0x30, 0x40, 0x50, 0x60, 0x70, 0x80] .map(key) .into_iter() .chain([sibling]) .enumerate() .map(|(i, key)| (key, Some(make_value(i as u64)))) .collect(); let db = hold(db, &seed).await; let mut model = Model::default(); model.apply(&seed); // The grandparent updates, creates, and deletes a key. The parent updates the // grandparent's create and a seeded key and deletes the sibling. The batch updates a // seeded key, deletes the smallest, and creates the largest. let grand = [ (key(0x20), Some(make_value(102))), (key(0x45), Some(make_value(145))), (key(0x70), None), ]; let middle = [ (key(0x45), Some(make_value(245))), (key(0x30), Some(make_value(230))), (sibling, None), ]; let writes = [ (key(0x50), Some(make_value(350))), (key(0x10), None), (key(0x85), Some(make_value(385))), ]; let grandparent = hold_batch(&db, db.new_batch(), &grand).await; let parent = hold_batch(&db, D::child(&grandparent), &middle).await; let with = |batch: D::Batch| { writes .iter() .fold(batch, |batch, &(key, value)| batch.write(key, value)) }; let batches: Vec<_> = (0..5).map(|_| with(D::child(&parent))).collect(); // The policy evicts the grandparent's update, a key whose ordered predecessor the // grandparent creates, and the batch's own update, replaces the parent's update of the // grandparent's create, and stops at the batch's create. let evicted = [0x20, 0x40, 0x50].map(key); let (replaced, replacement, stop) = (key(0x45), make_value(445), key(0x85)); let choose = |k: &Digest| { if evicted.contains(k) { Choice::Evict } else if *k == replaced { Choice::Replace(replacement) } else if *k == stop { Choice::Stop } else { Choice::Keep } }; let state = written(&db, &[&grandparent, &parent], D::child(&parent), &writes).await; let mut batches = batches.into_iter(); let mut twin = async |db: &D| { let mut policy = Script::new(usize::MAX, u64::MAX, choose); let merkleized = batches .next() .unwrap() .merkleize(db, None, &mut policy) .await .unwrap(); (merkleized, policy.visited) }; // With both ancestors pending, the walk visits every evicted and replaced update before // the batch's create, where it stops and leaves the floor. let (pending, visited) = twin(&db).await; let stopped = state.live[&stop]; assert_eq!(visited.last().map(|(loc, ..)| *loc), Some(stopped)); assert_eq!(D::span(&pending).inactivity_floor, stopped); for target in evicted.iter().chain([&replaced]) { assert!(visited.iter().any(|(_, key, _)| key == target)); } // Apply each ancestor while holding a reference to it, then drop the reference. Each // batch matches the one over pending ancestors. let retained = grandparent.clone(); let db = db.apply_batch(grandparent).await.unwrap(); let mut twins = vec![twin(&db).await]; drop(retained); twins.push(twin(&db).await); let retained = parent.clone(); let db = db.apply_batch(parent).await.unwrap(); twins.push(twin(&db).await); drop(retained); twins.push(twin(&db).await); for (merkleized, decided) in &twins { assert_eq!(*decided, visited); assert_same(&db, &pending, merkleized); } let last = twins.pop().unwrap().0; drop((pending, twins)); // Apply the batch. It serves the model's values with an exact bitmap. let db = db.apply_batch(last).await.unwrap(); model.apply(&grand); model.apply(&middle); model.apply(&writes); model.apply(&[ (key(0x20), None), (key(0x40), None), (key(0x50), None), (key(0x45), Some(make_value(445))), ]); assert_eq!(db.inactivity_floor_loc(), stopped); assert_values(&db, &model).await; assert!(db.live().await.keys().eq(model.values.keys())); db.assert_exact().await; db.destroy().await.unwrap(); } /// Assert that `db` holds exactly the `live` keys with their values, that each links to the /// next live key and the largest to the smallest, that every `absent` key is absent, and that /// the activity bitmap is [exact](assert_exact). pub(crate) async fn assert_links>( db: &D, live: &BTreeMap, absent: &[Digest], ) { assert!(db.live().await.keys().eq(live.keys()), "live keys diverged"); let next = live.keys().cycle().skip(1); for ((key, value), next) in live.iter().zip(next) { db.assert_link(*key, *value, *next).await; } for key in absent { db.assert_absent(*key).await; } db.assert_exact().await; } /// A diff entry of a batch: the update of its key in the database the batch's chain was built /// on, and its location in the batch (`None` when deleted). type Change = (Option>, Option>); /// Batch internals the ordered repair test needs beyond [`Inspect`]. pub(crate) trait Changes: Inspect { /// Return each entry the own diff of `batch` holds for `key`. fn changes(batch: &Self::Merkleized, key: &Digest) -> Vec>; /// Return a reference to `batch` that does not keep it alive. fn downgrade(batch: &Self::Merkleized) -> Weak>; } /// Return each entry `diff` holds for `key`, as [`Changes::changes`] returns them. fn changes( diff: &[(Digest, batch::DiffEntry)], key: &Digest, ) -> Vec> { diff.iter() .filter(|(k, _)| k == key) .map(|(_, entry)| (entry.base_old_loc(), entry.loc())) .collect() } impl Changes for Db where F: Family, C: Mutable>, I: UnorderedIndex> + 'static, U: Update, S: Strategy, Operation: Codec, Self: Inspect + DbAny>>, { fn changes(batch: &Self::Merkleized, key: &Digest) -> Vec> { changes(&batch.diff, key) } fn downgrade(batch: &Self::Merkleized) -> Weak> { Arc::downgrade(batch) } } impl Changes for crate::qmdb::current::db::Db where F: crate::merkle::Graftable, C: Mutable>, I: UnorderedIndex> + 'static, U: Update, S: Strategy, Operation: Codec, Self: Inspect + DbAny< F, Merkleized = Arc>, >, { fn changes(batch: &Self::Merkleized, key: &Digest) -> Vec> { changes(&batch.inner.diff, key) } fn downgrade(batch: &Self::Merkleized) -> Weak> { Arc::downgrade(batch) } } /// Ordered database access the eviction tests need beyond [`Links`] and [`Changes`]. pub(crate) trait Neighbors: Links + Changes { /// Return the next and the previous live key of `key`, without wrapping, through `batch`, /// or through the database without one. async fn neighbors( &self, batch: Option<&Self::Merkleized>, key: &Digest, ) -> (Option, Option); } impl Neighbors for Db, N, S> where F: Family, C: Mutable>>, I: crate::index::Ordered> + 'static, V: ValueEncoding, S: Strategy, Operation>: Codec, Self: Links + DbAny< F, Merkleized = Arc< batch::MerkleizedBatch, S>, >, >, { async fn neighbors( &self, batch: Option<&Self::Merkleized>, key: &Digest, ) -> (Option, Option) { match batch { Some(batch) => ( batch.get_next_key(key, self).await.unwrap(), batch.get_prev_key(key, self).await.unwrap(), ), None => ( self.get_next_key(key).await.unwrap(), self.get_prev_key(key).await.unwrap(), ), } } } /// A case of [`test_any_ordered_policy_eviction_matrix`]. Each starts from an empty state. struct Case { /// Labels the case in assertion messages and its reopen context. name: &'static str, /// The keys one applied batch creates. seed: Vec, /// The writes one batch applies after the seed. applied: Vec<(Digest, Option)>, /// The writes of a pending parent, if the batch is its child. parent: Option)>>, /// The batch's own writes. writes: Vec<(Digest, Option)>, /// The policy's entries and skips. limits: (usize, u64), /// The policy's choice for each listed key. It keeps every other update. choices: Vec<(Digest, Choice)>, /// The predecessors of evicted keys the walk leaves undecided, in key order. The batch /// rewrites each once after its decisions. rewritten: Vec, } /// Assert that `db` holds exactly the `live` keys, linked in a cycle, that the other `keys` /// are absent, and that the database's neighbor queries for every key in `keys` match `live`. async fn assert_neighbors>( db: &D, batch: Option<&D::Merkleized>, live: &BTreeMap, keys: &BTreeSet, name: &str, ) { if batch.is_none() { let absent: Vec<_> = keys .iter() .filter(|key| !live.contains_key(key)) .copied() .collect(); assert_links(db, live, &absent).await; } for key in keys { let next = live .range(( core::ops::Bound::Excluded(*key), core::ops::Bound::Unbounded, )) .next() .map(|(next, _)| *next); let prev = live.range(..*key).next_back().map(|(prev, _)| *prev); assert_eq!( db.neighbors(batch, key).await, (next, prev), "{name}: neighbors of {key}", ); } } /// Run `case` on `db` and return the database reopened after applying it. /// /// The oracle replays the batch's writes into the [state its walk sees](written) and walks it /// with [`walk_model`]. The final key set applies the decisions to that state. The /// walk appends its decisions in walk order, then rewrites each evicted key's predecessor in /// the final key set once, in key order, unless the walk kept or replaced it, which carries /// the link in its decision. The batch commits the walk's floor, or its commit location if no /// key remains. /// /// Asserts that the modeled walk reaches every stop, then the decisions, the appended /// operations, the floor, and the diff entry of each key the batch creates and evicts, then /// the links and neighbor queries of the final key set through the batch, after applying it, /// and after reopening. #[boxed] async fn run_case( context: &Context, db: D, reopen: &impl Fn(Context) -> Fut, make_value: &impl Fn(u64) -> Digest, case: Case, ) -> D where F: Family, D: Neighbors, Operation: Codec, Fut: Future, { let name = case.name; // Delete every key, then seed, apply the writes after the seed, and stage the parent. let deletes: Vec<_> = db.live().await.into_keys().map(|key| (key, None)).collect(); let db = if deletes.is_empty() { db } else { hold(db, &deletes).await }; let seed: Vec<_> = case .seed .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); let mut model = Model::default(); model.apply(&seed); let mut db = hold(db, &seed).await; if !case.applied.is_empty() { model.apply(&case.applied); db = hold(db, &case.applied).await; } let mut parent = None; if let Some(writes) = &case.parent { model.apply(writes); parent = Some(hold_batch(&db, db.new_batch(), writes).await); } let start = || parent.as_ref().map_or_else(|| db.new_batch(), D::child); let floor = parent.as_ref().map_or_else( || db.inactivity_floor_loc(), |parent| D::span(parent).inactivity_floor, ); let ancestors: Vec<_> = parent.iter().collect(); let state = written(&db, &ancestors, start(), &case.writes).await; let prior: BTreeSet = model.values.keys().copied().collect(); model.apply(&case.writes); // Walk the state after the writes. let choose = |key: &Digest| { case.choices .iter() .find(|(k, _)| k == key) .map_or(Choice::Keep, |(_, choice)| *choice) }; let at: BTreeMap = state.live.iter().map(|(key, loc)| (**loc, *key)).collect(); let active: Vec = at.keys().copied().collect(); let stops: Vec = at .iter() .filter(|(_, key)| matches!(choose(key), Choice::Stop)) .map(|(loc, _)| *loc) .collect(); let (entries, skips) = case.limits; let (walk_floor, visited) = walk_model(&active, *floor, *state.tip, entries, skips, &stops); assert!( stops.iter().all(|loc| visited.contains(loc)), "{name}: the walk reaches every stop", ); // Apply the decisions, then find the evicted keys' predecessors in the final key set. let mut live = model.values.clone(); let mut appended_ops = Vec::new(); let mut evicted = Vec::new(); let mut decided = BTreeSet::new(); for loc in &visited { let key = at[loc]; match choose(&key) { Choice::Keep => { appended_ops.push((key, Some(live[&key]))); decided.insert(key); } Choice::Replace(value) => { live.insert(key, value); appended_ops.push((key, Some(value))); decided.insert(key); } Choice::Evict => { live.remove(&key); appended_ops.push((key, None)); evicted.push(key); } Choice::Stop => {} } } let rewritten: BTreeSet = evicted .iter() .filter_map(|key| { live.range(..*key) .next_back() .or_else(|| live.iter().next_back()) .map(|(prev, _)| *prev) }) .filter(|prev| !decided.contains(prev)) .collect(); assert!( rewritten.iter().eq(&case.rewritten), "{name}: rewritten predecessors {rewritten:?}", ); appended_ops.extend(rewritten.iter().map(|key| (*key, Some(live[key])))); let floor = if live.is_empty() { *state.tip + appended_ops.len() as u64 } else { walk_floor }; // The batch makes the modeled decisions, appends the modeled operations after its writes, // and commits the modeled floor. let mut policy = Script::new(entries, skips, choose); let batch = case .writes .iter() .fold(start(), |batch, &(key, value)| batch.write(key, value)); let merkleized = batch.merkleize(&db, None, &mut policy).await.unwrap(); let expected: Vec<_> = visited .iter() .map(|loc| (GenericLocation::new(*loc), at[loc], model.values[&at[loc]])) .collect(); assert_eq!(policy.visited, expected, "{name}: decisions"); assert_eq!( appended::(&merkleized, &state), appended_ops, "{name}: appended operations", ); assert_eq!( *D::span(&merkleized).inactivity_floor, floor, "{name}: floor" ); // A key the batch creates and the walk evicts ends deleted with no base location. for key in evicted.iter().filter(|key| !prior.contains(key)) { assert_eq!( D::changes(&merkleized, key), [(None, None)], "{name}: created and evicted {key}", ); } // The final key set holds through the batch, after applying it, and after reopening. let keys: BTreeSet<_> = model.keys.iter().copied().collect(); assert_neighbors(&db, Some(&merkleized), &live, &keys, name).await; let db = match parent { Some(parent) => db.apply_batch(parent).await.unwrap(), None => db, }; let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(*db.inactivity_floor_loc(), floor, "{name}: applied floor"); assert_neighbors(&db, None, &live, &keys, name).await; let db = db.commit().await.unwrap(); drop(db); let db = reopen(context.child(name)).await; assert_eq!(*db.inactivity_floor_loc(), floor, "{name}: reopened floor"); assert_neighbors(&db, None, &live, &keys, name).await; db } /// Evicting ordered keys rewrites each evicted key's predecessor in the final key set to link /// past it, whatever holds the predecessor's update: an applied update the walk leaves /// undecided, kept, or stops at; a write of the batch the walk leaves undecided (written /// twice), or one it creates; a pending parent's update; a sibling in the evicted key's /// translated-key bucket; or the largest key, for the smallest evicted key. /// /// The cases also evict chains of two and three adjacent keys, the largest key, the batch's own /// writes, a key whose predecessor the walk decides later or replaces, every key but one, and /// every key. /// /// Under tight skips, the walk never reaches the predecessor, or stops at it with its last skip. /// /// Each case checks the decisions, the operations the walk appends, and the floor against a /// [model](run_case) of the state after the batch's writes, and the links and neighbor queries /// of the final key set through the batch, after applying it, and after reopening. /// /// A key the batch creates and the walk evicts ends with no base location in the batch's diff. #[boxed] pub(crate) async fn test_any_ordered_policy_eviction_matrix( context: Context, db: D, reopen: impl Fn(Context) -> Fut, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Neighbors, Operation: Codec, Fut: Future, { let key = |prefix: u8| colliding_digest(prefix, 0); let [a, b, c, d, e, f, g] = [0x10, 0x20, 0x30, 0x40, 0x50, 0x60, 0x70].map(key); let [x0, x1, x2] = [0, 1, 2].map(|suffix| colliding_digest(0x30, suffix)); let value = |i: u64| Some(make_value(1000 + i)); let evict = |keys: &[Digest]| -> Vec<_> { keys.iter().map(|key| (*key, Choice::Evict)).collect() }; let unbounded = (usize::MAX, u64::MAX); let case = |name, seed: &[Digest], limits, choices, rewritten: &[Digest]| Case { name, seed: seed.to_vec(), applied: Vec::new(), parent: None, writes: Vec::new(), limits, choices, rewritten: rewritten.to_vec(), }; let five = [a, b, c, d, e]; let eight = [a, b, c, x1, d, e, f, g]; let cases = [ // The kept smallest key links past the adjacent second and third, and the replaced // fourth links past the evicted largest key to the smallest. Case { choices: vec![ (b, Choice::Evict), (c, Choice::Evict), (d, Choice::Replace(make_value(1004))), (e, Choice::Evict), ], ..case("kept_and_replaced", &five, unbounded, Vec::new(), &[]) }, // The smallest key's update lies past the two updates the walk decides. Case { applied: vec![(a, value(1))], ..case("undecided_applied", &five, (2, u64::MAX), evict(&[b]), &[a]) }, // The walk stops at the smallest key's update, past the evicted second. Case { applied: vec![(a, value(1))], choices: vec![(b, Choice::Evict), (a, Choice::Stop)], ..case("stopped", &five, unbounded, Vec::new(), &[a]) }, // Emptying the previous case's state moves the floor to its commit, so the seed lies // just past the floor. The inactive locations below the smallest key's last update are // that commit, the key's superseded update, and the seed commit. Two skips reach the // second key, and after keeping the third through fifth keys the walk has no skip left // to pass the seed commit, so it never reaches the smallest key's update. Case { applied: vec![(a, value(10))], ..case( "unreached_predecessor", &five, (usize::MAX, 2), evict(&[b]), &[a], ) }, // Three skips exactly reach the smallest key's update, where the walk stops after // evicting the second. Case { applied: vec![(a, value(11))], choices: vec![(b, Choice::Evict), (a, Choice::Stop)], ..case( "stopped_at_last_skip", &five, (usize::MAX, 3), Vec::new(), &[a], ) }, // The walk keeps the smallest key's update after evicting its successor. Case { applied: vec![(a, value(1))], ..case("decided_later", &five, unbounded, evict(&[b]), &[]) }, // The same, with the smallest key's update in a pending parent. Case { parent: Some(vec![(a, value(1))]), ..case("parent_decided_later", &five, unbounded, evict(&[b]), &[]) }, // The batch updates the smallest key, and the walk decides only the second. Case { writes: vec![(a, value(2))], ..case("undecided_write", &five, (1, u64::MAX), evict(&[b]), &[a]) }, // The batch creates the second key, and the walk decides only the third. Case { writes: vec![(b, value(3))], ..case("created", &[a, c, d, e], (1, u64::MAX), evict(&[c]), &[b]) }, // A pending parent creates the second key, and the walk decides only the third. Case { parent: Some(vec![(b, value(3))]), ..case( "parent_created", &[a, c, d, e], (1, u64::MAX), evict(&[c]), &[b], ) }, // The evicted key's predecessor shares its bucket, and its update lies past the two // updates the walk decides. Case { applied: vec![(x0, value(5))], ..case( "sibling", &[a, x0, x1, x2, e], (2, u64::MAX), evict(&[x1]), &[x0], ) }, // The three smallest keys share the undecided largest key as their predecessor. case( "wrapped_chain", &five, (3, u64::MAX), evict(&[a, b, c]), &[e], ), // The adjacent third and fourth keys share the undecided second key as predecessor, // whose update lies past the three updates the walk decides. The repair reaches it // through the fourth key's previous bucket, the evicted third key's, and then that // key's previous bucket. Case { applied: vec![(b, value(9))], ..case( "two_bucket_chain", &five, (3, u64::MAX), evict(&[c, d]), &[b], ) }, // The middle key links to itself. case("all_but_one", &five, unbounded, evict(&[a, b, d, e]), &[]), case("every_key", &five, unbounded, evict(&five), &[]), // The batch updates the second key and creates the sixth, rewriting the largest as // its predecessor. The walk evicts both writes and keeps the rewrite. Case { writes: vec![(b, value(6)), (f, value(7))], ..case("own_writes", &five, unbounded, evict(&[b, f]), &[]) }, // A child of a pending parent evicts the parent's update of the third key and the // adjacent fourth. Case { parent: Some(vec![(c, value(8))]), ..case("parent_chain", &five, unbounded, evict(&[c, d]), &[]) }, // Of eight keys, the third and fourth share a translated-key bucket. The walk evicts // the second, fourth, sixth, and largest keys, so each predecessor it keeps carries // the link, including the fourth key's sibling. It stops at the smallest key's update, // which it rewrites to link past the second. Case { applied: vec![(a, value(20))], choices: vec![ (b, Choice::Evict), (x1, Choice::Evict), (e, Choice::Evict), (g, Choice::Evict), (a, Choice::Stop), ], ..case("links_stop_smallest", &eight, unbounded, Vec::new(), &[a]) }, // The same evictions from a child of a pending parent that updates the fifth key, // which the walk keeps. Case { parent: Some(vec![(d, value(21))]), ..case( "links_parent_kept", &eight, unbounded, evict(&[b, x1, e, g]), &[], ) }, // The two smallest keys share the largest as their predecessor, which the walk keeps // and links to the third key. case("links_two_smallest", &eight, unbounded, evict(&[a, b]), &[]), // A child evicts a pending parent's update of the fourth key, so its sibling, the // third key, links to the fifth. Case { parent: Some(vec![(x1, value(22))]), ..case("links_parent_evicted", &eight, unbounded, evict(&[x1]), &[]) }, ]; let mut db = db; for case in cases { db = run_case(&context, db, &reopen, &make_value, case).await; } db.destroy().await.unwrap(); } /// Evicting the first live update a child of two pending ancestors reaches rewrites the key's /// predecessor, an update of the parent the walk leaves undecided, to link past it. Keys /// A < B < C < D < E share one translated-key bucket. The grandparent writes B, the parent /// updates it, and the child, with one entry, evicts C. The child appends the delete and then /// B with the parent's value, and its diff holds one entry for B whose base is the update of B /// in the database the ancestors were built on: none when the grandparent creates B /// (rewriting A), and B's last applied update when the database holds it. /// /// The child makes the same decision and appends the same operations under the same floor and /// root whether it applies directly over both pending ancestors, merkleizes after the /// grandparent is applied and freed, or applies after the parent is applied. Each run links A, /// B, D, and E through the batch, after applying it, and after reopening, with an exact /// activity bitmap. #[boxed] pub(crate) async fn test_any_ordered_policy_repair_across_ancestors( context: Context, db: D, reopen: impl Fn(Context) -> Fut, make_value: impl Fn(u64) -> Digest, ) where F: Family, D: Neighbors, Operation: Codec, Fut: Future, { /// When the child merkleizes and applies relative to its ancestors. #[derive(Clone, Copy, PartialEq, Eq)] enum Lifecycle { /// Merkleize and apply over both pending ancestors. Pending, /// Apply and free the grandparent, then merkleize and apply over the pending parent. Freed, /// Merkleize over both pending ancestors, apply the parent, then apply the child. Applied, } let keys: [Digest; 5] = core::array::from_fn(|i| colliding_digest(0xAA, i as u64)); assert!(keys.is_sorted()); let [_, b, c, _, _] = keys; let all: BTreeSet<_> = keys.into_iter().collect(); let parent_value = make_value(301); // Each variant seeds keys, then updates some in later batches, all with a held floor. let variants: [(&'static str, &[usize], &[usize]); 2] = [ ("created", &[0, 2, 3, 4], &[]), ("applied", &[0, 1, 2, 3, 4], &[0, 1]), ]; let lifecycles = [ ("pending", Lifecycle::Pending), ("freed", Lifecycle::Freed), ("applied", Lifecycle::Applied), ]; let mut db = db; for (variant, seeded, updated) in variants { let mut outcomes = Vec::new(); for (name, lifecycle) in lifecycles { let label = format!("{variant} {name}"); let ctx = context.child(variant).child(name); db.destroy().await.unwrap(); let mut fresh = reopen(ctx.child("fresh")).await; let seed: Vec<_> = seeded .iter() .map(|&i| (keys[i], Some(make_value(i as u64)))) .collect(); fresh = hold(fresh, &seed).await; for &i in updated { fresh = hold(fresh, &[(keys[i], Some(make_value(100 + i as u64)))]).await; } let applied = fresh.live().await; let mut live: BTreeMap<_, _> = seed .iter() .map(|(key, value)| (*key, value.unwrap())) .collect(); for &i in updated { live.insert(keys[i], make_value(100 + i as u64)); } live.insert(b, parent_value); live.remove(&c); // The grandparent writes B and the parent updates it. With one entry, the walk // decides only C, the first live update it reaches. let grandparent = hold_batch(&fresh, fresh.new_batch(), &[(b, Some(make_value(201)))]).await; let parent = hold_batch(&fresh, D::child(&grandparent), &[(b, Some(parent_value))]).await; let state = written(&fresh, &[&grandparent, &parent], D::child(&parent), &[]).await; let active: Vec = state.active().iter().map(|loc| **loc).collect(); let floor = *D::span(&parent).inactivity_floor; let (floor, visited) = walk_model(&active, floor, *state.tip, 1, u64::MAX, &[]); assert_eq!( visited, [*state.live[&c]], "{label}: C is not the first update" ); // Merkleize the child, after applying and freeing the grandparent when the // lifecycle requires it. Only the parent's batch references the grandparent, and // only weakly. let child = D::child(&parent); let (fresh, grandparent) = if lifecycle == Lifecycle::Freed { let weak = D::downgrade(&grandparent); let fresh = fresh.apply_batch(grandparent).await.unwrap(); assert_eq!(weak.strong_count(), 0, "{label}: the grandparent is alive"); (fresh, None) } else { (fresh, Some(grandparent)) }; let choose = |key: &Digest| { assert_eq!(*key, c, "{label}: decided a key other than C"); Choice::Evict }; let mut policy = Script::new(1, u64::MAX, choose); let merkleized = child.merkleize(&fresh, None, &mut policy).await.unwrap(); assert_eq!(policy.locations(), [state.live[&c]], "{label}: decisions"); assert_eq!( appended::(&merkleized, &state), [(c, None), (b, Some(parent_value))], "{label}: appended operations", ); assert_eq!( *D::span(&merkleized).inactivity_floor, floor, "{label}: floor" ); assert_eq!( D::changes(&merkleized, &b), [(applied.get(&b).copied(), Some(state.tip + 1))], "{label}: B's diff entries", ); assert_eq!( D::changes(&merkleized, &c), [(Some(applied[&c]), None)], "{label}: C's diff entries", ); assert_neighbors(&fresh, Some(&merkleized), &live, &all, &label).await; let (start, ops) = D::ops(&merkleized); let ops: Vec<_> = ops.iter().map(|op| op.encode().to_vec()).collect(); outcomes.push((start, ops, floor, MerkleizedTrait::root(&merkleized))); // Apply the child, after applying the parent when the lifecycle requires it. let (fresh, parent) = if lifecycle == Lifecycle::Applied { (fresh.apply_batch(parent).await.unwrap(), None) } else { (fresh, Some(parent)) }; let fresh = fresh.apply_batch(merkleized).await.unwrap(); drop((grandparent, parent)); assert_eq!( *fresh.inactivity_floor_loc(), floor, "{label}: applied floor" ); assert_neighbors(&fresh, None, &live, &all, &label).await; drop(fresh.commit().await.unwrap()); db = reopen(ctx.child("reopened")).await; assert_eq!(*db.inactivity_floor_loc(), floor, "{label}: reopened floor"); assert_neighbors(&db, None, &live, &all, &label).await; } for outcome in &outcomes[1..] { assert_eq!(*outcome, outcomes[0], "{variant}: the lifecycles diverged"); } } db.destroy().await.unwrap(); } /// A policy over uncached updates reads every update it decides in one batched read of /// exactly those updates, and merkleize reads nothing more. /// /// The decided updates lie among superseded ones, superseded either by applied updates or by /// the batch's writes. Writes add one batched read before the first decision: the /// resolution's read of the written keys. pub(crate) async fn test_any_policy_reads_in_one_read( context: Context, db: D, reopen_db: impl Fn(Context) -> Pin + Send>>, make_value: impl Fn(u64) -> Digest, ) where D: DbAny, { let names = [ "log_journal_read_calls_total", "log_journal_read_many_calls_total", "log_journal_items_read_total", ]; let mut db = db; for in_batch in [false, true] { let ctx = context.child(if in_batch { "in_batch" } else { "applied" }); // Seed the 64 smallest keys in key order with a held floor. Then either supersede all // but every eighth one, or seed the 64 larger keys after them. Only the largest seeded // key precedes a larger key, so an ordered database rewrites no other seeded update. let mut keys: Vec<_> = (0..128).map(to_digest).collect(); keys.sort(); let seed: Vec<_> = keys[..64] .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(i as u64)))) .collect(); db = hold(db, &seed).await; let later: Vec<_> = if in_batch { keys[64..] .iter() .enumerate() .map(|(i, key)| (*key, Some(make_value(64 + i as u64)))) .collect() } else { seed.iter() .enumerate() .filter(|(i, _)| i % 8 != 0) .map(|(i, (key, _))| (*key, Some(make_value(i as u64 + 100)))) .collect() }; db = hold(db, &later).await; drop(db.commit().await.unwrap()); // Reopen so no operation is cached. db = reopen_db(ctx.child("cold")).await; // Decide every eighth update, or write every odd one of the first sixteen updates and // decide the eight even ones, which spends every entry before the writes. Every // decision follows the same batched reads. Without writes, those reads hold exactly // the eight decided updates. let (stride, batch, deltas): (_, _, &[u64]) = if in_batch { let batch = (1..16).step_by(2).fold(db.new_batch(), |batch, i| { batch.write(keys[i], Some(make_value(i as u64 + 100))) }); (2, batch, &[0, 2]) } else { (8, db.new_batch(), &[0, 1, 8]) }; let sample = || -> Vec<_> { names[..deltas.len()] .iter() .map(|name| counter(&context, name)) .collect() }; let after: Vec<_> = sample().iter().zip(deltas).map(|(n, d)| n + d).collect(); let mut probes = Vec::new(); let mut policy = Script::new(8, u64::MAX, |_: &Digest| { probes.push(sample()); Choice::Keep }); batch.merkleize(&db, None, &mut policy).await.unwrap(); let expected: Vec<_> = (0..8 * stride) .step_by(stride) .map(|i| (GenericLocation::::new(1 + i as u64), keys[i])) .collect(); let decided: Vec<_> = policy .visited .iter() .map(|(loc, key, _)| (*loc, *key)) .collect(); assert_eq!(decided, expected, "in_batch={in_batch}"); assert_eq!(probes, vec![after.clone(); 8], "in_batch={in_batch}"); assert_eq!(sample(), after, "in_batch={in_batch}"); // Start the next case from an empty database. db.destroy().await.unwrap(); db = reopen_db(ctx.child("fresh")).await; } db.destroy().await.unwrap(); } /// Merkleizing an unordered batch whose policy keeps, evicts, and replaces updates reads no /// operation past the policy's reads. pub(crate) async fn test_any_policy_unordered_reads_nothing( context: Context, db: D, make_value: impl Fn(u64) -> Digest, ) where D: DbAny, { // Seed sixteen updates with a held floor. let seed: Vec<_> = (0..16) .map(|i| (to_digest(i), Some(make_value(i)))) .collect(); let db = hold(db, &seed).await; // Evict one key, replace another, and keep the rest. Merkleize reads nothing after the // last decision. let replacement = make_value(200); let mut probes = Vec::new(); let mut policy = Script::new(usize::MAX, u64::MAX, |key: &Digest| { probes.push(counter(&context, "log_journal_items_read_total")); if *key == to_digest(1) { Choice::Evict } else if *key == to_digest(2) { Choice::Replace(replacement) } else { Choice::Keep } }); let merkleized = db .new_batch() .merkleize(&db, None, &mut policy) .await .unwrap(); assert_eq!(policy.visited.len(), 16); assert_eq!( probes.last().copied(), Some(counter(&context, "log_journal_items_read_total")) ); // Published values reflect the keeps, replacement, and eviction. let db = db.apply_batch(merkleized).await.unwrap(); for (i, (key, value)) in seed.into_iter().enumerate() { let expected = match i { 1 => None, 2 => Some(replacement), _ => value, }; assert_eq!(db.get(&key).await.unwrap(), expected); } db.destroy().await.unwrap(); } /// Skips that run out before a live update leave it undecided, and a window that ends at the /// update leaves it unread. A policy that stops at the update commits the floor at its /// location. #[test_traced("INFO")] fn test_any_policy_skips_stop_before_live_update() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("policy-skips"), None, ) .await .unwrap(); // Seed two keys, then delete the first, with a held floor. let mut keys = [key(0), key(1)]; keys.sort(); let seed = db .new_batch() .write(keys[0], Some(val(0))) .write(keys[1], Some(val(1))) .merkleize(&db, None, &mut Hold) .await .unwrap(); let seed_start = db.bounds().end; let db = db.apply_batch(seed).await.unwrap(); let deleted = db .new_batch() .write(keys[0], None) .merkleize(&db, None, &mut Hold) .await .unwrap(); let db = db.apply_batch(deleted).await.unwrap(); let reads = || counter(&context, "log_journal_items_read_total"); let live = seed_start + 1; let gap = *live - *db.inactivity_floor_loc(); // One skip short of the live update, the policy decides nothing. With one entry, the // window ends at the update, so the walk reads nothing. With unbounded entries, the // window reaches the tip, and the walk reads the update it cannot reach: the limits // bound what the walk passes and decides, and the window bounds what it reads. for (entries, read) in [(1, 0), (usize::MAX, 1)] { let before = reads(); let mut policy = Script::new(entries, gap - 1, keep); let merkleized = db .new_batch() .merkleize(&db, None, &mut policy) .await .unwrap(); assert!(policy.visited.is_empty()); assert_eq!(*merkleized.bounds().inactivity_floor, *live - 1); assert_eq!(reads(), before + read, "entries={entries}"); } // A policy with unbounded limits reads only the live update. The inactive suffix and // the last commit need no read. let before = reads(); let mut policy = Script::new(usize::MAX, u64::MAX, keep); let merkleized = db .new_batch() .merkleize(&db, None, &mut policy) .await .unwrap(); assert_eq!(policy.locations(), [live]); assert_eq!(merkleized.bounds().inactivity_floor, db.size()); assert_eq!(reads(), before + 1, "only the live update is read"); drop(merkleized); // With exactly enough skips, a policy that stops at the live update commits the // floor at its location. let mut policy = Script::new(usize::MAX, gap, |_: &Digest| Choice::Stop); let merkleized = db .new_batch() .merkleize(&db, None, &mut policy) .await .unwrap(); assert_eq!(policy.locations(), [live]); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), live); assert_eq!(db.get(&keys[0]).await.unwrap(), None); assert_eq!(db.get(&keys[1]).await.unwrap(), Some(val(1))); db.destroy().await.unwrap(); }); } fn key(i: u64) -> Digest { Sha256::hash(&[&i.to_be_bytes()]) } fn val(i: u64) -> Digest { Sha256::hash(&[&(i + 10000).to_be_bytes()]) } /// Commit a batch of key-value writes. async fn commit_writes( db: UnorderedVariable, writes: impl IntoIterator)>, metadata: Option, ) -> UnorderedVariable { let mut batch = db.new_batch(); for (k, v) in writes { batch = batch.write(k, v); } let merkleized = batch .merkleize(&db, metadata, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap() } /// Dropping an in-flight parallel init (e.g. losing a select against a timeout) aborts /// its snapshot workers and decoders rather than leaving them running until they happen /// to observe a closed channel. #[test_traced("INFO")] fn test_parallel_init_aborted_on_cancel() { /// Sum of the runtime's running-task gauges for snapshot build tasks. fn running_build_tasks(metrics: &str) -> u64 { metrics .lines() .filter(|line| { line.starts_with("runtime_tasks_running{") && (line.contains("snapshot_worker") || line.contains("snapshot_decoder")) }) .filter_map(|line| line.rsplit_once(' ')?.1.trim().parse::().ok()) .sum() } let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let cfg = || fixed_db_config_full::("cancel", Sequential, NZUsize!(4)); // Persist enough operations that the reopen's build spans many decode chunks. let db: UnorderedFixedP1 = UnorderedFixedP1::init(ctx.child("storage"), cfg(), None) .await .unwrap(); let mut batch = db.new_batch(); for i in 0..2_000u64 { batch = batch.write(key(i), Some(key(i))); } 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(); drop(db); { // Poll until snapshot tasks own the pending build, then cancel before init returns. let init = UnorderedFixedP1::init(ctx.child("storage"), cfg(), None); pin_mut!(init); let mut spawned = false; for _ in 0..10_000 { if poll!(&mut init).is_ready() { panic!("init completed before it could be cancelled"); } context.sleep(Duration::from_millis(1)).await; if running_build_tasks(&context.encode()) > 0 { spawned = true; break; } } assert!(spawned, "build tasks never spawned"); } // Leaving the scope dropped the init future. Give the runtime a beat to reap // the aborted tasks. context.sleep(Duration::from_millis(10)).await; let metrics = context.encode(); assert_eq!(running_build_tasks(&metrics), 0, "{metrics}"); }); } /// An empty batch (no mutations) still produces a valid commit. #[test_traced("INFO")] fn test_any_batch_empty() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("e"), None, ) .await .unwrap(); let root_before = db.root(); let batch = db.new_batch(); let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // A CommitFloor op was appended, so root must change. assert_ne!(db.root(), root_before); // DB should still be functional. let db = commit_writes(db, [(key(0), Some(val(0)))], None).await; assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); db.destroy().await.unwrap(); }); } /// Metadata propagates through merkleize and clears with None. #[test_traced("INFO")] fn test_any_batch_metadata() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("m"), None, ) .await .unwrap(); let metadata = val(42); // Batch with metadata. let db = commit_writes(db, [(key(0), Some(val(0)))], Some(metadata)).await; assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); // Batch without metadata clears it. let batch = db.new_batch(); let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), None); db.destroy().await.unwrap(); }); } /// batch.get() reads through: pending mutations -> base DB. /// Updates shadow the base value; deletes hide the key. #[test_traced("INFO")] fn test_any_batch_get_read_through() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("g"), None, ) .await .unwrap(); // Pre-populate with key A. let ka = key(0); let va = val(0); let db = commit_writes(db, [(ka, Some(va))], None).await; let kb = key(1); let vb = val(1); let kc = key(2); let mut batch = db.new_batch(); // Read-through to base DB. assert_eq!(batch.get(&ka, &db).await.unwrap(), Some(va)); // Pending mutation visible. batch = batch.write(kb, Some(vb)); assert_eq!(batch.get(&kb, &db).await.unwrap(), Some(vb)); // Nonexistent key. assert_eq!(batch.get(&kc, &db).await.unwrap(), None); // Update shadows base DB value. let va2 = val(100); batch = batch.write(ka, Some(va2)); assert_eq!(batch.get(&ka, &db).await.unwrap(), Some(va2)); // Delete hides the key. batch = batch.write(ka, None); assert_eq!(batch.get(&ka, &db).await.unwrap(), None); db.destroy().await.unwrap(); }); } /// merkleized.get() reflects the resolved diff after merkleize. #[test_traced("INFO")] fn test_any_batch_get_on_merkleized() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("mg"), None, ) .await .unwrap(); let ka = key(0); let kb = key(1); let kc = key(2); let kd = key(3); // Pre-populate A and B. let db = commit_writes(db, [(ka, Some(val(0))), (kb, Some(val(1)))], None).await; // Batch: update A, delete B, create C. let va2 = val(100); let vc = val(2); let mut batch = db.new_batch(); batch = batch.write(ka, Some(va2)); batch = batch.write(kb, None); batch = batch.write(kc, Some(vc)); let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); assert_eq!(merkleized.get(&ka, &db).await.unwrap(), Some(va2)); assert_eq!(merkleized.get(&kb, &db).await.unwrap(), None); assert_eq!(merkleized.get(&kc, &db).await.unwrap(), Some(vc)); assert_eq!(merkleized.get(&kd, &db).await.unwrap(), None); db.destroy().await.unwrap(); }); } /// Child batch reads through: child mutations -> parent diff -> base DB. #[test_traced("INFO")] fn test_any_batch_stacked_get() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("sg"), None, ) .await .unwrap(); let ka = key(0); let kb = key(1); // Parent batch writes A. let mut batch = db.new_batch(); batch = batch.write(ka, Some(val(0))); let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); // Child reads parent's A. let mut child = merkleized.new_batch::(); assert_eq!(child.get(&ka, &db).await.unwrap(), Some(val(0))); // Child overwrites A. child = child.write(ka, Some(val(100))); assert_eq!(child.get(&ka, &db).await.unwrap(), Some(val(100))); // Child writes new key B. child = child.write(kb, Some(val(1))); assert_eq!(child.get(&kb, &db).await.unwrap(), Some(val(1))); // Child deletes A. child = child.write(ka, None); assert_eq!(child.get(&ka, &db).await.unwrap(), None); db.destroy().await.unwrap(); }); } /// Parent deletes a base-DB key, child re-creates it. #[test_traced("INFO")] fn test_any_batch_stacked_delete_recreate() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("dr"), None, ) .await .unwrap(); let ka = key(0); // Pre-populate with key A. let db = commit_writes(db, [(ka, Some(val(0)))], None).await; // Parent batch deletes A. let mut parent = db.new_batch(); parent = parent.write(ka, None); let parent_m = parent .merkleize(&db, None, &mut Proportional) .await .unwrap(); assert_eq!(parent_m.get(&ka, &db).await.unwrap(), None); // Child re-creates A with a new value. let mut child = parent_m.new_batch::(); child = child.write(ka, Some(val(200))); let child_m = child.merkleize(&db, None, &mut Proportional).await.unwrap(); assert_eq!(child_m.get(&ka, &db).await.unwrap(), Some(val(200))); // Apply and verify DB state. let db = db.apply_batch(child_m).await.unwrap(); assert_eq!(db.get(&ka).await.unwrap(), Some(val(200))); db.destroy().await.unwrap(); }); } /// The floor walk during merkleize moves active operations to the tip. All keys remain /// accessible with correct values. #[test_traced("INFO")] fn test_any_batch_floor_walk() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("fr"), None, ) .await .unwrap(); // Pre-populate with 100 keys. let init: Vec<_> = (0..100).map(|i| (key(i), Some(val(i)))).collect(); let db = commit_writes(db, init, None).await; let floor_before = db.inactivity_floor_loc(); // Update 30 keys. let updates: Vec<_> = (0..30).map(|i| (key(i), Some(val(i + 500)))).collect(); let db = commit_writes(db, updates, None).await; // Floor should have advanced. assert!(db.inactivity_floor_loc() > floor_before); // All keys should still be accessible with correct values. for i in 0..30 { assert_eq!( db.get(&key(i)).await.unwrap(), Some(val(i + 500)), "updated key {i} mismatch" ); } for i in 30..100 { assert_eq!( db.get(&key(i)).await.unwrap(), Some(val(i)), "untouched key {i} mismatch" ); } db.destroy().await.unwrap(); }); } /// Each applied batch grows the log by its writes, commit, and any floor-walk operations. #[test_traced("INFO")] fn test_any_batch_apply_grows_log() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let mut db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("ar"), None, ) .await .unwrap(); assert_eq!(db.bounds().end, crate::mmr::Location::new(1)); for first_key in [0, 5] { let before = db.bounds().end; let writes = (first_key..first_key + 5).map(|i| (key(i), Some(val(i)))); db = commit_writes(db, writes, None).await; // Five writes, a CommitFloor, and any floor-walk operations. assert!(db.bounds().end.saturating_sub(*before) >= 6); } db.destroy().await.unwrap(); }); } /// 3-level chain: parent -> child -> grandchild, merkleize grandchild and apply. #[test_traced("INFO")] fn test_any_batch_deep_chain() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("dc"), None, ) .await .unwrap(); // Pre-populate with keys 0..5. let init: Vec<_> = (0..5).map(|i| (key(i), Some(val(i)))).collect(); let db = commit_writes(db, init, None).await; // Parent: overwrite key 0, add key 5. let mut parent = db.new_batch(); parent = parent.write(key(0), Some(val(100))); parent = parent.write(key(5), Some(val(5))); let parent_m = parent .merkleize(&db, None, &mut Proportional) .await .unwrap(); // Child: overwrite key 1, add key 6. let mut child = parent_m.new_batch::(); child = child.write(key(1), Some(val(101))); child = child.write(key(6), Some(val(6))); let child_m = child.merkleize(&db, None, &mut Proportional).await.unwrap(); // Grandchild: delete key 2, add key 7. let mut grandchild = child_m.new_batch::(); grandchild = grandchild.write(key(2), None); grandchild = grandchild.write(key(7), Some(val(7))); let grandchild_m = grandchild .merkleize(&db, None, &mut Proportional) .await .unwrap(); // Verify reads through the chain. assert_eq!( grandchild_m.get(&key(0), &db).await.unwrap(), Some(val(100)) ); assert_eq!( grandchild_m.get(&key(1), &db).await.unwrap(), Some(val(101)) ); assert_eq!(grandchild_m.get(&key(2), &db).await.unwrap(), None); assert_eq!(grandchild_m.get(&key(7), &db).await.unwrap(), Some(val(7))); // Apply. let db = db.apply_batch(grandchild_m).await.unwrap(); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(100))); assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(101))); assert_eq!(db.get(&key(2)).await.unwrap(), None); assert_eq!(db.get(&key(3)).await.unwrap(), Some(val(3))); assert_eq!(db.get(&key(4)).await.unwrap(), Some(val(4))); assert_eq!(db.get(&key(5)).await.unwrap(), Some(val(5))); assert_eq!(db.get(&key(6)).await.unwrap(), Some(val(6))); assert_eq!(db.get(&key(7)).await.unwrap(), Some(val(7))); db.destroy().await.unwrap(); }); } /// Chained batch produces the same DB state as sequential apply_batch calls. #[test_traced("INFO")] fn test_any_batch_chain_matches_sequential() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); // DB A: sequential apply. let ctx_a = ctx.child("a"); let db_a: UnorderedVariable = UnorderedVariableDb::init( ctx_a.child("db"), variable_db_config::("cms-a"), None, ) .await .unwrap(); // DB B: chained batch. let ctx_b = ctx.child("b"); let db_b: UnorderedVariable = UnorderedVariableDb::init( ctx_b.child("db"), variable_db_config::("cms-b"), None, ) .await .unwrap(); // Batch 1 operations: create keys 0..5. let writes1: Vec<_> = (0..5).map(|i| (key(i), Some(val(i)))).collect(); // Batch 2 operations: update key 0, delete key 1, create key 5. let writes2 = vec![ (key(0), Some(val(100))), (key(1), None), (key(5), Some(val(5))), ]; // DB A: apply sequentially. let db_a = commit_writes(db_a, writes1.clone(), None).await; let db_a = commit_writes(db_a, writes2.clone(), None).await; // DB B: apply as chain. let mut parent = db_b.new_batch(); for (k, v) in &writes1 { parent = parent.write(*k, *v); } let parent_m = parent .merkleize(&db_b, None, &mut Proportional) .await .unwrap(); let mut child = parent_m.new_batch::(); for (k, v) in &writes2 { child = child.write(*k, *v); } let child_m = child .merkleize(&db_b, None, &mut Proportional) .await .unwrap(); let db_b = db_b.apply_batch(child_m).await.unwrap(); // Both DBs must have the same state. assert_eq!(db_a.root(), db_b.root()); for i in 0..6 { assert_eq!( db_a.get(&key(i)).await.unwrap(), db_b.get(&key(i)).await.unwrap(), "key {i} mismatch" ); } db_a.destroy().await.unwrap(); db_b.destroy().await.unwrap(); }); } /// Create and delete the same key in a single batch produces no net change for that key. #[test_traced("INFO")] fn test_any_batch_create_then_delete_same_batch() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("cd"), None, ) .await .unwrap(); // Pre-populate key A. let db = commit_writes(db, [(key(0), Some(val(0)))], None).await; // In one batch: create B then delete B, also create C and delete A. let mut batch = db.new_batch(); batch = batch.write(key(1), Some(val(1))); // create B batch = batch.write(key(1), None); // delete B (net: no B) batch = batch.write(key(2), Some(val(2))); // create C batch = batch.write(key(0), None); // delete A let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get(&key(0)).await.unwrap(), None); assert_eq!(db.get(&key(1)).await.unwrap(), None); assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2))); db.destroy().await.unwrap(); }); } /// Deleting all keys exercises the empty-state floor rule: the floor moves to the commit. #[test_traced("INFO")] fn test_any_batch_delete_all_keys() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("da"), None, ) .await .unwrap(); // Pre-populate 5 keys. let init: Vec<_> = (0..5).map(|i| (key(i), Some(val(i)))).collect(); let db = commit_writes(db, init, None).await; // Delete all 5. let deletes: Vec<_> = (0..5).map(|i| (key(i), None)).collect(); let db = commit_writes(db, deletes, None).await; for i in 0..5 { assert_eq!(db.get(&key(i)).await.unwrap(), None, "key {i} not deleted"); } // DB should still be functional after deleting everything. let db = commit_writes(db, [(key(10), Some(val(10)))], None).await; assert_eq!(db.get(&key(10)).await.unwrap(), Some(val(10))); db.destroy().await.unwrap(); }); } /// Two independent batches from the same DB do not interfere with each other. #[test_traced("INFO")] fn test_any_batch_parallel_forks() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("pf"), None, ) .await .unwrap(); // Pre-populate. let db = commit_writes(db, [(key(0), Some(val(0)))], None).await; let root_before = db.root(); // Fork A: update key 0 and create key 1. let fork_a_m = db .new_batch() .write(key(0), Some(val(100))) .write(key(1), Some(val(1))) .merkleize(&db, None, &mut Proportional) .await .unwrap(); // Fork B: delete key 0 and create key 2. let fork_b_m = db .new_batch() .write(key(0), None) .write(key(2), Some(val(2))) .merkleize(&db, None, &mut Proportional) .await .unwrap(); // Different mutations must produce different roots. assert_ne!(fork_a_m.root(), fork_b_m.root()); // DB is unchanged (neither batch applied). assert_eq!(db.root(), root_before); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); assert_eq!(db.get(&key(1)).await.unwrap(), None); // Apply fork A only. let db = db.apply_batch(fork_a_m).await.unwrap(); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(100))); assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1))); assert_eq!(db.get(&key(2)).await.unwrap(), None); db.destroy().await.unwrap(); }); } /// The floor walk advances the floor correctly across a chained batch. #[test_traced("INFO")] fn test_any_batch_floor_walk_chained() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("frc"), None, ) .await .unwrap(); // Pre-populate with 50 keys. let init: Vec<_> = (0..50).map(|i| (key(i), Some(val(i)))).collect(); let db = commit_writes(db, init, None).await; let floor_before = db.inactivity_floor_loc(); // Parent: update keys 0..20. let mut parent = db.new_batch(); for i in 0..20 { parent = parent.write(key(i), Some(val(i + 500))); } let parent_m = parent .merkleize(&db, None, &mut Proportional) .await .unwrap(); // Child: update keys 20..30. let mut child = parent_m.new_batch::(); for i in 20..30 { child = child.write(key(i), Some(val(i + 500))); } let child_m = child.merkleize(&db, None, &mut Proportional).await.unwrap(); let db = db.apply_batch(child_m).await.unwrap(); // Floor must have advanced. assert!(db.inactivity_floor_loc() > floor_before); // All keys should be accessible. for i in 0..30 { assert_eq!( db.get(&key(i)).await.unwrap(), Some(val(i + 500)), "updated key {i} mismatch" ); } for i in 30..50 { assert_eq!( db.get(&key(i)).await.unwrap(), Some(val(i)), "untouched key {i} mismatch" ); } db.destroy().await.unwrap(); }); } /// Dropping a batch without applying it leaves the DB unchanged. #[test_traced("INFO")] fn test_any_batch_abandoned() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("ab"), None, ) .await .unwrap(); let db = commit_writes(db, [(key(0), Some(val(0)))], None).await; let root_before = db.root(); // Create, populate, merkleize, then drop without apply. { let mut batch = db.new_batch(); batch = batch.write(key(0), Some(val(999))); batch = batch.write(key(1), Some(val(1))); let _merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); // dropped here } // DB state is identical. assert_eq!(db.root(), root_before); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); assert_eq!(db.get(&key(1)).await.unwrap(), None); db.destroy().await.unwrap(); }); } /// Applying without `commit()` publishes in memory but is not recovered after reopen. #[test_traced("INFO")] fn test_any_batch_apply_requires_commit_for_recovery() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let partition = "apply_requires_commit"; let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::(partition), None, ) .await .unwrap(); let committed_root = db.root(); let merkleized = db .new_batch() .write(key(0), Some(val(0))) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); drop(db); let reopened: UnorderedVariable = UnorderedVariableDb::init( context.child("reopen"), variable_db_config::(partition), None, ) .await .unwrap(); assert_eq!(reopened.root(), committed_root); assert_eq!(reopened.get(&key(0)).await.unwrap(), None); reopened.destroy().await.unwrap(); }); } /// Reopening at a pruned target returns an error. #[test_traced("INFO")] fn test_any_bounded_initialization_pruned_target_errors() { let executor = deterministic::Runner::default(); executor.start(|context| async move { const KEYS: u64 = 64; let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("rp"), None, ) .await .unwrap(); let initial: Vec<_> = (0..KEYS).map(|i| (key(i), Some(val(i)))).collect(); let first_start = db.bounds().end; let mut db = commit_writes(db, initial, None).await; let mut round = 0u64; loop { round += 1; assert!( round <= 64, "failed to prune enough history for the bounded initialization test" ); let updates: Vec<_> = (0..KEYS) .map(|i| (key(i), Some(val(1000 + round * KEYS + i)))) .collect(); db = commit_writes(db, updates, None).await; let boundary = db.sync_boundary(); db = db.prune(boundary).await.unwrap(); let bounds = db.bounds(); if bounds.start > first_start { break; } } let oldest_retained = db.bounds().start; _ = db.sync().await.unwrap(); let Err(boundary_err) = UnorderedVariable::init( ctx.child("cap"), variable_db_config::("rp"), Some(oldest_retained), ) .await else { panic!("expected bounded initialization at retained boundary to fail"); }; assert!( matches!( boundary_err, crate::qmdb::Error::Journal(crate::journal::Error::ItemPruned(_)) ), "unexpected bounded initialization error at retained boundary: {boundary_err:?}" ); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("reopen"), variable_db_config::("rp"), None, ) .await .unwrap(); _ = db.sync().await.unwrap(); let Err(err) = UnorderedVariable::init( ctx.child("cap"), variable_db_config::("rp"), Some(first_start), ) .await else { panic!("expected bounded initialization to pruned target to fail"); }; assert!( matches!( err, crate::qmdb::Error::Journal(crate::journal::Error::ItemPruned(_)) ), "unexpected bounded initialization error: {err:?}" ); }); } /// Zero is rejected. Equal and above-end bounds preserve the committed state. #[test_traced("INFO")] fn test_any_bounded_initialization_zero_equal_and_above_end() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("ri"), None, ) .await .unwrap(); // Commit one key so the reopen checks below verify real state. let db = commit_writes(db, [(key(0), Some(val(0)))], None).await; let root_before = db.root(); let size_before = db.size(); let db = { _ = db.sync().await.unwrap(); UnorderedVariable::init( ctx.child("cap"), variable_db_config::("ri"), Some(size_before), ) .await } .unwrap(); assert_eq!(db.root(), root_before); assert_eq!(db.size(), size_before); _ = db.sync().await.unwrap(); let Err(zero_err) = UnorderedVariable::init( ctx.child("cap"), variable_db_config::("ri"), Some(Location::new(0)), ) .await else { panic!("expected bounded initialization to zero to fail"); }; assert!( matches!(zero_err, QmdbError::InvalidInitializationBound), "unexpected bounded initialization error: {zero_err:?}" ); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("reopen"), variable_db_config::("ri"), None, ) .await .unwrap(); assert_eq!(db.root(), root_before); assert_eq!(db.size(), size_before); let too_large_target = size_before + 1; drop(db); let db = UnorderedVariable::init( ctx.child("above"), variable_db_config::("ri"), Some(too_large_target), ) .await .unwrap(); assert_eq!(db.root(), root_before); drop(db); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("reopen2"), variable_db_config::("ri"), None, ) .await .unwrap(); assert_eq!(db.root(), root_before); assert_eq!(db.size(), size_before); db.destroy().await.unwrap(); }); } /// Bounded initialization fails when the target commit's inactivity floor has been pruned, even /// if the target commit location is still retained. #[test_traced("INFO")] fn test_any_bounded_initialization_rejects_target_with_pruned_floor() { let executor = deterministic::Runner::default(); executor.start(|context| async move { const KEYS: u64 = 64; let ctx = context.child("db"); let db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("rf"), None, ) .await .unwrap(); let db = commit_writes(db, (0..KEYS).map(|i| (key(i), Some(val(i)))), None).await; let mut db = commit_writes( db, (0..KEYS).map(|i| (key(i), Some(val(1_000 + i)))), None, ) .await; let initialization_bound = db.size(); let target_floor = db.inactivity_floor_loc(); let prune_loc = target_floor + (KEYS / 2); assert!( initialization_bound > *prune_loc, "test setup expected target size > prune_loc. target={initialization_bound:?}, \ floor={target_floor:?}" ); let mut round = 0u64; while db.inactivity_floor_loc() < prune_loc { round += 1; assert!( round <= 8, "failed to advance inactivity floor enough for floor-pruned initialization test" ); db = commit_writes( db, (0..KEYS).map(|i| (key(i), Some(val(10_000 + round * KEYS + i)))), None, ) .await; } let db = db.prune(prune_loc).await.unwrap(); let bounds = db.bounds(); assert!( bounds.start > *target_floor, "test setup expected pruned start beyond target floor; bounds={bounds:?}, target_floor={target_floor:?}" ); assert!( initialization_bound > bounds.start, "test setup expected target commit retained. target={initialization_bound:?}, \ bounds={bounds:?}" ); let original_root = db.root(); _ = db.sync().await.unwrap(); let Err(err) = UnorderedVariable::init( ctx.child("cap"), variable_db_config::("rf"), Some(initialization_bound), ) .await else { panic!("expected bounded initialization to floor-pruned target to fail"); }; assert!( matches!( err, QmdbError::HistoricalFloorPruned(_) ), "unexpected bounded initialization error: {err:?}" ); let db = UnorderedVariable::init( ctx.child("unchanged"), variable_db_config::("rf"), None, ) .await.unwrap(); assert_eq!(db.root(), original_root); }); } #[test_traced("INFO")] fn test_any_bounded_reopen_below_pruned_bitmap() { deterministic::Runner::default().start(|context| async move { let ctx = context.child("db"); let mut cfg = variable_db_config::("coarse-prune"); cfg.journal_config.items_per_section = NZU64!(2048); let db = UnorderedVariable::init(ctx.child("storage"), cfg.clone(), None) .await .unwrap(); let db = commit_writes(db, (0..100).map(|i| (key(i), Some(val(i)))), None).await; let target = db.size(); let root = db.root(); let db = commit_writes(db, (0..700).map(|i| (key(i), Some(val(1_000 + i)))), None).await; let db = commit_writes(db, (0..700).map(|i| (key(i), Some(val(10_000 + i)))), None).await; let prune_loc = Location::new(600); assert!(db.inactivity_floor_loc() >= prune_loc); let live_root = db.root(); let db = db.prune(prune_loc).await.unwrap(); assert_eq!(db.root(), live_root); assert_eq!(db.bounds().start, Location::new(0)); assert!(*target < db.bitmap.pruned_bits()); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(10_000))); drop(db.sync().await.unwrap()); for cap in [Some(target), None] { let db = UnorderedVariable::init(ctx.child("reopen"), cfg.clone(), cap) .await .unwrap(); assert_eq!(db.size(), target); assert_eq!(db.root(), root); for i in 0..100 { assert_eq!(db.get(&key(i)).await.unwrap(), Some(val(i))); } assert_eq!(db.get(&key(100)).await.unwrap(), None); drop(db); } }); } /// Valid children of an applied parent retain their state across bitmap pruning, /// including children merkleized before pruning and children constructed afterward. #[test_traced("INFO")] fn test_any_live_child_across_coarse_prune() { deterministic::Runner::default().start(|context| async move { const CHUNK_BITS: u64 = commonware_utils::bitmap::Prunable::::CHUNK_SIZE_BITS; const { assert!(CHUNK_BITS <= 600) }; for child_after_prune in [false, true] { let mut reference = None; for prune in [false, true] { let suffix = format!("any-live-child-{child_after_prune}-{prune}"); let ctx = context.child("db").with_attribute("case", &suffix); let mut cfg = variable_db_config::(&suffix); cfg.journal_config.items_per_section = NZU64!(2048); let db = UnorderedVariable::init(ctx.child("storage"), cfg, None) .await .unwrap(); let mut db = commit_writes(db, (0..700).map(|i| (key(i), Some(val(i)))), None).await; let mut parent = db.new_batch(); for i in 0..700 { parent = parent.write(key(i), (i != 1).then(|| val(10_000 + i))); } let parent = parent .merkleize(&db, None, &mut Proportional) .await .unwrap(); db = db.apply_batch(Arc::clone(&parent)).await.unwrap(); db = db.commit().await.unwrap(); let make_child = || { parent .new_batch::() .write(key(0), Some(val(20_000))) .write(key(1), Some(val(20_001))) .write(key(2), None) }; let child = if child_after_prune { None } else { Some( make_child() .merkleize(&db, None, &mut Proportional) .await .unwrap(), ) }; let prune_loc = Location::new(600); assert!(db.inactivity_floor_loc() >= prune_loc); let parent_root = db.root(); if prune { db = db.prune(prune_loc).await.unwrap(); } assert_eq!(db.root(), parent_root); assert_eq!(db.bounds().start, Location::new(0)); assert_eq!( db.bitmap.pruned_bits(), if prune { 600 / CHUNK_BITS * CHUNK_BITS } else { 0 }, ); let child = match child { Some(child) => child, None => make_child() .merkleize(&db, None, &mut Proportional) .await .unwrap(), }; for (i, expected) in [ (0, Some(val(20_000))), (1, Some(val(20_001))), (2, None), (3, Some(val(10_003))), (700, None), ] { assert_eq!(child.get(&key(i), &db).await.unwrap(), expected); } let child_root = child.root(); let operations = child.operations(); db = db.apply_batch(child).await.unwrap(); assert_eq!(db.root(), child_root); let mut values = Vec::new(); for i in 0..=700 { let expected = match i { 0 => Some(val(20_000)), 1 => Some(val(20_001)), 2 | 700 => None, _ => Some(val(10_000 + i)), }; let actual = db.get(&key(i)).await.unwrap(); assert_eq!(actual, expected); values.push(actual); } let observed = (db.root(), operations, values); if let Some(reference) = &reference { assert_eq!(&observed, reference); } else { reference = Some(observed); } db.destroy().await.unwrap(); } } }); } // // The tests above use MMR-backed databases (via the concrete Db type aliases). The tests // below verify the same core operations work with the MMB family, exercising the generic // `init_fixed`/`init_variable` path with `mmb::Family`. type MmbVariable = super::db::Db< crate::merkle::mmb::Family, Context, crate::journal::contiguous::variable::Journal< Context, super::operation::Operation< crate::merkle::mmb::Family, super::operation::update::Unordered>, >, >, crate::index::unordered::Index>, Sha256, super::operation::update::Unordered>, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, Sequential, >; async fn open_mmb_db(context: Context, suffix: &str) -> MmbVariable { let cfg = variable_db_config::(suffix); super::init(context, cfg, None).await.unwrap() } async fn commit_writes_mmb( db: MmbVariable, writes: impl IntoIterator)>, metadata: Option, ) -> MmbVariable { let mut batch = db.new_batch(); for (k, v) in writes { batch = batch.write(k, v); } let merkleized = batch .merkleize(&db, metadata, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap() } #[test_traced("INFO")] fn test_mmb_batch_crud() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let db = open_mmb_db(context.child("db"), "crud").await; // Insert and read back. let db = commit_writes_mmb(db, [(key(0), Some(val(0)))], None).await; assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); // Update existing key. let db = commit_writes_mmb(db, [(key(0), Some(val(1)))], None).await; assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(1))); // Delete key. let db = commit_writes_mmb(db, [(key(0), None)], None).await; assert!(db.get(&key(0)).await.unwrap().is_none()); // Multiple keys. let db = commit_writes_mmb(db, [(key(1), Some(val(1))), (key(2), Some(val(2)))], None).await; assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1))); assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2))); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_mmb_batch_empty() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let db = open_mmb_db(context.child("db"), "empty").await; let root_before = db.root(); let merkleized = db .new_batch() .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_ne!(db.root(), root_before); let db = commit_writes_mmb(db, [(key(0), Some(val(0)))], None).await; assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_mmb_batch_metadata() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let db = open_mmb_db(context.child("db"), "meta").await; let metadata = val(42); let db = commit_writes_mmb(db, [(key(0), Some(val(0)))], Some(metadata)).await; assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); let merkleized = db .new_batch() .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), None); db.destroy().await.unwrap(); }); } #[test_traced("WARN")] fn test_mmb_recovery() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let db = open_mmb_db(context.child("db").with_attribute("index", 0), "recovery").await; let db = commit_writes_mmb(db, [(key(0), Some(val(0)))], Some(val(99))).await; let db = commit_writes_mmb(db, [(key(1), Some(val(1)))], None).await; let root = db.root(); let bounds = db.bounds(); db.sync().await.unwrap(); // Reopen and verify state. let db = open_mmb_db(context.child("db").with_attribute("index", 1), "recovery").await; assert_eq!(db.root(), root); assert_eq!(db.bounds(), bounds); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1))); assert_eq!(db.get_metadata().await.unwrap(), None); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_mmb_prune() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let mut db = open_mmb_db(context.child("db"), "prune").await; for i in 0u64..20 { db = commit_writes_mmb(db, [(key(i), Some(val(i)))], None).await; } let floor = db.inactivity_floor_loc(); let db = db.prune(floor).await.unwrap(); // All keys still accessible. for i in 0u64..20 { assert_eq!(db.get(&key(i)).await.unwrap(), Some(val(i))); } db.destroy().await.unwrap(); }); } /// One-stage pipelining lets the next batch be built while the prior batch commits. #[test_traced("INFO")] fn test_any_batch_single_stage_pipeline() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let ctx = context.child("db"); let mut db: UnorderedVariable = UnorderedVariableDb::init( ctx.child("storage"), variable_db_config::("pipe"), None, ) .await .unwrap(); { let mut batch = db.new_batch(); batch = batch.write(key(0), Some(val(0))); let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let child_merkleized = { assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); let mut child = db.new_batch(); child = child.write(key(1), Some(val(1))); child.merkleize(&db, None, &mut Proportional).await.unwrap() }; let db = db.commit().await.unwrap(); let db = db.apply_batch(child_merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0))); assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1))); db.destroy().await.unwrap(); }); } } #[cfg(test)] mod bitmap_tests { //! Regression tests for activity-bitmap maintenance in `any::Db`. The mutation code in //! `apply_batch`, `prune_bitmap`, and initialization is independent of the snapshot index //! variant, so one variant (`unordered::variable`) suffices as the test bed. use crate::{ merkle::Location, qmdb::{ any::{ BITMAP_CHUNK_BYTES, test::assert_rebuild_matches, unordered::variable::test::{AnyTest, create_test_config}, }, floor::Proportional, }, }; use commonware_cryptography::{Hasher as _, Sha256}; use commonware_macros::test_traced; use commonware_runtime::{ Runner as _, Supervisor as _, deterministic::{self, Context}, }; use commonware_utils::bitmap::{Prunable, Readable as _}; /// Bits per chunk of the test database's activity bitmap. const CHUNK_BITS: u64 = Prunable::::CHUNK_SIZE_BITS; /// Open a fresh test DB. async fn open_db(context: Context) -> AnyTest { let cfg = create_test_config(0); AnyTest::init(context, cfg, None).await.unwrap() } /// CommitFloor convention: only the *current* last commit carries bit=1; every earlier /// (now intermediate) commit boundary carries bit=0. /// /// Maintained by `apply_batch`'s explicit demote-then-promote pair on CommitFloor bits. If /// the demote step were missed, intermediate commits would persist at bit=1. #[test_traced] fn current_commit_floor_bit_is_one_others_zero() { deterministic::Runner::default().start(|context| async move { let mut db = open_db(context.child("db")).await; // Apply three single-write batches; each produces one CommitFloor op. let keys: Vec<_> = (0..3u64) .map(|i| Sha256::hash(&[&i.to_be_bytes()])) .collect(); let mut commit_locs = Vec::new(); for (i, key) in keys.iter().enumerate() { let batch = db .new_batch() .write(*key, Some(vec![i as u8])) .merkleize(&db, None, &mut Proportional) .await .unwrap(); commit_locs.push(batch.bounds.tip.size - 1); db = db.apply_batch(batch).await.unwrap(); } let db = db.commit().await.unwrap(); // Setup sanity: three strictly-increasing commit locations, all within the bitmap. assert_eq!(commit_locs.len(), 3); assert!(*commit_locs[0] < *commit_locs[1]); assert!(*commit_locs[1] < *commit_locs[2]); assert!(*commit_locs[2] < db.bitmap.len()); // Earlier two commits are intermediate -> bit=0. assert!(!db.bitmap.get_bit(*commit_locs[0])); assert!(!db.bitmap.get_bit(*commit_locs[1])); // Most recent commit is current -> bit=1. assert!(db.bitmap.get_bit(*commit_locs[2])); let reopen = open_db( context .child("reopen") .with_attribute("case", "commit_floor"), ); let db = assert_rebuild_matches(db, reopen, &keys).await; db.destroy().await.unwrap(); }); } /// Bounded initialization rebuilds bitmap activity for the selected commit. Retained keys are /// active, discarded keys are absent, and the selected commit is active. An ordinary reopen /// must reconstruct the same bitmap. #[test_traced] fn bounded_initialization_restores_bitmap_to_target_commit() { deterministic::Runner::default().start(|context| async move { let db = open_db(context.child("db")).await; let k1 = Sha256::hash(&[&[1]]); let k2 = Sha256::hash(&[&[2]]); // Two committed batches; remember the size after the first. let b1 = db .new_batch() .write(k1, Some(vec![10])) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(b1).await.unwrap().commit().await.unwrap(); let size_after_first = db.bounds().end; let b2 = db .new_batch() .write(k2, Some(vec![20])) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(b2).await.unwrap(); // Setup sanity: both keys present, db has advanced past size_after_first. assert_eq!(db.get(&k1).await.unwrap(), Some(vec![10])); assert_eq!(db.get(&k2).await.unwrap(), Some(vec![20])); assert!(*db.bounds().end > *size_after_first); // Reopen at the state after the first commit. let db = { _ = db.sync().await.unwrap(); AnyTest::init( context.child("cap"), create_test_config(0), Some(size_after_first), ) .await } .unwrap(); // After recovery, k2 is gone and k1 remains. assert_eq!(db.get(&k1).await.unwrap(), Some(vec![10])); assert!(db.get(&k2).await.unwrap().is_none()); let reopen = open_db(context.child("reopen").with_attribute("case", "rewind")); let db = assert_rebuild_matches(db, reopen, &[k1, k2]).await; db.destroy().await.unwrap(); }); } /// The floor walk crosses the end of the applied bitmap into pending ancestor operations. /// /// A child overwrites its parent's update of an applied key, so the walk must pass the /// superseded ancestor location as inactive. The anchor overwrite and the previous commit /// provide the walk's entries. /// /// Applying the child publishes its latest values. Rebuilding independently checks the root /// and activity state after the walk. #[test_traced] fn floor_scan_falls_through_to_uncommitted_tail() { deterministic::Runner::default().start(|context| async move { let db = open_db(context.child("db")).await; let anchor = Sha256::hash(&[&[0xAA]]); // Commit one key. let b = db .new_batch() .write(anchor, Some(vec![1])) .merkleize(&db, None, &mut Proportional) .await .unwrap(); let db = db.apply_batch(b).await.unwrap(); // Setup sanity: anchor in committed snapshot. assert_eq!(db.get(&anchor).await.unwrap(), Some(vec![1])); let committed_bitmap_len = db.bitmap.len(); // Uncommitted parent: re-touch anchor at a location above the committed bitmap. let parent = db .new_batch() .write(anchor, Some(vec![2])) .merkleize(&db, None, &mut Proportional) .await .unwrap(); assert!( parent.bounds.tip.size > committed_bitmap_len, "parent must extend past committed bitmap to exercise the tail path", ); // The pending child supersedes the anchor and creates 16 other keys. let others: Vec<_> = (0..16u64) .map(|i| Sha256::hash(&[&(1000 + i).to_be_bytes()])) .collect(); let mut child_batch = parent.new_batch::(); child_batch = child_batch.write(anchor, Some(vec![3])); for (i, k) in others.iter().enumerate() { child_batch = child_batch.write(*k, Some(vec![i as u8])); } let child = child_batch .merkleize(&db, None, &mut Proportional) .await .unwrap(); assert!( child.bounds.tip.size > committed_bitmap_len, "child must include an uncommitted tail beyond committed bitmap", ); let expected_root = child.root(); let db = db.apply_batch(child).await.unwrap(); assert_eq!(db.root(), expected_root); assert_eq!(db.get(&anchor).await.unwrap(), Some(vec![3])); let keys: Vec<_> = [anchor].into_iter().chain(others).collect(); let reopen = open_db(context.child("reopen").with_attribute("case", "tail")); let db = assert_rebuild_matches(db, reopen, &keys).await; db.destroy().await.unwrap(); }); } /// Pruning two chunks retains a log start inside the second chunk. A reopen rebuilds the /// pruned prefix from that start, one chunk below the live prefix. #[test_traced] fn pruned_prefix_derives_from_retained_start_on_reopen() { deterministic::Runner::default().start(|context| async move { let mut db = open_db(context.child("db")).await; // Write the same 700 keys in three commits so the inactivity floor passes two chunks. let keys: Vec<_> = (0..700u64) .map(|i| Sha256::hash(&[&i.to_be_bytes()])) .collect(); for round in 0..3u8 { let mut batch = db.new_batch(); for (i, key) in keys.iter().enumerate() { batch = batch.write(*key, Some(vec![round, i as u8])); } let batch = batch.merkleize(&db, None, &mut Proportional).await.unwrap(); db = db.apply_batch(batch).await.unwrap().commit().await.unwrap(); } let loc = Location::new(2 * CHUNK_BITS); assert!(db.inactivity_floor_loc() >= loc); // Prune two chunks. The log retains the section containing the prune location, which // starts inside the second chunk, while the live bitmap drops both chunks. let db = db.prune(loc).await.unwrap(); let start = *db.bounds().start; assert!( (CHUNK_BITS..2 * CHUNK_BITS).contains(&start), "retained start {start} must lie in the second chunk", ); assert_eq!(db.bitmap.pruned_bits(), 2 * CHUNK_BITS); // Reopen. The rebuilt prefix keeps the second chunk. let reopen = open_db(context.child("reopen").with_attribute("case", "coarse")); let db = assert_rebuild_matches(db, reopen, &keys).await; assert_eq!(db.bitmap.pruned_bits(), CHUNK_BITS); db.destroy().await.unwrap(); }); } }