//! An authenticated database that only supports adding new keyed values (no updates or //! deletions). //! //! Two variants are available based on value encoding: //! - [fixed]: For fixed-size values. //! - [variable]: For variable-size values. //! //! # Inactivity floor //! //! Each commit carries an inactivity floor: a location before which the application //! declares operations are no longer needed. The floor is embedded in the operation //! log and included in the Merkle root, so all replicas processing the same operations //! arrive at the same floor. //! //! The floor controls two things: //! - **Pruning**: [`Immutable::prune`] only allows pruning up to the floor. //! - **Reconstruction**: on restart or sync, the snapshot is rebuilt from the floor //! onward. Keys set before the floor are not loaded into memory. //! //! A running database keeps serving keys set before the floor until their operations are //! pruned, while a restarted or synced one does not know them, so only reads at or above the //! floor are consistent across replicas. Treat declaring a floor as retiring every key set //! before it. //! //! The floor must be monotonically non-decreasing across commits and must not exceed the //! location of the batch's own commit operation. Pass `db.inactivity_floor_loc()` to keep the //! floor unchanged, or a higher value to advance it. //! //! # Examples //! //! ```ignore //! // Simple mode: apply a batch, then durably commit it. //! // The third argument to merkleize is the inactivity floor. Operations //! // before this location are declared inactive by the application. //! let floor = db.inactivity_floor_loc(); //! let merkleized = db.new_batch() //! .set(key, value) //! .merkleize(&db, None, floor).await.unwrap(); //! let db = db.apply_batch(merkleized).await?.commit().await?; //! ``` //! //! ```ignore //! // Several batches can be built on one parent before any is applied. //! let floor = db.inactivity_floor_loc(); //! let parent = db.new_batch() //! .set(key_a, value_a) //! .merkleize(&db, None, floor).await.unwrap(); //! //! let child_a = parent.new_batch::() //! .set(key_b, value_b) //! .merkleize(&db, None, floor).await.unwrap(); //! //! let child_b = parent.new_batch::() //! .set(key_c, value_c) //! .merkleize(&db, None, floor).await.unwrap(); //! //! let db = db.apply_batch(child_a).await?.commit().await?; //! ``` //! //! ```ignore //! // Apply a parent batch, commit, then build a child batch from the newly //! // published state and apply it. //! let floor = db.inactivity_floor_loc(); //! let parent = db.new_batch() //! .set(key_a, value_a) //! .merkleize(&db, None, floor).await.unwrap(); //! let db = db.apply_batch(parent).await?.commit().await?; //! //! let child = db.new_batch() //! .set(key_b, value_b) //! .merkleize(&db, None, floor).await.unwrap(); //! let db = db.apply_batch(child).await?.commit().await?; //! ``` use crate::{ Context, index::immutable::Index, journal::{ authenticated, contiguous::{Contiguous, Mutable}, }, merkle::{Family, Location, Proof, full::Config as MerkleConfig}, qmdb::{ Error, any::ValueEncoding, chain, metrics::Metrics, operation::Key, single_operation_root, sync::source, }, translator::Translator, }; use commonware_codec::EncodeShared; use commonware_cryptography::Hasher; use commonware_macros::boxed; use commonware_parallel::Strategy; use commonware_runtime::{Handle, ReadOptions}; use core::num::{NonZeroU64, NonZeroUsize}; use futures::{StreamExt, pin_mut}; use std::{ops::Range, sync::Arc}; use tracing::warn; pub mod batch; mod compact; pub mod fixed; mod operation; pub mod sync; pub mod variable; pub use compact::{ Config as CompactConfig, Db as CompactDb, MerkleizedBatch as CompactMerkleizedBatch, UnmerkleizedBatch as CompactUnmerkleizedBatch, }; pub use operation::Operation; /// Build the snapshot by replaying the log from `inactivity_floor_loc`, inserting the location of /// every retained [Operation::Set]. Assumes the log is not pruned beyond the inactivity floor. /// /// `init_buffer` sizes the replay read buffer (in bytes). async fn build_snapshot( context: E, translator: T, inactivity_floor_loc: Location, log: &C, init_buffer: NonZeroUsize, ) -> Result, Error> where F: Family, E: Context, K: Key, V: ValueEncoding, C: Contiguous>, T: Translator, { // Size the table for every replayed operation, an upper bound on the sets among them, so the // replay neither counts them first nor grows the table as it goes. Commits are not indexed, // so a log of tiny batches can leave the table one doubling larger than its entries need. let end = log.bounds().end; let mut snapshot = Index::new( context, translator, *inactivity_floor_loc, usize::try_from(end - *inactivity_floor_loc) .expect("operations since the floor must fit in usize"), ); // Init reads every operation once, so the replayed pages are not kept in the OS page cache: // the init cache, not the OS cache, decides which probes hit. let stream = log .replay( *inactivity_floor_loc..end, init_buffer, ReadOptions::DONT_CACHE, ) .await?; pin_mut!(stream); while let Some(batch) = stream.next().await { for (loc, op) in batch? { if let Operation::Set(key, _) = op { snapshot.insert(&key, loc); } } } Ok(snapshot) } /// Compute the authenticated root of a newly initialized database without opening storage. /// /// The initial commit never carries metadata, so this root always represents `Commit(None, 0)`. pub fn initial_root() -> H::Digest where F: Family, K: Key, V: ValueEncoding, H: Hasher, Operation: EncodeShared, { single_operation_root::(&Operation::::Commit(None, Location::new(0))) } /// Configuration for an [Immutable] 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 log: J, /// The translator used by the compressed index. pub translator: T, /// Size (in bytes) of the read buffer used to replay the log during init. pub init_buffer: NonZeroUsize, } /// An authenticated database that only supports adding new keyed values (no updates or /// deletions). /// /// # Invariant /// /// A key must be set at most once across the database history. /// /// Use [fixed::Db] or [variable::Db] for concrete instantiations. pub struct Immutable< F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, > where C::Item: EncodeShared, { /// Authenticated journal of operations. journal: authenticated::Journal, /// Cached canonical operations root. pub(crate) root: H::Digest, /// The location of the [Operation::Set] of every key set at or after the inactivity floor /// the snapshot was built from, excluding pruned locations. pub(crate) snapshot: Index, /// The inactivity floor declared by the last committed batch. /// Operations before this location are considered inactive by the application. pub(crate) inactivity_floor_loc: Location, /// Metrics for this database. metrics: Metrics, } impl std::fmt::Debug for Immutable where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, H: Hasher, T: Translator, S: Strategy, { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Immutable") .field("bounds", &self.bounds()) .field("inactivity_floor_loc", &self.inactivity_floor_loc()) .finish_non_exhaustive() } } // Shared read-only functionality. impl Immutable where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, H: Hasher, T: Translator, S: Strategy, { /// Initialize from the latest retained commit, discarding every operation after it. /// `Some(max_size)` selects the latest retained commit with at most `max_size` operations. /// `None` selects the latest retained state. #[boxed] pub async fn init( context: E, cfg: Config, max_size: Option>, ) -> Result> where C: authenticated::Backing, { // Snapshot reconstruction replays from the selected commit's inactivity floor, so that // floor must remain in the retained operation prefix. let mut journal = crate::qmdb::init_journal::( context.child("journal"), cfg.merkle_config, cfg.log, max_size, true, ) .await?; if journal.size() == 0 { warn!("Authenticated log is empty, initialized new db."); (journal, _) = journal .append(&Operation::Commit(None, Location::new(0))) .await?; journal = journal.sync().await?; } // Rebuild the in-memory state the last commit defines. Its inactivity floor is where // the snapshot replay starts, so keys set before the floor are not loaded. let (last_commit_loc, inactivity_floor_loc, snapshot) = { let bounds = journal.journal.bounds(); let last_commit_loc = Location::new(bounds.end.checked_sub(1).expect("commit should exist")); // Read the floor from the last commit operation. let last_op = journal.journal.read(*last_commit_loc).await?; let inactivity_floor_loc = last_op .has_floor() .expect("last operation should be a commit with floor"); // Replay the log from the inactivity floor to build the snapshot. let snapshot = build_snapshot( context.child("snapshot"), cfg.translator, inactivity_floor_loc, &journal.journal, cfg.init_buffer, ) .await?; (last_commit_loc, inactivity_floor_loc, snapshot) }; let inactive_peaks = F::inactive_peaks(last_commit_loc + 1, inactivity_floor_loc); let root = journal.root(inactive_peaks)?; let metrics = Metrics::new(context); let db = Self { journal, root, snapshot, inactivity_floor_loc, metrics, }; db.update_metrics(); Ok(db) } /// Return the inactivity floor location declared by the last committed batch. pub const fn inactivity_floor_loc(&self) -> Location { self.inactivity_floor_loc } /// Return the Location of the next operation appended to this db. pub fn size(&self) -> Location { self.bounds().end } /// Return the retained operation range `[start, end)`. /// /// Proof generation also requires the necessary Merkle nodes to be retained. pub fn bounds(&self) -> Range> { Location::new(self.journal.bounds().start)..Location::new(self.journal.bounds().end) } /// Update state gauges from the current database state. fn update_metrics(&self) { let bounds = self.journal.bounds(); self.metrics.update( bounds.end, bounds.start, *self.inactivity_floor_loc, bounds.end - 1, ); } /// Return the most recent location from which this database can safely be synced, and the /// upper bound on [`Self::prune`]'s `loc`. For immutable databases, this equals the /// inactivity floor declared by the last committed batch. pub const fn sync_boundary(&self) -> Location { self.inactivity_floor_loc } /// Get the value of `key` in the db, or None if it has no value or its corresponding operation /// has been pruned. /// /// Unlike batch reads, this may return a value written below the inactivity floor, until the /// database prunes that operation or rebuilds its index from the floor on restart or sync. pub async fn get(&self, key: &K) -> Result, Error> { self.get_from(key, Location::new(0)).await } /// Like [`Self::get`], but also ignores operations below `floor`. pub(crate) async fn get_from( &self, key: &K, floor: Location, ) -> Result, Error> { let _timer = self.metrics.get_timer(); self.metrics.get_calls.inc(); self.metrics.lookups_requested.inc(); for loc in self.snapshot.get(key) { let loc = Location::new(loc); if loc < floor { continue; } if let Some(v) = Self::get_from_loc(&self.journal, key, loc).await? { return Ok(Some(v)); } } Ok(None) } /// Batch read multiple keys. /// /// Returns results in the same order as the input keys. Like [`Self::get`], this may return /// values written below the inactivity floor. pub async fn get_many(&self, keys: &[&K]) -> Result>, Error> { self.get_many_from(keys, Location::new(0)).await } /// Like [`Self::get_many`], but also ignores operations below `floor`. pub(crate) async fn get_many_from( &self, keys: &[&K], floor: Location, ) -> Result>, Error> { if keys.is_empty() { return Ok(Vec::new()); } let _timer = self.metrics.get_many_timer(); self.metrics.get_many_calls.inc(); self.metrics.lookups_requested.inc_by(keys.len() as u64); let mut candidates: Vec<(usize, u64)> = Vec::with_capacity(keys.len()); let mut results: Vec> = vec![None; keys.len()]; // Every location at or above `floor` whose translated key matches a requested key is a // candidate for it, including collisions. They are read in one batch below, each location // once, and resolved by comparing full keys. for (key_idx, key) in keys.iter().enumerate() { for loc in self.snapshot.get(key) { if loc < *floor { continue; } candidates.push((key_idx, loc)); } } if candidates.is_empty() { return Ok(results); } candidates.sort_unstable_by_key(|&(_, pos)| pos); let same_pos = |a: &(usize, u64), b: &(usize, u64)| a.1 == b.1; let positions: Vec = candidates .chunk_by(same_pos) .map(|group| group[0].1) .collect(); let ops = self.journal.read_many(&positions).await?; assert_eq!( ops.len(), positions.len(), "read_many returns one operation per position" ); for (op, group) in ops.into_iter().zip(candidates.chunk_by(same_pos)) { let Operation::Set(k, v) = op else { return Err(Error::UnexpectedData(Location::new(group[0].1))); }; let mut pending = None; for &(key_idx, _) in group { if results[key_idx].is_some() || k != *keys[key_idx] { continue; } if let Some(prev) = pending.replace(key_idx) { results[prev] = Some(v.clone()); } } if let Some(last) = pending { results[last] = Some(v); } } Ok(results) } /// Get the value of the operation with location `loc` in the db if it matches `key`. Returns /// [`crate::qmdb::Error::OperationPruned`] if loc precedes the oldest retained location. The /// location is otherwise assumed valid. async fn get_from_loc( reader: &impl Contiguous>, key: &K, loc: Location, ) -> Result, Error> { if loc < reader.bounds().start { return Err(Error::OperationPruned(loc)); } let Operation::Set(k, v) = reader.read(*loc).await? else { return Err(Error::UnexpectedData(loc)); }; if k != *key { Ok(None) } else { Ok(Some(v)) } } /// Get the metadata associated with the last commit. pub async fn get_metadata(&self) -> Result, Error> { // The journal always ends with a commit operation. let last_commit_loc = self.size() - 1; let Operation::Commit(metadata, _floor) = self.journal.journal.read(*last_commit_loc).await? else { unreachable!("no commit operation at location of last commit {last_commit_loc}"); }; Ok(metadata) } /// Analogous to proof but with respect to the state of the database when it had `op_count` /// operations. /// /// # Contract /// /// `op_count` must be a commit-boundary size: the operation at `op_count - 1` must /// itself be a commit op. Non-commit-boundary sizes are not supported because the /// inactivity floor governing them is not directly retrievable. /// /// # Errors /// /// Returns [crate::merkle::Error::RangeOutOfBounds] if `op_count` > number of operations, or /// if `start_loc` >= `op_count`. /// Returns [`Error::Journal`] with [`crate::journal::Error::ItemPruned`] or [`Error::Merkle`] /// with [`crate::merkle::Error::ElementPruned`] if a required operation or Merkle node has /// been pruned. /// Returns [`Error::HistoricalFloorPruned`] if `op_count - 1` is retained but is not a /// commit op, either because the caller passed a non-commit-boundary `op_count` or /// because pruning removed the commit that would have governed `op_count`. #[allow(clippy::type_complexity)] #[tracing::instrument( name = "qmdb.immutable.db.historical_proof", level = "info", skip_all, fields( op_count = *op_count, start_loc = *start_loc, max_ops = max_ops.get(), ), )] pub async fn historical_proof( &self, op_count: Location, start_loc: Location, max_ops: NonZeroU64, ) -> Result<(Proof, Vec>), Error> { if op_count > self.journal.size() { return Err(crate::merkle::Error::RangeOutOfBounds(op_count).into()); } let inactive_peaks = crate::qmdb::inactive_peaks_at::(&self.journal, op_count).await?; Ok(self .journal .historical_proof(op_count, start_loc, max_ops, inactive_peaks) .await?) } /// Generate and return: /// 1. a proof of all operations applied to the db in the range starting at (and including) /// location `start_loc`, and ending at the first of either: /// - the last operation performed, or /// - the operation `max_ops` from the start. /// 2. the operations corresponding to the leaves in this range. pub async fn proof( &self, start_index: Location, max_ops: NonZeroU64, ) -> Result<(Proof, Vec>), Error> { let op_count = self.bounds().end; self.historical_proof(op_count, start_index, max_ops).await } /// Prune operations prior to `loc` without changing the db's root. Keys whose operations /// are no longer retained cannot be retrieved. /// /// The retained start in [`Self::bounds`] can remain below `loc`. /// /// Pruning is irreversible and requires no prior commit. After a crash, the database remains /// recoverable; uncommitted operations are not guaranteed to survive. /// /// # Errors /// /// - Returns [Error::PruneBeyondMinRequired] if `loc` > inactivity floor. #[tracing::instrument(name = "qmdb.immutable.db.prune", level = "info", skip_all)] #[boxed] pub async fn prune(mut self, loc: Location) -> Result> { let _timer = self.metrics.prune_timer(); self.metrics.prune_calls.inc(); if loc > self.inactivity_floor_loc { return Err(Error::PruneBeyondMinRequired( loc, self.inactivity_floor_loc, )); } (self.journal, _) = self.journal.prune(loc).await?; self.snapshot.prune(*self.bounds().start); self.update_metrics(); Ok(self) } /// Return the canonical QMDB root of the db. pub const fn root(&self) -> H::Digest { self.root } /// Return a reference to the merkleization strategy. pub const fn strategy(&self) -> &S { self.journal.strategy() } /// Return the pinned Merkle nodes at the given location. pub async fn pinned_nodes_at(&self, loc: Location) -> Result, Error> { self.journal .merkle .pinned_nodes_at(loc) .await .map_err(Into::into) } /// Sync all database state to disk. While this isn't necessary to ensure durability of /// committed operations, periodic invocation may reduce memory usage and the time required to /// recover the database on restart. #[tracing::instrument(name = "qmdb.immutable.db.sync", level = "info", skip_all)] pub async fn sync(mut self) -> Result> { let _timer = self.metrics.sync_timer(); self.metrics.sync_calls.inc(); self.journal = self.journal.sync().await?; Ok(self) } /// Begin durably persisting the journal state published by prior [`Immutable::apply_batch`] /// calls. /// /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit], /// plus a best-effort attempt to bound the recovery needed on startup. Use [Self::sync] to /// guarantee none is needed. A new sync waits for the prior sync before starting. Failures /// of the deferred durability work surface on the returned handle. A failed data sync also /// fails the next durability operation. A failed recovery-watermark sync is not observed by /// [Self::commit], and a failed merkle-node sync may not be. Both resurface on the next /// [Self::sync]. #[tracing::instrument(name = "qmdb.immutable.db.start_sync", level = "info", skip_all)] pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error> { self.metrics.start_sync_calls.inc(); let handle; (self.journal, handle) = self.journal.start_sync().await?; Ok((self, handle)) } /// Durably commit the journal state published by prior [`Immutable::apply_batch`] calls. #[tracing::instrument(name = "qmdb.immutable.db.commit", level = "info", skip_all)] pub async fn commit(mut self) -> Result> { let _timer = self.metrics.commit_timer(); self.metrics.commit_calls.inc(); self.journal = self.journal.commit().await?; Ok(self) } /// Destroy the db, removing all data from disk. #[boxed] pub async fn destroy(self) -> Result<(), Error> { Ok(self.journal.destroy().await?) } /// The [`Commitment`](chain::Commitment) for the database's current state. pub(crate) fn commitment(&self) -> chain::Commitment { chain::Commitment::new(self.size(), self.root) } /// Create a new speculative batch of operations with this database as its parent. pub fn new_batch(&self) -> batch::UnmerkleizedBatch { batch::UnmerkleizedBatch::new(self, self.commitment()) } /// Check that `batch` can be applied to the database in its current state, without /// applying it. /// /// [`Self::apply_batch`] runs the same validation but consumes the database when it /// fails; callers that want to reject a bad batch and keep the handle can check first. pub fn validate_batch( &self, batch: &batch::MerkleizedBatch, ) -> Result<(), Error> { batch.bounds.validate_apply_to(self.commitment()) } /// Apply a [`batch::MerkleizedBatch`] to the database. /// /// This publishes the batch to the in-memory database state and appends it to the journal. /// Call [`Immutable::commit`] or [`Immutable::sync`], or await the handle returned by /// [`Immutable::start_sync`], to make the applied state durable. /// /// # Errors /// /// Returns [`Error::StaleBatch`] if the database is at neither the batch's recorded database /// boundary nor an ancestor's tip. #[tracing::instrument(name = "qmdb.immutable.db.apply_batch", level = "info", skip_all)] pub async fn apply_batch( mut self, batch: Arc>, ) -> Result> { let _timer = self.metrics.apply_batch_timer(); self.metrics.apply_batch_calls.inc(); self.validate_batch(&batch)?; let db_size = self.size(); // Journal writes and snapshot inserts are independent until batch publication. let apply_snapshot = async { for (key, entry) in batch.diff.iter() { self.snapshot.insert(key, *entry.loc); } for (i, ancestor_diff) in batch.ancestor_diffs.iter().enumerate() { if batch.bounds.ancestors[i].size <= db_size { continue; } for (key, entry) in ancestor_diff.iter() { self.snapshot.insert(key, *entry.loc); } } }; let (journal, ()) = futures::join!( self.journal.apply_batch(&batch.journal_batch), apply_snapshot ); self.journal = journal?; // Update state. self.inactivity_floor_loc = batch.bounds.inactivity_floor; self.root = batch.root(); self.update_metrics(); self.metrics .operations_applied .inc_by(*batch.bounds.tip.size - *db_size); Ok(self) } } impl crate::qmdb::sync::Source for Immutable where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, Operation: EncodeShared, H: Hasher, T: Translator, S: Strategy, { type Family = F; type Digest = H::Digest; type Op = Operation; type Error = Error; async fn serve(&self, request: crate::qmdb::sync::Request) -> source::Result { self.journal.serve(request).await } } #[cfg(test)] pub(super) mod tests { use super::*; use crate::{ merkle::{Family, Location}, qmdb::{verify_proof, verify_proof_and_pinned_nodes}, translator::TwoCap, }; use commonware_codec::EncodeShared; use commonware_cryptography::{Sha256, sha256, sha256::Digest}; use commonware_runtime::{Supervisor as _, deterministic}; use commonware_utils::NZU64; use core::{future::Future, pin::Pin}; const ITEMS_PER_SECTION: u64 = 5; /// Emits the named test against `mmr::Family` and `mmb::Family`. macro_rules! immutable_tests { ($($name:ident => $scenario:ident, $open:ident;)*) => { $( #[test_traced] fn $name() { deterministic::Runner::default().start(|ctx| async move { immutable_tests!(@fixture $open, $scenario, mmr, ctx); }); } )* paste::paste! { $( #[test_traced] fn [<$name _mmb>]() { deterministic::Runner::default().start(|ctx| async move { immutable_tests!(@fixture $open, $scenario, mmb, ctx); }); } )* } }; (@fixture pair, $scenario:ident, $family:ident, $ctx:ident) => { let db = Db::<$family::Family, _, Digest, Digest, Sha256, TwoCap, Sequential>::init( $ctx.child("db"), config("db"), None, ).await.unwrap(); let foreign = Db::<$family::Family, _, Digest, Digest, Sha256, TwoCap, Sequential>::init( $ctx.child("foreign"), config("foreign"), None, ).await.unwrap(); tests::$scenario(db, foreign).await; }; (@fixture $open:ident, $scenario:ident, $family:ident, $ctx:ident) => { tests::$scenario($ctx, $open::<$family::Family>).await; }; } pub(super) use immutable_tests; type TestDb = Immutable< F, deterministic::Context, Digest, V, C, Sha256, TwoCap, commonware_parallel::Sequential, >; /// Reads stay exact after the ancestor batches are dropped, because merkleization /// retains their diffs. #[boxed] pub(crate) async fn run_dropped_ancestor_reads( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Seed the database with a key the chain never touches. let untouched = Sha256::fill(0u8); let untouched_value = Sha256::fill(100u8); let seed = db .new_batch() .set(untouched, untouched_value) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(seed).await.unwrap(); // A three-level unapplied chain, one key per level. let a = db .new_batch() .set(Sha256::fill(1u8), Sha256::fill(101u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let b = a .new_batch::() .set(Sha256::fill(2u8), Sha256::fill(102u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let c = b .new_batch::() .set(Sha256::fill(3u8), Sha256::fill(103u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = c.new_batch::(); drop(a); drop(b); // The merkleized batch and its unmerkleized child both resolve every // ancestor write from the retained diffs. for i in 0..4u8 { let key = Sha256::fill(i); let expected = Some(Sha256::fill(i.wrapping_add(100))); assert_eq!(c.get(&key, &db).await.unwrap(), expected); assert_eq!(child.get(&key, &db).await.unwrap(), expected); } let keys: Vec<_> = (0..4u8).map(Sha256::fill).collect(); let key_refs: Vec<_> = keys.iter().collect(); let expected: Vec<_> = (0..4u8) .map(|i| Some(Sha256::fill(i.wrapping_add(100)))) .collect(); assert_eq!(c.get_many(&key_refs, &db).await.unwrap(), expected); assert_eq!(child.get_many(&key_refs, &db).await.unwrap(), expected); // Merkleizing the child returns `StaleBatch`, because its dropped ancestors are unapplied. assert!(matches!( child.merkleize(&db, None, Location::new(0)).await, Err(Error::StaleBatch) )); // Applying the chain lands the same values the reads reported. let db = db.apply_batch(c).await.unwrap(); for i in 0..4u8 { assert_eq!( db.get(&Sha256::fill(i)).await.unwrap(), Some(Sha256::fill(i.wrapping_add(100))) ); } assert_eq!(db.get(&untouched).await.unwrap(), Some(untouched_value)); db.destroy().await.unwrap(); } /// Speculative reads binary-search a merkleized batch's diff, so every key of a multi-key /// batch must resolve before apply regardless of the order the keys were set in. #[boxed] pub(crate) async fn run_multi_key_batch_reads( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Set the keys in descending order so the batch itself must order the diff. let value = |i: u8| Sha256::fill(i.wrapping_add(100)); let mut batch = db.new_batch(); for i in (0..8u8).rev() { batch = batch.set(Sha256::fill(i), value(i)); } let parent = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); let keys: Vec<_> = (0..8u8).map(Sha256::fill).collect(); let key_refs: Vec<_> = keys.iter().collect(); let expected: Vec<_> = (0..8u8).map(|i| Some(value(i))).collect(); // The batch resolves its own diff, an unmerkleized child resolves it through the parent, // and a merkleized child resolves it as a retained ancestor diff. assert_eq!(parent.get_many(&key_refs, &db).await.unwrap(), expected); let draft = parent.new_batch::(); assert_eq!(draft.get_many(&key_refs, &db).await.unwrap(), expected); let child = draft.merkleize(&db, None, Location::new(0)).await.unwrap(); assert_eq!(child.get_many(&key_refs, &db).await.unwrap(), expected); for i in 0..8u8 { assert_eq!( child.get(&Sha256::fill(i), &db).await.unwrap(), Some(value(i)) ); } db.destroy().await.unwrap(); } /// A batch's proof and pinned nodes return `StaleRead` once a sibling is applied. #[boxed] pub(crate) async fn run_proof_refused_after_sibling_apply( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let batch = db .new_batch() .set(Sha256::fill(1u8), Sha256::fill(101u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let sibling = db .new_batch() .set(Sha256::fill(1u8), Sha256::fill(102u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert!(batch.proof(&db).is_ok()); assert!(batch.pinned_nodes(&db).is_ok()); // Applying the sibling moves the database to a state outside the batch's chain. let db = db.apply_batch(sibling).await.unwrap(); assert!(matches!(batch.proof(&db), Err(Error::StaleRead))); assert!(matches!(batch.pinned_nodes(&db), Err(Error::StaleRead))); db.destroy().await.unwrap(); } /// Batch reads cover only operations at or above the batch's inactivity floor, so a long-lived, /// a pruned, and a reopened instance at the same commitment all answer alike. #[boxed] pub(crate) async fn run_reads_ignore_inactive_operations( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let (old, fresh, own) = (Sha256::fill(1u8), Sha256::fill(2u8), Sha256::fill(3u8)); // `old` lands at location 1, and the commit at 2 keeps the floor at 0. let seed = db .new_batch() .set(old, Sha256::fill(101u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(seed).await.unwrap(); // Filler keys push the next write past a blob boundary, so pruning can remove `old`. let mut filler = db.new_batch(); for i in 10..18u8 { filler = filler.set(Sha256::fill(i), Sha256::fill(i)); } let filler = filler.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(filler).await.unwrap(); // A parent writes `fresh` at `base` and raises the floor there, leaving `old` below it. let base = db.bounds().end; let parent = db .new_batch() .set(fresh, Sha256::fill(102u8)) .merkleize(&db, None, base) .await .unwrap(); assert_eq!(parent.get(&old, &db).await.unwrap(), None); assert_eq!( parent.get(&fresh, &db).await.unwrap(), Some(Sha256::fill(102u8)) ); let child = parent.new_batch::(); assert_eq!(child.get(&old, &db).await.unwrap(), None); assert_eq!( child.get_many(&[&old, &fresh], &db).await.unwrap(), vec![None, Some(Sha256::fill(102u8))] ); // A child raising the floor to its commit, past every write including its own, sees none. let grandchild = child .set(own, Sha256::fill(103u8)) .merkleize(&db, None, base + 3) .await .unwrap(); assert_eq!( grandchild .get_many(&[&old, &fresh, &own], &db) .await .unwrap(), vec![None, None, None] ); drop(grandchild); // Apply the parent. The long-lived instance still indexes `old`, a pruned one has removed // it, and a reopened one rebuilds its index from the floor. Batch reads agree on all three. let db = db.apply_batch(parent).await.unwrap(); let db = db.commit().await.unwrap(); let expected = vec![None, Some(Sha256::fill(102u8))]; assert_eq!( db.new_batch().get_many(&[&old, &fresh], &db).await.unwrap(), expected ); assert_eq!( db.new_batch().get(&fresh, &db).await.unwrap(), Some(Sha256::fill(102u8)) ); // `twin` shares `old`'s translated key, so the index yields both locations for either // key, and only the one at or above the floor is read. let mut twin = [1u8; 32]; twin[31] = 9; let twin = Digest(twin); let collision = db .new_batch() .set(twin, Sha256::fill(104u8)) .merkleize(&db, None, base) .await .unwrap(); let db = db.apply_batch(collision).await.unwrap(); let db = db.commit().await.unwrap(); let collided = vec![None, Some(Sha256::fill(104u8))]; assert_eq!(db.new_batch().get(&old, &db).await.unwrap(), None); assert_eq!( db.new_batch().get(&twin, &db).await.unwrap(), Some(Sha256::fill(104u8)) ); assert_eq!( db.new_batch().get_many(&[&old, &twin], &db).await.unwrap(), collided ); let db = db.prune(base).await.unwrap(); assert!(db.bounds().start > Location::new(1), "`old` must be pruned"); assert_eq!( db.new_batch().get_many(&[&old, &fresh], &db).await.unwrap(), expected ); assert_eq!( db.new_batch().get_many(&[&old, &twin], &db).await.unwrap(), collided ); drop(db); let db = open_db(context.child("db")).await; assert_eq!( db.new_batch().get_many(&[&old, &fresh], &db).await.unwrap(), expected ); assert_eq!(db.new_batch().get(&old, &db).await.unwrap(), None); assert_eq!( db.new_batch().get_many(&[&old, &twin], &db).await.unwrap(), collided ); assert_eq!( db.new_batch().get(&twin, &db).await.unwrap(), Some(Sha256::fill(104u8)) ); db.destroy().await.unwrap(); } /// A child built before its parent was applied merkleizes to the same root after the /// database prunes past the chain's recorded database boundary. #[boxed] pub(crate) async fn run_merkleize_across_prune( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Seed committed state spanning several journal blobs, so the prune below removes // history and the chain's recorded boundary sits above zero. let mut seed = db.new_batch(); for i in 0..40u8 { seed = seed.set(Sha256::fill(i), Sha256::fill(i.wrapping_add(100))); } let seed = seed.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(seed).await.unwrap(); let db = db.commit().await.unwrap(); // The parent declares a floor at its own commit, past the chain's recorded boundary. let parent_floor = db.bounds().end + 1; let parent = db .new_batch() .set(Sha256::fill(200u8), Sha256::fill(212u8)) .merkleize(&db, None, parent_floor) .await .unwrap(); // Two identical children built before the apply. One merkleizes now to record the expected // root. let child = parent .new_batch::() .set(Sha256::fill(201u8), Sha256::fill(213u8)); let twin = parent .new_batch::() .set(Sha256::fill(201u8), Sha256::fill(213u8)); let expected = twin .merkleize(&db, None, parent_floor) .await .unwrap() .root(); // Apply the parent and prune to its floor. let db = db.apply_batch(Arc::clone(&parent)).await.unwrap(); let db = db.commit().await.unwrap(); let floor = db.inactivity_floor_loc(); let db = db.prune(floor).await.unwrap(); assert!(*db.bounds().start > 0, "the prune must remove history"); // The child built before the apply still merkleizes to the same root. let merkleized = child.merkleize(&db, None, parent_floor).await.unwrap(); assert_eq!(merkleized.root(), expected); db.destroy().await.unwrap(); } /// Once a sibling batch is applied, reads through the other sibling and its child return /// `StaleRead`, while merkleizing the child and applying the other sibling return `StaleBatch`. #[boxed] pub(crate) async fn run_stale_fork_refuses( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let k1 = Sha256::fill(1u8); let k2 = Sha256::fill(2u8); let floor = db.inactivity_floor_loc(); let winner = db .new_batch() .set(k1, Sha256::fill(3u8)) .merkleize(&db, None, floor) .await .unwrap(); let loser = db .new_batch() .set(k1, Sha256::fill(4u8)) .merkleize(&db, None, floor) .await .unwrap(); assert_eq!(winner.bounds().tip.size, loser.bounds().tip.size); assert_ne!(winner.root(), loser.root()); let child = loser.new_batch::().set(k2, Sha256::fill(5u8)); let direct = db.new_batch().set(k2, Sha256::fill(6u8)); let db = db.apply_batch(winner).await.unwrap(); // A batch built directly on the database before the apply is stale too. assert!(matches!(direct.get(&k1, &db).await, Err(Error::StaleRead))); assert!(matches!( direct.get_many(&[&k1], &db).await, Err(Error::StaleRead) )); assert!(matches!(child.get(&k1, &db).await, Err(Error::StaleRead))); assert!(matches!( child.get_many(&[&k1], &db).await, Err(Error::StaleRead) )); assert!(matches!(loser.get(&k1, &db).await, Err(Error::StaleRead))); // The compatibility check precedes every other read argument check, so a read of no keys // returns `StaleRead` too. let no_keys: [&Digest; 0] = []; assert!(matches!( child.get_many(&no_keys, &db).await, Err(Error::StaleRead) )); assert!(matches!( loser.get_many(&no_keys, &db).await, Err(Error::StaleRead) )); assert!(matches!( child.merkleize(&db, None, floor).await, Err(Error::StaleBatch) )); assert!(matches!(db.validate_batch(&loser), Err(Error::StaleBatch))); db.destroy().await.unwrap(); } /// Applying a batch's child moves the database to a state outside the parent's chain, so /// reads through the parent return `StaleRead` while the child keeps reading. #[boxed] pub(crate) async fn run_descendant_apply_makes_parent_reads_stale( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let (k1, k2) = (Sha256::fill(1u8), Sha256::fill(2u8)); let floor = db.inactivity_floor_loc(); let parent = db .new_batch() .set(k1, Sha256::fill(3u8)) .merkleize(&db, None, floor) .await .unwrap(); assert_eq!(parent.get(&k1, &db).await.unwrap(), Some(Sha256::fill(3u8))); let child = parent .new_batch::() .set(k2, Sha256::fill(4u8)) .merkleize(&db, None, floor) .await .unwrap(); let db = db.apply_batch(child.clone()).await.unwrap(); assert_eq!(child.get(&k1, &db).await.unwrap(), Some(Sha256::fill(3u8))); assert!(matches!(parent.get(&k1, &db).await, Err(Error::StaleRead))); assert!(matches!( parent.get_many(&[&k1], &db).await, Err(Error::StaleRead) )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_empty( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first")).await; let bounds = db.bounds(); assert_eq!(bounds.end, 1); assert_eq!(bounds.start, Location::new(0)); assert_eq!(db.inactivity_floor_loc(), Location::new(0)); assert!(db.get_metadata().await.unwrap().is_none()); // Make sure closing/reopening gets us back to the same state, even after adding an uncommitted op. let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(2u8); let root = db.root(); { let _batch = db.new_batch().set(k1, v1); // Don't merkleize/apply. This simulates a failed commit. } drop(db); let db = open_db(context.child("second")).await; assert_eq!(db.root(), root); assert_eq!(db.bounds().end, 1); // Test calling commit on an empty db which should make it (durably) non-empty. let merkleized = db .new_batch() .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.bounds().end, 2); // commit op added let root = db.root(); drop(db); let db = open_db(context.child("third")).await; assert_eq!(db.root(), root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_commit_after_sync_recovery( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first")).await; let k1 = Sha256::fill(1u8); let k2 = Sha256::fill(2u8); let v1 = Sha256::fill(3u8); let v2 = Sha256::fill(4u8); // Commit and sync the first key so recovery has an older persisted boundary. let db = commit_sets(db, [(k1, v1)], None).await; let db = db.sync().await.unwrap(); // Commit a second key without syncing; reopen must replay it from journal data. let db = commit_sets(db, [(k2, v2)], None).await; let committed_bounds = db.bounds(); let committed_root = db.root(); drop(db); let db = open_db(context.child("second")).await; assert_eq!(db.bounds(), committed_bounds); assert_eq!(db.root(), committed_root); assert_eq!(db.get(&k1).await.unwrap(), Some(v1)); assert_eq!(db.get(&k2).await.unwrap(), Some(v2)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_build_basic( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { // Build a db with 2 keys. let db = open_db(context.child("first")).await; let k1 = Sha256::fill(1u8); let k2 = Sha256::fill(2u8); let v1 = Sha256::fill(3u8); let v2 = Sha256::fill(4u8); assert!(db.get(&k1).await.unwrap().is_none()); assert!(db.get(&k2).await.unwrap().is_none()); // Set and commit the first key. let metadata = Some(Sha256::fill(99u8)); let merkleized = db .new_batch() .set(k1, v1) .merkleize(&db, metadata, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1); assert!(db.get(&k2).await.unwrap().is_none()); assert_eq!(db.bounds().end, 3); assert_eq!(db.get_metadata().await.unwrap(), Some(Sha256::fill(99u8))); // Set and commit the second key. let merkleized = db .new_batch() .set(k2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1); assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2); assert_eq!(db.bounds().end, 5); assert_eq!(db.get_metadata().await.unwrap(), None); // Capture state. let root = db.root(); // Add an uncommitted op then simulate failure. let k3 = Sha256::fill(5u8); let v3 = Sha256::fill(6u8); { let _batch = db.new_batch().set(k3, v3); // Don't merkleize/apply. This simulates a failed commit. } // Reopen, make sure state is restored to last commit point. drop(db); // Simulate failed commit let db = open_db(context.child("second")).await; assert!(db.get(&k3).await.unwrap().is_none()); assert_eq!(db.root(), root); assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1); assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2); assert_eq!(db.bounds().end, 5); assert_eq!(db.get_metadata().await.unwrap(), None); // Cleanup. db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_proof_verify( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first")).await; let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(10u8); let merkleized = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap(); let root = db.root(); assert!(verify_proof::( &proof, Location::new(0), &ops, &root )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_prune( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let mut db = open_db(context.child("first")).await; for i in 0..20u8 { let key = Sha256::fill(i); let value = Sha256::fill(i.wrapping_add(100)); let floor = db.bounds().end; let merkleized = db .new_batch() .set(key, value) .merkleize(&db, None, floor) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); db = db.commit().await.unwrap(); } let root_before = db.root(); let bounds_before = db.bounds(); let prune_loc = bounds_before.end - 5; let db = db.prune(prune_loc).await.unwrap(); assert_eq!(db.root(), root_before); let key_0 = Sha256::fill(0u8); assert!(db.get(&key_0).await.unwrap().is_none()); let key_19 = Sha256::fill(19u8); assert_eq!( db.get(&key_19).await.unwrap(), Some(Sha256::fill(19u8.wrapping_add(100))) ); db.destroy().await.unwrap(); } /// Pruning immediately after an uncommitted batch must leave the database recoverable. Since /// prune is not a durability boundary, recovery may return either the durable baseline or the /// buffered state, but never a mixture whose floor references pruned operations. #[boxed] pub(crate) async fn run_prune_after_uncommitted_apply_batch_recovery( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first")).await; // Fill more than one journal blob and establish a durable baseline whose floor still // requires the oldest blob. let mut batch = db.new_batch(); for i in 0..6u8 { batch = batch.set(Sha256::fill(i), Sha256::fill(i.wrapping_add(10))); } let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.sync().await.unwrap(); let durable_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end); // Apply, but do not commit, a batch that advances the floor far enough for prune to // remove the oldest blob. let buffered_floor = db.bounds().end; let key = Sha256::fill(100); let value = Sha256::fill(101); let merkleized = db .new_batch() .set(key, value) .merkleize(&db, None, buffered_floor) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let buffered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end); assert_ne!(buffered_state, durable_state); let db = db.prune(buffered_floor).await.unwrap(); assert!(db.bounds().start > Location::new(0)); drop(db); // Reopen must produce one coherent state. In particular, it must not recover the old // floor after the prune has removed operations that floor still needs. let db = open_db(context.child("second")).await; assert!(db.bounds().start <= db.inactivity_floor_loc()); let recovered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end); assert!( recovered_state == durable_state || recovered_state == buffered_state, "recovered state is neither the durable baseline nor the buffered state" ); db.destroy().await.unwrap(); } /// `operations()` must cover exactly the batch's own applied range and match the /// operations a post-apply `historical_proof` recovers from the log, for a db-based /// batch and for a chained batch applied after its ancestor. #[boxed] pub(crate) async fn run_operations_match_applied_log( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared + PartialEq + core::fmt::Debug, { let db = open_db(context.child("db")).await; let seed = db .new_batch() .set(Sha256::fill(1u8), Sha256::fill(11u8)) .set(Sha256::fill(2u8), Sha256::fill(12u8)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let (seed_start, seed_ops) = seed.operations(); let seed_root = seed.root(); let seed_proof = seed.proof(&db).unwrap(); let seed_pins = seed.pinned_nodes(&db).unwrap(); let applied_start = db.bounds().end; let db = db.apply_batch(seed).await.unwrap(); let seed_range = applied_start..db.bounds().end; assert_eq!(seed_start, seed_range.start); assert_eq!(*seed_start + seed_ops.len() as u64, *seed_range.end); // A chained batch's operations are its own suffix only. let parent = db .new_batch() .set(Sha256::fill(3u8), Sha256::fill(13u8)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::fill(4u8), Sha256::fill(14u8)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let (parent_start, parent_ops) = parent.operations(); let (child_start, child_ops) = child.operations(); let (parent_root, child_root) = (parent.root(), child.root()); let (parent_pins, child_pins) = ( parent.pinned_nodes(&db).unwrap(), child.pinned_nodes(&db).unwrap(), ); let (parent_proof, child_proof) = (parent.proof(&db).unwrap(), child.proof(&db).unwrap()); let applied_start = db.bounds().end; let db = db.apply_batch(parent).await.unwrap(); let parent_range = applied_start..db.bounds().end; // With the database at the parent's tip, the child's proof and pinned nodes are // unchanged. assert_eq!(child.proof(&db).unwrap(), child_proof); assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins); let applied_start = db.bounds().end; let db = db.apply_batch(child).await.unwrap(); let child_range = applied_start..db.bounds().end; assert_eq!(parent_start, parent_range.start); assert_eq!(*parent_start + parent_ops.len() as u64, *parent_range.end); assert_eq!(child_start, child_range.start); assert_eq!(*child_start + child_ops.len() as u64, *child_range.end); // A write-free batch still captures its commit-only suffix. let empty = db .new_batch() .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let (empty_start, empty_ops) = empty.operations(); let (empty_root, empty_proof) = (empty.root(), empty.proof(&db).unwrap()); let empty_pins = empty.pinned_nodes(&db).unwrap(); let applied_start = db.bounds().end; let db = db.apply_batch(empty).await.unwrap(); let empty_range = applied_start..db.bounds().end; assert_eq!(empty_start, empty_range.start); assert_eq!(*empty_start + empty_ops.len() as u64, *empty_range.end); // Every captured delta and proof must match what the log recovers for its // range, and verify against the batch's own root with and without the pins. for (start, ops, proof, pins, root) in [ (seed_start, seed_ops, seed_proof, seed_pins, seed_root), ( parent_start, parent_ops, parent_proof, parent_pins, parent_root, ), (child_start, child_ops, child_proof, child_pins, child_root), (empty_start, empty_ops, empty_proof, empty_pins, empty_root), ] { let len = core::num::NonZeroU64::new(ops.len() as u64).unwrap(); let end = Location::new(*start + ops.len() as u64); let (log_proof, log_ops) = db.historical_proof(end, start, len).await.unwrap(); assert_eq!(log_ops, *ops); assert_eq!(log_proof, proof); assert!(verify_proof::(&proof, start, &ops, &root)); assert!(verify_proof_and_pinned_nodes::( &proof, start, &ops, &pins, &root )); } // Flushing the applied batch prunes the store to its peaks. The late batch's base is // mid-mountain, so its artifacts return `ElementPruned` rather than unverifiable data. let late = db .new_batch() .set(Sha256::fill(5u8), Sha256::fill(15u8)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(Arc::clone(&late)).await.unwrap(); let db = db.commit().await.unwrap(); assert!(matches!( late.proof(&db), Err(crate::qmdb::Error::Merkle( crate::merkle::Error::ElementPruned(_) )) )); assert!(matches!( late.pinned_nodes(&db), Err(crate::qmdb::Error::Merkle( crate::merkle::Error::ElementPruned(_) )) )); // A batch built on the flushed store reads every node below it from the pinned peaks. let flushed = db .new_batch() .set(Sha256::fill(6u8), Sha256::fill(16u8)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let (flushed_start, flushed_ops) = flushed.operations(); let flushed_root = flushed.root(); let flushed_proof = flushed.proof(&db).unwrap(); let flushed_pins = flushed.pinned_nodes(&db).unwrap(); assert!(verify_proof_and_pinned_nodes::( &flushed_proof, flushed_start, &flushed_ops, &flushed_pins, &flushed_root )); let applied_start = db.bounds().end; let db = db.apply_batch(flushed).await.unwrap(); let flushed_range = applied_start..db.bounds().end; assert_eq!(flushed_start, flushed_range.start); assert_eq!( *flushed_start + flushed_ops.len() as u64, *flushed_range.end ); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_chain( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first")).await; let k1 = Sha256::fill(1u8); let k2 = Sha256::fill(2u8); let k3 = Sha256::fill(3u8); let v1 = Sha256::fill(11u8); let v2 = Sha256::fill(12u8); let v3 = Sha256::fill(13u8); let parent = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent .new_batch::() .set(k2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert_eq!(child.get(&k1, &db).await.unwrap(), Some(v1)); assert_eq!(child.get(&k2, &db).await.unwrap(), Some(v2)); assert!(child.get(&k3, &db).await.unwrap().is_none()); let db = db.apply_batch(child).await.unwrap().commit().await.unwrap(); assert_eq!(db.get(&k1).await.unwrap(), Some(v1)); assert_eq!(db.get(&k2).await.unwrap(), Some(v2)); let merkleized = db .new_batch() .set(k3, v3) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.get(&k3).await.unwrap(), Some(v3)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_build_and_authenticate( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { // Build a db with `ELEMENTS` key/value pairs and prove ranges over them. let db = open_db(context.child("first")).await; let mut batch = db.new_batch(); for i in 0u64..2_000 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = Sha256::fill(i as u8); batch = batch.set(k, v); } let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.bounds().end, 2_000 + 2); // Drop & reopen the db, making sure it has exactly the same state. let root = db.root(); drop(db); let db = open_db(context.child("second")).await; assert_eq!(root, db.root()); assert_eq!(db.bounds().end, 2_000 + 2); for i in 0u64..2_000 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = Sha256::fill(i as u8); assert_eq!(db.get(&k).await.unwrap().unwrap(), v); } // Make sure all ranges of 5 operations are provable, including truncated ranges at the // end. let max_ops = NZU64!(5); for i in 0..*db.bounds().end { let (proof, log) = db.proof(Location::new(i), max_ops).await.unwrap(); assert!(verify_proof::( &proof, Location::new(i), &log, &root )); } db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_recovery_from_failed_merkle_sync( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { // Insert 1000 keys then sync. const ELEMENTS: u64 = 1000; let db = open_db(context.child("first")).await; let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = Sha256::fill(i as u8); batch = batch.set(k, v); } let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.bounds().end, ELEMENTS + 2); let db = db.sync().await.unwrap(); let halfway_root = db.root(); // Insert another 1000 keys (different from the first batch) then commit. let mut batch = db.new_batch(); for i in ELEMENTS..ELEMENTS * 2 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = Sha256::fill(i as u8); batch = batch.set(k, v); } let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap(); // Drop before syncing // Recovery should replay the log to regenerate the merkle structure. // op_count = 1002 (first batch + commit) + 1000 (second batch) + 1 (second commit) = 2003 let db = open_db(context.child("second")).await; assert_eq!(db.bounds().end, 2003); let root = db.root(); assert_ne!(root, halfway_root); // Drop & reopen could preserve the final commit. drop(db); let db = open_db(context.child("third")).await; assert_eq!(db.bounds().end, 2003); assert_eq!(db.root(), root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_recovery_from_failed_log_sync( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first")).await; // Insert a single key and then commit to create a first commit point. let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(3u8); let merkleized = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let first_commit_root = db.root(); // Simulate failure. Sets that are never merkleized/applied are lost. // Recovery should restore the last commit point. drop(db); // Recovery should back up to previous commit point. let db = open_db(context.child("second")).await; assert_eq!(db.bounds().end, 3); let root = db.root(); assert_eq!(root, first_commit_root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_pruning( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { // Build a db with `ELEMENTS` key/value pairs then prune some of them. const ELEMENTS: u64 = 2_000; let db = open_db(context.child("first")).await; // Batch writes keys in BTreeMap-sorted order, so build the sorted key // list to map between journal locations and keys. let mut sorted_keys: Vec = (1u64..ELEMENTS + 1) .map(|i| Sha256::hash(&[&i.to_be_bytes()])) .collect(); sorted_keys.sort(); // Location 0: initial commit; locations 1..=ELEMENTS: Set ops in sorted // key order; location ELEMENTS+1: batch commit. // key_at_loc(L) = sorted_keys[L - 1] for 1 <= L <= ELEMENTS. let mut batch = db.new_batch(); for i in 1u64..ELEMENTS + 1 { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = Sha256::fill(i as u8); batch = batch.set(k, v); } // The inactivity floor must cover both prune targets in this test. // Second prune request is at ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1. let inactivity_floor = Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1); let merkleized = batch.merkleize(&db, None, inactivity_floor).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.bounds().end, ELEMENTS + 2); // Prune the db to the first half of the operations. let db = db.prune(Location::new((ELEMENTS + 2) / 2)).await.unwrap(); let bounds = db.bounds(); assert_eq!(bounds.end, ELEMENTS + 2); // items_per_section is 5, so half should be exactly at a blob boundary, in which case // the actual pruning location should match the requested. let oldest_retained_loc = bounds.start; assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2)); // Try to fetch a pruned key (at location oldest_retained - 1). let pruned_key = sorted_keys[*oldest_retained_loc as usize - 2]; assert!(db.get(&pruned_key).await.unwrap().is_none()); // Try to fetch unpruned key (at location oldest_retained). let unpruned_key = sorted_keys[*oldest_retained_loc as usize - 1]; assert!(db.get(&unpruned_key).await.unwrap().is_some()); // Drop & reopen the db, making sure it has exactly the same state. let root = db.root(); db.sync().await.unwrap(); let db = open_db(context.child("second")).await; assert_eq!(root, db.root()); let bounds = db.bounds(); assert_eq!(bounds.end, ELEMENTS + 2); let oldest_retained_loc = bounds.start; assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2)); // Prune to a non-blob boundary. let loc = Location::new(ELEMENTS / 2 + (ITEMS_PER_SECTION * 2 - 1)); let db = db.prune(loc).await.unwrap(); // Actual boundary should be a multiple of 5. let oldest_retained_loc = db.bounds().start; assert_eq!( oldest_retained_loc, Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION) ); // Confirm boundary persists across restart. db.sync().await.unwrap(); let db = open_db(context.child("third")).await; let oldest_retained_loc = db.bounds().start; assert_eq!( oldest_retained_loc, Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION) ); // Try to fetch a key before the inactivity floor (not in snapshot after reopen). let floor_val = ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1; let inactive_key = sorted_keys[floor_val as usize - 2]; assert!(db.get(&inactive_key).await.unwrap().is_none()); // Try to fetch a key at the inactivity floor (in snapshot after reopen). let active_key = sorted_keys[floor_val as usize - 1]; assert!(db.get(&active_key).await.unwrap().is_some()); // Confirm behavior of trying to create a proof of pruned items is as expected. let pruned_pos = ELEMENTS / 2; let proof_result = db .proof(Location::new(pruned_pos), NZU64!(pruned_pos + 100)) .await; assert!( matches!(proof_result, Err(Error::Journal(crate::journal::Error::ItemPruned(pos))) if pos == pruned_pos) ); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_prune_beyond_floor( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // Test pruning empty database (floor=0, so prune(1) fails) let result = db.prune(Location::new(1)).await; assert!( matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor)) if prune_loc == Location::new(1) && floor == Location::new(0)) ); let db = open_db(context.child("test")).await; // Add key-value pairs and commit let k1 = Digest::from(*b"12345678901234567890123456789012"); let k2 = Digest::from(*b"abcdefghijklmnopqrstuvwxyz123456"); let k3 = Digest::from(*b"99999999999999999999999999999999"); let v1 = Sha256::fill(1u8); let v2 = Sha256::fill(2u8); let v3 = Sha256::fill(3u8); // First batch with floor=3 (the commit location). let merkleized = db .new_batch() .set(k1, v1) .set(k2, v2) .merkleize(&db, None, Location::new(3)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // op_count is 4 (initial_commit, k1, k2, commit), last_commit is at location 3 assert_eq!(*db.size() - 1, 3); // Second batch with floor=5 (the new commit location). let merkleized = db .new_batch() .set(k3, v3) .merkleize(&db, None, Location::new(5)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // Test valid prune (3 <= floor of 5) let db = db.prune(Location::new(3)).await.unwrap(); // Test pruning beyond inactivity floor let floor = db.inactivity_floor_loc(); let beyond = floor + 1; let result = db.prune(beyond).await; assert!( matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, f)) if prune_loc == beyond && f == floor) ); } async fn commit_sets( db: TestDb, sets: impl IntoIterator, metadata: Option, ) -> TestDb where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { commit_sets_with_floor(db, sets, metadata, Location::new(0)).await } async fn commit_sets_with_floor( db: TestDb, sets: impl IntoIterator, metadata: Option, floor: Location, ) -> TestDb where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let mut batch = db.new_batch(); for (key, value) in sets { batch = batch.set(key, value); } let merkleized = batch.merkleize(&db, metadata, floor).await.unwrap(); let expected_end = merkleized.bounds().tip.size; let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.bounds().end, expected_end); db.commit().await.unwrap() } #[boxed] pub(crate) async fn run_bounded_initialization_recovery( context: deterministic::Context, open_db: impl Fn( deterministic::Context, Option>, ) -> Pin, Error>> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db"), None).await.unwrap(); let key1 = Sha256::hash(&[&1u64.to_be_bytes()]); let key2 = Sha256::hash(&[&2u64.to_be_bytes()]); let key3 = Sha256::hash(&[&3u64.to_be_bytes()]); let key4 = Sha256::hash(&[&4u64.to_be_bytes()]); let value1 = Sha256::fill(11u8); let value2 = Sha256::fill(22u8); let value3 = Sha256::fill(33u8); let value4 = Sha256::fill(66u8); let metadata_a = Sha256::fill(44u8); let db = commit_sets(db, [(key1, value1), (key2, value2)], Some(metadata_a)).await; let size_before = db.bounds().end; let root_before = db.root(); let last_commit_before = db.size() - 1; let metadata_b = Sha256::fill(55u8); let db = commit_sets(db, [(key3, value3), (key4, value4)], Some(metadata_b)).await; assert_ne!(db.root(), root_before); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_b)); assert_eq!(db.get(&key3).await.unwrap(), Some(value3)); assert_eq!(db.get(&key4).await.unwrap(), Some(value4)); _ = db.sync().await.unwrap(); let db = open_db(context.child("cap"), Some(size_before)) .await .unwrap(); assert_eq!(db.root(), root_before); assert_eq!(db.bounds().end, size_before); assert_eq!(db.size() - 1, last_commit_before); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a)); assert_eq!(db.get(&key1).await.unwrap(), Some(value1)); assert_eq!(db.get(&key2).await.unwrap(), Some(value2)); assert_eq!(db.get(&key3).await.unwrap(), None); assert_eq!(db.get(&key4).await.unwrap(), None); db.commit().await.unwrap(); let db = open_db(context.child("reopen"), None).await.unwrap(); assert_eq!(db.root(), root_before); assert_eq!(db.bounds().end, size_before); assert_eq!(db.size() - 1, last_commit_before); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a)); assert_eq!(db.get(&key1).await.unwrap(), Some(value1)); assert_eq!(db.get(&key2).await.unwrap(), Some(value2)); assert_eq!(db.get(&key3).await.unwrap(), None); assert_eq!(db.get(&key4).await.unwrap(), None); db.destroy().await.unwrap(); } /// A key retained by the initialization bound must survive even when it translator-collides /// with a key in the discarded suffix. #[boxed] pub(crate) async fn run_bounded_initialization_preserves_collision_bucket( context: deterministic::Context, open_db: impl Fn( deterministic::Context, Option>, ) -> Pin, Error>> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db"), None).await.unwrap(); // Two keys sharing the first two bytes collide under TwoCap. let mut k1_bytes = [0u8; 32]; let mut k2_bytes = [0u8; 32]; k1_bytes[0] = 0xAA; k1_bytes[1] = 0xBB; k2_bytes[0] = 0xAA; k2_bytes[1] = 0xBB; k1_bytes[31] = 0x01; k2_bytes[31] = 0x02; let key1 = Digest::from(k1_bytes); let key2 = Digest::from(k2_bytes); let value1 = Sha256::fill(11u8); let value2 = Sha256::fill(22u8); let db = commit_sets(db, [(key1, value1)], None).await; let size_after_first = db.bounds().end; let db = commit_sets(db, [(key2, value2)], None).await; assert_eq!(db.get(&key1).await.unwrap(), Some(value1)); assert_eq!(db.get(&key2).await.unwrap(), Some(value2)); _ = db.sync().await.unwrap(); let db = open_db(context.child("cap"), Some(size_after_first)) .await .unwrap(); // The retained key must still be readable; pre-fix this returned None because the // translator bucket was wiped by the suffix-key remove. assert_eq!(db.get(&key1).await.unwrap(), Some(value1)); assert_eq!(db.get(&key2).await.unwrap(), None); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_bounded_initialization_pruned_target_errors( context: deterministic::Context, open_small_sections_db: impl Fn( deterministic::Context, Option>, ) -> Pin< Box, Error>> + Send>, >, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_small_sections_db(context.child("db"), None) .await .unwrap(); let first_start = db.bounds().end; let mut db = commit_sets( db, (0u64..16).map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8))), None, ) .await; let mut round = 0u64; loop { round += 1; assert!( round <= 64, "failed to prune enough history for the bounded initialization test" ); // Floor must be >= last_commit_loc for prune to succeed. // With 16 sets, commit is at current end + 16. let floor = db.bounds().end + 16; db = commit_sets_with_floor( db, (0u64..16).map(|i| { let seed = round * 100 + i; ( Sha256::hash(&[&seed.to_be_bytes()]), Sha256::fill(seed as u8), ) }), None, floor, ) .await; let last_commit = db.size() - 1; db = db.prune(last_commit).await.unwrap(); if db.bounds().start > first_start { break; } } let oldest_retained = db.bounds().start; _ = db.sync().await.unwrap(); let Err(boundary_err) = open_small_sections_db(context.child("cap_error"), Some(oldest_retained)).await else { panic!("expected bounded initialization to fail"); }; assert!( matches!( boundary_err, Error::Journal(crate::journal::Error::ItemPruned(_)) ), "unexpected bounded initialization error at retained boundary: {boundary_err:?}" ); let db = open_small_sections_db(context.child("db"), None) .await .unwrap(); _ = db.sync().await.unwrap(); let Err(err) = open_small_sections_db(context.child("cap_error"), Some(first_start)).await else { panic!("expected bounded initialization to fail"); }; assert!( matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))), "unexpected bounded initialization error: {err:?}" ); } /// batch.get() reads pending mutations, then the database. #[boxed] pub(crate) async fn run_batch_get_read_through( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Pre-populate with key A. let key_a = Sha256::hash(&[&0u64.to_be_bytes()]); let val_a = Sha256::fill(1u8); let merkleized = db .new_batch() .set(key_a, val_a) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // batch.get(&A) should return DB value. let mut batch = db.new_batch(); assert_eq!(batch.get(&key_a, &db).await.unwrap(), Some(val_a)); // Set B in batch, batch.get(&B) returns the value. let key_b = Sha256::hash(&[&1u64.to_be_bytes()]); let val_b = Sha256::fill(2u8); batch = batch.set(key_b, val_b); assert_eq!(batch.get(&key_b, &db).await.unwrap(), Some(val_b)); // Nonexistent key. let key_c = Sha256::hash(&[&2u64.to_be_bytes()]); assert_eq!(batch.get(&key_c, &db).await.unwrap(), None); db.destroy().await.unwrap(); } /// Child batch reads parent diff and adds its own mutations. #[boxed] pub(crate) async fn run_batch_stacked_get( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Parent batch: set A. let key_a = Sha256::hash(&[&0u64.to_be_bytes()]); let val_a = Sha256::fill(10u8); let parent = db.new_batch().set(key_a, val_a); let parent_m = parent.merkleize(&db, None, Location::new(0)).await.unwrap(); // Child reads parent's A. let mut child = parent_m.new_batch::(); assert_eq!(child.get(&key_a, &db).await.unwrap(), Some(val_a)); // Child sets B. let key_b = Sha256::hash(&[&1u64.to_be_bytes()]); let val_b = Sha256::fill(20u8); child = child.set(key_b, val_b); assert_eq!(child.get(&key_b, &db).await.unwrap(), Some(val_b)); // Nonexistent key. let key_c = Sha256::hash(&[&2u64.to_be_bytes()]); assert_eq!(child.get(&key_c, &db).await.unwrap(), None); db.destroy().await.unwrap(); } /// Two-level stacked batch apply works end-to-end. #[boxed] pub(crate) async fn run_batch_stacked_apply( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Sort keys so operations are in BTreeMap order (same as merkleize writes). let mut kvs_first: Vec<(Digest, Digest)> = (0u64..5) .map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8))) .collect(); kvs_first.sort_by_key(|a| a.0); let mut kvs_second: Vec<(Digest, Digest)> = (5u64..10) .map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8))) .collect(); kvs_second.sort_by_key(|a| a.0); // Parent batch: set keys 0..5. let mut parent = db.new_batch(); for (k, v) in &kvs_first { parent = parent.set(*k, *v); } let parent_m = parent.merkleize(&db, None, Location::new(0)).await.unwrap(); // Child batch: set keys 5..10. let mut child = parent_m.new_batch::(); for (k, v) in &kvs_second { child = child.set(*k, *v); } let child_m = child.merkleize(&db, None, Location::new(0)).await.unwrap(); let expected_root = child_m.root(); let db = db.apply_batch(child_m).await.unwrap(); assert_eq!(db.root(), expected_root); // All 10 keys should be accessible. for (k, v) in kvs_first.iter().chain(kvs_second.iter()) { assert_eq!(db.get(k).await.unwrap(), Some(*v)); } db.destroy().await.unwrap(); } /// MerkleizedBatch::root() matches db.root() after apply_batch(). #[boxed] pub(crate) async fn run_batch_speculative_root( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let mut batch = db.new_batch(); for i in 0u8..10 { let k = Sha256::hash(&[&[i]]); batch = batch.set(k, Sha256::fill(i)); } let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); let speculative = merkleized.root(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.root(), speculative); // Second batch with metadata. let metadata = Some(Sha256::fill(55u8)); let mut batch = db.new_batch(); let k = Sha256::hash(&[&[0xAA]]); batch = batch.set(k, Sha256::fill(0xAA)); let merkleized = batch .merkleize(&db, metadata, Location::new(0)) .await .unwrap(); let speculative = merkleized.root(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.root(), speculative); db.destroy().await.unwrap(); } /// MerkleizedBatch::get() reads from diff and base DB. #[boxed] pub(crate) async fn run_merkleized_batch_get( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Pre-populate base DB. let key_a = Sha256::hash(&[&0u64.to_be_bytes()]); let val_a = Sha256::fill(10u8); let merkleized = db .new_batch() .set(key_a, val_a) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // Create a merkleized batch with a new key. let key_b = Sha256::hash(&[&1u64.to_be_bytes()]); let val_b = Sha256::fill(20u8); let merkleized = db .new_batch() .set(key_b, val_b) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Read base DB value through merkleized batch. assert_eq!(merkleized.get(&key_a, &db).await.unwrap(), Some(val_a)); // Read this batch's key from the diff. assert_eq!(merkleized.get(&key_b, &db).await.unwrap(), Some(val_b)); // Nonexistent key. let key_c = Sha256::hash(&[&2u64.to_be_bytes()]); assert_eq!(merkleized.get(&key_c, &db).await.unwrap(), None); db.destroy().await.unwrap(); } /// Independent sequential batches applied one at a time. #[boxed] pub(crate) async fn run_batch_sequential_apply( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key_a = Sha256::hash(&[&0u64.to_be_bytes()]); let val_a = Sha256::fill(1u8); // First batch. let m = db .new_batch() .set(key_a, val_a) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let root1 = m.root(); let db = db.apply_batch(m).await.unwrap(); assert_eq!(db.root(), root1); assert_eq!(db.get(&key_a).await.unwrap(), Some(val_a)); // Second independent batch. let key_b = Sha256::hash(&[&1u64.to_be_bytes()]); let val_b = Sha256::fill(2u8); let m = db .new_batch() .set(key_b, val_b) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let root2 = m.root(); let db = db.apply_batch(m).await.unwrap(); assert_eq!(db.root(), root2); assert_eq!(db.get(&key_b).await.unwrap(), Some(val_b)); db.destroy().await.unwrap(); } /// Many sequential batches accumulate correctly. #[boxed] pub(crate) async fn run_batch_many_sequential( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let mut db = open_db(context.child("db")).await; const BATCHES: u64 = 20; const KEYS_PER_BATCH: u64 = 5; let mut all_kvs: Vec<(Digest, Digest)> = Vec::new(); for batch_idx in 0..BATCHES { let mut batch = db.new_batch(); for j in 0..KEYS_PER_BATCH { let seed = batch_idx * 100 + j; let k = Sha256::hash(&[&seed.to_be_bytes()]); let v = Sha256::fill(seed as u8); batch = batch.set(k, v); all_kvs.push((k, v)); } let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } // Verify all key-values are readable. for (k, v) in &all_kvs { assert_eq!(db.get(k).await.unwrap(), Some(*v)); } // Verify proof over the full range. let root = db.root(); let (proof, ops) = db.proof(Location::new(0), NZU64!(10000)).await.unwrap(); assert!(verify_proof::( &proof, Location::new(0), &ops, &root )); // Expected: 1 initial commit + BATCHES * (KEYS_PER_BATCH + 1 commit). let expected = 1 + BATCHES * (KEYS_PER_BATCH + 1); assert_eq!(db.bounds().end, expected); db.destroy().await.unwrap(); } /// Empty batch (zero mutations) produces correct speculative root. #[boxed] pub(crate) async fn run_batch_empty_batch( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Apply a non-empty batch first. let k = Sha256::hash(&[&[1u8]]); let merkleized = db .new_batch() .set(k, Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let root_before = db.root(); let size_before = db.bounds().end; // Empty batch with no mutations. let merkleized = db .new_batch() .merkleize(&db, None, Location::new(0)) .await .unwrap(); let speculative = merkleized.root(); let db = db.apply_batch(merkleized).await.unwrap(); // Root changed (a new Commit op was appended). assert_ne!(db.root(), root_before); assert_eq!(db.root(), speculative); // Size grew by exactly 1 (the Commit op). assert_eq!(db.bounds().end, size_before + 1); db.destroy().await.unwrap(); } /// MerkleizedBatch::get() works on a chained child's merkleized batch. #[boxed] pub(crate) async fn run_batch_chained_merkleized_get( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Pre-populate base DB. let key_a = Sha256::hash(&[&0u64.to_be_bytes()]); let val_a = Sha256::fill(10u8); let merkleized = db .new_batch() .set(key_a, val_a) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // Parent batch sets key B. let key_b = Sha256::hash(&[&1u64.to_be_bytes()]); let val_b = Sha256::fill(1u8); let parent_m = db .new_batch() .set(key_b, val_b) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Child batch sets key C. let key_c = Sha256::hash(&[&2u64.to_be_bytes()]); let val_c = Sha256::fill(2u8); let child_m = parent_m .new_batch::() .set(key_c, val_c) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Child's MerkleizedBatch can read all three layers: // base DB value assert_eq!(child_m.get(&key_a, &db).await.unwrap(), Some(val_a)); // parent diff value assert_eq!(child_m.get(&key_b, &db).await.unwrap(), Some(val_b)); // child's own value assert_eq!(child_m.get(&key_c, &db).await.unwrap(), Some(val_c)); // nonexistent key let key_d = Sha256::hash(&[&3u64.to_be_bytes()]); assert_eq!(child_m.get(&key_d, &db).await.unwrap(), None); db.destroy().await.unwrap(); } /// Large single batch, verifying all values and proof. #[boxed] pub(crate) async fn run_batch_large( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; const N: u64 = 500; let mut kvs: Vec<(Digest, Digest)> = Vec::new(); let mut batch = db.new_batch(); for i in 0..N { let k = Sha256::hash(&[&i.to_be_bytes()]); let v = Sha256::fill((i % 256) as u8); batch = batch.set(k, v); kvs.push((k, v)); } let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // Verify every value. for (k, v) in &kvs { assert_eq!(db.get(k).await.unwrap(), Some(*v)); } // Verify proof over the full range. let root = db.root(); let (proof, ops) = db.proof(Location::new(0), NZU64!(1000)).await.unwrap(); assert!(verify_proof::( &proof, Location::new(0), &ops, &root )); // Expected: 1 initial commit + N sets + 1 commit. assert_eq!(db.bounds().end, 1 + N + 1); db.destroy().await.unwrap(); } /// Two keys that collide under the translator are set in sequential applied batches. Pruning /// the first write retires only its location, and `get()` still serves the second key. /// /// `open_db_small_sections` must return a DB whose log has `items_per_section=1` /// so pruning is per-item. #[boxed] pub(crate) async fn run_prune_collision_bucket( context: deterministic::Context, open_db_small_sections: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db_small_sections(context.child("db")).await; // Two keys sharing the first two bytes collide under TwoCap. let mut k1_bytes = [0u8; 32]; let mut k2_bytes = [0u8; 32]; k1_bytes[0] = 0xAA; k1_bytes[1] = 0xBB; k2_bytes[0] = 0xAA; k2_bytes[1] = 0xBB; k1_bytes[31] = 0x01; k2_bytes[31] = 0x02; let key1 = Digest::from(k1_bytes); let key2 = Digest::from(k2_bytes); let key3 = Sha256::fill(3u8); let v1 = Sha256::fill(1u8); let v2 = Sha256::fill(2u8); let v3 = Sha256::fill(3u8); // The log holds 0=initial commit, 1=Set(key1,v1), 2=Commit, 3-4=Set(key2,v2) and // Set(key3,v3), 5=Commit(floor=4). Floor=4 permits prune(2). let merkleized = db .new_batch() .set(key1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let merkleized = db .new_batch() .set(key2, v2) .set(key3, v3) .merkleize(&db, None, Location::new(4)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), Some(v2)); assert_eq!(db.snapshot.len(), 3); // Prune past the first Set (loc 1). With items_per_section=1, pruning to loc 2 removes // the blob containing loc 1 and retires only that location from the snapshot. let db = db.prune(Location::new(2)).await.unwrap(); assert_eq!(db.bounds().start, Location::new(2)); assert_eq!(db.snapshot.len(), 2); assert_eq!(db.get(&key1).await.unwrap(), None); assert_eq!(db.get(&key2).await.unwrap(), Some(v2)); assert_eq!(db.get(&key3).await.unwrap(), Some(v3)); db.destroy().await.unwrap(); } /// Every prune retires exactly the sets below the retained start, and a set at the retained /// start survives and stays readable. /// /// `open_db_small_sections` must return a DB whose log has `items_per_section=1` /// so pruning is per-item. #[boxed] pub(crate) async fn run_prune_retires_pruned_sets( context: deterministic::Context, open_db_small_sections: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let mut db = open_db_small_sections(context.child("db")).await; // The log holds 0=initial commit, then Set(key i) at 2i+1 and Commit at 2i+2. for i in 0..10u8 { let floor = db.bounds().end; let merkleized = db .new_batch() .set(Sha256::fill(i), Sha256::fill(i)) .merkleize(&db, None, floor) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } assert_eq!(db.bounds(), Location::new(0)..Location::new(21)); assert_eq!(db.snapshot.len(), 10); // Pruning to a commit retires the sets before it: one per two pruned operations. for loc in [2u64, 4, 6] { db = db.prune(Location::new(loc)).await.unwrap(); assert_eq!(db.bounds().start, Location::new(loc)); assert_eq!(db.snapshot.len(), 10 - loc as usize / 2); } assert!(db.get(&Sha256::fill(0)).await.unwrap().is_none()); // Pruning to the set at 9 retires the set at 7 and keeps the one at the retained start. db = db.prune(Location::new(9)).await.unwrap(); assert_eq!(db.snapshot.len(), 6); assert_eq!( db.get(&Sha256::fill(4)).await.unwrap(), Some(Sha256::fill(4)) ); // Pruning past it retires it. db = db.prune(Location::new(10)).await.unwrap(); assert_eq!(db.snapshot.len(), 5); assert!(db.get(&Sha256::fill(4)).await.unwrap().is_none()); assert_eq!( db.get(&Sha256::fill(5)).await.unwrap(), Some(Sha256::fill(5)) ); db.destroy().await.unwrap(); } /// Metadata propagates through merkleize and clears with None. #[boxed] pub(crate) async fn run_batch_metadata( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Batch with metadata. let metadata = Sha256::fill(42u8); let k = Sha256::hash(&[&[1u8]]); let merkleized = db .new_batch() .set(k, Sha256::fill(1u8)) .merkleize(&db, Some(metadata), Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); // Second batch clears metadata. let merkleized = db .new_batch() .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), None); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_merkleize_foreign_db( db: TestDb, foreign: TestDb, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let key = Sha256::fill(1); let batch = db .new_batch() .set(key, Sha256::fill(11)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap(); let batch = foreign .new_batch() .set(key, Sha256::fill(99)) .merkleize(&foreign, None, Location::new(0)) .await .unwrap(); let foreign = foreign.apply_batch(batch).await.unwrap(); assert_eq!(db.size(), foreign.size()); assert_ne!(db.root(), foreign.root()); let direct = db.new_batch().set(Sha256::fill(2), Sha256::fill(22)); let direct = direct.merkleize(&foreign, None, Location::new(0)).await; let parent = db .new_batch() .set(Sha256::fill(3), Sha256::fill(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::fill(4), Sha256::fill(44)); let child = child.merkleize(&foreign, None, Location::new(0)).await; assert!(matches!(direct, Err(Error::StaleBatch))); assert!(matches!(child, Err(Error::StaleBatch))); db.destroy().await.unwrap(); foreign.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_merkleize_stale_sibling( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let direct = db.new_batch().set(Sha256::fill(1), Sha256::fill(11)); let sibling = db .new_batch() .set(Sha256::fill(2), Sha256::fill(22)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling).await.unwrap(); let direct = direct.merkleize(&db, None, Location::new(0)).await; assert!(matches!(direct, Err(Error::StaleBatch))); let parent = db .new_batch() .set(Sha256::fill(3), Sha256::fill(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::fill(4), Sha256::fill(44)); let sibling = db .new_batch() .set(Sha256::fill(5), Sha256::fill(55)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling).await.unwrap(); let child = child.merkleize(&db, None, Location::new(0)).await; assert!(matches!(child, Err(Error::StaleBatch))); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_stale_batch_rejected( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let v1 = Sha256::fill(10u8); let v2 = Sha256::fill(20u8); // Create two batches from the same DB state. let batch_a = db .new_batch() .set(key1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let batch_b = db .new_batch() .set(key2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Applying and committing the first must succeed. let db = db.apply_batch(batch_a).await.unwrap(); assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), None); assert_eq!(db.get_metadata().await.unwrap(), None); let db = db.commit().await.unwrap(); let root = db.root(); let size = db.size(); // Applying the second must fail because the DB was modified. let result = db.apply_batch(batch_b).await; assert!(matches!(result, Err(Error::StaleBatch))); // The rejection mutated nothing: reopening recovers the committed state. let db = open_db(context.child("reopen")).await; assert_eq!(db.root(), root); assert_eq!(db.size(), size); assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), None); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_stale_batch_chained( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let key3 = Sha256::hash(&[&[3]]); let common_parent = db .new_batch() .set(Sha256::hash(&[&[10]]), Sha256::fill(10u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let sibling_a = common_parent .new_batch::() .set(Sha256::hash(&[&[11]]), Sha256::fill(11u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let sibling_b = common_parent .new_batch::() .set(Sha256::hash(&[&[12]]), Sha256::fill(12u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling_a).await.unwrap(); assert!(matches!( db.validate_batch(&sibling_b), Err(Error::StaleBatch) )); // Build equal-size sibling parents, then extend only one sibling. let parent_a = db .new_batch() .set(key1, Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let parent_b = db .new_batch() .set(key2, Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child_b = parent_b .new_batch::() .set(key3, Sha256::fill(3u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(parent_a).await.unwrap(); assert!(matches!( db.validate_batch(&child_b), Err(Error::StaleBatch) )); db.destroy().await.unwrap(); } /// Applying a batch over an applied ancestor and unapplied ones, including an empty one, /// inserts one snapshot location per key, matching the snapshot rebuilt on reopen. #[boxed] pub(crate) async fn run_partial_ancestor_commit( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let key3 = Sha256::hash(&[&[3]]); let v1 = Sha256::fill(1u8); let v2 = Sha256::fill(2u8); let v3 = Sha256::fill(3u8); // Chain: DB <- A <- E <- B <- C, where E is empty. let a = db .new_batch() .set(key1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let e = a .new_batch::() .merkleize(&db, None, Location::new(0)) .await .unwrap(); let b = e .new_batch::() .set(key2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let c = b .new_batch::() .set(key3, v3) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let expected_root = c.root(); // Apply only A, then apply C directly (E and B unapplied). let db = db.apply_batch(a).await.unwrap(); let db = db.apply_batch(c).await.unwrap(); assert_eq!(db.root(), expected_root); assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), Some(v2)); assert_eq!(db.get(&key3).await.unwrap(), Some(v3)); // The snapshot holds one location per key. let keys = [key1, key2, key3]; let locations = |db: &TestDb| { keys.iter() .map(|key| { let mut locations: Vec<_> = db.snapshot.get(key).collect(); locations.sort(); locations }) .collect::>() }; assert_eq!(db.snapshot.len(), keys.len()); let live = locations(&db); // Reopen: the rebuilt snapshot holds the same locations. db.sync().await.unwrap(); let db = open_db(context.child("reopen")).await; assert_eq!(db.snapshot.len(), keys.len()); assert_eq!(locations(&db), live); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_delayed_merkleize_after_ancestor_apply( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let key3 = Sha256::hash(&[&[3]]); let v1 = Sha256::fill(1u8); let v2 = Sha256::fill(2u8); let v3 = Sha256::fill(3u8); let a = db .new_batch() .set(key1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let b = a .new_batch::() .set(key2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let pending = b .new_batch::() .set(key3, v3) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let expected_root = pending.root(); let c = b.new_batch::().set(key3, v3); // The database may advance to a held intermediate ancestor. let db = db.apply_batch(Arc::clone(&a)).await.unwrap(); let live = b .new_batch::() .set(key3, v3) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert_eq!(live.root(), expected_root); // Retiring that ancestor advances the effective database boundary. drop(a); let retired = c.merkleize(&db, None, Location::new(0)).await.unwrap(); assert_eq!(retired.bounds().db, db.commitment()); assert_eq!(retired.root(), expected_root); let c = b.new_batch::().set(key3, v3); let db = db.apply_batch(b).await.unwrap(); let c = c.merkleize(&db, None, Location::new(0)).await.unwrap(); assert_eq!(c.root(), expected_root); let db = db.apply_batch(c).await.unwrap(); assert_eq!(db.root(), expected_root); assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), Some(v2)); assert_eq!(db.get(&key3).await.unwrap(), Some(v3)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_sequential_commit_parent_then_child( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let v1 = Sha256::fill(1u8); let v2 = Sha256::fill(2u8); // Parent batch. let parent_m = db .new_batch() .set(key1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Child batch built on parent. let child_m = parent_m .new_batch::() .set(key2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Apply parent first, then child, as a sequential chain. let db = db.apply_batch(parent_m).await.unwrap(); let db = db.apply_batch(child_m).await.unwrap(); // Both keys present. assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), Some(v2)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_child_root_matches_pending_and_committed( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); // Build the child while the parent is unapplied. let parent = db .new_batch() .set(key1, Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let pending_child = parent .new_batch::() .set(key2, Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Commit the parent, then rebuild the same logical child from the // committed DB state and compare roots. let db = db.apply_batch(parent).await.unwrap(); let db = db.commit().await.unwrap(); let committed_child = db .new_batch() .set(key2, Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert_eq!(pending_child.root(), committed_child.root()); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_stale_batch_child_applied_before_parent( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); // Parent batch. let parent_m = db .new_batch() .set(key1, Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Child batch. let child_m = parent_m .new_batch::() .set(key2, Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Apply child first (it carries all parent ops too). let db = db.apply_batch(child_m).await.unwrap(); // Parent is stale. let result = db.apply_batch(parent_m).await; assert!(matches!(result, Err(Error::StaleBatch))); } /// to_batch() creates an owned batch whose root matches the DB's. /// A child batch chained from it can be applied. #[boxed] pub(crate) async fn run_to_batch( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; // Populate. let key1 = Sha256::hash(&[&[1]]); let v1 = Sha256::fill(10u8); let merkleized = db .new_batch() .set(key1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // to_batch root matches the DB's root. let snapshot = db.to_batch(); assert_eq!(snapshot.root(), db.root()); // Chain a child from `snapshot` and apply it. let key2 = Sha256::hash(&[&[2]]); let v2 = Sha256::fill(20u8); let child = snapshot .new_batch::() .set(key2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(child).await.unwrap(); assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), Some(v2)); db.destroy().await.unwrap(); } /// Regression: applying a batch after its unapplied ancestor's Arc is /// dropped must still apply the ancestor's snapshot diffs. #[boxed] pub(crate) async fn run_apply_after_ancestor_dropped( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let key3 = Sha256::hash(&[&[3]]); let v1 = Sha256::fill(1u8); let v2 = Sha256::fill(2u8); let v3 = Sha256::fill(3u8); // Chain: DB <- A <- B <- C let a = db .new_batch() .set(key1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let b = a .new_batch::() .set(key2, v2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let c = b .new_batch::() .set(key3, v3) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Drop A and B without committing. Their Weak refs in C are now dead. drop(a); drop(b); // Apply only the tip. This is !skip_ancestors (DB hasn't changed). let db = db.apply_batch(c).await.unwrap(); // All three keys must be in the snapshot. assert_eq!(db.get(&key1).await.unwrap(), Some(v1)); assert_eq!(db.get(&key2).await.unwrap(), Some(v2)); assert_eq!(db.get(&key3).await.unwrap(), Some(v3)); db.destroy().await.unwrap(); } /// Verify the inactivity floor is zero for a fresh empty database and is /// correctly set after applying batches with specific floor values. #[boxed] pub(crate) async fn run_inactivity_floor_tracking( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // Empty DB has floor=0. assert_eq!(db.inactivity_floor_loc(), Location::new(0)); // Apply batch with floor=0, floor stays 0. let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(2u8); let merkleized = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(0)); // Apply batch with floor=3, floor advances. let k2 = Sha256::fill(3u8); let v2 = Sha256::fill(4u8); let merkleized = db .new_batch() .set(k2, v2) .merkleize(&db, None, Location::new(3)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(3)); // Floor persists across restart. let db = db.commit().await.unwrap(); db.sync().await.unwrap(); let db = open_db(context.child("reopen")).await; assert_eq!(db.inactivity_floor_loc(), Location::new(3)); db.destroy().await.unwrap(); } /// Verify that applying a batch with a floor equal to the current floor succeeds, /// and that a higher floor also succeeds. #[boxed] pub(crate) async fn run_floor_monotonicity( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // DB starts with 1 op (initial commit). // First batch: 1 set + 1 commit = total_size 3. Use floor=2 (the commit loc). let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(2u8); let merkleized = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(2)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(2)); // Same floor is OK. Second batch: 1 set + 1 commit = total_size 5. floor=2 < 5. let k2 = Sha256::fill(3u8); let v2 = Sha256::fill(4u8); let merkleized = db .new_batch() .set(k2, v2) .merkleize(&db, None, Location::new(2)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(2)); // Higher floor also succeeds. Third batch: 1 set + 1 commit = total_size 7. floor=5 < 7. let k3 = Sha256::fill(5u8); let v3 = Sha256::fill(6u8); let merkleized = db .new_batch() .set(k3, v3) .merkleize(&db, None, Location::new(5)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(5)); db.destroy().await.unwrap(); } /// Verify that the inactivity floor is correctly restored after a recovery. #[boxed] pub(crate) async fn run_bounded_initialization_restores_floor( context: deterministic::Context, open_db: impl Fn( deterministic::Context, Option>, ) -> Pin, Error>> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test"), None).await.unwrap(); // Apply first batch with floor=2. let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(2u8); let merkleized = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(2)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let first_size = db.bounds().end; assert_eq!(db.inactivity_floor_loc(), Location::new(2)); // Apply second batch with floor=4 (the new commit's location). let k2 = Sha256::fill(3u8); let v2 = Sha256::fill(4u8); let merkleized = db .new_batch() .set(k2, v2) .merkleize(&db, None, Location::new(4)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(4)); // Reopen at the first batch. _ = db.sync().await.unwrap(); let db = open_db(context.child("cap"), Some(first_size)) .await .unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(2)); db.destroy().await.unwrap(); } /// Verify that applying a batch with a floor lower than the current floor /// returns an error. #[boxed] pub(crate) async fn run_floor_monotonicity_violation( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // DB starts with 1 op. First batch: 1 set + 1 commit = total_size 3. floor=2. let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(2u8); let merkleized = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(2)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // Merkleize a batch with floor=1 (regression). Should return an error. let k2 = Sha256::fill(3u8); let v2 = Sha256::fill(4u8); let result = db .new_batch() .set(k2, v2) .merkleize(&db, None, Location::new(1)) .await; assert!(matches!(result, Err(Error::FloorRegressed(new, current)) if new == Location::new(1) && current == Location::new(2))); } /// Verify that merkleizing a batch with a floor beyond its commit location returns an error. #[boxed] pub(crate) async fn run_floor_beyond_size( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // DB has 1 op (initial commit). A batch with 1 set + 1 commit = total_size 3. // Setting floor=100 exceeds total_size. let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(2u8); let result = db .new_batch() .set(k1, v1) .merkleize(&db, None, Location::new(100)) .await; assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit)) if floor == Location::new(100) && commit == Location::new(2))); // Boundary: floor == total_size must also be rejected. The commit op is // at total_size - 1, so a floor equal to total_size would allow a later // prune to remove the commit and leave the db unrecoverable. let k2 = Sha256::fill(3u8); let v2 = Sha256::fill(4u8); let result = db .new_batch() .set(k2, v2) .merkleize(&db, None, Location::new(3)) .await; assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit)) if floor == Location::new(3) && commit == Location::new(2))); // Floor == total_size - 1 (the commit location) is the maximum valid. let merkleized = db .new_batch() .set(k2, v2) .merkleize(&db, None, Location::new(2)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); db.destroy().await.unwrap(); } /// Merkleize returns `FloorRegressed` for a chained batch whose floor is below its parent's /// floor, even though that floor is still at or above the database's floor. #[boxed] pub(crate) async fn run_chained_floor_regression( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // The database's floor is 0 (from the seeded initial commit). // a: 1 set + commit at loc 2, floor=2 (valid: >= 0, == commit_loc). // b: 1 set + commit at loc 4, floor=1 (regresses below a's floor=2, but still >= 0). // Merkleizing b must fail with FloorRegressed(1, 2). The bound is a's floor, not the // database's floor. let a = db .new_batch() .set(Sha256::fill(1u8), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(2)) .await .unwrap(); let Err(err) = a .new_batch::() .set(Sha256::fill(3u8), Sha256::fill(4u8)) .merkleize(&db, None, Location::new(1)) .await else { panic!("merkleize must reject the invalid floor"); }; assert!( matches!(err, Error::FloorRegressed(new, prev) if new == Location::new(1) && prev == Location::new(2)), "unexpected error: {err:?}" ); db.destroy().await.unwrap(); } /// Merkleize returns `FloorBeyondSize` for a chained batch whose floor exceeds its own commit /// location. The commit location includes the parent's operations. The floor still satisfies /// monotonicity, so only the commit-location bound rejects it. #[boxed] pub(crate) async fn run_chained_floor_beyond_commit( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // a: 1 set + commit at loc 2, floor=2. let a = db .new_batch() .set(Sha256::fill(1u8), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(2)) .await .unwrap(); // b: 1 set + commit at loc 4, declaring floor=5 (one past its commit). The error names // commit location 4, so the bound counts a's operations. let Err(err) = a .new_batch::() .set(Sha256::fill(3u8), Sha256::fill(4u8)) .merkleize(&db, None, Location::new(5)) .await else { panic!("merkleize must reject the invalid floor"); }; assert!( matches!(err, Error::FloorBeyondSize(floor, commit) if floor == Location::new(5) && commit == Location::new(4)), "unexpected error: {err:?}" ); db.destroy().await.unwrap(); } /// Reinitializing at an earlier commit restores every key active at that commit, even after an /// ordinary reopen rebuilds the snapshot from the latest floor. #[boxed] pub(crate) async fn run_bounded_initialization_after_reopen_with_floor_change( context: deterministic::Context, open_db: impl Fn( deterministic::Context, Option>, ) -> Pin, Error>> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first"), None).await.unwrap(); let k1 = Sha256::fill(1u8); let k2 = Sha256::fill(2u8); let k3 = Sha256::fill(3u8); let v1 = Sha256::fill(11u8); let v2 = Sha256::fill(12u8); let v3 = Sha256::fill(13u8); // Commit A: 3 keys with floor=0. let db = commit_sets(db, [(k1, v1), (k2, v2), (k3, v3)], None).await; let first_size = db.bounds().end; let first_root = db.root(); // Commit B: 3 more keys with floor=first_size (declares batch A inactive). let k4 = Sha256::fill(4u8); let k5 = Sha256::fill(5u8); let k6 = Sha256::fill(6u8); let v4 = Sha256::fill(14u8); let v5 = Sha256::fill(15u8); let v6 = Sha256::fill(16u8); let db = commit_sets_with_floor(db, [(k4, v4), (k5, v5), (k6, v6)], None, first_size).await; db.sync().await.unwrap(); // Reopen: snapshot rebuilt from floor=first_size, batch A keys excluded. let db = open_db(context.child("second"), None).await.unwrap(); // Verify batch A keys are NOT in the reopened snapshot (expected). assert!(db.get(&k1).await.unwrap().is_none()); // Reopen at commit A. _ = db.sync().await.unwrap(); let db = open_db(context.child("cap"), Some(first_size)) .await .unwrap(); // All batch A keys must be accessible after recovery. assert_eq!(db.get(&k1).await.unwrap(), Some(v1)); assert_eq!(db.get(&k2).await.unwrap(), Some(v2)); assert_eq!(db.get(&k3).await.unwrap(), Some(v3)); assert_eq!(db.root(), first_root); assert_eq!(db.inactivity_floor_loc(), Location::new(0)); // Batch B keys must NOT be accessible. assert!(db.get(&k4).await.unwrap().is_none()); db.destroy().await.unwrap(); } /// Opening an earlier commit rebuilds the snapshot from that commit's floor, even when /// the database was previously opened at a later floor. #[boxed] pub(crate) async fn run_bounded_initialization_after_reopen_partial_floor_gap( context: deterministic::Context, open_db: impl Fn( deterministic::Context, Option>, ) -> Pin, Error>> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("first"), None).await.unwrap(); let k1 = Sha256::fill(1u8); let v1 = Sha256::fill(11u8); // Commit A: 1 key, floor=0. let db = commit_sets(db, [(k1, v1)], None).await; let first_size = db.bounds().end; let first_root = db.root(); // Commit B: 1 key, floor=first_size. let k2 = Sha256::fill(2u8); let v2 = Sha256::fill(12u8); let db = commit_sets_with_floor(db, [(k2, v2)], None, first_size).await; let second_size = db.bounds().end; // Commit C: 1 key, floor=second_size. This raises the floor // above commit B's keys, so reopen excludes both A and B keys. let k3 = Sha256::fill(3u8); let v3 = Sha256::fill(13u8); let db = commit_sets_with_floor(db, [(k3, v3)], None, second_size).await; db.sync().await.unwrap(); // Reopen: snapshot rebuilt from floor=second_size. Only k3 is in snapshot. let db = open_db(context.child("second"), None).await.unwrap(); assert!(db.get(&k1).await.unwrap().is_none()); assert!(db.get(&k2).await.unwrap().is_none()); assert_eq!(db.get(&k3).await.unwrap(), Some(v3)); // Commit B's snapshot contains k2. Its floor excludes k1, and its end excludes k3. _ = db.sync().await.unwrap(); let db = open_db(context.child("cap"), Some(second_size)) .await .unwrap(); assert!(db.get(&k1).await.unwrap().is_none()); // below B's floor assert_eq!(db.get(&k2).await.unwrap(), Some(v2)); assert!(db.get(&k3).await.unwrap().is_none()); // in suffix, removed // Open commit A and rebuild its snapshot from floor zero. _ = db.sync().await.unwrap(); let db = open_db(context.child("cap"), Some(first_size)) .await .unwrap(); assert_eq!(db.get(&k1).await.unwrap(), Some(v1)); assert!(db.get(&k2).await.unwrap().is_none()); // above first_size, truncated assert_eq!(db.root(), first_root); assert_eq!(db.inactivity_floor_loc(), Location::new(0)); db.destroy().await.unwrap(); } /// After committing with `floor = commit_loc` and pruning down to it, exactly one operation, /// the commit itself, is at or above the floor. This is the smallest nonempty such set the /// per-commit bound allows. The DB must remain fully usable: /// /// - `prune(commit_loc + 1)` is rejected (the floor is a hard ceiling). /// - `prune` does not affect the root (documented invariant). /// - Reopen reconstructs `inactivity_floor_loc` from the sole remaining commit op, and the /// in-memory snapshot is empty (all Sets were below the floor). /// - A follow-on batch applies cleanly on top from the floor-at-max state. #[boxed] pub(crate) async fn run_single_commit_live_set( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("test")).await; // Initial commit is at loc 0. 3 sets + 1 commit -> commit lands at loc 4. // Declare floor = 4 (= commit_loc), the tight maximum. let metadata = Sha256::fill(42u8); let commit_loc = Location::::new(4); let k1 = Sha256::fill(1u8); let k2 = Sha256::fill(2u8); let k3 = Sha256::fill(3u8); let v1 = Sha256::fill(11u8); let v2 = Sha256::fill(12u8); let v3 = Sha256::fill(13u8); let merkleized = db .new_batch() .set(k1, v1) .set(k2, v2) .set(k3, v3) .merkleize(&db, Some(metadata), commit_loc) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.size() - 1, commit_loc); assert_eq!(db.inactivity_floor_loc(), commit_loc); let root_after_commit = db.root(); // All three keys are in the in-memory snapshot pre-prune. assert_eq!(db.get(&k1).await.unwrap(), Some(v1)); assert_eq!(db.get(&k2).await.unwrap(), Some(v2)); assert_eq!(db.get(&k3).await.unwrap(), Some(v3)); // Prune at the floor, the maximum prune allowed. // Pruning is blob-aligned, so `bounds.start` may not physically advance all the way // to `commit_loc`. What matters semantically is that the floor authorizes pruning // of everything below the commit and that any further prune is rejected. let db = db.prune(commit_loc).await.unwrap(); let bounds = db.bounds(); assert!( bounds.start <= commit_loc, "prune must not advance bounds.start past the floor" ); assert_eq!(bounds.end, commit_loc + 1); // State preserved across the prune; root unchanged; commit metadata still readable. assert_eq!(db.size() - 1, commit_loc); assert_eq!(db.inactivity_floor_loc(), commit_loc); assert_eq!(db.root(), root_after_commit); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); // Persist, then verify pruning one past the floor is rejected, because the floor is // the hard ceiling. let db = db.sync().await.unwrap(); let Err(err) = db.prune(commit_loc + 1).await else { panic!("expected prune to fail"); }; assert!(matches!(err, Error::PruneBeyondMinRequired(p, f) if *p == *commit_loc + 1 && *f == *commit_loc)); // Reopening rebuilds the snapshot from the inactivity floor. Only the commit // remains at or above that floor, so the rebuilt snapshot contains no keys. let db = open_db(context.child("reopened")).await; assert_eq!(db.size() - 1, commit_loc); assert_eq!(db.inactivity_floor_loc(), commit_loc); assert_eq!(db.root(), root_after_commit); // The commit op at `commit_loc` is the anchor that survived pruning. Its metadata // must come back through `get_metadata` after the snapshot rebuild. assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); // Keys set below the floor are excluded from the rebuilt snapshot. assert!(db.get(&k1).await.unwrap().is_none()); assert!(db.get(&k2).await.unwrap().is_none()); assert!(db.get(&k3).await.unwrap().is_none()); // A follow-on batch applies on top. Monotonicity requires the new floor to be at // least `commit_loc` (= 4). Advancing to the new tight max (= 6) exercises the // floor-at-max -> new-batch transition. let k4 = Sha256::fill(4u8); let v4 = Sha256::fill(14u8); let next_commit_loc = Location::::new(6); let merkleized = db .new_batch() .set(k4, v4) .merkleize(&db, None, next_commit_loc) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.size() - 1, next_commit_loc); assert_eq!(db.inactivity_floor_loc(), next_commit_loc); // New key readable; keys from the pre-prune batch remain excluded. assert_eq!(db.get(&k4).await.unwrap(), Some(v4)); assert!(db.get(&k1).await.unwrap().is_none()); // Follow-on commit replaced the anchor: its metadata was `None`, so `get_metadata` // should no longer return the original metadata. assert_eq!(db.get_metadata().await.unwrap(), None); db.destroy().await.unwrap(); } /// `get_many` on the DB and on unmerkleized/merkleized batches returns results /// that match individual `get` calls. #[boxed] pub(crate) async fn run_get_many( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let k1 = Sha256::fill(1u8); let k2 = Sha256::fill(2u8); let k3 = Sha256::fill(3u8); let k_missing = Sha256::fill(99u8); let v1 = Sha256::fill(11u8); let v2 = Sha256::fill(12u8); let v3 = Sha256::fill(13u8); // Commit k1 and k2 to disk. let merkleized = db .new_batch() .set(k1, v1) .set(k2, v2) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); // DB-level get_many. let results = db.get_many(&[&k1, &k2, &k_missing]).await.unwrap(); assert_eq!(results, vec![Some(v1), Some(v2), None]); // Empty input. let results = db.get_many(&([] as [&Digest; 0])).await.unwrap(); assert!(results.is_empty()); // Unmerkleized batch: mutations + DB fallthrough. let batch = db.new_batch().set(k3, v3); let results = batch.get_many(&[&k3, &k1, &k_missing], &db).await.unwrap(); assert_eq!(results, vec![Some(v3), Some(v1), None]); // Merkleized batch: diff + parent chain + DB fallthrough. let parent = db .new_batch() .set(k3, v3) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let results = parent.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap(); assert_eq!(results, vec![Some(v1), Some(v3), None]); // Child of merkleized parent reads parent diff. let k4 = Sha256::fill(4u8); let v4 = Sha256::fill(14u8); let child = parent.new_batch::().set(k4, v4); let results = child .get_many(&[&k1, &k3, &k4, &k_missing], &db) .await .unwrap(); assert_eq!(results, vec![Some(v1), Some(v3), Some(v4), None]); db.destroy().await.unwrap(); } /// `get_many` fills every slot of a key listed more than once in the input. #[boxed] pub(crate) async fn run_get_many_duplicate_keys( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key = Sha256::fill(1u8); let value = Sha256::fill(11u8); let k_missing = Sha256::fill(99u8); let merkleized = db .new_batch() .set(key, value) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let results = db.get_many(&[&key, &k_missing, &key, &key]).await.unwrap(); assert_eq!(results, vec![Some(value), None, Some(value), Some(value)]); db.destroy().await.unwrap(); } /// `get_many` reports unexpected data when the snapshot points at a non-`Set` operation. #[boxed] pub(crate) async fn run_get_many_unexpected_data( context: deterministic::Context, open_db: impl Fn( deterministic::Context, ) -> Pin> + Send>>, ) where V: ValueEncoding, C: Mutable>, C::Item: EncodeShared, { let db = open_db(context.child("db")).await; let key = Sha256::fill(1u8); let value = Sha256::fill(11u8); let merkleized = db .new_batch() .set(key, value) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let mut db = db.commit().await.unwrap(); let bad_key = Sha256::fill(99u8); let bad_loc = db.size() - 1; db.snapshot.insert(&bad_key, *bad_loc); let err = db.get(&bad_key).await.unwrap_err(); assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc)); let err = db.get_many(&[&bad_key]).await.unwrap_err(); assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc)); db.destroy().await.unwrap(); } }