//! An immutable authenticated db that discards historical operations, retaining only a witness //! for each applied batch. //! //! Mirrors the API of [`crate::qmdb::immutable::Immutable`] (`new_batch -> merkleize -> //! apply_batch -> commit / sync / start_sync`, pipelined batch chains, `StaleBatch` validation) //! but is backed by the peak-only [`crate::merkle::compact`]. Because history is discarded, //! the db has no `get` / `proof` / `bounds` methods. A merkleized batch can prove only its own //! operations, and only until it is applied. Use the full variant for historical proofs. //! //! # Witness journal //! //! The witness journal holds a complete snapshot of every applied batch. [`Db::init`] //! restores a retained applied state within its operation cap. [`Db::prune`] bounds the retained //! history. Initialization restores the db's in-memory state from an entry. The Merkle is rebuilt //! from the stored pinned nodes and operation, and the commit fields are decoded from the //! operation. An entry that cannot rebuild surfaces as [`Error::DataCorrupted`]. Each retained //! witness also serves compact sync at its size without retaining historical operations. //! //! # Inactivity floor //! //! Commits carry an inactivity floor for wire-format compatibility with //! [`crate::qmdb::immutable::Immutable`]: the root is computed over the encoded operation //! sequence, and that sequence must include the same floor to produce the same root as the //! full variant. The floor has no effect on pruning or snapshot rebuilding here; all //! historical in-memory state is discarded whenever a batch is applied. use super::operation::Operation; pub use crate::qmdb::compact::Config; use crate::{ Context, merkle::{Family, Location, Proof, batch, compact as compact_merkle}, qmdb::{ self, Error, any::value::ValueEncoding, chain::{self, Bounds, Commitment}, compact::{batch as compact_batch, witness}, operation::Key, sync::{CompactTarget, Request, Source, source}, }, }; use commonware_codec::{Encode, EncodeShared, Read}; use commonware_cryptography::{Digest, Hasher}; use commonware_macros::boxed; use commonware_parallel::Strategy; use commonware_runtime::Handle; use core::marker::PhantomData; use std::{ collections::BTreeMap, sync::{Arc, Weak}, }; /// An immutable authenticated db that discards historical operations, retaining only a witness /// for each applied batch. pub struct Db where F: Family, E: Context, K: Key, V: ValueEncoding, H: Hasher, Operation: EncodeShared, Operation: Read, { merkle: compact_merkle::Merkle, last_commit_metadata: Option, inactivity_floor_loc: Location, commit_codec_config: as Read>::Cfg, witness: witness::Store, _key: PhantomData, } impl std::fmt::Debug for Db where F: Family, E: Context, K: Key, V: ValueEncoding, H: Hasher, Operation: EncodeShared, Operation: Read, { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Db") .field("size", &self.size()) .field("inactivity_floor_loc", &self.inactivity_floor_loc()) .finish_non_exhaustive() } } /// Result of merkleizing a batch. type MerkleizeResult = Result>, Error>; /// A speculative batch for a compact immutable db. #[allow(clippy::type_complexity)] pub struct UnmerkleizedBatch where F: Family, K: Key, V: ValueEncoding, H: Hasher, Operation: EncodeShared, { merkle_batch: compact_merkle::UnmerkleizedBatch, mutations: BTreeMap, parent: Option>>, base: chain::Commitment, } /// A speculative batch whose root digest has been computed. #[derive(Clone)] pub struct MerkleizedBatch where Operation: EncodeShared, { pub(super) merkle_batch: Arc>, operations: Arc>>, pub(super) commit_metadata: Option, pub(super) parent: Option>, pub(super) bounds: chain::Bounds, } impl MerkleizedBatch where Operation: EncodeShared, { pub(super) fn ancestors(&self) -> impl Iterator> + use { chain::ancestors(self.parent.clone(), |batch| batch.parent.as_ref()) } /// The [`Commitment`] this batch commits to. pub(super) const fn commitment(&self) -> Commitment { self.bounds.tip } /// Return the root digest after this batch is applied. pub const fn root(&self) -> D { self.bounds.tip.root } /// Return the [`Bounds`] of the batch. pub const fn bounds(&self) -> &Bounds { &self.bounds } /// Return the operations this batch appends to the log and the location of the first. #[allow(clippy::type_complexity)] pub fn operations(&self) -> (Location, Arc>>) { (self.bounds.base.size, Arc::clone(&self.operations)) } /// Inclusion proof for the operations returned by [`Self::operations`], anchored at /// this batch's tip. The pair verifies against [`Self::root`] via /// [`crate::qmdb::verify_proof`]. Together with [`Self::pinned_nodes`] they verify via /// [`crate::qmdb::verify_proof_and_pinned_nodes`]. /// /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be /// held. Nodes below the chain are read from `db`'s [Merkle store][crate::merkle::mem::Mem], /// which retains them at least until this batch's changes are applied (applying it or a /// descendant prunes the store to its frontier). /// /// # Errors /// /// Returns [`Error::StaleRead`] if `db` is not at a state of this batch's chain, /// [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or belongs to a /// dropped unapplied ancestor, and [`crate::merkle::Error::Empty`] if the batch has no /// operations (such as a batch from [`Db::to_batch`]). pub fn proof(&self, db: &Db) -> Result, Error> where E: Context, H: Hasher, Operation: Read, { let db = self.bounds.compatible(db, db.commitment())?; let inactive_peaks = F::inactive_peaks(self.bounds.tip.size, self.bounds.inactivity_floor); let hasher = qmdb::hasher::(); self.merkle_batch .range_proof( db.merkle.mem(), &hasher, self.bounds.base.size..self.bounds.tip.size, inactive_peaks, ) .map_err(Into::into) } /// The Merkle frontier at the first operation returned by [`Self::operations`] /// ([`Family::nodes_to_pin`]), which lets a consumer holding only this batch's base rebuild /// compact state and replay the operations. The operations, [`Self::proof`], and pinned /// nodes verify against [`Self::root`] via [`crate::qmdb::verify_proof_and_pinned_nodes`]. /// /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be /// held. Nodes below the chain are read from `db`'s [Merkle store][crate::merkle::mem::Mem], /// which retains them at least until this batch's changes are applied (applying it or a /// descendant prunes the store to its frontier). /// /// # Errors /// /// Returns [`Error::StaleRead`] if `db` is not at a state of this batch's chain, and /// [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or belongs to a /// dropped unapplied ancestor. pub fn pinned_nodes(&self, db: &Db) -> Result, Error> where E: Context, H: Hasher, Operation: Read, { let db = self.bounds.compatible(db, db.commitment())?; let base = db.merkle.mem(); F::nodes_to_pin(self.bounds.base.size) .map(|pos| { self.merkle_batch .get_node(pos) .or_else(|| base.get_node(pos)) .ok_or(crate::merkle::Error::ElementPruned(pos)) }) .collect::, _>>() .map_err(Into::into) } /// Create a new speculative batch with this one as its parent. /// /// Every unapplied ancestor in the chain must be held until the child (or any descendant) is /// merkleized. Otherwise, `merkleize` returns [`Error::StaleBatch`]. pub fn new_batch(self: &Arc) -> UnmerkleizedBatch where H: Hasher, { UnmerkleizedBatch { merkle_batch: compact_merkle::UnmerkleizedBatch::wrap(self.merkle_batch.new_batch()), mutations: BTreeMap::new(), parent: Some(Arc::clone(self)), base: self.commitment(), } } } impl UnmerkleizedBatch where F: Family, K: Key, V: ValueEncoding, H: Hasher, S: Strategy, Operation: EncodeShared, { pub(super) fn new(db: &Db, base: chain::Commitment) -> Self where E: Context, Operation: Read, { Self { merkle_batch: db.merkle.new_batch(), mutations: BTreeMap::new(), parent: None, base, } } /// The database boundary for this batch chain. /// /// A batch created from the database uses its base. A child inherits its parent's `db`. fn db(&self) -> Commitment { self.parent .as_ref() .map_or(self.base, |parent| parent.bounds.db) } pub fn set(mut self, key: K, value: V::Value) -> Self { self.mutations.insert(key, value); self } /// Resolve mutations into operations, merkleize, and return the batch. /// /// `inactivity_floor` is threaded through the commit operation for wire-format parity with /// [`crate::qmdb::immutable::Immutable`]. It must be at least the floor this batch builds on /// (its parent's, or the database's) and at most the batch's commit location /// (`total_size - 1`). These bounds are validated, but the floor does not drive any local /// pruning or retention in this variant. /// /// # Errors /// /// - Returns [`Error::StaleBatch`] if `db` is not at a state of this batch's chain or an /// unapplied ancestor was dropped. /// - Returns [`Error::FloorRegressed`] if `inactivity_floor` is below the floor this batch /// builds on, and [`Error::FloorBeyondSize`] if it is past the commit location. #[tracing::instrument( name = "qmdb.immutable.compact.batch.merkleize", level = "info", skip_all )] pub async fn merkleize( self, db: &Db, metadata: Option, inactivity_floor: Location, ) -> MerkleizeResult where E: Context, Operation: Read, { let live_ancestors: Vec<_> = chain::parent_and_ancestors(self.parent.as_ref(), |parent| parent.ancestors()) .collect(); let boundary = chain::effective_boundary( self.db(), live_ancestors.last().map(|oldest| oldest.bounds.base), ); let ancestors: Vec<_> = live_ancestors .iter() .map(|batch| batch.commitment()) .collect(); let db = chain::merkleizable(db, db.commitment(), boundary, ancestors.iter().copied())?; let mut ops: Vec> = Vec::with_capacity(self.mutations.len() + 1); for (key, value) in self.mutations { ops.push(Operation::Set(key, value)); } ops.push(Operation::Commit(metadata.clone(), inactivity_floor)); let operations = Arc::new(ops); let total_size = self.base.size + operations.len() as u64; chain::validate_merkleize_floor::( self.parent.as_ref().map_or_else( || db.inactivity_floor_loc(), |parent| parent.bounds.inactivity_floor, ), inactivity_floor, total_size - 1, )?; let inactive_peaks = F::inactive_peaks(total_size, inactivity_floor); let (merkle, root) = compact_batch::merkleize_ops::( &db.merkle, self.merkle_batch, Arc::clone(&operations), inactive_peaks, ) .await?; // Hold ancestor batches until their operations and nodes have been captured. drop(live_ancestors); Ok(Arc::new(MerkleizedBatch { merkle_batch: merkle, operations, commit_metadata: metadata, parent: self.parent.as_ref().map(Arc::downgrade), bounds: chain::Bounds { base: self.base, db: boundary, tip: Commitment::new(total_size, root), ancestors, inactivity_floor, }, })) } } impl Db where F: Family, E: Context, K: Key, V: ValueEncoding, H: Hasher, S: Strategy, Operation: EncodeShared, Operation: Read, { /// Initialize from a retained witness. /// `Some(max_size)` selects the latest retained witness for at most `max_size` operations. /// `None` selects the latest retained state. /// /// Fresh storage receives a durable bootstrap commit and witness. #[boxed] pub async fn init( context: E, cfg: Config< as Read>::Cfg, S>, max_size: Option>, ) -> Result> { let Config { strategy, witness: witness_config, commit_codec_config, } = cfg; // Reconstruct and verify the selected witness before exposing any state derived from it. let mut merkle = compact_merkle::Merkle::new(strategy); let (witness, last_commit_op) = witness::init::>( context.child("witness"), witness_config, max_size, &mut merkle, &commit_codec_config, Operation::::Commit(None, Location::new(0)) .encode() .to_vec(), ) .await?; let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else { return Err(Error::DataCorrupted("last operation was not a commit")); }; Ok(Self { merkle, last_commit_metadata, inactivity_floor_loc, commit_codec_config, witness, _key: PhantomData, }) } fn encode_commit_op(metadata: Option, inactivity_floor_loc: Location) -> Vec { Operation::::Commit(metadata, inactivity_floor_loc) .encode() .to_vec() } /// Build a compact db from state fetched by the sync engine. /// /// The imported witness lives only in memory until the first [`Self::apply_batch`], /// [`Self::commit`], [`Self::sync`], or [`Self::start_sync`]. Applying a batch replaces it with /// the newly applied journal checkpoint. A durability method journals it directly. Until one of /// those operations succeeds, prune is rejected. pub(crate) fn init_from_sync( strategy: S, journal: witness::Journal, commit_codec_config: as Read>::Cfg, last_commit_loc: Location, pinned_nodes: Vec, last_commit_op: Operation, ) -> Result> { let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else { return Err(Error::UnexpectedData(last_commit_loc)); }; witness::validate_inactivity_floor(inactivity_floor_loc, last_commit_loc)?; let op_bytes = Self::encode_commit_op(last_commit_metadata.clone(), inactivity_floor_loc); let mut merkle = compact_merkle::Merkle::from_compact_state(strategy, last_commit_loc, pinned_nodes)?; let hasher = qmdb::hasher::(); merkle.append_leaf(&hasher, &op_bytes)?; let imported = witness::build_witness::(&merkle, inactivity_floor_loc, op_bytes)?; merkle.prune_to_frontier(); let witness = witness::Store::from_import(journal, imported); Ok(Self { merkle, last_commit_metadata, inactivity_floor_loc, commit_codec_config, witness, _key: PhantomData, }) } /// Return the root of the db. pub const fn root(&self) -> H::Digest { self.witness.tip().root } /// Return the inactivity floor declared by the last committed batch. pub const fn inactivity_floor_loc(&self) -> Location { self.inactivity_floor_loc } /// Return the location of the next operation appended to this db. pub const fn size(&self) -> Location { self.witness.tip().size() } /// Get the metadata associated with the last commit. pub fn get_metadata(&self) -> Option { self.last_commit_metadata.clone() } /// Return the compact-sync target described by the current witness. /// /// This reflects the most recently applied batch. The target remains non-durable until a /// covering [`Self::commit`], [`Self::sync`], or [`Self::start_sync`] completes. pub const fn target(&self) -> CompactTarget { self.witness.tip().target() } /// The [`Commitment`] for the database's current state. pub(crate) const fn commitment(&self) -> chain::Commitment { chain::Commitment::new(self.size(), self.root()) } /// Create a new speculative batch of operations with this database as its parent. pub fn new_batch(&self) -> UnmerkleizedBatch { UnmerkleizedBatch::new(self, self.commitment()) } /// Create an owned merkleized batch representing the current applied state. pub fn to_batch(&self) -> Arc> { Arc::new(MerkleizedBatch { merkle_batch: self.merkle.to_batch(), operations: Arc::new(Vec::new()), commit_metadata: self.last_commit_metadata.clone(), 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: &MerkleizedBatch, ) -> Result<(), Error> { batch.bounds.validate_apply_to(self.commitment()) } /// Apply a merkleized batch to the database. /// /// The state is updated in memory and appended to the witness journal. Call [`Self::commit`] /// or [`Self::sync`], or await the handle returned by [`Self::start_sync`], to make the /// applied state durable. /// /// # Errors /// /// Returns [`Error::StaleBatch`] if the database is at neither the batch's recorded database /// boundary nor an ancestor's tip. #[tracing::instrument( name = "qmdb.immutable.compact.db.apply_batch", level = "info", skip_all )] pub async fn apply_batch( mut self, batch: Arc>, ) -> Result> { self.validate_batch(&batch)?; self.merkle.apply_batch(&batch.merkle_batch)?; self.last_commit_metadata = batch.commit_metadata.clone(); self.inactivity_floor_loc = batch.bounds.inactivity_floor; let last_commit_metadata = self.last_commit_metadata.clone(); let inactivity_floor_loc = self.inactivity_floor_loc; self.witness = self .witness .apply::(&mut self.merkle, inactivity_floor_loc, || { Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc) }) .await?; assert_eq!(self.commitment(), batch.bounds.tip); Ok(self) } /// Begin durably persisting the current db state to disk. /// /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit], /// plus a best-effort attempt to bound the recovery needed on reopen. 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 and the next durability /// operation. #[tracing::instrument( name = "qmdb.immutable.compact.db.start_sync", level = "info", skip_all )] pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error> { let last_commit_metadata = self.last_commit_metadata.clone(); let inactivity_floor_loc = self.inactivity_floor_loc; let handle; (self.witness, handle) = self .witness .start_sync::(&mut self.merkle, inactivity_floor_loc, || { Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc) }) .await?; Ok((self, handle)) } /// Durably persist the current db state to disk. This is faster than [`Self::sync`] but /// reopen may need to replay the witness journal's tail to recover. #[tracing::instrument(name = "qmdb.immutable.compact.db.commit", level = "info", skip_all)] pub async fn commit(mut self) -> Result> { let last_commit_metadata = self.last_commit_metadata.clone(); let inactivity_floor_loc = self.inactivity_floor_loc; self.witness = self .witness .commit::(&mut self.merkle, inactivity_floor_loc, || { Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc) }) .await?; Ok(self) } /// Durably persist the current db state to disk, also persisting journal metadata to /// minimize recovery work on reopen. #[tracing::instrument(name = "qmdb.immutable.compact.db.sync", level = "info", skip_all)] pub async fn sync(mut self) -> Result> { let last_commit_metadata = self.last_commit_metadata.clone(); let inactivity_floor_loc = self.inactivity_floor_loc; self.witness = self .witness .sync::(&mut self.merkle, inactivity_floor_loc, || { Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc) }) .await?; Ok(self) } /// Drop witnesses for commits with fewer than `pruning_boundary` operations. Some witness below /// the boundary may survive. /// /// Bounded initialization and compact sync can only use witnesses that survive. The current /// commit's witness always survives. The prune is made durable before this method returns. /// /// # Errors /// /// Fails if a compact-sync import has not yet been applied to the witness journal. #[tracing::instrument(name = "qmdb.immutable.compact.db.prune", level = "info", skip_all)] pub async fn prune(mut self, pruning_boundary: Location) -> Result> { self.witness = self.witness.prune(pruning_boundary).await?; Ok(self) } /// Destroy all persisted state associated with this database. #[boxed] pub async fn destroy(self) -> Result<(), Error> { self.witness.destroy().await?; Ok(()) } } impl Source for Db where F: Family, E: Context, K: Key, V: ValueEncoding, H: Hasher, Operation: EncodeShared + Read, S: Strategy, { type Family = F; type Digest = H::Digest; type Op = Operation; type Error = qmdb::Error; async fn serve(&self, request: Request) -> source::Result { let response = self .witness .compact_state::( self.merkle.strategy(), &self.commit_codec_config, request, ) .await?; Ok((response, None)) } } #[cfg(test)] mod tests { use super::*; use crate::{ journal::contiguous::{Contiguous as _, fixed, variable::Config as JournalConfig}, merkle::{mmb, mmr}, metadata::{Config as MetadataConfig, Metadata}, qmdb::{ any::value::FixedEncoding, compact::witness, verify_proof, verify_proof_and_pinned_nodes, }, }; use commonware_cryptography::{Sha256, sha256::Digest}; use commonware_macros::test_traced; use commonware_parallel::Sequential; use commonware_runtime::{ Blob as _, Runner as _, Storage as _, Supervisor as _, buffer::paged::CacheRef, deterministic, mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs}, }; use commonware_utils::{NZU16, NZU64, NZUsize, sequence::VecU64}; use core::future::Future; use std::num::{NonZeroU16, NonZeroUsize}; type TestDb = Db, Sha256, Sequential>; const WITNESS_PAGE_SIZE: NonZeroU16 = NZU16!(77); const WITNESS_PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9); fn witness_config(partition: &str) -> JournalConfig<()> { JournalConfig { partition: format!("{partition}-witness"), items_per_section: NZU64!(64), compression: None, codec_config: (), page_cache: CacheRef::new(WITNESS_PAGE_SIZE, WITNESS_PAGE_CACHE_SIZE), write_buffer: NZUsize!(1024), replay_buffer: NZUsize!(1024), } } async fn open_db(context: deterministic::Context, partition: &str) -> TestDb { let witness_cfg = witness_config(partition); let cfg = Config { strategy: Sequential, witness: witness_cfg, commit_codec_config: (), }; Db::init(context, cfg, None).await.unwrap() } async fn compact_merkleize_foreign_db_inner(context: deterministic::Context) { let db = open_db::(context.child("db"), "immutable-merkleize-foreign-db").await; let foreign = open_db::( context.child("foreign"), "immutable-merkleize-foreign-db-foreign", ) .await; let batch = db .new_batch() .set(Sha256::fill(1), Sha256::fill(11)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap(); let batch = foreign .new_batch() .set(Sha256::fill(2), Sha256::fill(99)) .merkleize(&foreign, None, Location::new(0)) .await .unwrap(); let foreign = foreign.apply_batch(batch).await.unwrap(); assert_eq!(db.size(), foreign.size()); assert_ne!(db.root(), foreign.root()); let batch = db.new_batch().set(Sha256::fill(3), Sha256::fill(22)); assert!(matches!( batch.merkleize(&foreign, None, Location::new(0)).await, Err(Error::StaleBatch) )); let parent = db .new_batch() .set(Sha256::fill(4), Sha256::fill(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::fill(5), Sha256::fill(44)); assert!(matches!( child.merkleize(&foreign, None, Location::new(0)).await, Err(Error::StaleBatch) )); db.destroy().await.unwrap(); foreign.destroy().await.unwrap(); } async fn compact_merkleize_stale_sibling_inner(context: deterministic::Context) { let db = open_db::(context.child("db"), "immutable-merkleize-stale-sibling").await; let direct = db.new_batch().set(Sha256::fill(1), Sha256::fill(11)); let sibling = db .new_batch() .set(Sha256::fill(2), Sha256::fill(22)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling).await.unwrap(); assert!(matches!( direct.merkleize(&db, None, Location::new(0)).await, Err(Error::StaleBatch) )); let parent = db .new_batch() .set(Sha256::fill(3), Sha256::fill(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::fill(4), Sha256::fill(44)); let sibling = db .new_batch() .set(Sha256::fill(5), Sha256::fill(55)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling).await.unwrap(); assert!(matches!( child.merkleize(&db, None, Location::new(0)).await, Err(Error::StaleBatch) )); db.destroy().await.unwrap(); } async fn compact_merkleize_ancestor_states_inner(context: deterministic::Context) { let db = open_db::(context.child("db"), "immutable-merkleize-ancestor-states").await; let grandparent = db .new_batch() .set(Sha256::fill(1), Sha256::fill(11)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let parent = grandparent .new_batch::() .set(Sha256::fill(2), Sha256::fill(22)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let pending = parent .new_batch::() .set(Sha256::fill(3), Sha256::fill(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(Arc::clone(&grandparent)).await.unwrap(); let applied = parent .new_batch::() .set(Sha256::fill(3), Sha256::fill(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert_eq!(pending.root(), applied.root()); drop(grandparent); let retired = parent .new_batch::() .set(Sha256::fill(3), Sha256::fill(33)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); assert_eq!(retired.bounds().db, db.commitment()); assert_eq!(pending.root(), retired.root()); let child = parent .new_batch::() .set(Sha256::fill(3), Sha256::fill(33)); 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 expected_root = child.root(); let db = db.apply_batch(child).await.unwrap(); assert_eq!(db.root(), expected_root); db.destroy().await.unwrap(); } #[test_traced] fn test_compact_merkleize_foreign_db_mmr() { deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::); } #[test_traced] fn test_compact_merkleize_foreign_db_mmb() { deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::); } #[test_traced] fn test_compact_merkleize_stale_sibling_mmr() { deterministic::Runner::default() .start(compact_merkleize_stale_sibling_inner::); } #[test_traced] fn test_compact_merkleize_stale_sibling_mmb() { deterministic::Runner::default() .start(compact_merkleize_stale_sibling_inner::); } #[test_traced] fn test_compact_merkleize_ancestor_states_mmr() { deterministic::Runner::default() .start(compact_merkleize_ancestor_states_inner::); } #[test_traced] fn test_compact_merkleize_ancestor_states_mmb() { deterministic::Runner::default() .start(compact_merkleize_ancestor_states_inner::); } async fn open_bounded( context: deterministic::Context, witness_cfg: JournalConfig<()>, cap: Location, ) -> Result, Error> { let cfg = Config { strategy: Sequential, witness: witness_cfg, commit_codec_config: (), }; Db::init(context, cfg, Some(cap)).await } /// A batch's proof and pinned nodes return `StaleRead` after a bounded initialization and /// another batch bring the database to the batch's base size with a different root. async fn proof_refused_after_off_chain_reopen(context: deterministic::Context) { let db = open_db::(context.child("db"), "off-chain-proof").await; let mut seed = db.new_batch(); for value in 1..=6 { seed = seed.set(Sha256::fill(value), Sha256::fill(value)); } let seed = seed.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(seed).await.unwrap(); let db = db.sync().await.unwrap(); let batch = db .new_batch() .set(Sha256::fill(100), Sha256::fill(100)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let (start, ops) = batch.operations(); let root = batch.root(); let original_proof = batch.proof(&db).unwrap(); let original_pins = batch.pinned_nodes(&db).unwrap(); assert!(verify_proof_and_pinned_nodes::( &original_proof, start, &ops, &original_pins, &root )); drop(db); let db = open_bounded::( context.child("reopen"), witness_config("off-chain-proof"), Location::new(1), ) .await .unwrap(); let mut other = db.new_batch(); for value in 11..=16 { other = other.set(Sha256::fill(value), Sha256::fill(value)); } let other = other.merkleize(&db, None, Location::new(0)).await.unwrap(); let db = db.apply_batch(other).await.unwrap(); assert_eq!(db.size(), start); assert!(matches!( batch .new_batch::() .merkleize(&db, None, Location::new(0)) .await, Err(Error::StaleBatch) )); let proof = batch.proof(&db); let pins = batch.pinned_nodes(&db); assert!( matches!(proof, Err(Error::StaleRead)), "proof must return StaleRead at a state outside the batch's chain" ); assert!( matches!(pins, Err(Error::StaleRead)), "pinned nodes must return StaleRead at a state outside the batch's chain" ); } #[test] fn compact_mmr_proof_refused_after_off_chain_reopen() { deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::); } #[test] fn compact_mmb_proof_refused_after_off_chain_reopen() { deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::); } /// Batch artifacts (operations, range proof, pinned frontier) verify against the batch root, /// survive applying and dropping ancestors, and return `ElementPruned` once the batch itself is /// applied and the compact store is pruned past them. async fn compact_operations_and_proof_inner(context: deterministic::Context) { let db = open_db::(context.child("db"), "immutable-operations-and-proof").await; let value = |key: u8| Sha256::fill(key.wrapping_add(100)); // Seed committed state so the chain below starts above a pruned frontier. let mut seed = db.new_batch(); for key in 1u8..=6 { seed = seed.set(Sha256::fill(key), value(key)); } let seed = seed .merkleize(&db, Some(Sha256::fill(7)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(seed).await.unwrap().sync().await.unwrap(); let floor = db.size(); assert_eq!(floor, Location::new(8)); // A batch from `to_batch` has no operations to prove. assert!(matches!( db.to_batch().proof(&db), Err(Error::Merkle(crate::merkle::Error::Empty)) )); // A two-deep unapplied chain. The child's artifacts read the parent's nodes through the // chain. let mut parent = db.new_batch(); for key in 8u8..=12 { parent = parent.set(Sha256::fill(key), value(key)); } let parent = parent .merkleize(&db, Some(Sha256::fill(13)), floor) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::fill(14), value(14)) .merkleize(&db, Some(Sha256::fill(15)), floor) .await .unwrap(); // The operation suffix is the batch's own sets plus its commit, handed out zero-copy. let (child_start, child_ops) = child.operations(); let (_, child_ops_again) = child.operations(); let child_end = child.bounds().tip.size; assert!(Arc::ptr_eq(&child_ops, &child_ops_again)); assert_eq!(child_start, parent.bounds().tip.size); assert_eq!(*child_start + child_ops.len() as u64, *child_end); assert_eq!(child_end, Location::new(16)); assert!(matches!( child_ops.as_slice(), [Operation::Set(key, set_value), Operation::Commit(Some(metadata), operation_floor)] if key == &Sha256::fill(14) && set_value == &value(14) && metadata == &Sha256::fill(15) && operation_floor == &floor )); // The proof is anchored at the batch tip and verifies with or without the pins. let child_root = child.root(); let child_proof = child.proof(&db).unwrap(); let child_pins = child.pinned_nodes(&db).unwrap(); assert_eq!(child_proof.leaves, child_end); assert_eq!( child_proof.inactive_peaks, F::inactive_peaks(child_end, floor), ); assert!(verify_proof::( &child_proof, child_start, &child_ops, &child_root )); assert!(verify_proof_and_pinned_nodes::( &child_proof, child_start, &child_ops, &child_pins, &child_root )); // The pins are order-sensitive. assert!(child_pins.len() > 1); let mut reordered_child_pins = child_pins.clone(); reordered_child_pins.swap(0, 1); assert!(!verify_proof_and_pinned_nodes::( &child_proof, child_start, &child_ops, &reordered_child_pins, &child_root )); // Pipelined consumer: applying the parent prunes the store to the parent's tip, which is // exactly the frontier the child's base pins, so the artifacts survive dropping the // parent. let db = db.apply_batch(parent).await.unwrap(); let child_proof_after = child.proof(&db).unwrap(); assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins); assert!(verify_proof_and_pinned_nodes::( &child_proof_after, child_start, &child_ops, &child_pins, &child_root )); let applied_start = db.size(); let db = db.apply_batch(child).await.unwrap(); let child_range = applied_start..db.size(); assert_eq!(child_range, child_start..child_end); // A commit-only batch proves exactly its commit operation. let commit_floor = db.size(); let commit_only = db .new_batch() .merkleize(&db, Some(Sha256::fill(16)), commit_floor) .await .unwrap(); let (commit_start, commit_ops) = commit_only.operations(); let commit_end = commit_only.bounds().tip.size; let commit_root = commit_only.root(); let commit_proof = commit_only.proof(&db).unwrap(); let commit_pins = commit_only.pinned_nodes(&db).unwrap(); assert_eq!(commit_start, commit_floor); assert!(matches!( commit_ops.as_slice(), [Operation::Commit(Some(metadata), operation_floor)] if metadata == &Sha256::fill(16) && operation_floor == &commit_floor )); assert_eq!(*commit_start + commit_ops.len() as u64, *commit_end); assert_eq!(commit_proof.leaves, commit_end); assert_eq!( commit_proof.inactive_peaks, F::inactive_peaks(commit_end, commit_floor) ); assert!(verify_proof_and_pinned_nodes::( &commit_proof, commit_start, &commit_ops, &commit_pins, &commit_root )); let applied_start = db.size(); let db = db.apply_batch(commit_only).await.unwrap(); let commit_range = applied_start..db.size(); assert_eq!(commit_range, commit_start..commit_end); let db = db.sync().await.unwrap(); // Applying a batch built mid-mountain prunes its own artifacts. The accessors return // `ElementPruned` rather than a proof that fails to verify, while a child merkleized // before the apply still finds its base frontier in the store. let late_parent = db .new_batch() .set(Sha256::fill(17), value(17)) .merkleize(&db, Some(Sha256::fill(18)), db.size()) .await .unwrap(); let late = late_parent .new_batch::() .set(Sha256::fill(19), value(19)) .merkleize(&db, Some(Sha256::fill(20)), db.size()) .await .unwrap(); let db = db.apply_batch(Arc::clone(&late_parent)).await.unwrap(); assert!(matches!( late_parent.proof(&db), Err(Error::Merkle(crate::merkle::Error::ElementPruned(_))) )); assert!(matches!( late_parent.pinned_nodes(&db), Err(Error::Merkle(crate::merkle::Error::ElementPruned(_))) )); drop(late_parent); let (late_start, late_ops) = late.operations(); let late_root = late.root(); let late_proof = late.proof(&db).unwrap(); let late_pins = late.pinned_nodes(&db).unwrap(); assert!(verify_proof_and_pinned_nodes::( &late_proof, late_start, &late_ops, &late_pins, &late_root )); let db = db.apply_batch(late).await.unwrap(); db.destroy().await.unwrap(); } #[test_traced] fn test_compact_operations_and_proof_mmr() { deterministic::Runner::default().start(compact_operations_and_proof_inner::); } #[test_traced] fn test_compact_operations_and_proof_mmb() { deterministic::Runner::default().start(compact_operations_and_proof_inner::); } /// Open the persisted witness journal directly so tests can corrupt the tip entry. async fn open_witness_journal( context: deterministic::Context, partition: &str, ) -> witness::Journal { let cfg = witness_config(partition); witness::Journal::init(context, cfg).await.unwrap() } /// A compact db over a delayed-sync storage backend. type DelayedDb = Db< mmr::Family, DelayedSyncContext, Digest, FixedEncoding, Sha256, Sequential, >; /// Open a [DelayedDb] whose blob syncs park on `pending`. /// /// Init durably persists the bootstrap witness, so while syncs park the returned future /// must be driven with `drive_pending_syncs` (or the mock unblocked first). fn open_delayed_db( context: deterministic::Context, partition: &str, pending: &PendingSyncs, ) -> impl Future>> { let witness_cfg = witness_config(partition); let cfg = Config { strategy: Sequential, witness: witness_cfg, commit_codec_config: (), }; let context = DelayedSyncContext { inner: context, pending: pending.clone(), }; DelayedDb::init(context, cfg, None) } /// Apply a single-key batch writing `key -> value` with `metadata`. async fn apply_set(db: DelayedDb, key: Digest, value: Digest, metadata: Digest) -> DelayedDb { let floor = db.inactivity_floor_loc(); let batch = db .new_batch() .set(key, value) .merkleize(&db, Some(metadata), floor) .await .unwrap(); db.apply_batch(batch).await.unwrap() } /// State persisted via an awaited start_sync handle is recovered on reopen. #[test_traced] fn test_compact_start_sync_recovery() { deterministic::Runner::default().start(|ctx| async move { let partition = "immutable-start-sync-recovery"; let pending = PendingSyncs::default(); pending.unblock(); let mut db = open_delayed_db(ctx.child("delayed"), partition, &pending) .await .unwrap(); let metadata = Sha256::fill(9u8); db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), metadata).await; let handle; (db, handle) = db.start_sync().await.unwrap(); handle.await.unwrap(); let root = db.root(); drop(db); let db = open_delayed_db(ctx.child("reopen"), partition, &pending) .await .unwrap(); assert_eq!(db.root(), root); assert_eq!(db.get_metadata(), Some(metadata)); db.destroy().await.unwrap(); }); } /// A sync begun by `start_sync` that fails in flight surfaces the error through both the /// returned handle and the next durability operation, even when that operation has nothing /// new to persist. #[test_traced] fn test_compact_start_sync_failure_propagates() { deterministic::Runner::default().start(|ctx| async move { let pending = PendingSyncs::default(); pending.unblock(); let mut db = open_delayed_db(ctx.child("delayed"), "immutable-start-sync-fail", &pending) .await .unwrap(); db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), Sha256::fill(9u8)).await; // Arm all future syncs to resolve to an injected error. pending.arm_fail(); let handle; (db, handle) = db.start_sync().await.unwrap(); assert!( handle.await.is_err(), "the sync handle surfaces the failure" ); let starts_before = pending.starts(); // The witness entry was already appended, so this commit has nothing to stage. // It must still observe the retained failure rather than no-op. assert!( db.commit().await.is_err(), "the next durability op surfaces the failed in-flight sync" ); assert_eq!( pending.starts(), starts_before, "the surfaced error is the retained failure, not a fresh sync's" ); }); } #[test_traced("INFO")] fn test_compact_stale_batch_rejected() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-stale").await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let value1 = Sha256::fill(10u8); let value2 = Sha256::fill(20u8); let batch_a = db .new_batch() .set(key1, value1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let batch_b = db .new_batch() .set(key2, value2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let stale = db.new_batch().set(key2, value2); let expected_root = batch_a.root(); let db = db.apply_batch(batch_a).await.unwrap(); assert_eq!(db.root(), expected_root); // After the apply, merkleizing a batch built on the pre-apply state and applying the // merkleized sibling both return `StaleBatch`. assert!(matches!( stale.merkleize(&db, None, Location::new(0)).await, Err(Error::StaleBatch) )); assert!(matches!( db.apply_batch(batch_b).await, Err(Error::StaleBatch) )); }); } #[test_traced("INFO")] fn test_compact_delayed_merkleize_after_ancestor_apply() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-delayed-child").await; let key1 = Sha256::hash(&[&[1]]); let key2 = Sha256::hash(&[&[2]]); let key3 = Sha256::hash(&[&[3]]); let value1 = Sha256::fill(10u8); let value2 = Sha256::fill(20u8); let value3 = Sha256::fill(30u8); let a = db .new_batch() .set(key1, value1) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let b = a .new_batch::() .set(key2, value2) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let c = b.new_batch::().set(key3, value3); let db = db.apply_batch(a).await.unwrap(); let c = c.merkleize(&db, None, Location::new(0)).await.unwrap(); let expected_root = c.root(); let db = db.apply_batch(c).await.unwrap(); assert_eq!(db.root(), expected_root); }); } /// `to_batch()` reflects the current applied state before it becomes durable. #[test_traced("INFO")] fn test_compact_to_batch_reflects_live_state() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-to-batch-live").await; let pre_apply_root = db.root(); let pre_snapshot = db.to_batch(); assert_eq!( pre_snapshot.root(), pre_apply_root, "snapshot before any mutation should match the live root" ); let key = Sha256::hash(&[&[1]]); let value = Sha256::fill(10u8); let batch = db .new_batch() .set(key, value) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap(); // Observe the applied state before making it durable. let live_root = db.root(); assert_ne!( live_root, pre_apply_root, "applying a non-empty batch must change the live root" ); let snapshot = db.to_batch(); assert_eq!( snapshot.root(), live_root, "to_batch().root() must match the live db.root() even before sync" ); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_stale_batch_chained() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-chained-stale").await; let common_parent = db .new_batch() .set(Sha256::hash(&[&[10]]), Sha256::fill(10u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let sibling_a = common_parent .new_batch::() .set(Sha256::hash(&[&[11]]), Sha256::fill(11u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let sibling_b = common_parent .new_batch::() .set(Sha256::hash(&[&[12]]), Sha256::fill(12u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(sibling_a).await.unwrap(); assert!(matches!( db.validate_batch(&sibling_b), Err(Error::StaleBatch) )); let parent_a = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let parent_b = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child_b = parent_b .new_batch::() .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(parent_a).await.unwrap(); assert!(matches!( db.validate_batch(&child_b), Err(Error::StaleBatch) )); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_stale_parent_after_child_applied() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-child-before-parent").await; let parent = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(child).await.unwrap(); assert!(matches!( db.apply_batch(parent).await, Err(Error::StaleBatch) )); }); } #[test_traced("INFO")] fn test_compact_sequential_commit_parent_then_child() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-parent-child").await; let parent = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let expected_root = child.root(); let db = db.apply_batch(parent).await.unwrap(); let db = db.apply_batch(child).await.unwrap().sync().await.unwrap(); assert_eq!(db.root(), expected_root); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_floor_regressed() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-floor-regressed").await; let advance_floor = db.new_batch().set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)); let advance_floor = advance_floor .merkleize(&db, None, Location::new(1)) .await .unwrap(); let db = db.apply_batch(advance_floor).await.unwrap(); let db = db.sync().await.unwrap(); let regressed = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await; assert!(matches!( regressed, Err(Error::FloorRegressed(new, current)) if new == Location::new(0) && current == Location::new(1) )); }); } // A chained batch whose tip floor is below its parent's floor must be rejected: // the parent's Commit participates in the per-commit monotonicity invariant even // before it is applied. #[test_traced("INFO")] fn test_compact_chained_floor_regression() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-regressed-ancestor-floor") .await; let parent = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(1)) .await .unwrap(); let child = parent .new_batch::() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await; assert!(matches!( child, Err(Error::FloorRegressed(new, prev)) if new == Location::new(0) && prev == Location::new(1) )); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_restores_commit_metadata_and_floor() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-rewind-meta").await; let k1 = Sha256::hash(&[&[1]]); let v1 = Sha256::fill(11u8); let meta1 = Sha256::fill(0xaa); let floor1 = Location::new(0); let batch = db .new_batch() .set(k1, v1) .merkleize(&db, Some(meta1), floor1) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let root_after_first = db.root(); let size_after_first = db.size(); let k2 = Sha256::hash(&[&[2]]); let v2 = Sha256::fill(22u8); let meta2 = Sha256::fill(0xbb); // Advance the floor to the commit of the first batch (loc 1). let floor2 = Location::new(1); let batch = db .new_batch() .set(k2, v2) .merkleize(&db, Some(meta2), floor2) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); assert_eq!(db.get_metadata(), Some(meta2)); assert_eq!(db.inactivity_floor_loc(), floor2); let db = { _ = db.sync().await.unwrap(); open_bounded::( context.child("cap"), witness_config("immutable-rewind-meta"), size_after_first, ) .await } .unwrap(); assert_eq!(db.root(), root_after_first); assert_eq!(db.get_metadata(), Some(meta1)); assert_eq!(db.inactivity_floor_loc(), floor1); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_persists_across_reopen() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-rewind-reopen"; let meta1 = Sha256::fill(0xaa); let floor1 = Location::new(0); let meta2 = Sha256::fill(0xbb); let floor2 = Location::new(1); let root_after_first = { let db = open_db::(context.child("first"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8)) .merkleize(&db, Some(meta1), floor1) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let root = db.root(); let size_after_first = db.size(); let batch = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8)) .merkleize(&db, Some(meta2), floor2) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let _db = { _ = db.sync().await.unwrap(); open_bounded::( context.child("cap"), witness_config(partition), size_after_first, ) .await } .unwrap(); root }; let db = open_db::(context.child("second"), partition).await; assert_eq!(db.root(), root_after_first); assert_eq!(db.get_metadata(), Some(meta1)); assert_eq!(db.inactivity_floor_loc(), floor1); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_commit_persists_across_reopen() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-commit-reopen"; let meta1 = Sha256::fill(0xaa); let meta2 = Sha256::fill(0xbb); let root_after_second = { let db = open_db::(context.child("first"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8)) .merkleize(&db, Some(meta1), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap(); let batch = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8)) .merkleize(&db, Some(meta2), Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap(); db.root() }; // Reopen recovers the committed tip even though the journal was never synced. let db = open_db::(context.child("second"), partition).await; assert_eq!(db.root(), root_after_second); assert_eq!(db.get_metadata(), Some(meta2)); assert_eq!(db.inactivity_floor_loc(), Location::new(1)); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_to_committed_entry_after_reopen() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-commit-rewind-reopen"; let meta1 = Sha256::fill(0xaa); let meta2 = Sha256::fill(0xbb); let (root_a, size_a) = { let db = open_db::(context.child("first"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8)) .merkleize(&db, Some(meta1), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap(); let root_a = db.root(); let size_a = db.size(); let batch = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8)) .merkleize(&db, Some(meta2), Location::new(1)) .await .unwrap(); let _db = db.apply_batch(batch).await.unwrap().commit().await.unwrap(); (root_a, size_a) }; // Both committed witnesses survive the crash. Reopen recovers the tip, and the earlier // commit remains a valid initialization target. let db = open_db::(context.child("second"), partition).await; let db = { _ = db.sync().await.unwrap(); open_bounded::(context.child("cap"), witness_config(partition), size_a) .await } .unwrap(); assert_eq!(db.root(), root_a); assert_eq!(db.get_metadata(), Some(meta1)); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_sync_after_commit() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-sync-after-commit"; let meta = Sha256::fill(0xaa); let root = { let db = open_db::(context.child("first"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8)) .merkleize(&db, Some(meta), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap(); // Commit persists witness data. Sync must also persist recovery metadata. let db = db.commit().await.unwrap(); let db = db.sync().await.unwrap(); db.root() }; // Check the watermark before reopening can rebuild the offsets journal. let metadata = Metadata::<_, u64, VecU64>::init( context.child("checkpoint"), MetadataConfig { partition: format!("{partition}-witness_offsets-metadata"), codec_config: (), }, ) .await .unwrap(); // Key 3 records the durable prefix: the bootstrap witness and the applied batch. assert_eq!(metadata.get(&3).copied().map(u64::from), Some(2)); drop(metadata); let db = open_db::(context.child("second"), partition).await; assert_eq!(db.root(), root); assert_eq!(db.get_metadata(), Some(meta)); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_reopen_rejects_tampered_witness() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-witness-tamper"; let db = open_db::(context.child("db"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8)) .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); drop(db); // Corrupt the entry structurally. An extra pinned node cannot rebuild the Merkle. let journal = open_witness_journal(context.child("tamper"), partition).await; let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await; pinned_nodes.push(Sha256::fill(0xff)); witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await; let cfg = Config { strategy: Sequential, witness: witness_config(partition), commit_codec_config: (), }; let reopened = TestDb::::init(context.child("reopen_witness"), cfg, None).await; assert!(matches!(reopened, Err(Error::DataCorrupted(_)))); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_rejects_corrupt_target_entry() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-corrupt-rewind-target"; let db = open_db::(context.child("db"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let initialization_bound = db.target().size; let batch = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let tip_target = db.target(); drop(db); // Corrupt the initialization target's entry (the journal holds bootstrap, target, tip). let mut journal = open_witness_journal(context.child("corrupt"), partition).await; journal = witness::tests::corrupt_entry(journal, 1, |entry| { entry.pinned_nodes.push(Sha256::fill(0xff)); }) .await; drop(journal); // The tip entry is intact, so reopen succeeds. let cfg = Config { strategy: Sequential, witness: witness_config(partition), commit_codec_config: (), }; let reopened = TestDb::::init(context.child("reopen"), cfg, None) .await .unwrap(); assert_eq!(reopened.target(), tip_target); // The corrupt entry fails the recovery before any truncation. assert!(matches!( { drop(reopened); open_bounded::( context.child("cap"), witness_config(partition), initialization_bound, ) .await }, Err(Error::DataCorrupted(_)) )); // The newer history survives: reopen still lands on the original tip. let cfg = Config { strategy: Sequential, witness: witness_config(partition), commit_codec_config: (), }; let reopened = TestDb::::init(context.child("reopen2"), cfg, None) .await .unwrap(); assert_eq!(reopened.target(), tip_target); reopened.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_reopen_rejects_interrupted_import() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-interrupted-import"; let db = open_db::(context.child("db"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8)) .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); drop(db); // Simulate a crash between an import's journal clear and its entry append: the // journal is empty but its size is nonzero. let journal = open_witness_journal(context.child("clear"), partition).await; let size = journal.size(); let journal = journal.clear_to_size(size.max(1)).await.unwrap(); drop(journal); // Reopen must fail rather than bootstrap a fresh db. let cfg = Config { strategy: Sequential, witness: witness_config(partition), commit_codec_config: (), }; let reopened = TestDb::::init(context.child("reopen_witness"), cfg, None).await; assert!(matches!( reopened, Err(Error::DataCorrupted("witness journal has no tip")) )); }); } #[test_traced("INFO")] fn test_compact_reopen_rejects_commit_floor_beyond_tip() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-invalid-persisted-floor"; let db = open_db::(context.child("db"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8)) .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); drop(db); let oversized_floor = Location::new(10); // Overwrite the persisted commit op with a floor beyond its own commit location. let journal = open_witness_journal(context.child("tamper"), partition).await; let (_, size, pinned_nodes) = witness::tests::tip(&journal).await; let bad_op = Operation::>::Commit( Some(Sha256::fill(0xaa)), oversized_floor, ) .encode() .to_vec(); witness::tests::overwrite_tip(journal, bad_op, size, pinned_nodes).await; let cfg = Config { strategy: Sequential, witness: witness_config(partition), commit_codec_config: (), }; let reopened = TestDb::::init(context.child("reopen_witness"), cfg, None).await; assert!(matches!( reopened, Err(Error::DataCorrupted("invalid compact witness")) )); }); } #[test_traced("INFO")] fn test_compact_reopen_rejects_tampered_pinned_nodes() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-pinned-nodes-tamper"; let db = open_db::(context.child("db"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8)) .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let tampered_target = db.target(); drop(db); // Flip one pinned-node digest. There is no stored proof to cross-check against, so the // rebuild succeeds and yields a different root, the same way a bit-flipped replay // journal reopens with a different root. let journal = open_witness_journal(context.child("tamper"), partition).await; let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await; pinned_nodes[0] = Sha256::fill(0xff); witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await; let cfg = Config { strategy: Sequential, witness: witness_config(partition), commit_codec_config: (), }; let reopened = TestDb::::init(context.child("reopen_witness"), cfg, None) .await .unwrap(); assert_ne!(reopened.target(), tampered_target); reopened.destroy().await.unwrap(); }); } /// A witness entry appended but not synced (a commit interrupted before its journal sync) /// must be dropped on reopen, recovering the last synced commit. #[test_traced("INFO")] fn test_compact_reopen_drops_unsynced_witness() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-witness-unsynced"; let db = open_db::(context.child("db"), partition).await; // Commit state A. let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let target_a = db.target(); drop(db); // Simulate the crash window: append an entry ahead of the tip without syncing it, // then drop the journal. The unsynced tail must not survive reopen. let journal = open_witness_journal(context.child("crash"), partition).await; let (op_bytes, mut size, pinned_nodes) = witness::tests::tip(&journal).await; size += 2; witness::tests::append_unsynced(journal, op_bytes, size, pinned_nodes).await; // Reopen must drop the unsynced entry and recover state A. let reopened = open_db::(context.child("reopen"), partition).await; assert_eq!(reopened.target(), target_a); reopened.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_initialization_zero_and_above_end() { deterministic::Runner::default().start(|context| async move { let cfg = witness_config("immutable-caps"); assert!(matches!( open_bounded::(context.child("zero"), cfg.clone(), Location::new(0)) .await, Err(Error::InvalidInitializationBound) )); let db = open_bounded::( context.child("fresh"), cfg.clone(), Location::new(100), ) .await .unwrap(); let root = db.root(); assert_eq!(db.size(), Location::new(1)); drop(db); let db = open_bounded::(context.child("above"), cfg, Location::new(100)) .await .unwrap(); assert_eq!(db.root(), root); assert_eq!(db.size(), Location::new(1)); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_between_commits() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-rewind-between").await; let initial_root = db.root(); // A multi-op commit jumps the committed size from 1 (bootstrap) to 4. let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let root_a = db.root(); let size_a = db.size(); assert_eq!(size_a, Location::new(4)); // A second commit moves the size to 6. let batch = db .new_batch() .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8)) .merkleize(&db, Some(Sha256::fill(0xb1)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let cfg = witness_config("immutable-rewind-between"); drop(db); for (target, size, root) in [(5, 4, root_a), (3, 1, initial_root), (2, 1, initial_root)] { let db = open_bounded::( context.child("cap"), cfg.clone(), Location::new(target), ) .await .unwrap(); assert_eq!(db.size(), Location::new(size)); assert_eq!(db.root(), root); drop(db); let db = open_db::(context.child("restart"), "immutable-rewind-between") .await; assert_eq!(db.root(), root); } }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_multiple_commits() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-rewind-multi"; let db = open_db::(context.child("db"), partition).await; // Commit A, B, C, recording the state after A. let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let root_a = db.root(); let size_a = db.size(); let target_a = db.target(); let mut db = db; for i in [2u8, 3] { let batch = db .new_batch() .set(Sha256::hash(&[&[i]]), Sha256::fill(i)) .merkleize(&db, Some(Sha256::fill(i)), Location::new(0)) .await .unwrap(); db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); } assert_ne!(db.root(), root_a); // Reopen two commits earlier. let db = { _ = db.sync().await.unwrap(); open_bounded::(context.child("cap"), witness_config(partition), size_a) .await } .unwrap(); assert_eq!(db.root(), root_a); assert_eq!(db.size(), size_a); assert_eq!(db.get_metadata(), Some(Sha256::fill(0xa1))); assert_eq!(db.target(), target_a); drop(db); // The recovery is durable: reopen recovers state A. let db = open_db::(context.child("reopen"), partition).await; assert_eq!(db.root(), root_a); assert_eq!(db.target(), target_a); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_preserves_current_state() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-rewind-noop").await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let root = db.root(); let size = db.size(); let db = { _ = db.sync().await.unwrap(); open_bounded::( context.child("cap"), witness_config("immutable-rewind-noop"), size, ) .await } .unwrap(); assert_eq!(db.root(), root); assert_eq!(db.size(), size); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_prune_then_bounded_initialization() { deterministic::Runner::default().start(|context| async move { // One entry per section so pruning takes effect at entry granularity (pruning is // section-aligned and never drops a partial section). let mut witness_cfg = witness_config("immutable-prune-rewind"); witness_cfg.items_per_section = NZU64!(1); let cfg = Config { strategy: Sequential, witness: witness_cfg.clone(), commit_codec_config: (), }; let mut db: TestDb = Db::init(context.child("db"), cfg, None).await.unwrap(); // Commit A, B, C. let mut sizes = Vec::new(); for i in [1u8, 2, 3] { let batch = db .new_batch() .set(Sha256::hash(&[&[i]]), Sha256::fill(i)) .merkleize(&db, Some(Sha256::fill(i)), Location::new(0)) .await .unwrap(); db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); sizes.push(db.size()); } // Reopen retained state B before the terminal check that pruned state A is unavailable. let db = db.prune(sizes[1]).await.unwrap().sync().await.unwrap(); drop(db); let db = open_bounded::( context.child("retained"), witness_cfg.clone(), sizes[1], ) .await .unwrap(); assert_eq!(db.size(), sizes[1]); assert_eq!(db.get_metadata(), Some(Sha256::fill(2))); drop(db); assert!(matches!( open_bounded::(context.child("pruned"), witness_cfg, sizes[0]).await, Err(Error::HistoricalFloorPruned(_)) )); }); } /// Witness config holding one witness per section, so every commit occupies its own blobs. fn sectioned_witness_config(partition: &str) -> JournalConfig<()> { let mut cfg = witness_config(partition); cfg.items_per_section = NZU64!(1); cfg } /// Seed a db with the bootstrap witness plus `commits` synced metadata-only commits, /// returning the size and root after each commit. Every commit appends exactly its commit /// operation, so the witness at position `p` has size `p + 1`. async fn seed_witness_sections( context: deterministic::Context, witness_cfg: JournalConfig<()>, commits: u8, ) -> Vec<(Location, Digest)> { let cfg = Config { strategy: Sequential, witness: witness_cfg, commit_codec_config: (), }; let mut db: TestDb = Db::init(context, cfg, None).await.unwrap(); let mut states = Vec::new(); for i in 1..=commits { let batch = db .new_batch() .merkleize(&db, Some(Sha256::fill(i)), db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); states.push((db.size(), db.root())); } states } /// Leave witness data blob `blob` with a partial trailing page, as a crash mid-write would. /// Paged tail recovery repairs (resizes and syncs) such a tail when the blob is opened. async fn tear_witness_data( context: &deterministic::Context, witness_cfg: &JournalConfig<()>, blob: u64, ) { // The variable journal keeps data blobs in `{partition}_data`, named by the big-endian // section index. let partition = format!("{}_data", witness_cfg.partition); let (handle, len) = context.open(&partition, &blob.to_be_bytes()).await.unwrap(); handle.resize(len - 1).await.unwrap(); handle.sync().await.unwrap(); } /// Bounded init through a delayed-sync backend, returning the db and the durability calls /// initialization made. async fn open_bounded_counting( context: deterministic::Context, witness_cfg: JournalConfig<()>, cap: Location, ) -> (DelayedDb, usize) { // Arming counts every durability call from here on. The gate blocks the first call and // every later started sync parks. `drive_pending_syncs` releases them whenever // initialization stalls. let pending = PendingSyncs::default(); pending.arm(); let delayed = DelayedSyncContext { inner: context, pending: pending.clone(), }; let cfg = Config { strategy: Sequential, witness: witness_cfg, commit_codec_config: (), }; let db = drive_pending_syncs(&pending, DelayedDb::init(delayed, cfg, Some(cap))) .await .unwrap(); (db, pending.calls()) } /// Bounded initialization opens no witness section the bound discards, so a torn tail there /// costs no repair. #[test_traced] fn test_compact_bounded_initialization_ignores_discarded_witness_sections() { deterministic::Runner::default().start(|context| async move { // Build twin journals with one witness per section, so position `p` lives in data // blob `p`. Each holds the bootstrap witness (position 0, size 1), a synced commit // (position 1, size 2), and a commit without sync (position 2, size 3). The control // twin stays intact and sets the baseline durability count. let control_cfg = sectioned_witness_config("immutable-skip-discarded-control"); let torn_cfg = sectioned_witness_config("immutable-skip-discarded-torn"); let mut states = Vec::new(); for (label, witness_cfg) in [("control", &control_cfg), ("torn", &torn_cfg)] { let context = context.child(label); states.push( seed_witness_sections(context.child("seed"), witness_cfg.clone(), 1).await[0], ); // Commit without syncing so the witness at position 2 lies above the recovery // watermark, where a torn tail is a crash shape rather than corruption. let cfg = Config { strategy: Sequential, witness: witness_cfg.clone(), commit_codec_config: (), }; let db: TestDb = Db::init(context.child("extend"), cfg, None).await.unwrap(); let batch = db .new_batch() .merkleize(&db, Some(Sha256::fill(2)), db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap(); drop(db.commit().await.unwrap()); } // The twins share one history, so the synced commit's size and root are the expected // recovery for both. assert_eq!(states[0], states[1]); let (size, root) = states[0]; // The witness at position 1 has size `size`, so the bound discards position 2 and // must not open it. Tear its data blob: repairing the tail would sync it. tear_witness_data(&context, &torn_cfg, 2).await; // Both twins recover the synced commit under the bound. They differ only in data // blob 2, so equal durability counts show the torn blob was not repaired. let (control, control_calls) = open_bounded_counting(context.child("control"), control_cfg, size).await; let (torn, torn_calls) = open_bounded_counting(context.child("torn"), torn_cfg, size).await; assert_eq!(control.size(), size); assert_eq!(control.root(), root); assert_eq!(torn.size(), size); assert_eq!(torn.root(), root); assert_eq!( torn_calls, control_calls, "the discarded torn section must not be repaired" ); control.destroy().await.unwrap(); torn.destroy().await.unwrap(); }); } /// A torn tail in a retained witness section is still repaired under a bound. The witness it /// held is lost and initialization falls back to the previous one. #[test_traced] fn test_compact_bounded_initialization_repairs_retained_witness_section() { deterministic::Runner::default().start(|context| async move { // Both witnesses share a section, so the bootstrap witness survives the torn tail // page and ends mid-page, where recovery must rewrite rather than only shrink. let clean_cfg = witness_config("immutable-repair-retained-clean"); let torn_cfg = witness_config("immutable-repair-retained-torn"); let mut states = Vec::new(); for (label, witness_cfg) in [("clean", &clean_cfg), ("torn", &torn_cfg)] { // Fresh storage bootstraps the witness at position 0 (size 1), the state the torn // twin falls back to. let cfg = Config { strategy: Sequential, witness: witness_cfg.clone(), commit_codec_config: (), }; let db: TestDb = Db::init(context.child(label), cfg, None).await.unwrap(); let genesis = db.root(); // Commit without syncing so the witness at position 1 lies above the recovery // watermark, where a torn tail is a crash shape rather than corruption. let batch = db .new_batch() .merkleize(&db, Some(Sha256::fill(1)), db.inactivity_floor_loc()) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap(); states.push((genesis, db.size(), db.root())); } assert_eq!(states[0], states[1]); let (genesis, size, root) = states[0]; // The torn twin's last page in section 0 holds the end of the witness at position 1, // so tearing it leaves that witness incomplete. tear_witness_data(&context, &torn_cfg, 0).await; // Section 0 lies below the bound, so both twins open it. The torn twin trims its // tail and republishes the journal without the lost witness. let (clean, clean_calls) = open_bounded_counting(context.child("clean"), clean_cfg, size).await; let (torn, torn_calls) = open_bounded_counting(context.child("torn"), torn_cfg, size).await; // The clean twin recovers the witness at position 1. The torn twin recovers the // bootstrap witness and spends extra durability calls on the repair. assert_eq!(clean.size(), size); assert_eq!(clean.root(), root); assert_eq!(torn.size(), Location::new(1)); assert_eq!(torn.root(), genesis); assert!( torn_calls > clean_calls, "the retained torn section is repaired" ); clean.destroy().await.unwrap(); torn.destroy().await.unwrap(); }); } /// A compact-sync import lands at the journal end whatever its size, so afterwards witness /// positions can exceed sizes. Bounded initialization still selects by size, widening its /// view when the positions below the bound cannot settle the selection. #[test_traced("INFO")] fn test_compact_bounded_initialization_after_import_below_positions() { deterministic::Runner::default().start(|context| async move { // A source state of size 2, captured from its witness journal. Its tip witness holds // the pinned nodes one operation below the commit, which a compact-sync boundary // response carries. let src_cfg = sectioned_witness_config("immutable-import-below-src"); let (src_size, src_root) = seed_witness_sections(context.child("src"), src_cfg.clone(), 1).await[0]; let journal = witness::Journal::<_, mmr::Family, Digest>::init(context.child("tip"), src_cfg) .await .unwrap(); let (_, _, src_pinned) = witness::tests::tip(&journal).await; drop(journal); // The destination has used positions 0 through 3, so the import lands at position 4. let dst_cfg = sectioned_witness_config("immutable-import-below-dst"); seed_witness_sections(context.child("dst"), dst_cfg.clone(), 3).await; // Import the source state from its last commit location, pinned nodes, and commit // operation. `seed_witness_sections` committed it with metadata `fill(1)` at floor 0, // so the rebuilt root matches the source root. let journal = witness::Journal::init(context.child("import"), dst_cfg.clone()) .await .unwrap(); let imported = TestDb::::init_from_sync( Sequential, journal, (), src_size - 1, src_pinned, Operation::Commit(Some(Sha256::fill(1)), Location::new(0)), ) .unwrap(); assert_eq!(imported.root(), src_root); // Committing applies the pending import. It clears the destination journal at its end // and appends the imported witness there, so its only entry has size 2 at position 4. drop(imported.commit().await.unwrap()); let journal = witness::Journal::<_, mmr::Family, Digest>::init( context.child("placed"), dst_cfg.clone(), ) .await .unwrap(); assert_eq!(journal.bounds(), 4..5); drop(journal); // Reopening recovers the imported state. Three more commits occupy positions 5 through // 7 with sizes 3 through 5. `states` holds the size and root at positions 4 through 7. let cfg = Config { strategy: Sequential, witness: dst_cfg.clone(), commit_codec_config: (), }; let mut db: TestDb = Db::init(context.child("reopen"), cfg, None).await.unwrap(); assert_eq!(db.size(), src_size); let mut states = vec![(src_size, src_root)]; for i in 2..=4u8 { let batch = db .new_batch() .merkleize(&db, Some(Sha256::fill(i)), db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); states.push((db.size(), db.root())); } drop(db); // Open at each recorded size from the tip down. The cap-5 bounded view holds only // position 4 (size 2). It ends at the bound with a tip size below the bound, so // recovery widens and selects position 7 by size. Caps 4, 3, and 2 lie at or below the // retained start (position 4), so recovery opens unbounded and selects positions 6, 5, // and 4. Each open discards the witnesses above its selection, so the caps must // descend. for (size, root) in states.into_iter().rev() { let db = open_bounded::( context.child("bounded").with_attribute("cap", *size), dst_cfg.clone(), size, ) .await .unwrap(); assert_eq!(db.size(), size); assert_eq!(db.root(), root); } // Only the imported witness remains, and its size 2 exceeds cap 1. assert!(matches!( open_bounded::(context.child("pruned"), dst_cfg, Location::new(1)) .await, Err(Error::HistoricalFloorPruned(pruned)) if pruned == Location::new(1) )); }); } /// A bounded initialization whose selection fails leaves the witness offsets watermark /// acknowledging every synced witness. #[test_traced] fn test_compact_failed_bounded_selection_preserves_acknowledged_offsets() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-failed-bounded-offsets"; let cap = Location::new(5); let witness_cfg = sectioned_witness_config(partition); // The witness journal lives in `{partition}-witness`. Its offsets journal, whose // checkpoint persists the recovery watermark, lives in `{partition}-witness_offsets`. let offsets_partition = format!("{partition}-witness_offsets"); let cfg = Config { strategy: Sequential, witness: witness_cfg.clone(), commit_codec_config: (), }; let mut db: TestDb = Db::init(context.child("seed"), cfg, None).await.unwrap(); // The first batch appends six sets and a commit after the bootstrap commit, so the // witness at position 1 has size 8, above the cap used below. let mut batch = db.new_batch(); for key in 1..=6u8 { batch = batch.set(Sha256::fill(key), Sha256::fill(key)); } let batch = batch.merkleize(&db, None, Location::new(0)).await.unwrap(); db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let first_retained_size = db.size(); assert!(first_retained_size > cap); // Eight synced empty commits occupy positions 2 through 9 with sizes 9 through 16. // One witness per section puts positions 1 through 4 in the sections the cap-5 view // opens, and positions 5 through 9 in sections it discards. for _ in 0..8 { let batch = db .new_batch() .merkleize(&db, None, db.inactivity_floor_loc()) .await .unwrap(); db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); } let tip = db.size(); // Pruning at the first commit removes the bootstrap section. The retained start // (position 1) lies below the cap while its size lies above it. A retained start at // or above the cap would open unbounded instead. let db = db .prune(first_retained_size) .await .unwrap() .sync() .await .unwrap(); drop(db); // Sync acknowledged all ten witnesses (positions 0 through 9). The watermark lies above // the cap, so any lowering to the cap is visible below. let watermark = fixed::Journal::<_, u64>::persisted_watermark( context.child("before"), &offsets_partition, ) .await .unwrap(); assert_eq!(watermark, Some(10)); // No compact-sync import put a smaller size at the journal end, so the bounded view // ends at the cap with a tip size above it and recovery stays bounded. The offsets open // clamps the watermark to the cap, so inspection anchors at the ceiling and leaves the // acknowledged offsets untouched. No retained witness fits under the cap, so selection // fails. assert!(matches!( open_bounded::(context.child("failed"), witness_cfg.clone(), cap) .await, Err(Error::HistoricalFloorPruned(found)) if found == cap )); // Selection fails before publication, and inspection anchored at the ceiling skips the // offsets truncate, so the failed attempt leaves the durable watermark at 10. let watermark = fixed::Journal::<_, u64>::persisted_watermark( context.child("after"), &offsets_partition, ) .await .unwrap(); assert_eq!(watermark, Some(10)); // Reinitializing at `tip` recovers every retained witness. let reopened = open_bounded::(context.child("retry"), witness_cfg, tip) .await .unwrap(); assert_eq!(reopened.size(), tip); reopened.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_prune_past_tip_keeps_tip() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-prune-past-tip"; let db = open_db::(context.child("db"), partition).await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let target = db.target(); // Prune with a boundary beyond the tip: the tip entry must survive. let boundary = db.size() + 100; let db = db.prune(boundary).await.unwrap(); assert_eq!(db.target(), target); drop(db); let reopened = open_db::(context.child("reopen"), partition).await; assert_eq!(reopened.target(), target); reopened.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_preserves_pre_advance_batch() { deterministic::Runner::default().start(|context| async move { let db = open_db::( context.child("db"), "immutable-rewind-preserves-pre-advance", ) .await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let size_after_first = db.size(); // Merkleize a batch against the post-commit-A state. let held = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); // Advance past that state and commit, then reopen at that state. let batch = db .new_batch() .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let db = { _ = db.sync().await.unwrap(); open_bounded::( context.child("cap"), witness_config("immutable-rewind-preserves-pre-advance"), size_after_first, ) .await } .unwrap(); // The recovery restored the state that `held` was merkleized against, so it still // matches the Merkle size and applies cleanly. let db = db.apply_batch(held).await.unwrap(); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_noop_commit_after_commit() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-noop-after-commit").await; let k1 = Sha256::hash(&[&[1]]); let v1 = Sha256::fill(11u8); let k2 = Sha256::hash(&[&[2]]); let v2 = Sha256::fill(22u8); let batch = db .new_batch() .set(k1, v1) .set(k2, v2) .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let root_after_first = db.root(); let size_after_first = db.size(); let db = db.sync().await.unwrap(); assert_eq!(db.size(), size_after_first); assert_eq!(db.root(), root_after_first); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_noop_commit_after_reopen() { deterministic::Runner::default().start(|context| async move { let partition = "immutable-noop-after-reopen"; let (root_before_drop, size_before_drop) = { let db = open_db::(context.child("first"), partition).await; let k1 = Sha256::hash(&[&[1]]); let v1 = Sha256::fill(11u8); let k2 = Sha256::hash(&[&[2]]); let v2 = Sha256::fill(22u8); let batch = db .new_batch() .set(k1, v1) .set(k2, v2) .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); (db.root(), db.size()) }; let db = open_db::(context.child("second"), partition).await; assert_eq!(db.root(), root_before_drop); assert_eq!(db.size(), size_before_drop); let db = db.sync().await.unwrap(); assert_eq!(db.size(), size_before_drop); assert_eq!(db.root(), root_before_drop); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_noop_commit_after_bounded_initialization() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-noop-after-rewind").await; let k1 = Sha256::hash(&[&[1]]); let v1 = Sha256::fill(11u8); let k2 = Sha256::hash(&[&[2]]); let v2 = Sha256::fill(22u8); let batch = db .new_batch() .set(k1, v1) .set(k2, v2) .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let root_after_first = db.root(); let size_after_first = db.size(); let k3 = Sha256::hash(&[&[3]]); let v3 = Sha256::fill(33u8); let batch = db .new_batch() .set(k3, v3) .merkleize(&db, Some(Sha256::fill(0xbb)), Location::new(1)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let db = { _ = db.sync().await.unwrap(); open_bounded::( context.child("cap"), witness_config("immutable-noop-after-rewind"), size_after_first, ) .await } .unwrap(); assert_eq!(db.size(), size_after_first); assert_eq!(db.root(), root_after_first); let db = db.sync().await.unwrap(); assert_eq!(db.size(), size_after_first); assert_eq!(db.root(), root_after_first); db.destroy().await.unwrap(); }); } #[test_traced("INFO")] fn test_compact_bounded_initialization_makes_post_advance_batch_stale() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-rewind-makes-stale").await; let batch = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); let size_after_first = db.size(); let batch = db .new_batch() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap(); // Merkleize a batch against the post-commit-B state, which the recovery will discard. let held = db .new_batch() .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8)) .merkleize(&db, None, Location::new(0)) .await .unwrap(); let db = { _ = db.sync().await.unwrap(); open_bounded::( context.child("cap"), witness_config("immutable-rewind-makes-stale"), size_after_first, ) .await } .unwrap(); // After recovery, mem.size reflects post-commit-A, but the held batch starts after // post-commit-B. Apply must be rejected with StaleBatch. assert!(matches!(db.apply_batch(held).await, Err(Error::StaleBatch))); }); } #[test_traced("INFO")] fn test_compact_floor_beyond_size() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-floor-beyond").await; let batch = db.new_batch().merkleize(&db, None, Location::new(2)).await; assert!(matches!( batch, Err(Error::FloorBeyondSize(floor, tip)) if floor == Location::new(2) && tip == Location::new(1) )); }); } // Merkleize returns `FloorBeyondSize` for a chained batch whose floor exceeds its own commit // location. The commit location includes the parent's operations. #[test_traced("INFO")] fn test_compact_chained_floor_beyond_commit() { deterministic::Runner::default().start(|context| async move { let db = open_db::(context.child("db"), "immutable-ancestor-floor-beyond") .await; // parent: set + commit at loc 2, floor=2. let parent = db .new_batch() .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8)) .merkleize(&db, None, Location::new(2)) .await .unwrap(); // child: set + commit at loc 4, floor=5 (one past its commit). let child = parent .new_batch::() .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8)) .merkleize(&db, None, Location::new(5)) .await; assert!(matches!( child, Err(Error::FloorBeyondSize(floor, commit)) if floor == Location::new(5) && commit == Location::new(4) )); }); } }