//! The [Keyless] qmdb allows for append-only storage of data that can later be retrieved by its //! location. Both fixed-size and variable-size values are supported via the [fixed] and [variable] //! submodules. //! //! # Examples //! //! ```ignore //! // Simple mode: apply a batch, then durably commit it. //! let batch = db.new_batch().append(value); //! let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await?; //! 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().append(value_a); //! let parent = parent.merkleize(&db, None, floor).await?; //! //! let child_a = parent.new_batch(); //! let child_a = child_a.append(value_b); //! let child_a = child_a.merkleize(&db, None, floor).await?; //! //! let child_b = parent.new_batch(); //! let child_b = child_b.append(value_c); //! let child_b = child_b.merkleize(&db, None, floor).await?; //! //! let db = db.apply_batch(child_a).await?.commit().await?; //! ``` //! //! ```ignore //! // Sequential commit: apply parent then child. //! let floor = db.inactivity_floor_loc(); //! let parent = db.new_batch().append(value_a); //! let parent_m = parent.merkleize(&db, None, floor).await?; //! let child = parent_m.new_batch().append(value_b); //! let child_m = child.merkleize(&db, None, floor).await?; //! //! let db = db.apply_batch(parent_m).await?; //! let db = db.apply_batch(child_m).await?.commit().await?; //! ``` use crate::{ Context, journal::{ authenticated, contiguous::{Contiguous, Mutable}, }, merkle::{Family, Location, Proof, full::Config as MerkleConfig}, qmdb::{ Error, any::value::ValueEncoding, chain, metrics::Metrics, single_operation_root, sync::source, }, }; use commonware_codec::EncodeShared; use commonware_cryptography::Hasher; use commonware_macros::boxed; use commonware_parallel::Strategy; use commonware_runtime::Handle; use std::{num::NonZeroU64, sync::Arc}; use tracing::{debug, warn}; pub mod batch; mod compact; pub mod fixed; mod operation; pub(crate) mod sync; pub mod variable; pub use compact::{ Config as CompactConfig, Db as CompactDb, MerkleizedBatch as CompactMerkleizedBatch, UnmerkleizedBatch as CompactUnmerkleizedBatch, }; pub use operation::{APPEND_CONTEXT, COMMIT_CONTEXT, Operation}; /// 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, V: ValueEncoding, H: Hasher, Operation: EncodeShared, { single_operation_root::(&Operation::::Commit(None, Location::new(0))) } /// Check that `locs` is strictly increasing and every location is below `end`. fn check_locations(locs: &[Location], end: Location) -> Result<(), Error> { if let Some(pair) = locs.windows(2).find(|pair| pair[0] >= pair[1]) { return Err(Error::LocationOutOfOrder(pair[0], pair[1])); } // Strictly increasing, so only the last location can reach `end`. if let Some(&last) = locs.last() && last >= end { return Err(Error::LocationOutOfBounds(last, end)); } Ok(()) } /// Configuration for a [Keyless] authenticated db. #[derive(Clone)] pub struct Config { /// Configuration for the Merkle structure backing the authenticated journal. pub merkle: MerkleConfig, /// Configuration for the operations log journal. pub log: J, } /// A keyless authenticated database. pub struct Keyless where F: Family, E: Context, V: ValueEncoding, C: Contiguous>, H: Hasher, S: Strategy, Operation: EncodeShared, { /// Authenticated journal of operations. journal: authenticated::Journal, /// Cached canonical operations root. root: H::Digest, /// The inactivity floor declared by the last committed batch. Operations at locations below /// this value are considered inactive by the application and may be pruned. inactivity_floor_loc: Location, /// Metrics for this database. metrics: Metrics, } impl std::fmt::Debug for Keyless where F: Family, E: Context, V: ValueEncoding, C: Mutable>, H: Hasher, S: Strategy, Operation: EncodeShared, { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Keyless") .field("bounds", &self.bounds()) .field("inactivity_floor_loc", &self.inactivity_floor_loc()) .finish_non_exhaustive() } } impl Keyless where F: Family, E: Context, V: ValueEncoding, C: Mutable>, H: Hasher, S: Strategy, Operation: EncodeShared, { /// 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. /// /// After a bounded initialization past a prune, or a sync of a range that starts above the /// floor, the selected commit's inactivity floor can be below the retained start. A batch /// read of a location between the two that no batch chain holds then returns /// [`crate::journal::Error::ItemPruned`], while a database that retains the location answers /// it. Databases at the same commitment therefore answer batch reads alike only when each /// retains every operation from its floor onward. #[boxed] pub async fn init( context: E, cfg: Config, max_size: Option>, ) -> Result> where C: authenticated::Backing, { // Keyless restores commit fields without replaying a keyed snapshot, so its logical floor // may precede the retained operation prefix. let mut journal = crate::qmdb::init_journal::( context.child("journal"), cfg.merkle, cfg.log, max_size, false, ) .await?; let metrics = Metrics::new(context); if journal.size() == 0 { warn!("no operations found in log, creating initial commit"); (journal, _) = journal .append(&Operation::Commit(None, Location::new(0))) .await?; journal = journal.sync().await?; } let (last_commit_loc, inactivity_floor_loc) = { let bounds = journal.bounds(); let last_commit_loc = Location::new( bounds .end .checked_sub(1) .expect("at least one commit should exist"), ); let op = journal.read(*last_commit_loc).await?; let inactivity_floor_loc = op .has_floor() .expect("last operation should be a commit with floor"); (last_commit_loc, inactivity_floor_loc) }; let inactive_peaks = F::inactive_peaks(last_commit_loc + 1, inactivity_floor_loc); let root = journal.root(inactive_peaks)?; let db = Self { journal, root, inactivity_floor_loc, metrics, }; db.update_metrics(); Ok(db) } /// Get the value at location `loc` in the database. /// /// # Errors /// /// - Returns [`Error::LocationOutOfBounds`] if `loc` >= `self.bounds().end`. /// - Returns [`Error::Journal`] with [`crate::journal::Error::ItemPruned`] if `loc` < /// `self.bounds().start`. pub async fn get(&self, loc: Location) -> Result, Error> { let _timer = self.metrics.get_timer(); self.metrics.get_calls.inc(); self.metrics.lookups_requested.inc(); let op_count = self.journal.bounds().end; if loc >= op_count { return Err(Error::LocationOutOfBounds(loc, Location::new(op_count))); } let op = self.journal.read(*loc).await?; let result = op.into_value(); Ok(result) } /// Batch read values at multiple locations. /// /// Locations must be strictly increasing. /// Returns results in the same order as the input locations. /// /// # Errors /// /// - Returns [`Error::LocationOutOfOrder`] if the locations are not strictly increasing. /// - Returns [`Error::LocationOutOfBounds`] if any location >= `self.bounds().end`. /// - Returns [`Error::Journal`] with [`crate::journal::Error::ItemPruned`] if any location < /// `self.bounds().start`. pub async fn get_many(&self, locs: &[Location]) -> Result>, Error> { if locs.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(locs.len() as u64); check_locations(locs, Location::new(self.journal.bounds().end))?; let positions: Vec = locs.iter().map(|loc| **loc).collect(); let ops = self.journal.read_many(&positions).await?; let result = ops.into_iter().map(|op| op.into_value()).collect(); Ok(result) } /// Returns the inactivity floor declared by the last committed batch. pub const fn inactivity_floor_loc(&self) -> Location { self.inactivity_floor_loc } /// Return the retained operation range `[start, end)`. /// /// Proof generation also requires the necessary Merkle nodes to be retained. pub fn bounds(&self) -> std::ops::Range> { let bounds = self.journal.bounds(); Location::new(bounds.start)..Location::new(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 inactivity floor declared by the last committed batch, which is the upper bound /// on [`Self::prune`]'s `loc`. /// /// This logical boundary may precede the retained start in [`Self::bounds`]. pub const fn sync_boundary(&self) -> Location { self.inactivity_floor_loc } /// 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 op = self.journal.read(*self.journal.size() - 1).await?; let Operation::Commit(metadata, _floor) = op else { return Ok(None); }; Ok(metadata) } /// Return the 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() } /// 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. /// /// # Errors /// /// - Returns [`Error::Merkle`] with [`crate::merkle::Error::RangeOutOfBounds`] if `start_loc` /// >= the number of operations. /// - 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. pub async fn proof( &self, start_loc: Location, max_ops: NonZeroU64, ) -> Result<(Proof, Vec>), Error> { self.historical_proof(self.bounds().end, start_loc, max_ops) .await } /// Analogous to proof, but with respect to the state of the Merkle structure when it had /// `op_count` operations. /// /// `op_count` must be the size of a commit boundary. /// /// # Errors /// /// - Returns [`Error::Merkle`] with [`crate::merkle::Error::RangeOutOfBounds`] if `start_loc` /// >= `op_count` or `op_count` > number of operations. /// - 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. #[allow(clippy::type_complexity)] #[tracing::instrument( name = "qmdb.keyless.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?) } /// Return the pinned Merkle nodes for a lower operation boundary of `loc`. pub async fn pinned_nodes_at(&self, loc: Location) -> Result, Error> { self.journal .merkle .pinned_nodes_at(loc) .await .map_err(Into::into) } /// Prune historical operations prior to `loc`. This does not affect the db's root. /// /// The retained start in [`Self::bounds`] can remain below `loc`. /// /// `prune` requires no prior commit. After a crash, the database remains recoverable; /// uncommitted operations are not guaranteed to survive. /// /// # Errors /// /// - Returns [`Error::PruneBeyondMinRequired`] if `loc` > the inactivity floor. #[tracing::instrument(name = "qmdb.keyless.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.update_metrics(); Ok(self) } /// 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.keyless.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 [`Keyless::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.keyless.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 [`Keyless::apply_batch`] calls. #[tracing::instrument(name = "qmdb.keyless.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.journal.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()) } /// Create an initial [`batch::MerkleizedBatch`] from the DB's current state. pub fn to_batch(&self) -> Arc> { Arc::new(batch::MerkleizedBatch { journal_batch: self.journal.to_merkleized_batch(), parent: None, bounds: chain::Bounds::from_db(self.commitment(), self.inactivity_floor_loc), }) } /// 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 [`Keyless::commit`] or [`Keyless::sync`], or await the handle returned by /// [`Keyless::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.keyless.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 start_loc = self.journal.size(); self.journal = self.journal.apply_batch(&batch.journal_batch).await?; self.inactivity_floor_loc = batch.bounds.inactivity_floor; self.root = batch.root(); let end_loc = batch.bounds.tip.size; debug!(size = ?end_loc, "applied batch"); self.update_metrics(); self.metrics .operations_applied .inc_by(*end_loc - *start_loc); Ok(self) } } impl crate::qmdb::sync::Source for Keyless where F: Family, E: Context, V: ValueEncoding, C: Mutable>, H: Hasher, S: Strategy, Operation: EncodeShared, { 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(crate) mod tests { use super::*; use crate::qmdb::{verify_proof, verify_proof_and_pinned_nodes}; use commonware_cryptography::Sha256; use commonware_parallel::Strategy; use commonware_runtime::{Supervisor as _, deterministic}; use commonware_utils::NZU64; use std::{future::Future, pin::Pin}; pub(crate) type Reopen = Box Pin + Send>>>; pub(crate) type BoundedOpen = Box< dyn Fn( deterministic::Context, Location, ) -> Pin>> + Send>>, >; type TestKeyless = Keyless; /// Test value factory: creates distinct values from an index. pub(crate) trait TestValue: Clone + PartialEq + std::fmt::Debug + Send + Sync { fn make(i: u64) -> Self; } impl TestValue for Vec { fn make(i: u64) -> Self { vec![(i % 255) as u8; ((i % 13) + 7) as usize] } } impl TestValue for commonware_utils::sequence::U64 { fn make(i: u64) -> Self { Self::new(i * 10 + 1) } } /// Emits the named test against `mmr::Family` and `mmb::Family`. macro_rules! keyless_tests { ($($name:ident => $scenario:ident, $fixture:ident;)*) => { $( #[test_traced] fn $name() { deterministic::Runner::default().start(|ctx| async move { keyless_tests!(@fixture $fixture, $scenario, mmr, ctx); }); } )* paste::paste! { $( #[test_traced] fn [<$name _mmb>]() { deterministic::Runner::default().start(|ctx| async move { keyless_tests!(@fixture $fixture, $scenario, mmb, ctx); }); } )* } }; (@fixture bounded, $scenario:ident, $family:ident, $ctx:ident) => { let db = open_db::<$family::Family>($ctx.child("db")).await; tests::$scenario($ctx, db, reopen::<$family::Family>(), bounded_open::<$family::Family>()).await; }; (@fixture bounded_floor, $scenario:ident, $family:ident, $ctx:ident) => { let db = open_db::<$family::Family>($ctx.child("db")).await; tests::$scenario($ctx, db, bounded_open::<$family::Family>()).await; }; (@fixture bounded_indexed, $scenario:ident, $family:ident, $ctx:ident) => { let db = open_db::<$family::Family>($ctx.child("db").with_attribute("index", 1)).await; tests::$scenario($ctx, db, reopen::<$family::Family>(), bounded_open::<$family::Family>()).await; }; (@fixture db, $scenario:ident, $family:ident, $ctx:ident) => { let db = open_db::<$family::Family>($ctx.child("db")).await; tests::$scenario(db).await; }; (@fixture pair, $scenario:ident, $family:ident, $ctx:ident) => { let db = open_db::<$family::Family>($ctx.child("db")).await; let foreign = open_db_with_suffix::<$family::Family>("foreign", $ctx.child("foreign")).await; tests::$scenario(db, foreign).await; }; (@fixture reopen, $scenario:ident, $family:ident, $ctx:ident) => { let db = open_db::<$family::Family>($ctx.child("db")).await; tests::$scenario($ctx, db, reopen::<$family::Family>()).await; }; (@fixture reopen_indexed, $scenario:ident, $family:ident, $ctx:ident) => { let db = open_db::<$family::Family>($ctx.child("db").with_attribute("index", 1)).await; tests::$scenario($ctx, db, reopen::<$family::Family>()).await; }; } pub(super) use keyless_tests; /// Reads stay exact after the ancestor batches are dropped, because the /// journal batch retains every ancestor's items. #[boxed] pub(crate) async fn run_dropped_ancestor_reads( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { // Commit one value so the chain has committed state below it. let seed = db .new_batch() .append(V::Value::make(0)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(seed).await.unwrap(); db = db.commit().await.unwrap(); let floor = db.inactivity_floor_loc(); // A three-level unapplied chain, two appends per level. Each level's // appends start at its parent's tip. let base_a = db.bounds().end; let a = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, floor) .await .unwrap(); let base_b = a.bounds().tip.size; let b = a .new_batch::() .append(V::Value::make(3)) .append(V::Value::make(4)) .merkleize(&db, None, floor) .await .unwrap(); let base_c = b.bounds().tip.size; let c = b .new_batch::() .append(V::Value::make(5)) .merkleize(&db, None, floor) .await .unwrap(); let child = c.new_batch::(); drop(a); drop(b); // Appended values across all three levels resolve from the retained items. let locs = [base_a, base_a + 1, base_b, base_b + 1, base_c]; let expected: Vec<_> = (1..=5u64).map(|i| Some(V::Value::make(i))).collect(); for (loc, expected) in locs.iter().zip(&expected) { assert_eq!(&c.get(*loc, &db).await.unwrap(), expected); assert_eq!(&child.get(*loc, &db).await.unwrap(), expected); } assert_eq!(c.get_many(&locs, &db).await.unwrap(), expected); assert_eq!(child.get_many(&locs, &db).await.unwrap(), expected); // Merkleizing the child returns `StaleBatch`, because its dropped ancestors are unapplied. assert!(matches!( child.merkleize(&db, None, floor).await, Err(Error::StaleBatch) )); // Applying the chain lands the same values the reads reported. let db = db.apply_batch(c).await.unwrap(); for (loc, expected) in locs.iter().zip(&expected) { assert_eq!(&db.get(*loc).await.unwrap(), expected); } db.destroy().await.unwrap(); } /// Batch reads below the batch's inactivity floor return `BelowInactivityFloor` however far /// the database has pruned, and a merkleized batch returns `LocationOutOfBounds` at or past /// its tip, like an unmerkleized one. #[boxed] pub(crate) async fn run_reads_below_floor_refused( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { // Locations 1 and 2 hold appends, and the commit at 3 raises the floor to 2. let seed = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, Location::new(2)) .await .unwrap(); db = db.apply_batch(seed).await.unwrap(); db = db.commit().await.unwrap(); // A batch built on the database uses its floor. let batch = db.new_batch(); assert!(matches!( batch.get(Location::new(1), &db).await, Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(1) )); assert_eq!( batch.get(Location::new(2), &db).await.unwrap(), Some(V::Value::make(2)) ); assert!(matches!( batch .get_many(&[Location::new(1), Location::new(2)], &db) .await, Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(1) )); // Unordered locations are reported before a location below the floor. assert!(matches!( batch .get_many(&[Location::new(1), Location::new(0)], &db) .await, Err(Error::LocationOutOfOrder(first, second)) if first == Location::new(1) && second == Location::new(0) )); // A merkleized batch appending at 4 and raising the floor to 4 returns // `BelowInactivityFloor` for location 2. let parent = db .new_batch() .append(V::Value::make(3)) .merkleize(&db, None, Location::new(4)) .await .unwrap(); assert!(matches!( parent.get(Location::new(2), &db).await, Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(2) )); assert_eq!( parent.get(Location::new(4), &db).await.unwrap(), Some(V::Value::make(3)) ); assert!(matches!( parent .get_many(&[Location::new(2), Location::new(4)], &db) .await, Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(2) )); assert!(matches!( parent .get_many(&[Location::new(3), Location::new(2)], &db) .await, Err(Error::LocationOutOfOrder(first, second)) if first == Location::new(3) && second == Location::new(2) )); // A merkleized batch bounds reads by its own tip, like an unmerkleized one. let tip = parent.bounds().tip.size; assert!(matches!( parent.get(tip, &db).await, Err(Error::LocationOutOfBounds(loc, end)) if loc == tip && end == tip )); assert!(matches!( parent.get_many(&[Location::new(4), tip], &db).await, Err(Error::LocationOutOfBounds(loc, end)) if loc == tip && end == tip )); // Its unmerkleized child builds on the parent's floor. let child = parent.new_batch::(); assert!(matches!( child.get(Location::new(3), &db).await, Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(3) )); assert_eq!( child.get(Location::new(4), &db).await.unwrap(), Some(V::Value::make(3)) ); drop(child); // A pruned instance returns `BelowInactivityFloor` for the location, not `ItemPruned`. db = db.apply_batch(parent).await.unwrap(); let mut batch = db.new_batch(); for i in 10..22 { batch = batch.append(V::Value::make(i)); } let floor = db.bounds().end + 12; let batch = batch.merkleize(&db, None, floor).await.unwrap(); db = db.apply_batch(batch).await.unwrap(); db = db.commit().await.unwrap(); db = db.prune(floor).await.unwrap(); assert!(db.bounds().start > Location::new(2)); assert!(matches!( db.new_batch().get(Location::new(2), &db).await, Err(Error::BelowInactivityFloor(loc)) if loc == Location::new(2) )); 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( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { // Seed committed state so the chain's recorded boundary sits above zero. let mut seed = db.new_batch(); for i in 0..80 { seed = seed.append(V::Value::make(i)); } let seed = seed .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(seed).await.unwrap(); 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() .append(V::Value::make(100)) .merkleize(&db, None, parent_floor) .await .unwrap(); // Two identical children built before the apply, appending enough that their merkleize // merges deep over the database's Merkle nodes. let build = |parent: &Arc>| { let mut child = parent.new_batch::(); for i in 200..280 { child = child.append(V::Value::make(i)); } child }; let child = build(&parent); let expected = build(&parent) .merkleize(&db, None, parent_floor) .await .unwrap() .root(); // Apply the parent and prune to its floor. let db = db.apply_batch(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( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { // Commit one value so the siblings have committed state to read. let merkleized = db .new_batch() .append(V::Value::make(0)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); db = db.commit().await.unwrap(); let floor = db.inactivity_floor_loc(); let winner = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, floor) .await .unwrap(); let loser = db .new_batch() .append(V::Value::make(2)) .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::().append(V::Value::make(3)); let direct = db.new_batch().append(V::Value::make(4)); let committed = Location::new(0); 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(committed, &db).await, Err(Error::StaleRead) )); assert!(matches!( direct.get_many(&[committed], &db).await, Err(Error::StaleRead) )); assert!(matches!( child.get(committed, &db).await, Err(Error::StaleRead) )); assert!(matches!( child.get_many(&[committed], &db).await, Err(Error::StaleRead) )); assert!(matches!( loser.get(committed, &db).await, Err(Error::StaleRead) )); // The compatibility check precedes every other read argument check, so reads of no // locations, of a location at the reader's end, and of unordered locations all return // `StaleRead`. assert!(matches!( child.get_many(&[], &db).await, Err(Error::StaleRead) )); assert!(matches!( loser.get_many(&[], &db).await, Err(Error::StaleRead) )); assert!(matches!( child.get(child.size(), &db).await, Err(Error::StaleRead) )); assert!(matches!( loser.get(loser.size(), &db).await, Err(Error::StaleRead) )); assert!(matches!( child .get_many(&[Location::new(1), Location::new(0)], &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< F: Family, V, C, S: Strategy, >( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { let floor = db.inactivity_floor_loc(); let parent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, floor) .await .unwrap(); let loc = Location::new(1); assert_eq!(parent.get(loc, &db).await.unwrap(), Some(V::Value::make(1))); let child = parent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, floor) .await .unwrap(); let db = db.apply_batch(child.clone()).await.unwrap(); assert_eq!(child.get(loc, &db).await.unwrap(), Some(V::Value::make(1))); assert!(matches!(parent.get(loc, &db).await, Err(Error::StaleRead))); assert!(matches!( parent.get_many(&[loc], &db).await, Err(Error::StaleRead) )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_merkleize_foreign_db( db: TestKeyless, foreign: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let batch = db .new_batch() .append(V::Value::make(11)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap(); let batch = foreign .new_batch() .append(V::Value::make(99)) .merkleize(&foreign, None, Location::new(0)) .await .unwrap(); let foreign = foreign.apply_batch(batch).await.unwrap(); assert_eq!(db.bounds().end, foreign.bounds().end); assert_ne!(db.root(), foreign.root()); let batch = db.new_batch().append(V::Value::make(22)); assert!(matches!( batch.merkleize(&foreign, None, Location::new(0)).await, Err(Error::StaleBatch) )); // A child must also reject a database outside its ancestor chain. let parent = db .new_batch() .append(V::Value::make(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent.new_batch::().append(V::Value::make(44)); assert!(matches!( child.merkleize(&foreign, None, Location::new(0)).await, Err(Error::StaleBatch) )); db.destroy().await.unwrap(); foreign.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_merkleize_stale_sibling( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let direct = db.new_batch().append(V::Value::make(11)); let sibling = db .new_batch() .append(V::Value::make(22)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling).await.unwrap(); let direct_result = direct.merkleize(&db, None, Location::new(0)).await; let parent = db .new_batch() .append(V::Value::make(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent.new_batch::().append(V::Value::make(44)); let sibling = db .new_batch() .append(V::Value::make(55)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling).await.unwrap(); let child_result = child.merkleize(&db, None, Location::new(0)).await; db.destroy().await.unwrap(); assert!(matches!(direct_result, Err(Error::StaleBatch))); assert!(matches!(child_result, Err(Error::StaleBatch))); } #[boxed] pub(crate) async fn run_merkleize_ancestor_states( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let grandparent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let parent = grandparent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let pending = parent .new_batch::() .append(V::Value::make(3)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // The database may advance to a held intermediate ancestor. let db = db.apply_batch(Arc::clone(&grandparent)).await.unwrap(); let applied = parent .new_batch::() .append(V::Value::make(3)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert_eq!(pending.root(), applied.root()); // Once that ancestor is dropped, its commitment becomes the effective boundary. drop(grandparent); let retired = parent .new_batch::() .append(V::Value::make(3)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert_eq!(retired.bounds().db, db.commitment()); assert_eq!(pending.root(), retired.root()); // A child created before its immediate parent is applied still merkleizes and applies // afterward. let child = parent.new_batch::().append(V::Value::make(3)); let db = db.apply_batch(parent).await.unwrap(); let child = child.merkleize(&db, None, Location::new(0)).await.unwrap(); assert_eq!(pending.root(), child.root()); let db = db.apply_batch(child).await.unwrap(); let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap(); assert!(verify_proof::( &proof, Location::new(0), &ops, &db.root() )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_empty( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let bounds = db.bounds(); assert_eq!(bounds.end, 1); // initial commit should exist assert_eq!(bounds.start, Location::new(0)); assert_eq!(db.get_metadata().await.unwrap(), None); assert_eq!(db.bounds().end - 1, Location::new(0)); // Make sure closing/reopening gets us back to the same state, even after adding an uncommitted op. let root = db.root(); { db.new_batch().append(V::Value::make(1)); // Don't merkleize/apply, to simulate a failed commit } drop(db); let db = reopen(context.child("db").with_attribute("index", 2)).await; assert_eq!(db.root(), root); assert_eq!(db.bounds().end, 1); assert_eq!(db.get_metadata().await.unwrap(), None); // Test calling commit on an empty db which should make it (durably) non-empty. let metadata = V::Value::make(99); let merkleized = db .new_batch() .merkleize(&db, Some(metadata.clone()), db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.bounds().end, 2); // 2 commit ops assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone())); assert_eq!( db.get(Location::new(1)).await.unwrap(), Some(metadata.clone()) ); // the commit op let root = db.root(); // Commit op should remain after reopen even without clean shutdown. drop(db); let db = reopen(context.child("db").with_attribute("index", 3)).await; assert_eq!(db.bounds().end, 2); // commit op should remain after re-open. assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); assert_eq!(db.root(), root); assert_eq!(db.bounds().end - 1, Location::new(1)); 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( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared + PartialEq + core::fmt::Debug, { let seed = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .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() .append(V::Value::make(3)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child = parent .new_batch::() .append(V::Value::make(4)) .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() .append(V::Value::make(5)) .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() .append(V::Value::make(6)) .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_commit_after_sync_recovery( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let value0 = V::Value::make(10); let value1 = V::Value::make(20); // Commit and sync the first append so recovery has an older durable boundary. let first_loc = Location::new(1); let merkleized = db .new_batch() .append(value0.clone()) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let db = db.sync().await.unwrap(); // Commit the next append without syncing; reopen must replay it from the journal. let second_loc = db.bounds().end; let merkleized = db .new_batch() .append(value1.clone()) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let committed_bounds = db.bounds(); let committed_root = db.root(); drop(db); let db = reopen(context.child("db").with_attribute("index", 2)).await; assert_eq!(db.bounds(), committed_bounds); assert_eq!(db.root(), committed_root); assert_eq!(db.get(first_loc).await.unwrap(), Some(value0)); assert_eq!(db.get(second_loc).await.unwrap(), Some(value1)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_build_basic( context: deterministic::Context, mut db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Build a db with 2 values and make sure we can get them back. let v1 = V::Value::make(1); let v2 = V::Value::make(2); { let batch = db.new_batch(); let loc1 = batch.size(); let batch = batch.append(v1.clone()); let loc2 = batch.size(); let batch = batch.append(v2.clone()); assert_eq!(loc1, Location::new(1)); assert_eq!(loc2, Location::new(2)); let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } // Make sure closing/reopening gets us back to the same state. assert_eq!(db.bounds().end, 4); // 2 appends, 1 commit + 1 initial commit assert_eq!(db.get_metadata().await.unwrap(), None); assert_eq!(db.get(Location::new(3)).await.unwrap(), None); // the commit op let root = db.root(); db.sync().await.unwrap(); let db = reopen(context.child("db").with_attribute("index", 2)).await; assert_eq!(db.bounds().end, 4); assert_eq!(db.root(), root); assert_eq!(db.get(Location::new(1)).await.unwrap().unwrap(), v1); assert_eq!(db.get(Location::new(2)).await.unwrap().unwrap(), v2); // Make sure commit operation remains after drop/reopen. drop(db); let db = reopen(context.child("db").with_attribute("index", 3)).await; assert_eq!(db.bounds().end, 4); assert_eq!(db.root(), root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_recovery( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let root = db.root(); const ELEMENTS: u64 = 100; // Create uncommitted appends then simulate failure. { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i)); } // Don't merkleize/apply, to simulate a failed commit } drop(db); // Should rollback to the previous root. let mut db = reopen(context.child("db").with_attribute("index", 2)).await; assert_eq!(root, db.root()); // Apply the updates and commit them this time. { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 100)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let db = db.commit().await.unwrap(); let root = db.root(); // Create uncommitted appends then simulate failure. { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 200)); } // Don't merkleize/apply, to simulate a failed commit } drop(db); // Should rollback to the previous root. let mut db = reopen(context.child("db").with_attribute("index", 3)).await; assert_eq!(root, db.root()); // Apply the updates and commit them this time. { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 300)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let db = db.commit().await.unwrap(); let root = db.root(); // Make sure we can reopen and get back to the same state. drop(db); let db = reopen(context.child("db").with_attribute("index", 4)).await; assert_eq!(db.bounds().end, 2 * ELEMENTS + 3); assert_eq!(db.root(), root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_proof( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { const ELEMENTS: u64 = 50; { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let root = db.root(); let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap(); assert!(verify_proof::( &proof, Location::new(0), &ops, &root, )); assert_eq!(ops.len() as u64, 1 + ELEMENTS + 1); let (proof, ops) = db.proof(Location::new(10), NZU64!(5)).await.unwrap(); assert!(verify_proof::( &proof, Location::new(10), &ops, &root, )); assert_eq!(ops.len(), 5); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_metadata( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let metadata = V::Value::make(99); let merkleized = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, Some(metadata.clone()), db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); let merkleized = db .new_batch() .merkleize(&db, None, db.inactivity_floor_loc()) .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_pruning( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Initial floor is 0, so pruning past 0 should fail. assert_eq!(db.inactivity_floor_loc(), Location::new(0)); 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 = reopen(context.child("reopen_empty")).await; // Add values and commit, advancing the floor to the new commit location. let first_commit_loc = Location::::new(3); let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, first_commit_loc) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.bounds().end - 1, first_commit_loc); assert_eq!(db.inactivity_floor_loc(), first_commit_loc); // Append one more, advancing the floor with it. let second_commit_loc = Location::::new(5); let merkleized = db .new_batch() .append(V::Value::make(3)) .merkleize(&db, None, second_commit_loc) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); // Valid prune: up to the floor (previous commit location). let root = db.root(); let db = db.prune(first_commit_loc).await.unwrap(); assert_eq!(db.root(), root); // Pruning beyond the current floor fails. let new_floor = db.inactivity_floor_loc(); let beyond = new_floor + 1; let result = db.prune(beyond).await; assert!( matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor)) if prune_loc == beyond && floor == new_floor) ); } #[boxed] pub(crate) async fn run_empty_db_recovery( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let root = db.root(); const ELEMENTS: u64 = 200; // Reopen DB without clean shutdown and make sure the state is the same. drop(db); let db = reopen(context.child("db").with_attribute("index", 2)).await; assert_eq!(db.bounds().end, 1); // initial commit should exist assert_eq!(db.root(), root); // Simulate failure after inserting operations without a commit. { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i)); } // Don't merkleize/apply, to simulate a failed commit } drop(db); let db = reopen(context.child("db").with_attribute("index", 3)).await; assert_eq!(db.bounds().end, 1); // initial commit should exist assert_eq!(db.root(), root); // Repeat: simulate failure after inserting operations without a commit. { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 500)); } // Don't merkleize/apply, to simulate a failed commit } drop(db); let db = reopen(context.child("db").with_attribute("index", 4)).await; assert_eq!(db.bounds().end, 1); // initial commit should exist assert_eq!(db.root(), root); // One last check: multiple batches of uncommitted appends. { let mut batch = db.new_batch(); for i in 0..ELEMENTS * 3 { batch = batch.append(V::Value::make(i + 1000)); } // Don't merkleize/apply, to simulate a failed commit } drop(db); let mut db = reopen(context.child("db").with_attribute("index", 5)).await; assert_eq!(db.bounds().end, 1); // initial commit should exist assert_eq!(db.root(), root); assert_eq!(db.bounds().end - 1, Location::new(0)); // Apply the ops one last time but fully commit them this time, then clean up. { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 2000)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } db.commit().await.unwrap(); let db = reopen(context.child("db").with_attribute("index", 6)).await; assert!(db.bounds().end > 1); assert_ne!(db.root(), root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_replay_with_trailing_appends( context: deterministic::Context, mut db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Add some initial operations and commit. { let mut batch = db.new_batch(); for i in 0..10u64 { batch = batch.append(V::Value::make(i)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let db = db.commit().await.unwrap(); let committed_root = db.root(); let committed_size = db.bounds().end; // Add exactly one more append (uncommitted). { db.new_batch().append(V::Value::make(99)); // Don't merkleize/apply, to simulate a failed commit } drop(db); // Reopen and verify correct recovery. let mut db = reopen(context.child("db").with_attribute("index", 2)).await; assert_eq!( db.bounds().end, committed_size, "Should recover to last commit" ); assert_eq!(db.root(), committed_root, "Root should match last commit"); assert_eq!( db.bounds().end - 1, committed_size - 1, "Last commit location should be correct" ); // Verify we can append and commit new data after recovery. let new_value = V::Value::make(77); { let batch = db.new_batch(); let loc = batch.size(); let batch = batch.append(new_value.clone()); assert_eq!( loc, committed_size, "New append should get the expected location" ); let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let db = db.commit().await.unwrap(); assert_eq!(db.get(committed_size).await.unwrap(), Some(new_value)); let new_committed_root = db.root(); let new_committed_size = db.bounds().end; // Add multiple uncommitted appends. { let mut batch = db.new_batch(); for i in 0..5u64 { batch = batch.append(V::Value::make(200 + i)); } // Don't merkleize/apply, to simulate a failed commit } drop(db); // Reopen and verify correct recovery. let db = reopen(context.child("db").with_attribute("index", 3)).await; assert_eq!( db.bounds().end, new_committed_size, "Should recover to last commit with multiple trailing appends" ); assert_eq!( db.root(), new_committed_root, "Root should match last commit after multiple appends" ); assert_eq!( db.bounds().end - 1, new_committed_size - 1, "Last commit location should be correct after multiple appends" ); db.destroy().await.unwrap(); } /// `get_many` on the DB and on unmerkleized/merkleized batches returns /// results consistent with individual `get` calls, and rejects locations that /// are not strictly increasing or out of bounds. #[boxed] pub(crate) async fn run_get_many( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let v1 = V::Value::make(1); let v2 = V::Value::make(2); let v3 = V::Value::make(3); // Commit v1 and v2 to disk. let batch = db.new_batch(); let loc1 = batch.size(); let batch = batch.append(v1.clone()); let loc2 = batch.size(); let batch = batch.append(v2.clone()); let merkleized = batch .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(&[loc1, loc2]).await.unwrap(); assert_eq!(results, vec![Some(v1.clone()), Some(v2.clone())]); // Empty input. let results = db.get_many(&[]).await.unwrap(); assert!(results.is_empty()); // Decreasing and repeated locations are rejected. assert!(matches!( db.get_many(&[loc2, loc1]).await, Err(Error::LocationOutOfOrder(first, second)) if first == loc2 && second == loc1 )); assert!(matches!( db.get_many(&[loc1, loc1]).await, Err(Error::LocationOutOfOrder(..)) )); // An out-of-bounds location is rejected even when it follows valid ones. let end = db.bounds().end; assert!(matches!( db.get_many(&[loc1, end]).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end )); // Unmerkleized batch: pending appends + DB fallthrough. let batch = db.new_batch(); let loc3 = batch.size(); let batch = batch.append(v3.clone()); let results = batch.get_many(&[loc1, loc3], &db).await.unwrap(); assert_eq!(results, vec![Some(v1.clone()), Some(v3.clone())]); assert!(matches!( batch.get_many(&[loc3, loc1], &db).await, Err(Error::LocationOutOfOrder(..)) )); let end = batch.size(); assert!(matches!( batch.get_many(&[loc1, end], &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end )); // Merkleized batch: parent chain + DB fallthrough. let parent = db.new_batch(); let parent_loc = parent.size(); let parent = parent .append(v3.clone()) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let results = parent.get_many(&[loc1, parent_loc], &db).await.unwrap(); assert_eq!(results, vec![Some(v1.clone()), Some(v3.clone())]); assert!(matches!( parent.get_many(&[parent_loc, loc1], &db).await, Err(Error::LocationOutOfOrder(..)) )); let end = parent.size(); assert!(matches!( parent.get_many(&[loc1, end], &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end )); let child = parent.new_batch::().append(V::Value::make(4)); let results = child.get_many(&[loc1, loc2], &db).await.unwrap(); assert_eq!(results, vec![Some(v1.clone()), Some(v2.clone())]); assert!(matches!( child.get_many(&[loc2, loc1], &db).await, Err(Error::LocationOutOfOrder(..)) )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_chained( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let v1 = V::Value::make(10); let v2 = V::Value::make(20); let v3 = V::Value::make(30); let parent = db.new_batch(); let loc1 = parent.size(); let parent = parent.append(v1.clone()); let parent_m = parent .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child = parent_m.new_batch::(); let loc2 = child.size(); let child = child.append(v2.clone()); let loc3 = child.size(); let child = child.append(v3.clone()); let child_m = child .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child_root = child_m.root(); let db = db.apply_batch(child_m).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.root(), child_root); assert_eq!(db.get(loc1).await.unwrap(), Some(v1)); assert_eq!(db.get(loc2).await.unwrap(), Some(v2)); assert_eq!(db.get(loc3).await.unwrap(), Some(v3)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_stale_batch( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let batch_a = db .new_batch() .append(V::Value::make(10)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let batch_b = db .new_batch() .append(V::Value::make(20)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(batch_a).await.unwrap(); let db = db.commit().await.unwrap(); let root = db.root(); let last_commit_loc = db.bounds().end - 1; 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 = reopen(context.child("reopen")).await; assert_eq!(db.root(), root); assert_eq!(db.bounds().end - 1, last_commit_loc); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_partial_ancestor_commit( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Chain: DB <- A <- B <- C let a = db .new_batch() .append(V::Value::make(10)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let b = a .new_batch::() .append(V::Value::make(20)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let c = b .new_batch::() .append(V::Value::make(30)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let expected_root = c.root(); // Apply only A, then apply C directly (B's items applied via ancestor batches). let db = db.apply_batch(a).await.unwrap(); let db = db.apply_batch(c).await.unwrap(); // Root must match what the full chain produces. assert_eq!(db.root(), expected_root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_delayed_merkleize_after_ancestor_apply< F: Family, V, C, H, S: Strategy, >( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let floor = db.inactivity_floor_loc(); let a = db .new_batch() .append(V::Value::make(10)) .merkleize(&db, None, floor) .await .unwrap(); let b = a .new_batch::() .append(V::Value::make(20)) .merkleize(&db, None, floor) .await .unwrap(); let c = b.new_batch::().append(V::Value::make(30)); let db = db.apply_batch(a).await.unwrap(); let c = c.merkleize(&db, None, floor).await.unwrap(); let expected_root = c.root(); let db = db.apply_batch(c).await.unwrap(); assert_eq!(db.root(), expected_root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_to_batch( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let batch = db.new_batch(); let loc1 = batch.size(); let batch = batch.append(V::Value::make(10)); let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let snapshot = db.to_batch(); assert_eq!(snapshot.root(), db.root()); let child_batch = snapshot.new_batch::(); let loc2 = child_batch.size(); let child_batch = child_batch.append(V::Value::make(20)); let merkleized = child_batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.get(loc1).await.unwrap(), Some(V::Value::make(10))); assert_eq!(db.get(loc2).await.unwrap(), Some(V::Value::make(20))); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_non_empty_recovery( context: deterministic::Context, mut db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Append many values then commit, advancing the floor to the new commit so we can // later prune up to it. const ELEMENTS: u64 = 200; { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i)); } let new_commit = db.bounds().end + ELEMENTS; let merkleized = batch.merkleize(&db, None, new_commit).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let db = db.commit().await.unwrap(); let root = db.root(); let op_count = db.bounds().end; // Reopen DB without clean shutdown and make sure the state is the same. drop(db); let db = reopen(context.child("db").with_attribute("index", 2)).await; assert_eq!(db.bounds().end, op_count); assert_eq!(db.root(), root); assert_eq!(db.bounds().end - 1, op_count - 1); drop(db); // Insert many operations without commit, then simulate failure. let db = reopen(context.child("recovery_a")).await; { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 1000)); } // Don't merkleize/apply, to simulate a failed commit } drop(db); let db = reopen(context.child("recovery_b")).await; assert_eq!(db.bounds().end, op_count); assert_eq!(db.root(), root); drop(db); // Repeat after pruning to the last commit. let db = reopen(context.child("db").with_attribute("index", 3)).await; let last_commit = db.bounds().end - 1; let db = db.prune(last_commit).await.unwrap(); assert_eq!(db.bounds().end, op_count); assert_eq!(db.root(), root); db.sync().await.unwrap(); let db = reopen(context.child("recovery_c")).await; { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 2000)); } } drop(db); let db = reopen(context.child("recovery_d")).await; assert_eq!(db.bounds().end, op_count); assert_eq!(db.root(), root); drop(db); // Apply the ops one last time but fully commit them this time, then clean up. let mut db = reopen(context.child("db").with_attribute("index", 4)).await; { let mut batch = db.new_batch(); for i in 0..ELEMENTS { batch = batch.append(V::Value::make(i + 3000)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } db.commit().await.unwrap(); let db = reopen(context.child("db").with_attribute("index", 5)).await; let bounds = db.bounds(); assert!(bounds.end > op_count); assert_ne!(db.root(), root); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_proof_comprehensive( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { // Build a db with some values. const ELEMENTS: u64 = 100; { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { batch = batch.append(V::Value::make(i)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } // Test that historical proof fails with op_count > number of operations. assert!(matches!( db.historical_proof(db.bounds().end + 1, Location::new(5), NZU64!(10)) .await, Err(Error::::Merkle(crate::merkle::Error::RangeOutOfBounds( _ ))) )); let root = db.root(); for (start_loc, max_ops) in [ (0, 10), (10, 5), (50, 20), (90, 15), (0, 1), (ELEMENTS - 1, 1), (ELEMENTS, 1), ] { let (proof, ops) = db .proof(Location::new(start_loc), NZU64!(max_ops)) .await .unwrap(); assert!( verify_proof::(&proof, Location::new(start_loc), &ops, &root,), "Failed to verify proof for range starting at {start_loc} with max {max_ops} ops", ); let expected_ops = std::cmp::min(max_ops, *db.bounds().end - start_loc); assert_eq!(ops.len() as u64, expected_ops); let wrong_root = Sha256::hash(&[&[0xFF; 32]]); assert!(!verify_proof::( &proof, Location::new(start_loc), &ops, &wrong_root, )); if start_loc > 0 { assert!(!verify_proof::( &proof, Location::new(start_loc - 1), &ops, &root, )); } } db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_proof_with_pruning( context: deterministic::Context, mut db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { const ELEMENTS: u64 = 100; { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { batch = batch.append(V::Value::make(i)); } let new_commit = db.bounds().end + ELEMENTS; let merkleized = batch.merkleize(&db, None, new_commit).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } { let mut batch = db.new_batch(); for i in ELEMENTS..ELEMENTS * 2 { batch = batch.append(V::Value::make(i)); } let new_commit = db.bounds().end + ELEMENTS; let merkleized = batch.merkleize(&db, None, new_commit).await.unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let root = db.root(); const PRUNE_LOC: u64 = 30; let db = db.prune(Location::new(PRUNE_LOC)).await.unwrap(); let oldest_retained = db.bounds().start; assert_eq!(db.root(), root); db.sync().await.unwrap(); let db = reopen(context).await; assert_eq!(db.root(), root); for (start_loc, max_ops) in [ (oldest_retained, 10), (Location::new(50), 20), (Location::new(150), 10), (Location::new(190), 15), ] { if start_loc < oldest_retained { continue; } let (proof, ops) = db.proof(start_loc, NZU64!(max_ops)).await.unwrap(); assert!(verify_proof::(&proof, start_loc, &ops, &root,)); } let aggressive_prune: Location = Location::new(150); let db = db.prune(aggressive_prune).await.unwrap(); let new_oldest = db.bounds().start; let (proof, ops) = db.proof(new_oldest, NZU64!(20)).await.unwrap(); assert!(verify_proof::( &proof, new_oldest, &ops, &root, )); let almost_all = db.bounds().end - 5; let db = db.prune(almost_all).await.unwrap(); let final_oldest = db.bounds().start; if final_oldest < db.bounds().end { let (final_proof, final_ops) = db.proof(final_oldest, NZU64!(10)).await.unwrap(); assert!(verify_proof::( &final_proof, final_oldest, &final_ops, &root, )); } db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_get_out_of_bounds( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { assert!(db.get(Location::new(0)).await.unwrap().is_none()); let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!( db.get(Location::new(1)).await.unwrap(), Some(V::Value::make(1)) ); assert!(db.get(Location::new(3)).await.unwrap().is_none()); assert!(matches!( db.get(Location::new(4)).await, Err(Error::LocationOutOfBounds(loc, size)) if loc == Location::new(4) && size == Location::new(4) )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_get_pruned( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let mut round = 0u64; while db.bounds().start == Location::new(0) { round += 1; assert!(round <= 64, "failed to prune any history"); db = commit_appends(db, (0..16).map(|i| V::Value::make(round * 100 + i)), None).await; let last_commit = db.bounds().end - 1; db = db.prune(last_commit).await.unwrap(); } let start = db.bounds().start; let pruned = start - 1; assert!(matches!( db.get(pruned).await, Err(Error::Journal(crate::journal::Error::ItemPruned(loc))) if loc == *pruned )); assert!(matches!( db.get_many(&[pruned, start]).await, Err(Error::Journal(crate::journal::Error::ItemPruned(loc))) if loc == *pruned )); db.get(start).await.unwrap(); db.get_many(&[start]).await.unwrap(); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_get( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let base_vals: Vec = (0..3).map(|i| V::Value::make(10 + i)).collect(); let mut base_locs = Vec::new(); { let mut batch = db.new_batch(); for v in &base_vals { let loc = batch.size(); batch = batch.append(v.clone()); base_locs.push(loc); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } let batch = db.new_batch(); for (i, loc) in base_locs.iter().enumerate() { assert_eq!( batch.get(*loc, &db).await.unwrap(), Some(base_vals[i].clone()), ); } let new_val = V::Value::make(99); let new_loc = batch.size(); let batch = batch.append(new_val.clone()); assert_eq!(batch.get(new_loc, &db).await.unwrap(), Some(new_val)); let end = new_loc + 1; assert!(matches!( batch.get(end, &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end )); assert!(matches!( batch.get_many(&[base_locs[0], end], &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == end && bound == end )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_stacked_get( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let v1 = V::Value::make(1); let v2 = V::Value::make(2); let parent = db.new_batch(); let loc1 = parent.size(); let parent = parent.append(v1.clone()); let parent_m = parent .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child = parent_m.new_batch::(); assert_eq!(child.get(loc1, &db).await.unwrap(), Some(v1)); let loc2 = child.size(); let child = child.append(v2.clone()); assert_eq!(child.get(loc2, &db).await.unwrap(), Some(v2)); assert!(matches!( child.get(Location::new(9999), &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == Location::new(9999) && bound == loc2 + 1 )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_speculative_root( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let mut batch = db.new_batch(); for i in 0u64..10 { batch = batch.append(V::Value::make(i)); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let speculative = merkleized.root(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.root(), speculative); let merkleized = db .new_batch() .append(V::Value::make(100)) .merkleize(&db, Some(V::Value::make(55)), db.inactivity_floor_loc()) .await .unwrap(); let speculative = merkleized.root(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.root(), speculative); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_merkleized_batch_get( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let base_val = V::Value::make(10); let merkleized = db .new_batch() .append(base_val.clone()) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let new_val = V::Value::make(20); let merkleized = db .new_batch() .append(new_val.clone()) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); assert_eq!( merkleized.get(Location::new(1), &db).await.unwrap(), Some(base_val), ); assert_eq!( merkleized.get(Location::new(3), &db).await.unwrap(), Some(new_val), ); assert_eq!(merkleized.get(Location::new(4), &db).await.unwrap(), None); assert!(matches!( merkleized.get(Location::new(5), &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == Location::new(5) && bound == Location::new(5) )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_chained_apply_sequential( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let v1 = V::Value::make(1); let v2 = V::Value::make(2); let parent = db.new_batch(); let loc1 = parent.size(); let parent = parent.append(v1.clone()); let parent_m = parent .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let parent_root = parent_m.root(); let db = db.apply_batch(parent_m).await.unwrap(); assert_eq!(db.root(), parent_root); assert_eq!(db.get(loc1).await.unwrap(), Some(v1)); let batch2 = db.new_batch(); let loc2 = batch2.size(); let batch2 = batch2.append(v2.clone()); let batch2_m = batch2 .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let batch2_root = batch2_m.root(); let db = db.apply_batch(batch2_m).await.unwrap(); assert_eq!(db.root(), batch2_root); assert_eq!(db.get(loc2).await.unwrap(), Some(v2)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_many_sequential( mut db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { const BATCHES: u64 = 20; const APPENDS_PER_BATCH: u64 = 5; let mut all_values: Vec = Vec::new(); let mut all_locs: Vec> = Vec::new(); for batch_idx in 0..BATCHES { let mut batch = db.new_batch(); for j in 0..APPENDS_PER_BATCH { let v = V::Value::make(batch_idx * 10 + j); let loc = batch.size(); batch = batch.append(v.clone()); all_values.push(v); all_locs.push(loc); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(merkleized).await.unwrap(); } for (i, loc) in all_locs.iter().enumerate() { assert_eq!(db.get(*loc).await.unwrap(), Some(all_values[i].clone())); } 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, )); assert_eq!(db.bounds().end, 1 + BATCHES * (APPENDS_PER_BATCH + 1)); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_empty( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let merkleized = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let root_before = db.root(); let size_before = db.bounds().end; let merkleized = db .new_batch() .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let speculative = merkleized.root(); let db = db.apply_batch(merkleized).await.unwrap(); assert_ne!(db.root(), root_before); assert_eq!(db.root(), speculative); assert_eq!(db.bounds().end, size_before + 1); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_chained_merkleized_get( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let base_val = V::Value::make(10); let floor = db.inactivity_floor_loc(); let merkleized = db .new_batch() .append(base_val.clone()) .merkleize(&db, None, floor) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let v1 = V::Value::make(1); let parent = db.new_batch(); let loc1 = parent.size(); let parent_m = parent .append(v1.clone()) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let v2 = V::Value::make(2); let child = parent_m.new_batch::(); let loc2 = child.size(); let child_m = child .append(v2.clone()) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); assert_eq!( child_m.get(Location::new(1), &db).await.unwrap(), Some(base_val), ); assert_eq!(child_m.get(loc1, &db).await.unwrap(), Some(v1.clone())); assert_eq!(child_m.get(loc2, &db).await.unwrap(), Some(v2.clone())); let child_end = child_m.size(); assert!(matches!( child_m.get(child_end, &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == child_end && bound == child_end )); // An ancestor's view ends at its own tip. let parent_end = parent_m.size(); assert!(matches!( parent_m.get(loc2, &db).await, Err(Error::LocationOutOfBounds(loc, bound)) if loc == loc2 && bound == parent_end )); // Applying the descendant moves the database off the ancestor's chain states. let db = db.apply_batch(Arc::clone(&parent_m)).await.unwrap(); let db = db.apply_batch(child_m).await.unwrap(); assert_eq!(db.get(loc2).await.unwrap(), Some(v2)); assert!(matches!( parent_m.get(loc1, &db).await, Err(Error::StaleRead) )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_batch_large( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared + std::fmt::Debug, { const N: u64 = 500; let mut values = Vec::new(); let mut locs = Vec::new(); let mut batch = db.new_batch(); for i in 0..N { let v = V::Value::make(i); locs.push(batch.size()); batch = batch.append(v.clone()); values.push(v); } let merkleized = batch .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); for (i, loc) in locs.iter().enumerate() { assert_eq!(db.get(*loc).await.unwrap(), Some(values[i].clone())); } 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, )); assert_eq!(db.bounds().end, 1 + N + 1); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_stale_batch_chained( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let common_parent = db .new_batch() .append(V::Value::make(10)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let sibling_a = common_parent .new_batch::() .append(V::Value::make(11)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let sibling_b = common_parent .new_batch::() .append(V::Value::make(12)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(sibling_a).await.unwrap(); assert!(matches!( db.validate_batch(&sibling_b), Err(Error::StaleBatch) )); let parent_a = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let parent_b = db .new_batch() .append(V::Value::make(2)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child_b = parent_b .new_batch::() .append(V::Value::make(3)) .merkleize(&db, None, db.inactivity_floor_loc()) .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(); } #[boxed] pub(crate) async fn run_sequential_commit_parent_then_child( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let parent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child = parent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(parent).await.unwrap(); let db = db.apply_batch(child).await.unwrap(); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_stale_batch_child_before_parent( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { let parent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let child = parent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(child).await.unwrap(); assert!(matches!( db.apply_batch(parent).await, Err(Error::StaleBatch) )); } #[boxed] pub(crate) async fn run_child_root_matches_pending_and_committed( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, Operation: EncodeShared, { // Build the child while the parent is unapplied. let parent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); let pending_child = parent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, db.inactivity_floor_loc()) .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() .append(V::Value::make(2)) .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); assert_eq!(pending_child.root(), committed_child.root()); db.destroy().await.unwrap(); } async fn commit_appends( db: TestKeyless, values: impl IntoIterator, metadata: Option, ) -> TestKeyless where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Tests that don't specifically exercise floor behavior advance the floor to the new // commit location, so pruning up to the last commit works analogously to the pre-floor // semantics. let base_size = *db.bounds().end; let appends_iter: Vec<_> = values.into_iter().collect(); let new_commit_loc = Location::new(base_size + appends_iter.len() as u64); let mut batch = db.new_batch(); for value in appends_iter { batch = batch.append(value); } let merkleized = batch .merkleize(&db, metadata, new_commit_loc) .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, db: TestKeyless, reopen: Reopen>, bounded: BoundedOpen, F>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let initial_root = db.root(); let initial_size = db.bounds().end; let value_a = V::Value::make(1); let value_b = V::Value::make(2); let metadata_a = V::Value::make(3); let db = commit_appends( db, [value_a.clone(), value_b.clone()], Some(metadata_a.clone()), ) .await; let root_before = db.root(); let size_before = db.bounds().end; let commit_before = db.bounds().end - 1; let value_c = V::Value::make(4); let metadata_b = V::Value::make(5); let db = commit_appends(db, [value_c.clone()], Some(metadata_b.clone())).await; assert_ne!(db.root(), root_before); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_b)); drop(db); let db = bounded(context.child("cap"), size_before).await.unwrap(); assert_eq!(db.root(), root_before); assert_eq!(db.bounds().end, size_before); assert_eq!(db.bounds().end - 1, commit_before); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a.clone())); assert_eq!( db.get(Location::new(1)).await.unwrap(), Some(value_a.clone()) ); assert_eq!( db.get(Location::new(2)).await.unwrap(), Some(value_b.clone()) ); assert!( matches!( db.get(Location::new(4)).await, Err(Error::LocationOutOfBounds(_, size)) if size == size_before ), "discarded append should be out of bounds", ); drop(db); let db = reopen(context.child("reopen")).await; assert_eq!(db.root(), root_before); assert_eq!(db.bounds().end, size_before); assert_eq!(db.bounds().end - 1, commit_before); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a)); assert_eq!( db.get(Location::new(1)).await.unwrap(), Some(value_a.clone()) ); assert_eq!( db.get(Location::new(2)).await.unwrap(), Some(value_b.clone()) ); assert!(matches!( db.get(Location::new(4)).await, Err(Error::LocationOutOfBounds(_, size)) if size == size_before )); drop(db); let db = bounded(context.child("cap"), initial_size).await.unwrap(); assert_eq!(db.root(), initial_root); assert_eq!(db.bounds().end, initial_size); assert_eq!(db.get_metadata().await.unwrap(), None); assert!(matches!( db.get(Location::new(1)).await, Err(Error::LocationOutOfBounds(_, size)) if size == initial_size )); drop(db); let db = reopen(context.child("reopen_initial_boundary")).await; assert_eq!(db.root(), initial_root); assert_eq!(db.bounds().end, initial_size); assert_eq!(db.get_metadata().await.unwrap(), None); assert!(matches!( db.get(Location::new(1)).await, Err(Error::LocationOutOfBounds(_, size)) if size == initial_size )); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_bounded_initialization_pruned_target_errors< F: Family, V, C, H, S: Strategy, >( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, bounded: BoundedOpen, F>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let first_start = db.bounds().end; let mut db = commit_appends(db, (0..16).map(V::Value::make), None).await; let mut round = 0u64; loop { round += 1; assert!( round <= 64, "failed to prune enough history for the initialization test" ); db = commit_appends(db, (0..16).map(|i| V::Value::make(round * 100 + i)), None).await; let last_commit = db.bounds().end - 1; db = db.prune(last_commit).await.unwrap(); if db.bounds().start > first_start { break; } } let oldest_retained = db.bounds().start; drop(db); let Err(boundary_err) = bounded(context.child("cap_error"), 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 = reopen(context.child("reopen_boundary")).await; drop(db); let Err(err) = bounded(context.child("cap_error"), first_start).await else { panic!("expected bounded initialization to fail"); }; assert!( matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))), "unexpected bounded initialization error: {err:?}" ); } #[boxed] pub(crate) async fn run_floor_tracking( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Freshly created db has floor = 0. assert_eq!(db.inactivity_floor_loc(), Location::new(0)); // Apply a batch with a declared floor; the db's floor should update. let floor_a = Location::::new(2); let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, floor_a) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.inactivity_floor_loc(), floor_a); // Reopen: floor should survive restart (it's part of the last commit operation). drop(db); let db = reopen(context.child("reopen")).await; assert_eq!(db.inactivity_floor_loc(), floor_a); // Floor may stay the same across a commit (monotonic non-decreasing). let merkleized = db .new_batch() .append(V::Value::make(3)) .merkleize(&db, None, floor_a) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), floor_a); // Floor may advance further. let floor_b = Location::::new(5); let merkleized = db .new_batch() .append(V::Value::make(4)) .merkleize(&db, None, floor_b) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), floor_b); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_floor_regression_rejected( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Advance floor to 3 and commit so the state is durable. let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, Location::new(3)) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.inactivity_floor_loc(), Location::new(3)); // Merkleize rejects a commit with a lower floor. let Err(err) = db .new_batch() .append(V::Value::make(3)) .merkleize(&db, None, Location::new(1)) .await else { panic!("merkleize must reject the invalid floor"); }; assert!( matches!(err, Error::FloorRegressed(new, current) if *new == 1 && *current == 3), "unexpected error: {err:?}" ); db.destroy().await.unwrap(); } #[boxed] pub(crate) async fn run_floor_beyond_commit_loc_rejected( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Batch of 2 appends + 1 commit lands at locations [1..4); commit at 3, total_size = 4. // A floor > 3 (the commit location) is invalid. Even floor == 4 (one past the commit) // is rejected so a subsequent prune cannot remove the last readable commit. let Err(err) = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, Location::new(999)) .await else { panic!("merkleize must reject the invalid floor"); }; assert!( matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 999 && *commit == 3), "unexpected error: {err:?}" ); // Boundary: floor == total_size (= commit_loc + 1) is also rejected. let Err(err) = db .new_batch() .append(V::Value::make(3)) .append(V::Value::make(4)) .merkleize(&db, None, Location::new(4)) .await else { panic!("merkleize must reject the invalid floor"); }; assert!( matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 4 && *commit == 3), "unexpected error: {err:?}" ); } #[boxed] pub(crate) async fn run_bounded_initialization_restores_floor( context: deterministic::Context, db: TestKeyless, bounded: BoundedOpen, F>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // First commit: floor advances to 3 (= commit location). let floor_a = Location::::new(3); let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, floor_a) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let initialization_bound = db.bounds().end; // Second commit: floor advances to 6. let floor_b = Location::::new(6); let merkleized = db .new_batch() .append(V::Value::make(3)) .append(V::Value::make(4)) .merkleize(&db, None, floor_b) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.inactivity_floor_loc(), floor_b); // Reopen at the first commit. The floor should restore to floor_a. drop(db); let db = bounded(context.child("cap"), initialization_bound) .await .unwrap(); assert_eq!(db.inactivity_floor_loc(), floor_a); // Prune is now gated at floor_a: pruning up to the floor works. let db = db.prune(floor_a).await.unwrap(); // Pruning past the floor fails. let beyond = floor_a + 1; let Err(err) = db.prune(beyond).await else { panic!("expected prune to fail"); }; assert!(matches!(err, Error::PruneBeyondMinRequired(_, _))); } /// Floor is embedded in the Commit operation and therefore in the Merkle root: two databases /// with identical appends but different floors must produce different roots. #[boxed] pub(crate) async fn run_floor_changes_root( db_a: TestKeyless, db_b: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { let appends = [V::Value::make(1), V::Value::make(2)]; // db_a commits with floor=0. let mut batch_a = db_a.new_batch(); for v in appends.iter() { batch_a = batch_a.append(v.clone()); } let merkleized = batch_a .merkleize(&db_a, None, Location::new(0)) .await .unwrap(); let db_a = db_a.apply_batch(merkleized).await.unwrap(); // db_b commits the same appends but with floor=3 (= commit location). let mut batch_b = db_b.new_batch(); for v in appends.iter() { batch_b = batch_b.append(v.clone()); } let merkleized = batch_b .merkleize(&db_b, None, Location::new(3)) .await .unwrap(); let db_b = db_b.apply_batch(merkleized).await.unwrap(); assert_ne!(db_a.root(), db_b.root()); db_a.destroy().await.unwrap(); db_b.destroy().await.unwrap(); } /// A floor equal to the commit operation's location is on the tight boundary of acceptance. #[boxed] pub(crate) async fn run_floor_at_commit_loc_accepted( db: TestKeyless, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // 2 appends + 1 commit on top of the initial commit: commit lands at location 3. // floor == 3 (= commit_loc) is the maximum accepted value under the per-commit bound. let commit_loc = Location::::new(3); let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, commit_loc) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); assert_eq!(db.inactivity_floor_loc(), commit_loc); db.destroy().await.unwrap(); } /// Commit, reinitialize at an earlier bound, then verify the restored floor survives another /// reopen. #[boxed] pub(crate) async fn run_bounded_initialization_after_reopen_with_floor< F: Family, V, C, H, S: Strategy, >( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, bounded: BoundedOpen, F>, ) where V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // First commit: 2 appends + commit, floor advances to 3. let floor_a = Location::::new(3); let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .merkleize(&db, None, floor_a) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); let initialization_bound = db.bounds().end; // Second commit: 2 appends + commit, floor advances to 6. let floor_b = Location::::new(6); let merkleized = db .new_batch() .append(V::Value::make(3)) .append(V::Value::make(4)) .merkleize(&db, None, floor_b) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); db.commit().await.unwrap(); // Reopen to simulate a crash after both commits were durable. let db = reopen(context.child("reopen")).await; assert_eq!(db.inactivity_floor_loc(), floor_b); // Reopen at the first commit. The floor should restore to floor_a. drop(db); let db = bounded(context.child("cap"), initialization_bound) .await .unwrap(); assert_eq!(db.inactivity_floor_loc(), floor_a); assert_eq!(db.bounds().end - 1, Location::new(3)); // Bounded initialization is durable. Reopen and confirm the floor again. drop(db); let db = reopen(context.child("reopen").with_attribute("index", 2)).await; assert_eq!(db.inactivity_floor_loc(), floor_a); 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_rejected( db: TestKeyless, ) where F: Family, V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // parent: 1 append + commit at loc 2 with floor=2 (the parent's commit_loc). let parent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, Location::new(2)) .await .unwrap(); // child: 1 append + commit at loc 4 with floor=1 (regressed from parent's floor=2). let Err(err) = parent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, Location::new(1)) .await else { panic!("merkleize must reject the invalid floor"); }; assert!( matches!(err, Error::FloorRegressed(new, prev) if *new == 1 && *prev == 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. #[boxed] pub(crate) async fn run_chained_floor_beyond_commit_rejected( db: TestKeyless, ) where F: Family, V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // parent: 1 append + commit at loc 2 with floor=2. let parent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, Location::new(2)) .await .unwrap(); // child: 1 append + commit at loc 4, declaring floor=5 (one past its commit). The error // names commit location 4, so the bound counts the parent's operations. let Err(err) = parent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, Location::new(5)) .await else { panic!("merkleize must reject the invalid floor"); }; assert!( matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 5 && *commit == 4), "unexpected error: {err:?}" ); } /// 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: the commit is readable, the root /// is preserved, reopen recovers `inactivity_floor_loc` from the sole remaining op, and a /// follow-on batch applies cleanly on top. #[boxed] pub(crate) async fn run_single_commit_live_set( context: deterministic::Context, db: TestKeyless, reopen: Reopen>, ) where F: Family, V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // Initial commit is at loc 0. 3 appends + commit -> commit lands at loc 4. // Declare floor = 4 (= commit_loc), the tight maximum. let metadata = V::Value::make(42); let commit_loc = Location::::new(4); let merkleized = db .new_batch() .append(V::Value::make(1)) .append(V::Value::make(2)) .append(V::Value::make(3)) .merkleize(&db, Some(metadata.clone()), commit_loc) .await .unwrap(); let db = db.apply_batch(merkleized).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.bounds().end - 1, commit_loc); assert_eq!(db.inactivity_floor_loc(), commit_loc); let root_after_commit = db.root(); // Prune at the floor, the maximum prune allowed under the invariant. Pruning is // blob-aligned, so `bounds.start` may not physically advance all the way to `commit_loc`. // What matters semantically is that the floor has authorized 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); // The commit op remains readable; its metadata is intact. assert_eq!(db.get(commit_loc).await.unwrap(), Some(metadata.clone())); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone())); assert_eq!(db.bounds().end - 1, commit_loc); assert_eq!(db.inactivity_floor_loc(), commit_loc); // Prune does not affect the root (documented invariant on `prune`). assert_eq!(db.root(), root_after_commit); // Persist the prune, then verify pruning one past the floor is rejected. 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 restores the inactivity floor from the last commit. let db = reopen(context.child("reopened")).await; let reopened_bounds = db.bounds(); assert_eq!(reopened_bounds.end, commit_loc + 1); assert_eq!(db.bounds().end - 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.clone())); // 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 (= 7) exercises the // ancestor-to-tip floor transition from a single-commit starting point. let next_commit_loc = Location::::new(7); let v5 = V::Value::make(5); let v6 = V::Value::make(6); let merkleized = db .new_batch() .append(v5.clone()) .append(v6.clone()) .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.bounds().end - 1, next_commit_loc); assert_eq!(db.inactivity_floor_loc(), next_commit_loc); // New appends are readable, and the journal still holds the original commit op (it was // not pruned again), so reading it still returns its metadata. assert_eq!(db.get(Location::new(5)).await.unwrap(), Some(v5)); assert_eq!(db.get(Location::new(6)).await.unwrap(), Some(v6)); assert_eq!(db.get(commit_loc).await.unwrap(), Some(metadata)); db.destroy().await.unwrap(); } /// A multi-level chain with strictly-monotonic, within-bounds floors applies cleanly. #[boxed] pub(crate) async fn run_chained_apply_with_valid_floors_succeeds( db: TestKeyless, ) where F: Family, V: ValueEncoding, C: Mutable>, H: Hasher, Operation: EncodeShared, { // parent: 1 append, commit at loc 2, floor = 2. // child: 1 append, commit at loc 4, floor = 3. // grandchild: 1 append, commit at loc 6, floor = 5. let parent = db .new_batch() .append(V::Value::make(1)) .merkleize(&db, None, Location::new(2)) .await .unwrap(); let child = parent .new_batch::() .append(V::Value::make(2)) .merkleize(&db, None, Location::new(3)) .await .unwrap(); let grandchild = child .new_batch::() .append(V::Value::make(3)) .merkleize(&db, None, Location::new(5)) .await .unwrap(); let db = db.apply_batch(grandchild).await.unwrap(); // Grandchild's commit is the last op, and the tip's floor is the database's floor. assert_eq!(db.bounds().end - 1, Location::new(6)); assert_eq!(db.inactivity_floor_loc(), Location::new(5)); db.destroy().await.unwrap(); } }