Skip to main content

commonware_storage/qmdb/immutable/
compact.rs

1//! An immutable authenticated db that discards historical operations, retaining only a witness
2//! for each applied batch.
3//!
4//! Mirrors the API of [`crate::qmdb::immutable::Immutable`] (`new_batch -> merkleize ->
5//! apply_batch -> commit / sync / start_sync`, pipelined batch chains, `StaleBatch` validation)
6//! but is backed by the peak-only [`crate::merkle::compact`]. Because history is discarded,
7//! the db has no `get` / `proof` / `bounds` methods. A merkleized batch can prove only its own
8//! operations, and only until it is applied. Use the full variant for historical proofs.
9//!
10//! # Witness journal
11//!
12//! The witness journal holds a complete snapshot of every applied batch, so [`Db::rewind`] can
13//! restore any retained applied state (history is bounded only by [`Db::prune`]). Reopen
14//! and rewind restore the db's in-memory state from an entry. The Merkle is rebuilt from the
15//! stored pinned nodes and operation, and the commit fields are decoded from the operation. An
16//! entry that cannot rebuild surfaces as [`Error::DataCorrupted`]. The witness is also what lets
17//! compact nodes serve compact sync without retaining historical operations.
18//!
19//! # Inactivity floor
20//!
21//! Commits carry an inactivity floor for wire-format compatibility with
22//! [`crate::qmdb::immutable::Immutable`]: the root is computed over the encoded operation
23//! sequence, and that sequence must include the same floor to produce the same root as the
24//! full variant. The floor has no effect on pruning or snapshot rebuilding here; all
25//! historical in-memory state is discarded whenever a batch is applied.
26
27use super::operation::Operation;
28pub use crate::qmdb::compact::Config;
29use crate::{
30    Context,
31    journal::contiguous::variable::{self, Config as JournalConfig},
32    merkle::{Family, Location, Proof, batch, compact as compact_merkle},
33    qmdb::{
34        self, Error,
35        any::value::ValueEncoding,
36        batch_chain::{self, Bounds, Commitment},
37        compact::{
38            batch as compact_batch,
39            witness::{self, VerifiedWitness},
40        },
41        operation::Key,
42        sync::{CompactTarget, FeedbackTx, Request, Response, Source},
43    },
44};
45use commonware_codec::{Encode, EncodeShared, Read};
46use commonware_cryptography::{Digest, Hasher};
47use commonware_macros::boxed;
48use commonware_parallel::Strategy;
49use commonware_runtime::Handle;
50use core::marker::PhantomData;
51use std::{
52    collections::BTreeMap,
53    sync::{Arc, Weak},
54};
55
56/// An immutable authenticated db that discards historical operations, retaining only a witness
57/// for each applied batch.
58pub struct Db<F, E, K, V, H, C, S: Strategy>
59where
60    F: Family,
61    E: Context,
62    K: Key,
63    V: ValueEncoding,
64    H: Hasher,
65    Operation<F, K, V>: EncodeShared,
66    Operation<F, K, V>: Read<Cfg = C>,
67    C: Clone + Send + Sync + 'static,
68{
69    merkle: compact_merkle::Merkle<F, H::Digest, S>,
70    root: H::Digest,
71    last_commit_loc: Location<F>,
72    last_commit_metadata: Option<V::Value>,
73    inactivity_floor_loc: Location<F>,
74    commit_codec_config: C,
75    witness: witness::Store<E, F, H::Digest>,
76    _key: PhantomData<K>,
77}
78
79impl<F, E, K, V, H, C, S: Strategy> std::fmt::Debug for Db<F, E, K, V, H, C, S>
80where
81    F: Family,
82    E: Context,
83    K: Key,
84    V: ValueEncoding,
85    H: Hasher,
86    Operation<F, K, V>: EncodeShared,
87    Operation<F, K, V>: Read<Cfg = C>,
88    C: Clone + Send + Sync + 'static,
89{
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        f.debug_struct("Db")
92            .field("size", &self.size())
93            .field("inactivity_floor_loc", &self.inactivity_floor_loc())
94            .finish_non_exhaustive()
95    }
96}
97
98/// A speculative batch for a compact immutable db.
99#[allow(clippy::type_complexity)]
100pub struct UnmerkleizedBatch<F, H, K, V, S: Strategy>
101where
102    F: Family,
103    K: Key,
104    V: ValueEncoding,
105    H: Hasher,
106    Operation<F, K, V>: EncodeShared,
107{
108    merkle_batch: compact_merkle::UnmerkleizedBatch<F, H::Digest, S>,
109    mutations: BTreeMap<K, V::Value>,
110    parent: Option<Arc<MerkleizedBatch<F, H::Digest, K, V, S>>>,
111    base: batch_chain::Commitment<F, H::Digest>,
112}
113
114/// A speculative batch whose root digest has been computed.
115#[derive(Clone)]
116pub struct MerkleizedBatch<F: Family, D: Digest, K: Key, V: ValueEncoding, S: Strategy>
117where
118    Operation<F, K, V>: EncodeShared,
119{
120    pub(super) merkle_batch: Arc<batch::MerkleizedBatch<F, D, S>>,
121    operations: Arc<Vec<Operation<F, K, V>>>,
122    pub(super) commit_metadata: Option<V::Value>,
123    pub(super) parent: Option<Weak<Self>>,
124    pub(super) bounds: batch_chain::Bounds<F, D>,
125    pub(super) _key: PhantomData<K>,
126}
127
128impl<F: Family, D: Digest, K: Key, V: ValueEncoding, S: Strategy> MerkleizedBatch<F, D, K, V, S>
129where
130    Operation<F, K, V>: EncodeShared,
131{
132    pub(super) fn ancestors(&self) -> impl Iterator<Item = Arc<Self>> + use<F, D, K, V, S> {
133        batch_chain::ancestors(self.parent.clone(), |batch| batch.parent.as_ref())
134    }
135
136    /// The [`Commitment`] this batch commits to.
137    pub(super) const fn commitment(&self) -> Commitment<F, D> {
138        self.bounds.tip
139    }
140
141    /// Return the root digest after this batch is applied.
142    pub const fn root(&self) -> D {
143        self.bounds.tip.root
144    }
145
146    /// Return the [`Bounds`] of the batch.
147    pub const fn bounds(&self) -> &Bounds<F, D> {
148        &self.bounds
149    }
150
151    /// Return the operations this batch appends to the log and the location of the first.
152    #[allow(clippy::type_complexity)]
153    pub fn operations(&self) -> (Location<F>, Arc<Vec<Operation<F, K, V>>>) {
154        (self.bounds.base.size, Arc::clone(&self.operations))
155    }
156
157    /// Inclusion proof for the operations returned by [`Self::operations`], anchored at
158    /// this batch's tip. The pair verifies against [`Self::root`] via
159    /// [`crate::qmdb::verify_proof`]. Together with [`Self::pinned_nodes`] they verify via
160    /// [`crate::qmdb::verify_proof_and_pinned_nodes`].
161    ///
162    /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be
163    /// alive. Nodes below the chain are read from `db`'s
164    /// [Merkle store][crate::merkle::mem::Mem], which retains them at least until this batch's
165    /// changes are applied (applying it or a descendant prunes the store to its frontier).
166    ///
167    /// # Errors
168    ///
169    /// Returns [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or
170    /// belongs to a dropped unapplied ancestor, and [`crate::merkle::Error::Empty`] if the batch
171    /// has no operations (a [`Db::to_batch`] snapshot).
172    pub fn proof<E, C, H>(&self, db: &Db<F, E, K, V, H, C, S>) -> Result<Proof<F, D>, Error<F>>
173    where
174        E: Context,
175        H: Hasher<Digest = D>,
176        C: Clone + Send + Sync + 'static,
177        Operation<F, K, V>: Read<Cfg = C>,
178    {
179        let inactive_peaks = F::inactive_peaks(self.bounds.tip.size, self.bounds.inactivity_floor);
180        let hasher = qmdb::hasher::<H>();
181        db.merkle
182            .with_mem(|base| {
183                self.merkle_batch.range_proof(
184                    base,
185                    &hasher,
186                    self.bounds.base.size..self.bounds.tip.size,
187                    inactive_peaks,
188                )
189            })
190            .map_err(Into::into)
191    }
192
193    /// The Merkle frontier at the first operation returned by [`Self::operations`]
194    /// ([`Family::nodes_to_pin`]), which lets a consumer holding only this batch's base rebuild
195    /// compact state and replay the operations. The operations, [`Self::proof`], and pinned
196    /// nodes verify against [`Self::root`] via [`crate::qmdb::verify_proof_and_pinned_nodes`].
197    ///
198    /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be
199    /// alive. Nodes below the chain are read from `db`'s
200    /// [Merkle store][crate::merkle::mem::Mem], which retains them at least until this batch's
201    /// changes are applied (applying it or a descendant prunes the store to its frontier).
202    ///
203    /// # Errors
204    ///
205    /// Returns [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or
206    /// belongs to a dropped unapplied ancestor.
207    pub fn pinned_nodes<E, C, H>(&self, db: &Db<F, E, K, V, H, C, S>) -> Result<Vec<D>, Error<F>>
208    where
209        E: Context,
210        H: Hasher<Digest = D>,
211        C: Clone + Send + Sync + 'static,
212        Operation<F, K, V>: Read<Cfg = C>,
213    {
214        db.merkle
215            .with_mem(|base| {
216                F::nodes_to_pin(self.bounds.base.size)
217                    .map(|pos| {
218                        self.merkle_batch
219                            .get_node(pos)
220                            .or_else(|| base.get_node(pos))
221                            .ok_or(crate::merkle::Error::ElementPruned(pos))
222                    })
223                    .collect::<Result<Vec<_>, _>>()
224            })
225            .map_err(Into::into)
226    }
227
228    /// Create a new speculative batch with this one as its parent.
229    pub fn new_batch<H>(self: &Arc<Self>) -> UnmerkleizedBatch<F, H, K, V, S>
230    where
231        H: Hasher<Digest = D>,
232    {
233        UnmerkleizedBatch {
234            merkle_batch: compact_merkle::UnmerkleizedBatch::wrap(self.merkle_batch.new_batch()),
235            mutations: BTreeMap::new(),
236            parent: Some(Arc::clone(self)),
237            base: self.commitment(),
238        }
239    }
240}
241
242impl<F, H, K, V, S> UnmerkleizedBatch<F, H, K, V, S>
243where
244    F: Family,
245    K: Key,
246    V: ValueEncoding,
247    H: Hasher,
248    S: Strategy,
249    Operation<F, K, V>: EncodeShared,
250{
251    pub(super) fn new<E, C>(
252        db: &Db<F, E, K, V, H, C, S>,
253        base: batch_chain::Commitment<F, H::Digest>,
254    ) -> Self
255    where
256        E: Context,
257        C: Clone + Send + Sync + 'static,
258        Operation<F, K, V>: Read<Cfg = C>,
259    {
260        Self {
261            merkle_batch: db.merkle.new_batch(),
262            mutations: BTreeMap::new(),
263            parent: None,
264            base,
265        }
266    }
267
268    /// The database boundary for this batch chain.
269    ///
270    /// A batch created from the database uses its base. A child inherits its parent's `db`.
271    fn db(&self) -> Commitment<F, H::Digest> {
272        self.parent
273            .as_ref()
274            .map_or(self.base, |parent| parent.bounds.db)
275    }
276
277    pub fn set(mut self, key: K, value: V::Value) -> Self {
278        self.mutations.insert(key, value);
279        self
280    }
281
282    /// Resolve mutations into operations, merkleize, and return an `Arc<MerkleizedBatch>`.
283    ///
284    /// `inactivity_floor` is threaded through the commit operation for wire-format parity with
285    /// [`crate::qmdb::immutable::Immutable`]. It must be >= the database's current floor
286    /// (monotonically non-decreasing) and at most the batch's commit location
287    /// (`total_size - 1`); these bounds are validated, but the floor does not drive any local
288    /// pruning or retention in this variant.
289    #[tracing::instrument(
290        name = "qmdb.immutable.compact.batch.merkleize",
291        level = "info",
292        skip_all
293    )]
294    pub async fn merkleize<E, C>(
295        self,
296        db: &Db<F, E, K, V, H, C, S>,
297        metadata: Option<V::Value>,
298        inactivity_floor: Location<F>,
299    ) -> Arc<MerkleizedBatch<F, H::Digest, K, V, S>>
300    where
301        F: Family,
302        E: Context,
303        C: Clone + Send + Sync + 'static,
304        Operation<F, K, V>: Read<Cfg = C>,
305    {
306        let live_ancestors: Vec<_> =
307            batch_chain::parent_and_ancestors(self.parent.as_ref(), |parent| parent.ancestors())
308                .collect();
309        let boundary = batch_chain::effective_boundary(
310            self.db(),
311            live_ancestors.last().map(|oldest| oldest.bounds.base),
312        );
313
314        let mut ops: Vec<Operation<F, K, V>> = Vec::with_capacity(self.mutations.len() + 1);
315        for (key, value) in self.mutations {
316            ops.push(Operation::Set(key, value));
317        }
318        ops.push(Operation::Commit(metadata.clone(), inactivity_floor));
319
320        let operations = Arc::new(ops);
321        let total_size = self.base.size + operations.len() as u64;
322        let inactive_peaks = F::inactive_peaks(total_size, inactivity_floor);
323        let (merkle, root) = compact_batch::merkleize_ops::<F, H, S, _>(
324            &db.merkle,
325            self.merkle_batch,
326            Arc::clone(&operations),
327            inactive_peaks,
328        )
329        .await
330        .expect("inactive_peaks computed from batch size");
331
332        let ancestors = batch_chain::collect_ancestor_bounds(
333            live_ancestors,
334            |batch| batch.bounds.inactivity_floor,
335            |batch| batch.commitment(),
336        );
337
338        Arc::new(MerkleizedBatch {
339            merkle_batch: merkle,
340            operations,
341            commit_metadata: metadata,
342            parent: self.parent.as_ref().map(Arc::downgrade),
343            bounds: batch_chain::Bounds {
344                base: self.base,
345                db: boundary,
346                tip: Commitment::new(total_size, root),
347                ancestors,
348                inactivity_floor,
349            },
350            _key: PhantomData,
351        })
352    }
353}
354
355impl<F, E, K, V, H, C, S> Db<F, E, K, V, H, C, S>
356where
357    F: Family,
358    E: Context,
359    K: Key,
360    V: ValueEncoding,
361    H: Hasher,
362    S: Strategy,
363    Operation<F, K, V>: EncodeShared,
364    Operation<F, K, V>: Read<Cfg = C>,
365    C: Clone + Send + Sync + 'static,
366{
367    fn encode_commit_op(metadata: Option<V::Value>, inactivity_floor_loc: Location<F>) -> Vec<u8> {
368        Operation::<F, K, V>::Commit(metadata, inactivity_floor_loc)
369            .encode()
370            .to_vec()
371    }
372
373    /// Build a compact db from state fetched by the sync engine.
374    ///
375    /// The imported witness lives only in memory until the first [`Self::apply_batch`],
376    /// [`Self::commit`], [`Self::sync`], or [`Self::start_sync`]. Applying a batch replaces it with
377    /// the newly applied journal checkpoint; a durability method journals it directly. Until one
378    /// of those operations succeeds, rewind and prune are rejected.
379    pub(crate) fn init_from_sync(
380        strategy: S,
381        journal: witness::Journal<E, F, H::Digest>,
382        commit_codec_config: C,
383        last_commit_loc: Location<F>,
384        pinned_nodes: Vec<H::Digest>,
385        last_commit_op: Operation<F, K, V>,
386    ) -> Result<Self, Error<F>> {
387        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
388            return Err(Error::UnexpectedData(last_commit_loc));
389        };
390        witness::validate_inactivity_floor(inactivity_floor_loc, last_commit_loc)?;
391
392        let op_bytes = Self::encode_commit_op(last_commit_metadata.clone(), inactivity_floor_loc);
393        let merkle =
394            compact_merkle::Merkle::from_compact_state(strategy, last_commit_loc, pinned_nodes)?;
395        let hasher = qmdb::hasher::<H>();
396        merkle.append_leaf(&hasher, &op_bytes)?;
397        let imported = witness::build_witness::<F, H, S>(&merkle, inactivity_floor_loc, op_bytes)?;
398        merkle.prune_to_frontier();
399
400        let witness = witness::Store::from_import(journal, imported);
401        let root = witness.with(|w| w.root);
402        Ok(Self {
403            merkle,
404            root,
405            last_commit_loc,
406            last_commit_metadata,
407            inactivity_floor_loc,
408            commit_codec_config,
409            witness,
410            _key: PhantomData,
411        })
412    }
413
414    /// Open a compact db from persisted compact state and rebuild its witness store.
415    ///
416    /// On first open, this bootstraps the initial commit and its witness so every later reopen and
417    /// rewind can assume the journal tip is a complete compact witness.
418    #[boxed]
419    pub(crate) async fn init_from_merkle(
420        mut merkle: compact_merkle::Merkle<F, H::Digest, S>,
421        witness_context: E,
422        witness_config: JournalConfig<()>,
423        commit_codec_config: C,
424    ) -> Result<Self, Error<F>>
425    where
426        F: Family,
427        Operation<F, K, V>: Read<Cfg = C>,
428    {
429        // Bootstrap: append an initial Commit(None, 0) on first open.
430        let journal: witness::Journal<E, F, H::Digest> =
431            variable::Journal::init(witness_context, witness_config).await?;
432        let (witness, last_commit_op) = witness::init::<E, F, H, S, Operation<F, K, V>>(
433            journal,
434            &mut merkle,
435            &commit_codec_config,
436            Operation::<F, K, V>::Commit(None, Location::new(0))
437                .encode()
438                .to_vec(),
439        )
440        .await?;
441        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
442            return Err(Error::DataCorrupted("last operation was not a commit"));
443        };
444        let last_commit_loc = witness.with(|w| w.size()) - 1;
445        let root = witness.with(|w| w.root);
446
447        Ok(Self {
448            merkle,
449            root,
450            last_commit_loc,
451            last_commit_metadata,
452            inactivity_floor_loc,
453            commit_codec_config,
454            witness,
455            _key: PhantomData,
456        })
457    }
458
459    /// Return the root of the db.
460    pub const fn root(&self) -> H::Digest {
461        self.root
462    }
463
464    /// Return a reference to the merkleization strategy.
465    pub const fn strategy(&self) -> &S {
466        self.merkle.strategy()
467    }
468
469    /// Return the location of the last commit.
470    pub const fn last_commit_loc(&self) -> Location<F> {
471        self.last_commit_loc
472    }
473
474    /// Return the inactivity floor declared by the last committed batch.
475    pub const fn inactivity_floor_loc(&self) -> Location<F> {
476        self.inactivity_floor_loc
477    }
478
479    /// Return the location of the next operation appended to this db.
480    pub fn size(&self) -> Location<F> {
481        self.last_commit_loc + 1
482    }
483
484    /// Get the metadata associated with the last commit.
485    pub fn get_metadata(&self) -> Option<V::Value> {
486        self.last_commit_metadata.clone()
487    }
488
489    /// Return the compact-sync target described by the current witness.
490    ///
491    /// This reflects the most recently applied batch. The target remains non-durable until a
492    /// covering [`Self::commit`], [`Self::sync`], or [`Self::start_sync`] completes.
493    pub fn target(&self) -> CompactTarget<F, H::Digest> {
494        self.witness.with(VerifiedWitness::target)
495    }
496
497    /// The [`Commitment`] for the database's current state.
498    pub(crate) fn commitment(&self) -> batch_chain::Commitment<F, H::Digest> {
499        batch_chain::Commitment::new(self.last_commit_loc + 1, self.root())
500    }
501
502    /// Create a new speculative batch of operations with this database as its parent.
503    pub fn new_batch(&self) -> UnmerkleizedBatch<F, H, K, V, S> {
504        UnmerkleizedBatch::new(self, self.commitment())
505    }
506
507    /// Create an owned merkleized batch representing the current applied state.
508    pub fn to_batch(&self) -> Arc<MerkleizedBatch<F, H::Digest, K, V, S>>
509    where
510        F: Family,
511    {
512        Arc::new(MerkleizedBatch {
513            merkle_batch: self.merkle.to_batch(),
514            operations: Arc::new(Vec::new()),
515            commit_metadata: self.last_commit_metadata.clone(),
516            parent: None,
517            bounds: batch_chain::Bounds::from_db(self.commitment(), self.inactivity_floor_loc),
518            _key: PhantomData,
519        })
520    }
521
522    /// Check that `batch` can be applied to the database in its current state, without
523    /// applying it.
524    ///
525    /// [`Self::apply_batch`] runs the same validation but consumes the database when it
526    /// fails; callers that want to reject a bad batch and keep the handle can check first.
527    pub fn validate_batch(
528        &self,
529        batch: &MerkleizedBatch<F, H::Digest, K, V, S>,
530    ) -> Result<(), Error<F>> {
531        batch
532            .bounds
533            .validate_apply_to(self.commitment(), self.inactivity_floor_loc)
534    }
535
536    /// Apply a merkleized batch to the database.
537    ///
538    /// Returns the range of locations written. The state is updated in memory and appended to the
539    /// witness journal. Call [`Self::commit`] or [`Self::sync`], or await the handle returned by
540    /// [`Self::start_sync`], to make the applied state durable.
541    ///
542    /// # Errors
543    ///
544    /// - [`Error::StaleBatch`] if the batch is detected as stale (see
545    ///   [`crate::qmdb::batch_chain`] for more details).
546    /// - [`Error::FloorRegressed`] if any commit in the chain declares a floor below the
547    ///   previous commit's floor.
548    /// - [`Error::FloorBeyondSize`] if any commit in the chain declares a floor beyond its own
549    ///   commit location.
550    #[tracing::instrument(
551        name = "qmdb.immutable.compact.db.apply_batch",
552        level = "info",
553        skip_all
554    )]
555    pub async fn apply_batch(
556        mut self,
557        batch: Arc<MerkleizedBatch<F, H::Digest, K, V, S>>,
558    ) -> Result<(Self, core::ops::Range<Location<F>>), Error<F>> {
559        self.validate_batch(&batch)?;
560
561        let start_loc = self.last_commit_loc + 1;
562        self.merkle.apply_batch(&batch.merkle_batch)?;
563        self.root = batch.root();
564        self.last_commit_loc = batch.bounds.tip.size - 1;
565        self.last_commit_metadata = batch.commit_metadata.clone();
566        self.inactivity_floor_loc = batch.bounds.inactivity_floor;
567        let last_commit_metadata = self.last_commit_metadata.clone();
568        let inactivity_floor_loc = self.inactivity_floor_loc;
569        self.witness = self
570            .witness
571            .apply::<H, S>(&self.merkle, inactivity_floor_loc, || {
572                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
573            })
574            .await?;
575        Ok((self, start_loc..batch.bounds.tip.size))
576    }
577
578    /// Begin durably persisting the current db state to disk.
579    ///
580    /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit],
581    /// plus a best-effort attempt to bound the recovery needed on reopen. Use [Self::sync] to
582    /// guarantee none is needed. A new sync waits for the prior sync before starting. Failures
583    /// of the deferred durability work surface on the returned handle and the next durability
584    /// operation.
585    #[tracing::instrument(
586        name = "qmdb.immutable.compact.db.start_sync",
587        level = "info",
588        skip_all
589    )]
590    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
591        let last_commit_metadata = self.last_commit_metadata.clone();
592        let inactivity_floor_loc = self.inactivity_floor_loc;
593        let handle;
594        (self.witness, handle) = self
595            .witness
596            .start_sync::<H, S>(&self.merkle, inactivity_floor_loc, || {
597                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
598            })
599            .await?;
600        Ok((self, handle))
601    }
602
603    /// Durably persist the current db state to disk. This is faster than [`Self::sync`] but
604    /// reopen may need to replay the witness journal's tail to recover.
605    #[tracing::instrument(name = "qmdb.immutable.compact.db.commit", level = "info", skip_all)]
606    pub async fn commit(mut self) -> Result<Self, Error<F>> {
607        let last_commit_metadata = self.last_commit_metadata.clone();
608        let inactivity_floor_loc = self.inactivity_floor_loc;
609        self.witness = self
610            .witness
611            .commit::<H, S>(&self.merkle, inactivity_floor_loc, || {
612                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
613            })
614            .await?;
615        Ok(self)
616    }
617
618    /// Durably persist the current db state to disk, also persisting journal metadata to
619    /// minimize recovery work on reopen.
620    #[tracing::instrument(name = "qmdb.immutable.compact.db.sync", level = "info", skip_all)]
621    pub async fn sync(mut self) -> Result<Self, Error<F>> {
622        let last_commit_metadata = self.last_commit_metadata.clone();
623        let inactivity_floor_loc = self.inactivity_floor_loc;
624        self.witness = self
625            .witness
626            .sync::<H, S>(&self.merkle, inactivity_floor_loc, || {
627                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
628            })
629            .await?;
630        Ok(self)
631    }
632
633    /// Rewind the db to the applied state with exactly `target` operations, discarding any
634    /// uncommitted batches and any later states. The rewind is made durable before this
635    /// method returns.
636    ///
637    /// # Errors
638    ///
639    /// Returns [`crate::merkle::Error::RewindBeyondHistory`] (wrapped as [`Error::Merkle`]) if
640    /// no retained applied state has exactly `target` operations (never applied, or pruned).
641    #[tracing::instrument(name = "qmdb.immutable.compact.db.rewind", level = "info", skip_all)]
642    pub async fn rewind(mut self, target: Location<F>) -> Result<Self, Error<F>>
643    where
644        F: Family,
645    {
646        // A clean current target only needs to settle its pipelined sync. An uncommitted target
647        // takes the regular rewind path so the witness journal becomes durable before return.
648        if self.size() == target
649            && self.witness.with(|w| w.size()) == target
650            && !self.witness.has_uncommitted_state()
651        {
652            self.witness.wait_for_sync().await?;
653            return Ok(self);
654        }
655
656        let last_commit_op;
657        (self.witness, last_commit_op) = self
658            .witness
659            .rewind::<H, S, Operation<F, K, V>>(&self.merkle, target, &self.commit_codec_config)
660            .await?;
661        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
662            return Err(Error::DataCorrupted("last operation was not a commit"));
663        };
664        self.last_commit_metadata = last_commit_metadata;
665        self.inactivity_floor_loc = inactivity_floor_loc;
666        self.last_commit_loc = target - 1;
667        self.root = self.witness.with(|w| w.root);
668        Ok(self)
669    }
670
671    /// Drop witnesses for commits with fewer than `pruning_boundary` operations. Some witness below
672    /// the boundary may survive.
673    ///
674    /// Pruning bounds how far back [`Self::rewind`] can reach; the current commit's witness always
675    /// survives. The prune is made durable before this method returns.
676    ///
677    /// # Errors
678    ///
679    /// Fails if a compact-sync import has not yet been applied to the witness journal.
680    #[tracing::instrument(name = "qmdb.immutable.compact.db.prune", level = "info", skip_all)]
681    pub async fn prune(mut self, pruning_boundary: Location<F>) -> Result<Self, Error<F>> {
682        self.witness = self.witness.prune(pruning_boundary).await?;
683        Ok(self)
684    }
685
686    /// Destroy all persisted state associated with this database.
687    #[boxed]
688    pub async fn destroy(self) -> Result<(), Error<F>> {
689        self.witness.destroy().await?;
690        Ok(())
691    }
692}
693
694impl<F, E, K, V, H, C, S> Source for Db<F, E, K, V, H, C, S>
695where
696    F: Family,
697    E: Context,
698    K: Key,
699    V: ValueEncoding,
700    H: Hasher,
701    Operation<F, K, V>: EncodeShared + Read<Cfg = C>,
702    C: Clone + Send + Sync + 'static,
703    S: Strategy,
704{
705    type Family = F;
706    type Digest = H::Digest;
707    type Op = Operation<F, K, V>;
708    type Error = qmdb::Error<F>;
709
710    async fn serve(
711        &self,
712        request: Request<F>,
713    ) -> Result<(Response<F, Self::Op, H::Digest>, FeedbackTx), Self::Error> {
714        Ok((
715            self.witness
716                .compact_state(&self.commit_codec_config, request)?,
717            None,
718        ))
719    }
720}
721
722#[cfg(test)]
723mod tests {
724    use super::*;
725    use crate::{
726        merkle::{mmb, mmr},
727        qmdb::{
728            any::value::FixedEncoding, compact::witness, verify_proof,
729            verify_proof_and_pinned_nodes,
730        },
731    };
732    use commonware_cryptography::{Sha256, sha256::Digest};
733    use commonware_macros::test_traced;
734    use commonware_parallel::Sequential;
735    use commonware_runtime::{
736        BufferPooler, Runner as _, Supervisor as _,
737        buffer::paged::CacheRef,
738        deterministic,
739        mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs},
740    };
741    use commonware_utils::{NZU16, NZU64, NZUsize};
742    use core::future::Future;
743    use futures::FutureExt as _;
744    use std::num::{NonZeroU16, NonZeroUsize};
745
746    type TestDb<F> =
747        Db<F, deterministic::Context, Digest, FixedEncoding<Digest>, Sha256, (), Sequential>;
748
749    const WITNESS_PAGE_SIZE: NonZeroU16 = NZU16!(77);
750    const WITNESS_PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9);
751
752    fn witness_config(partition: &str, pooler: &impl BufferPooler) -> JournalConfig<()> {
753        JournalConfig {
754            partition: format!("{partition}-witness"),
755            items_per_section: NZU64!(64),
756            compression: None,
757            codec_config: (),
758            page_cache: CacheRef::from_pooler(pooler, WITNESS_PAGE_SIZE, WITNESS_PAGE_CACHE_SIZE),
759            write_buffer: NZUsize!(1024),
760            replay_buffer: NZUsize!(1024),
761        }
762    }
763
764    async fn open_db<F: Family>(context: deterministic::Context, partition: &str) -> TestDb<F> {
765        let witness_cfg = witness_config(partition, &context);
766        let merkle = crate::merkle::compact::Merkle::new(Sequential);
767        Db::init_from_merkle(merkle, context.child("witness"), witness_cfg, ())
768            .await
769            .unwrap()
770    }
771
772    /// Batch artifacts (operations, range proof, pinned frontier) verify against the batch root,
773    /// survive applying and dropping ancestors, and are refused once the batch itself is applied
774    /// and the compact store is pruned past them.
775    async fn compact_operations_and_proof_inner<F: Family>(context: deterministic::Context) {
776        let db = open_db::<F>(context.child("db"), "immutable-operations-and-proof").await;
777        let value = |key: u8| Sha256::fill(key.wrapping_add(100));
778
779        // Seed committed state so the chain below forks above a pruned frontier.
780        let mut seed = db.new_batch();
781        for key in 1u8..=6 {
782            seed = seed.set(Sha256::fill(key), value(key));
783        }
784        let seed = seed
785            .merkleize(&db, Some(Sha256::fill(7)), Location::new(0))
786            .await;
787        let (db, _) = db.apply_batch(seed).await.unwrap();
788        let db = db.sync().await.unwrap();
789        let floor = db.size();
790        assert_eq!(floor, Location::new(8));
791
792        // A snapshot batch has no operations to prove.
793        assert!(matches!(
794            db.to_batch().proof(&db),
795            Err(Error::Merkle(crate::merkle::Error::Empty))
796        ));
797
798        // A two-deep unapplied chain: the child's artifacts read the parent's nodes through the
799        // live chain.
800        let mut parent = db.new_batch();
801        for key in 8u8..=12 {
802            parent = parent.set(Sha256::fill(key), value(key));
803        }
804        let parent = parent.merkleize(&db, Some(Sha256::fill(13)), floor).await;
805        let child = parent
806            .new_batch::<Sha256>()
807            .set(Sha256::fill(14), value(14))
808            .merkleize(&db, Some(Sha256::fill(15)), floor)
809            .await;
810
811        // The operation suffix is the batch's own sets plus its commit, handed out zero-copy.
812        let (child_start, child_ops) = child.operations();
813        let (_, child_ops_again) = child.operations();
814        let child_end = child.bounds().tip.size;
815        assert!(Arc::ptr_eq(&child_ops, &child_ops_again));
816        assert_eq!(child_start, parent.bounds().tip.size);
817        assert_eq!(*child_start + child_ops.len() as u64, *child_end);
818        assert_eq!(child_end, Location::new(16));
819        assert!(matches!(
820            child_ops.as_slice(),
821            [Operation::Set(key, set_value), Operation::Commit(Some(metadata), operation_floor)]
822                if key == &Sha256::fill(14)
823                    && set_value == &value(14)
824                    && metadata == &Sha256::fill(15)
825                    && operation_floor == &floor
826        ));
827
828        // The proof is anchored at the batch tip and verifies with or without the pins.
829        let child_root = child.root();
830        let child_proof = child.proof(&db).unwrap();
831        let child_pins = child.pinned_nodes(&db).unwrap();
832        assert_eq!(child_proof.leaves, child_end);
833        assert_eq!(
834            child_proof.inactive_peaks,
835            F::inactive_peaks(child_end, floor),
836        );
837        assert!(verify_proof::<Sha256, _, _>(
838            &child_proof,
839            child_start,
840            &child_ops,
841            &child_root
842        ));
843        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
844            &child_proof,
845            child_start,
846            &child_ops,
847            &child_pins,
848            &child_root
849        ));
850
851        // The pins are order-sensitive.
852        assert!(child_pins.len() > 1);
853        let mut reordered_child_pins = child_pins.clone();
854        reordered_child_pins.swap(0, 1);
855        assert!(!verify_proof_and_pinned_nodes::<Sha256, _, _>(
856            &child_proof,
857            child_start,
858            &child_ops,
859            &reordered_child_pins,
860            &child_root
861        ));
862
863        // Pipelined consumer: applying the parent prunes the store to the parent's tip, which is
864        // exactly the frontier the child's base pins, so the artifacts survive dropping the
865        // parent.
866        let (db, _) = db.apply_batch(parent).await.unwrap();
867        let child_proof_after = child.proof(&db).unwrap();
868        assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins);
869        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
870            &child_proof_after,
871            child_start,
872            &child_ops,
873            &child_pins,
874            &child_root
875        ));
876        let (db, child_range) = db.apply_batch(child).await.unwrap();
877        assert_eq!(child_range, child_start..child_end);
878
879        // A commit-only batch proves exactly its commit operation.
880        let commit_floor = db.size();
881        let commit_only = db
882            .new_batch()
883            .merkleize(&db, Some(Sha256::fill(16)), commit_floor)
884            .await;
885        let (commit_start, commit_ops) = commit_only.operations();
886        let commit_end = commit_only.bounds().tip.size;
887        let commit_root = commit_only.root();
888        let commit_proof = commit_only.proof(&db).unwrap();
889        let commit_pins = commit_only.pinned_nodes(&db).unwrap();
890        assert_eq!(commit_start, commit_floor);
891        assert!(matches!(
892            commit_ops.as_slice(),
893            [Operation::Commit(Some(metadata), operation_floor)]
894                if metadata == &Sha256::fill(16) && operation_floor == &commit_floor
895        ));
896        assert_eq!(*commit_start + commit_ops.len() as u64, *commit_end);
897        assert_eq!(commit_proof.leaves, commit_end);
898        assert_eq!(
899            commit_proof.inactive_peaks,
900            F::inactive_peaks(commit_end, commit_floor)
901        );
902        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
903            &commit_proof,
904            commit_start,
905            &commit_ops,
906            &commit_pins,
907            &commit_root
908        ));
909        let (db, commit_range) = db.apply_batch(commit_only).await.unwrap();
910        assert_eq!(commit_range, commit_start..commit_end);
911        let db = db.sync().await.unwrap();
912
913        // Applying a batch that forked mid-mountain prunes its own artifacts. The accessors
914        // refuse rather than returning a proof that fails to verify, while a child merkleized
915        // before the apply still finds its base frontier in the store.
916        let late_parent = db
917            .new_batch()
918            .set(Sha256::fill(17), value(17))
919            .merkleize(&db, Some(Sha256::fill(18)), db.size())
920            .await;
921        let late = late_parent
922            .new_batch::<Sha256>()
923            .set(Sha256::fill(19), value(19))
924            .merkleize(&db, Some(Sha256::fill(20)), db.size())
925            .await;
926        let (db, _) = db.apply_batch(Arc::clone(&late_parent)).await.unwrap();
927        assert!(matches!(
928            late_parent.proof(&db),
929            Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
930        ));
931        assert!(matches!(
932            late_parent.pinned_nodes(&db),
933            Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
934        ));
935        drop(late_parent);
936
937        let (late_start, late_ops) = late.operations();
938        let late_root = late.root();
939        let late_proof = late.proof(&db).unwrap();
940        let late_pins = late.pinned_nodes(&db).unwrap();
941        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
942            &late_proof,
943            late_start,
944            &late_ops,
945            &late_pins,
946            &late_root
947        ));
948        let (db, _) = db.apply_batch(late).await.unwrap();
949
950        db.destroy().await.unwrap();
951    }
952
953    #[test_traced]
954    fn test_compact_operations_and_proof_mmr() {
955        deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmr::Family>);
956    }
957
958    #[test_traced]
959    fn test_compact_operations_and_proof_mmb() {
960        deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmb::Family>);
961    }
962
963    /// Open the persisted witness journal directly so tests can corrupt the tip entry.
964    async fn open_witness_journal(
965        context: deterministic::Context,
966        partition: &str,
967    ) -> witness::Journal<deterministic::Context, mmr::Family, Digest> {
968        let cfg = witness_config(partition, &context);
969        witness::Journal::init(context, cfg).await.unwrap()
970    }
971
972    /// A compact db over a delayed-sync storage backend.
973    type DelayedDb = Db<
974        mmr::Family,
975        DelayedSyncContext<deterministic::Context>,
976        Digest,
977        FixedEncoding<Digest>,
978        Sha256,
979        (),
980        Sequential,
981    >;
982
983    /// Open a [DelayedDb] whose blob syncs park on `pending`.
984    ///
985    /// Init durably persists the bootstrap witness, so while syncs park the returned future
986    /// must be driven with `drive_pending_syncs` (or the mock unblocked first).
987    fn open_delayed_db(
988        context: &deterministic::Context,
989        label: &'static str,
990        partition: &str,
991        pending: &PendingSyncs,
992    ) -> impl Future<Output = Result<DelayedDb, Error<mmr::Family>>> {
993        let witness_cfg = witness_config(partition, context);
994        let merkle = crate::merkle::compact::Merkle::new(Sequential);
995        let context = DelayedSyncContext {
996            inner: context.child(label),
997            pending: pending.clone(),
998        };
999        DelayedDb::init_from_merkle(merkle, context.child("witness"), witness_cfg, ())
1000    }
1001
1002    /// Apply a single-key batch writing `key -> value` with `metadata`.
1003    async fn apply_set(db: DelayedDb, key: Digest, value: Digest, metadata: Digest) -> DelayedDb {
1004        let floor = db.inactivity_floor_loc();
1005        let batch = db
1006            .new_batch()
1007            .set(key, value)
1008            .merkleize(&db, Some(metadata), floor)
1009            .await;
1010        let (db, _) = db.apply_batch(batch).await.unwrap();
1011        db
1012    }
1013
1014    /// State persisted via an awaited start_sync handle is recovered on reopen.
1015    #[test_traced]
1016    fn test_compact_start_sync_recovery() {
1017        deterministic::Runner::default().start(|ctx| async move {
1018            let partition = "immutable-start-sync-recovery";
1019            let pending = PendingSyncs::default();
1020            pending.unblock();
1021            let mut db = open_delayed_db(&ctx, "delayed", partition, &pending)
1022                .await
1023                .unwrap();
1024            let metadata = Sha256::fill(9u8);
1025            db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), metadata).await;
1026
1027            let handle;
1028            (db, handle) = db.start_sync().await.unwrap();
1029            handle.await.unwrap();
1030            let root = db.root();
1031            drop(db);
1032
1033            let db = open_delayed_db(&ctx, "reopen", partition, &pending)
1034                .await
1035                .unwrap();
1036            assert_eq!(db.root(), root);
1037            assert_eq!(db.get_metadata(), Some(metadata));
1038            db.destroy().await.unwrap();
1039        });
1040    }
1041
1042    /// A sync begun by `start_sync` that fails in flight surfaces the error through both the
1043    /// returned handle and the next durability operation, even when that operation has nothing
1044    /// new to persist.
1045    #[test_traced]
1046    fn test_compact_start_sync_failure_propagates() {
1047        deterministic::Runner::default().start(|ctx| async move {
1048            let pending = PendingSyncs::default();
1049            pending.unblock();
1050            let mut db = open_delayed_db(&ctx, "delayed", "immutable-start-sync-fail", &pending)
1051                .await
1052                .unwrap();
1053            db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), Sha256::fill(9u8)).await;
1054
1055            // Arm all future syncs to resolve to an injected error.
1056            pending.arm_fail();
1057
1058            let handle;
1059            (db, handle) = db.start_sync().await.unwrap();
1060            assert!(
1061                handle.await.is_err(),
1062                "the sync handle surfaces the failure"
1063            );
1064            let starts_before = pending.starts();
1065
1066            // The witness entry was already appended, so this commit has nothing to stage.
1067            // It must still observe the retained failure rather than no-op.
1068            assert!(
1069                db.commit().await.is_err(),
1070                "the next durability op surfaces the failed in-flight sync"
1071            );
1072            assert_eq!(
1073                pending.starts(),
1074                starts_before,
1075                "the surfaced error is the retained failure, not a fresh sync's"
1076            );
1077        });
1078    }
1079
1080    /// A rewind to the current size waits for the in-flight sync and adopts its proof of
1081    /// durability instead of starting new journal work.
1082    #[test_traced]
1083    fn test_compact_start_sync_rewind_fast_path_drains() {
1084        deterministic::Runner::default().start(|ctx| async move {
1085            let partition = "immutable-start-sync-rewind-drain";
1086            let pending = PendingSyncs::default();
1087            let open = open_delayed_db(&ctx, "delayed", partition, &pending);
1088            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1089            db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), Sha256::fill(9u8)).await;
1090
1091            let handle;
1092            (db, handle) = db.start_sync().await.unwrap();
1093            let root = db.root();
1094            let size = db.size();
1095
1096            let starts_before = pending.starts();
1097            let db = {
1098                let mut rewind = std::pin::pin!(db.rewind(size));
1099                assert!(
1100                    rewind.as_mut().now_or_never().is_none(),
1101                    "rewind proceeded while the started sync was pending"
1102                );
1103                pending.unblock();
1104                rewind.await.unwrap()
1105            };
1106            handle.await.unwrap();
1107            assert_eq!(
1108                pending.starts(),
1109                starts_before,
1110                "the fast path started journal work instead of adopting the proven sync"
1111            );
1112            assert_eq!(db.root(), root);
1113            drop(db);
1114
1115            // The awaited pipelined sync made the witness entry durable.
1116            let db = open_delayed_db(&ctx, "reopen", partition, &pending)
1117                .await
1118                .unwrap();
1119            assert_eq!(db.root(), root);
1120            db.destroy().await.unwrap();
1121        });
1122    }
1123
1124    /// A rewind to the current size fails when the sync started for the tip witness has
1125    /// already failed, rather than reporting the unproven tip as durable.
1126    #[test_traced]
1127    fn test_compact_start_sync_rewind_fast_path_fails() {
1128        deterministic::Runner::default().start(|ctx| async move {
1129            let pending = PendingSyncs::default();
1130            pending.unblock();
1131            let mut db = open_delayed_db(
1132                &ctx,
1133                "delayed",
1134                "immutable-start-sync-rewind-fail",
1135                &pending,
1136            )
1137            .await
1138            .unwrap();
1139            db = apply_set(db, Sha256::fill(1u8), Sha256::fill(2u8), Sha256::fill(9u8)).await;
1140
1141            pending.arm_fail();
1142            let handle;
1143            (db, handle) = db.start_sync().await.unwrap();
1144            assert!(handle.await.is_err());
1145            let size = db.size();
1146            assert!(
1147                db.rewind(size).await.is_err(),
1148                "rewind reported an unproven tip as durable"
1149            );
1150        });
1151    }
1152
1153    #[test_traced("INFO")]
1154    fn test_compact_stale_batch_rejected() {
1155        deterministic::Runner::default().start(|context| async move {
1156            let db = open_db::<mmr::Family>(context.child("db"), "immutable-stale").await;
1157
1158            let key1 = Sha256::hash(&[&[1]]);
1159            let key2 = Sha256::hash(&[&[2]]);
1160            let value1 = Sha256::fill(10u8);
1161            let value2 = Sha256::fill(20u8);
1162
1163            let batch_a = db
1164                .new_batch()
1165                .set(key1, value1)
1166                .merkleize(&db, None, Location::new(0))
1167                .await;
1168            let batch_b = db
1169                .new_batch()
1170                .set(key2, value2)
1171                .merkleize(&db, None, Location::new(0))
1172                .await;
1173
1174            let expected_root = batch_a.root();
1175            let (db, _) = db.apply_batch(batch_a).await.unwrap();
1176            assert_eq!(db.root(), expected_root);
1177            assert!(matches!(
1178                db.apply_batch(batch_b).await,
1179                Err(Error::StaleBatch)
1180            ));
1181        });
1182    }
1183
1184    #[test_traced("INFO")]
1185    fn test_compact_delayed_merkleize_after_ancestor_apply() {
1186        deterministic::Runner::default().start(|context| async move {
1187            let db = open_db::<mmr::Family>(context.child("db"), "immutable-delayed-child").await;
1188            let key1 = Sha256::hash(&[&[1]]);
1189            let key2 = Sha256::hash(&[&[2]]);
1190            let key3 = Sha256::hash(&[&[3]]);
1191            let value1 = Sha256::fill(10u8);
1192            let value2 = Sha256::fill(20u8);
1193            let value3 = Sha256::fill(30u8);
1194
1195            let a = db
1196                .new_batch()
1197                .set(key1, value1)
1198                .merkleize(&db, None, Location::new(0))
1199                .await;
1200            let b = a
1201                .new_batch::<Sha256>()
1202                .set(key2, value2)
1203                .merkleize(&db, None, Location::new(0))
1204                .await;
1205            let c = b.new_batch::<Sha256>().set(key3, value3);
1206
1207            let (db, _) = db.apply_batch(a).await.unwrap();
1208            let c = c.merkleize(&db, None, Location::new(0)).await;
1209            let expected_root = c.root();
1210            let (db, _) = db.apply_batch(c).await.unwrap();
1211
1212            assert_eq!(db.root(), expected_root);
1213        });
1214    }
1215
1216    /// `to_batch()` reflects the current applied state before it becomes durable.
1217    #[test_traced("INFO")]
1218    fn test_compact_to_batch_reflects_live_state() {
1219        deterministic::Runner::default().start(|context| async move {
1220            let db = open_db::<mmr::Family>(context.child("db"), "immutable-to-batch-live").await;
1221
1222            let pre_apply_root = db.root();
1223            let pre_snapshot = db.to_batch();
1224            assert_eq!(
1225                pre_snapshot.root(),
1226                pre_apply_root,
1227                "snapshot before any mutation should match the live root"
1228            );
1229
1230            let key = Sha256::hash(&[&[1]]);
1231            let value = Sha256::fill(10u8);
1232            let batch = db
1233                .new_batch()
1234                .set(key, value)
1235                .merkleize(&db, None, Location::new(0))
1236                .await;
1237            let (db, _) = db.apply_batch(batch).await.unwrap();
1238
1239            // Observe the applied state before making it durable.
1240            let live_root = db.root();
1241            assert_ne!(
1242                live_root, pre_apply_root,
1243                "applying a non-empty batch must change the live root"
1244            );
1245
1246            let snapshot = db.to_batch();
1247            assert_eq!(
1248                snapshot.root(),
1249                live_root,
1250                "to_batch().root() must match the live db.root() even before sync"
1251            );
1252
1253            db.destroy().await.unwrap();
1254        });
1255    }
1256
1257    #[test_traced("INFO")]
1258    fn test_compact_stale_batch_chained() {
1259        deterministic::Runner::default().start(|context| async move {
1260            let db = open_db::<mmr::Family>(context.child("db"), "immutable-chained-stale").await;
1261
1262            let common_parent = db
1263                .new_batch()
1264                .set(Sha256::hash(&[&[10]]), Sha256::fill(10u8))
1265                .merkleize(&db, None, Location::new(0))
1266                .await;
1267            let sibling_a = common_parent
1268                .new_batch::<Sha256>()
1269                .set(Sha256::hash(&[&[11]]), Sha256::fill(11u8))
1270                .merkleize(&db, None, Location::new(0))
1271                .await;
1272            let sibling_b = common_parent
1273                .new_batch::<Sha256>()
1274                .set(Sha256::hash(&[&[12]]), Sha256::fill(12u8))
1275                .merkleize(&db, None, Location::new(0))
1276                .await;
1277            let (db, _) = db.apply_batch(sibling_a).await.unwrap();
1278            assert!(matches!(
1279                db.validate_batch(&sibling_b),
1280                Err(Error::StaleBatch)
1281            ));
1282
1283            let parent_a = db
1284                .new_batch()
1285                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1286                .merkleize(&db, None, Location::new(0))
1287                .await;
1288            let parent_b = db
1289                .new_batch()
1290                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
1291                .merkleize(&db, None, Location::new(0))
1292                .await;
1293            let child_b = parent_b
1294                .new_batch::<Sha256>()
1295                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
1296                .merkleize(&db, None, Location::new(0))
1297                .await;
1298
1299            let (db, _) = db.apply_batch(parent_a).await.unwrap();
1300            assert!(matches!(
1301                db.validate_batch(&child_b),
1302                Err(Error::StaleBatch)
1303            ));
1304            db.destroy().await.unwrap();
1305        });
1306    }
1307
1308    #[test_traced("INFO")]
1309    fn test_compact_stale_parent_after_child_applied() {
1310        deterministic::Runner::default().start(|context| async move {
1311            let db =
1312                open_db::<mmr::Family>(context.child("db"), "immutable-child-before-parent").await;
1313
1314            let parent = db
1315                .new_batch()
1316                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1317                .merkleize(&db, None, Location::new(0))
1318                .await;
1319            let child = parent
1320                .new_batch::<Sha256>()
1321                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
1322                .merkleize(&db, None, Location::new(0))
1323                .await;
1324
1325            let (db, _) = db.apply_batch(child).await.unwrap();
1326            assert!(matches!(
1327                db.apply_batch(parent).await,
1328                Err(Error::StaleBatch)
1329            ));
1330        });
1331    }
1332
1333    #[test_traced("INFO")]
1334    fn test_compact_sequential_commit_parent_then_child() {
1335        deterministic::Runner::default().start(|context| async move {
1336            let db = open_db::<mmr::Family>(context.child("db"), "immutable-parent-child").await;
1337
1338            let parent = db
1339                .new_batch()
1340                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1341                .merkleize(&db, None, Location::new(0))
1342                .await;
1343            let child = parent
1344                .new_batch::<Sha256>()
1345                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
1346                .merkleize(&db, None, Location::new(0))
1347                .await;
1348            let expected_root = child.root();
1349
1350            let (db, _) = db.apply_batch(parent).await.unwrap();
1351            let (db, _) = db.apply_batch(child).await.unwrap();
1352            let db = db.sync().await.unwrap();
1353
1354            assert_eq!(db.root(), expected_root);
1355
1356            db.destroy().await.unwrap();
1357        });
1358    }
1359
1360    #[test_traced("INFO")]
1361    fn test_compact_floor_regressed() {
1362        deterministic::Runner::default().start(|context| async move {
1363            let db = open_db::<mmr::Family>(context.child("db"), "immutable-floor-regressed").await;
1364
1365            let advance_floor = db.new_batch().set(Sha256::hash(&[&[1]]), Sha256::fill(1u8));
1366            let advance_floor = advance_floor.merkleize(&db, None, Location::new(1)).await;
1367            let (db, _) = db.apply_batch(advance_floor).await.unwrap();
1368            let db = db.sync().await.unwrap();
1369            let target = db.target();
1370
1371            let regressed = db
1372                .new_batch()
1373                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
1374                .merkleize(&db, None, Location::new(0))
1375                .await;
1376
1377            assert!(matches!(
1378                db.apply_batch(regressed).await,
1379                Err(Error::FloorRegressed(new, current))
1380                    if new == Location::new(0) && current == Location::new(1)
1381            ));
1382
1383            // Reopen and verify the rejected batch persisted nothing.
1384            let db =
1385                open_db::<mmr::Family>(context.child("reopen"), "immutable-floor-regressed").await;
1386            assert_eq!(db.target(), target);
1387        });
1388    }
1389
1390    // A chained batch whose tip floor is below its parent's floor must be rejected:
1391    // the parent's Commit participates in the per-commit monotonicity invariant even
1392    // before it is applied.
1393    #[test_traced("INFO")]
1394    fn test_compact_ancestor_floor_regressed() {
1395        deterministic::Runner::default().start(|context| async move {
1396            let db =
1397                open_db::<mmr::Family>(context.child("db"), "immutable-regressed-ancestor-floor")
1398                    .await;
1399
1400            let parent = db
1401                .new_batch()
1402                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1403                .merkleize(&db, None, Location::new(1))
1404                .await;
1405            let child = parent
1406                .new_batch::<Sha256>()
1407                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
1408                .merkleize(&db, None, Location::new(0))
1409                .await;
1410
1411            let target = db.target();
1412            assert!(matches!(
1413                db.apply_batch(child).await,
1414                Err(Error::FloorRegressed(new, prev))
1415                    if new == Location::new(0) && prev == Location::new(1)
1416            ));
1417
1418            // Reopen and verify the rejected chain persisted nothing.
1419            let db = open_db::<mmr::Family>(
1420                context.child("reopen"),
1421                "immutable-regressed-ancestor-floor",
1422            )
1423            .await;
1424            assert_eq!(db.target(), target);
1425        });
1426    }
1427
1428    #[test_traced("INFO")]
1429    fn test_compact_rewind_restores_commit_metadata_and_floor() {
1430        deterministic::Runner::default().start(|context| async move {
1431            let db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-meta").await;
1432
1433            let k1 = Sha256::hash(&[&[1]]);
1434            let v1 = Sha256::fill(11u8);
1435            let meta1 = Sha256::fill(0xaa);
1436            let floor1 = Location::new(0);
1437            let batch = db
1438                .new_batch()
1439                .set(k1, v1)
1440                .merkleize(&db, Some(meta1), floor1)
1441                .await;
1442            let (db, _) = db.apply_batch(batch).await.unwrap();
1443            let db = db.sync().await.unwrap();
1444            let root_after_first = db.root();
1445            let size_after_first = db.size();
1446
1447            let k2 = Sha256::hash(&[&[2]]);
1448            let v2 = Sha256::fill(22u8);
1449            let meta2 = Sha256::fill(0xbb);
1450            // Advance the floor to the commit of the first batch (loc 1).
1451            let floor2 = Location::new(1);
1452            let batch = db
1453                .new_batch()
1454                .set(k2, v2)
1455                .merkleize(&db, Some(meta2), floor2)
1456                .await;
1457            let (db, _) = db.apply_batch(batch).await.unwrap();
1458            let db = db.sync().await.unwrap();
1459            assert_eq!(db.get_metadata(), Some(meta2));
1460            assert_eq!(db.inactivity_floor_loc(), floor2);
1461
1462            let db = db.rewind(size_after_first).await.unwrap();
1463            assert_eq!(db.root(), root_after_first);
1464            assert_eq!(db.get_metadata(), Some(meta1));
1465            assert_eq!(db.inactivity_floor_loc(), floor1);
1466
1467            db.destroy().await.unwrap();
1468        });
1469    }
1470
1471    #[test_traced("INFO")]
1472    fn test_compact_rewind_persists_across_reopen() {
1473        deterministic::Runner::default().start(|context| async move {
1474            let partition = "immutable-rewind-reopen";
1475            let meta1 = Sha256::fill(0xaa);
1476            let floor1 = Location::new(0);
1477            let meta2 = Sha256::fill(0xbb);
1478            let floor2 = Location::new(1);
1479
1480            let root_after_first = {
1481                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
1482                let batch = db
1483                    .new_batch()
1484                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
1485                    .merkleize(&db, Some(meta1), floor1)
1486                    .await;
1487                let (db, _) = db.apply_batch(batch).await.unwrap();
1488                let db = db.sync().await.unwrap();
1489                let root = db.root();
1490                let size_after_first = db.size();
1491
1492                let batch = db
1493                    .new_batch()
1494                    .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8))
1495                    .merkleize(&db, Some(meta2), floor2)
1496                    .await;
1497                let (db, _) = db.apply_batch(batch).await.unwrap();
1498                let db = db.sync().await.unwrap();
1499
1500                let _db = db.rewind(size_after_first).await.unwrap();
1501                root
1502            };
1503
1504            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
1505            assert_eq!(db.root(), root_after_first);
1506            assert_eq!(db.get_metadata(), Some(meta1));
1507            assert_eq!(db.inactivity_floor_loc(), floor1);
1508
1509            db.destroy().await.unwrap();
1510        });
1511    }
1512
1513    #[test_traced("INFO")]
1514    fn test_compact_commit_persists_across_reopen() {
1515        deterministic::Runner::default().start(|context| async move {
1516            let partition = "immutable-commit-reopen";
1517            let meta1 = Sha256::fill(0xaa);
1518            let meta2 = Sha256::fill(0xbb);
1519
1520            let root_after_second = {
1521                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
1522                let batch = db
1523                    .new_batch()
1524                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
1525                    .merkleize(&db, Some(meta1), Location::new(0))
1526                    .await;
1527                let (db, _) = db.apply_batch(batch).await.unwrap();
1528                let db = db.commit().await.unwrap();
1529
1530                let batch = db
1531                    .new_batch()
1532                    .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8))
1533                    .merkleize(&db, Some(meta2), Location::new(1))
1534                    .await;
1535                let (db, _) = db.apply_batch(batch).await.unwrap();
1536                let db = db.commit().await.unwrap();
1537                db.root()
1538            };
1539
1540            // Reopen recovers the committed tip even though the journal was never synced.
1541            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
1542            assert_eq!(db.root(), root_after_second);
1543            assert_eq!(db.get_metadata(), Some(meta2));
1544            assert_eq!(db.inactivity_floor_loc(), Location::new(1));
1545            db.destroy().await.unwrap();
1546        });
1547    }
1548
1549    #[test_traced("INFO")]
1550    fn test_compact_rewind_to_committed_entry_after_reopen() {
1551        deterministic::Runner::default().start(|context| async move {
1552            let partition = "immutable-commit-rewind-reopen";
1553            let meta1 = Sha256::fill(0xaa);
1554            let meta2 = Sha256::fill(0xbb);
1555
1556            let (root_a, size_a) = {
1557                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
1558                let batch = db
1559                    .new_batch()
1560                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
1561                    .merkleize(&db, Some(meta1), Location::new(0))
1562                    .await;
1563                let (db, _) = db.apply_batch(batch).await.unwrap();
1564                let db = db.commit().await.unwrap();
1565                let root_a = db.root();
1566                let size_a = db.size();
1567
1568                let batch = db
1569                    .new_batch()
1570                    .set(Sha256::hash(&[&[2]]), Sha256::fill(22u8))
1571                    .merkleize(&db, Some(meta2), Location::new(1))
1572                    .await;
1573                let (db, _) = db.apply_batch(batch).await.unwrap();
1574                let _db = db.commit().await.unwrap();
1575                (root_a, size_a)
1576            };
1577
1578            // Both committed witnesses survive the crash: reopen recovers the tip, and the
1579            // earlier commit remains a valid rewind target.
1580            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
1581            let db = db.rewind(size_a).await.unwrap();
1582            assert_eq!(db.root(), root_a);
1583            assert_eq!(db.get_metadata(), Some(meta1));
1584            db.destroy().await.unwrap();
1585        });
1586    }
1587
1588    #[test_traced("INFO")]
1589    fn test_compact_sync_after_commit() {
1590        deterministic::Runner::default().start(|context| async move {
1591            let partition = "immutable-sync-after-commit";
1592            let meta = Sha256::fill(0xaa);
1593
1594            let root = {
1595                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
1596                let batch = db
1597                    .new_batch()
1598                    .set(Sha256::hash(&[&[1]]), Sha256::fill(11u8))
1599                    .merkleize(&db, Some(meta), Location::new(0))
1600                    .await;
1601                let (db, _) = db.apply_batch(batch).await.unwrap();
1602                let db = db.commit().await.unwrap();
1603                // The commit already made the state durable, so this is a no-op.
1604                let db = db.sync().await.unwrap();
1605                db.root()
1606            };
1607
1608            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
1609            assert_eq!(db.root(), root);
1610            assert_eq!(db.get_metadata(), Some(meta));
1611            db.destroy().await.unwrap();
1612        });
1613    }
1614
1615    #[test_traced("INFO")]
1616    fn test_compact_reopen_rejects_tampered_witness() {
1617        deterministic::Runner::default().start(|context| async move {
1618            let partition = "immutable-witness-tamper";
1619            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
1620            let batch = db
1621                .new_batch()
1622                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
1623                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1624                .await;
1625            let (db, _) = db.apply_batch(batch).await.unwrap();
1626            let db = db.sync().await.unwrap();
1627            drop(db);
1628
1629            // Corrupt the entry structurally. An extra pinned node cannot rebuild the Merkle.
1630            let journal = open_witness_journal(context.child("tamper"), partition).await;
1631            let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
1632            pinned_nodes.push(Sha256::fill(0xff));
1633            witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;
1634
1635            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1636            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1637                merkle,
1638                context.child("reopen_witness"),
1639                witness_config(partition, &context),
1640                (),
1641            )
1642            .await;
1643            assert!(matches!(reopened, Err(Error::DataCorrupted(_))));
1644        });
1645    }
1646
1647    #[test_traced("INFO")]
1648    fn test_compact_rewind_rejects_corrupt_target_entry() {
1649        deterministic::Runner::default().start(|context| async move {
1650            let partition = "immutable-corrupt-rewind-target";
1651            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
1652            let batch = db
1653                .new_batch()
1654                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1655                .merkleize(&db, None, Location::new(1))
1656                .await;
1657            let (db, _) = db.apply_batch(batch).await.unwrap();
1658            let db = db.sync().await.unwrap();
1659            let rewind_target = db.target().size;
1660            let batch = db
1661                .new_batch()
1662                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
1663                .merkleize(&db, None, Location::new(1))
1664                .await;
1665            let (db, _) = db.apply_batch(batch).await.unwrap();
1666            let db = db.sync().await.unwrap();
1667            let tip_target = db.target();
1668            drop(db);
1669
1670            // Corrupt the rewind target's entry (the journal holds bootstrap, target, tip).
1671            let mut journal = open_witness_journal(context.child("corrupt"), partition).await;
1672            journal = witness::tests::corrupt_entry(journal, 1, |entry| {
1673                entry.pinned_nodes.push(Sha256::fill(0xff));
1674            })
1675            .await;
1676            drop(journal);
1677
1678            // The tip entry is intact, so reopen succeeds.
1679            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1680            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1681                merkle,
1682                context.child("reopen"),
1683                witness_config(partition, &context),
1684                (),
1685            )
1686            .await
1687            .unwrap();
1688            assert_eq!(reopened.target(), tip_target);
1689
1690            // The corrupt entry fails the rewind before any truncation.
1691            assert!(matches!(
1692                reopened.rewind(rewind_target).await,
1693                Err(Error::DataCorrupted(_))
1694            ));
1695
1696            // The newer history survives: reopen still lands on the original tip.
1697            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1698            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1699                merkle,
1700                context.child("reopen2"),
1701                witness_config(partition, &context),
1702                (),
1703            )
1704            .await
1705            .unwrap();
1706            assert_eq!(reopened.target(), tip_target);
1707            reopened.destroy().await.unwrap();
1708        });
1709    }
1710
1711    #[test_traced("INFO")]
1712    fn test_compact_reopen_rejects_interrupted_import() {
1713        deterministic::Runner::default().start(|context| async move {
1714            let partition = "immutable-interrupted-import";
1715            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
1716            let batch = db
1717                .new_batch()
1718                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
1719                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1720                .await;
1721            let (db, _) = db.apply_batch(batch).await.unwrap();
1722            let db = db.sync().await.unwrap();
1723            drop(db);
1724
1725            // Simulate a crash between an import's journal clear and its entry append: the
1726            // journal is empty but its size is nonzero.
1727            let journal = open_witness_journal(context.child("clear"), partition).await;
1728            let size = journal.size();
1729            let journal = journal.clear_to_size(size.max(1)).await.unwrap();
1730            drop(journal);
1731
1732            // Reopen must fail rather than bootstrap a fresh db.
1733            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1734            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1735                merkle,
1736                context.child("reopen_witness"),
1737                witness_config(partition, &context),
1738                (),
1739            )
1740            .await;
1741            assert!(matches!(reopened, Err(Error::Journal(_))));
1742        });
1743    }
1744
1745    #[test_traced("INFO")]
1746    fn test_compact_reopen_rejects_commit_floor_beyond_tip() {
1747        deterministic::Runner::default().start(|context| async move {
1748            let partition = "immutable-invalid-persisted-floor";
1749            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
1750            let batch = db
1751                .new_batch()
1752                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
1753                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1754                .await;
1755            let (db, _) = db.apply_batch(batch).await.unwrap();
1756            let db = db.sync().await.unwrap();
1757            drop(db);
1758            let oversized_floor = Location::new(10);
1759
1760            // Overwrite the persisted commit op with a floor beyond its own commit location.
1761            let journal = open_witness_journal(context.child("tamper"), partition).await;
1762            let (_, size, pinned_nodes) = witness::tests::tip(&journal).await;
1763            let bad_op = Operation::<mmr::Family, Digest, FixedEncoding<Digest>>::Commit(
1764                Some(Sha256::fill(0xaa)),
1765                oversized_floor,
1766            )
1767            .encode()
1768            .to_vec();
1769            witness::tests::overwrite_tip(journal, bad_op, size, pinned_nodes).await;
1770
1771            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1772            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1773                merkle,
1774                context.child("reopen_witness"),
1775                witness_config(partition, &context),
1776                (),
1777            )
1778            .await;
1779            assert!(matches!(
1780                reopened,
1781                Err(Error::DataCorrupted("invalid compact witness"))
1782            ));
1783        });
1784    }
1785
1786    #[test_traced("INFO")]
1787    fn test_compact_reopen_rejects_tampered_pinned_nodes() {
1788        deterministic::Runner::default().start(|context| async move {
1789            let partition = "immutable-pinned-nodes-tamper";
1790            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
1791            let batch = db
1792                .new_batch()
1793                .set(Sha256::hash(&[&[7]]), Sha256::fill(7u8))
1794                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1795                .await;
1796            let (db, _) = db.apply_batch(batch).await.unwrap();
1797            let db = db.sync().await.unwrap();
1798            let tampered_target = db.target();
1799            drop(db);
1800
1801            // Flip one pinned-node digest. There is no stored proof to cross-check against, so the
1802            // rebuild succeeds and yields a different root, the same way a bit-flipped replay
1803            // journal reopens with a different root.
1804            let journal = open_witness_journal(context.child("tamper"), partition).await;
1805            let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
1806            pinned_nodes[0] = Sha256::fill(0xff);
1807            witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;
1808
1809            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1810            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1811                merkle,
1812                context.child("reopen_witness"),
1813                witness_config(partition, &context),
1814                (),
1815            )
1816            .await
1817            .unwrap();
1818            assert_ne!(reopened.target(), tampered_target);
1819            reopened.destroy().await.unwrap();
1820        });
1821    }
1822
1823    /// A witness entry appended but not synced (a commit interrupted before its journal sync)
1824    /// must be dropped on reopen, recovering the last synced commit.
1825    #[test_traced("INFO")]
1826    fn test_compact_reopen_drops_unsynced_witness() {
1827        deterministic::Runner::default().start(|context| async move {
1828            let partition = "immutable-witness-unsynced";
1829            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
1830
1831            // Commit state A.
1832            let batch = db
1833                .new_batch()
1834                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1835                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(1))
1836                .await;
1837            let (db, _) = db.apply_batch(batch).await.unwrap();
1838            let db = db.sync().await.unwrap();
1839            let target_a = db.target();
1840            drop(db);
1841
1842            // Simulate the crash window: append an entry ahead of the tip without syncing it,
1843            // then drop the journal. The unsynced tail must not survive reopen.
1844            let journal = open_witness_journal(context.child("crash"), partition).await;
1845            let (op_bytes, mut size, pinned_nodes) = witness::tests::tip(&journal).await;
1846            size += 2;
1847            witness::tests::append_unsynced(journal, op_bytes, size, pinned_nodes).await;
1848
1849            // Reopen must drop the unsynced entry and recover state A.
1850            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
1851            assert_eq!(reopened.target(), target_a);
1852            reopened.destroy().await.unwrap();
1853        });
1854    }
1855
1856    #[test_traced("INFO")]
1857    fn test_compact_rewind_beyond_history() {
1858        deterministic::Runner::default().start(|context| async move {
1859            let db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-beyond").await;
1860            // The bootstrap commit is the oldest retained state (one leaf); no commit with zero
1861            // operations exists to rewind to.
1862            assert!(matches!(
1863                db.rewind(Location::new(0)).await,
1864                Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
1865            ));
1866
1867            let db =
1868                open_db::<mmr::Family>(context.child("reopen"), "immutable-rewind-beyond").await;
1869            // A target past the tip is not a commit either.
1870            let beyond_tip = db.size() + 100;
1871            assert!(matches!(
1872                db.rewind(beyond_tip).await,
1873                Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
1874            ));
1875        });
1876    }
1877
1878    #[test_traced("INFO")]
1879    fn test_compact_rewind_between_commits() {
1880        deterministic::Runner::default().start(|context| async move {
1881            let db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-between").await;
1882
1883            // A multi-op commit jumps the committed size from 1 (bootstrap) to 4.
1884            let batch = db
1885                .new_batch()
1886                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1887                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
1888                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
1889                .await;
1890            let (db, _) = db.apply_batch(batch).await.unwrap();
1891            let db = db.sync().await.unwrap();
1892            let root_a = db.root();
1893            let size_a = db.size();
1894            assert_eq!(size_a, Location::new(4));
1895
1896            // A second commit moves the size to 6.
1897            let batch = db
1898                .new_batch()
1899                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
1900                .merkleize(&db, Some(Sha256::fill(0xb1)), Location::new(0))
1901                .await;
1902            let (db, _) = db.apply_batch(batch).await.unwrap();
1903            let db = db.sync().await.unwrap();
1904            let root_b = db.root();
1905
1906            // Targets inside a commit's span match no entry, even though entries exist on
1907            // both sides.
1908            let mut db = db;
1909            for target in [2u64, 3, 5] {
1910                assert!(matches!(
1911                    db.rewind(Location::new(target)).await,
1912                    Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
1913                ));
1914                db = open_db::<mmr::Family>(
1915                    context.child("reopen").with_attribute("target", target),
1916                    "immutable-rewind-between",
1917                )
1918                .await;
1919            }
1920            assert_eq!(db.root(), root_b);
1921
1922            // The exact commit boundary remains a valid target.
1923            let db = db.rewind(size_a).await.unwrap();
1924            assert_eq!(db.root(), root_a);
1925            assert_eq!(db.get_metadata(), Some(Sha256::fill(0xa1)));
1926            db.destroy().await.unwrap();
1927        });
1928    }
1929
1930    #[test_traced("INFO")]
1931    fn test_compact_rewind_multiple_commits() {
1932        deterministic::Runner::default().start(|context| async move {
1933            let partition = "immutable-rewind-multi";
1934            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
1935
1936            // Commit A, B, C, recording the state after A.
1937            let batch = db
1938                .new_batch()
1939                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1940                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
1941                .await;
1942            let (db, _) = db.apply_batch(batch).await.unwrap();
1943            let db = db.sync().await.unwrap();
1944            let root_a = db.root();
1945            let size_a = db.size();
1946            let target_a = db.target();
1947
1948            let mut db = db;
1949            for i in [2u8, 3] {
1950                let batch = db
1951                    .new_batch()
1952                    .set(Sha256::hash(&[&[i]]), Sha256::fill(i))
1953                    .merkleize(&db, Some(Sha256::fill(i)), Location::new(0))
1954                    .await;
1955                (db, _) = db.apply_batch(batch).await.unwrap();
1956                db = db.sync().await.unwrap();
1957            }
1958            assert_ne!(db.root(), root_a);
1959
1960            // Rewind two commits in one call.
1961            let db = db.rewind(size_a).await.unwrap();
1962            assert_eq!(db.root(), root_a);
1963            assert_eq!(db.size(), size_a);
1964            assert_eq!(db.get_metadata(), Some(Sha256::fill(0xa1)));
1965            assert_eq!(db.target(), target_a);
1966            drop(db);
1967
1968            // The rewind is durable: reopen recovers state A.
1969            let db = open_db::<mmr::Family>(context.child("reopen"), partition).await;
1970            assert_eq!(db.root(), root_a);
1971            assert_eq!(db.target(), target_a);
1972            db.destroy().await.unwrap();
1973        });
1974    }
1975
1976    #[test_traced("INFO")]
1977    fn test_compact_rewind_to_current_is_noop() {
1978        deterministic::Runner::default().start(|context| async move {
1979            let db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-noop").await;
1980            let batch = db
1981                .new_batch()
1982                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
1983                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
1984                .await;
1985            let (db, _) = db.apply_batch(batch).await.unwrap();
1986            let db = db.sync().await.unwrap();
1987            let root = db.root();
1988            let size = db.size();
1989
1990            let db = db.rewind(size).await.unwrap();
1991            assert_eq!(db.root(), root);
1992            assert_eq!(db.size(), size);
1993            db.destroy().await.unwrap();
1994        });
1995    }
1996
1997    #[test_traced("INFO")]
1998    fn test_compact_prune_then_rewind() {
1999        deterministic::Runner::default().start(|context| async move {
2000            // One entry per section so pruning takes effect at entry granularity (pruning is
2001            // section-aligned and never drops a partial section).
2002            let mut witness_cfg = witness_config("immutable-prune-rewind", &context);
2003            witness_cfg.items_per_section = NZU64!(1);
2004            let merkle = crate::merkle::compact::Merkle::new(Sequential);
2005            let mut db: TestDb<mmr::Family> =
2006                Db::init_from_merkle(merkle, context.child("witness"), witness_cfg.clone(), ())
2007                    .await
2008                    .unwrap();
2009
2010            // Commit A, B, C.
2011            let mut sizes = Vec::new();
2012            for i in [1u8, 2, 3] {
2013                let batch = db
2014                    .new_batch()
2015                    .set(Sha256::hash(&[&[i]]), Sha256::fill(i))
2016                    .merkleize(&db, Some(Sha256::fill(i)), Location::new(0))
2017                    .await;
2018                (db, _) = db.apply_batch(batch).await.unwrap();
2019                db = db.sync().await.unwrap();
2020                sizes.push(db.size());
2021            }
2022
2023            // Prune history below B: rewinding to B still works, rewinding to A does not.
2024            let db = db.prune(sizes[1]).await.unwrap();
2025            assert!(matches!(
2026                db.rewind(sizes[0]).await,
2027                Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
2028            ));
2029
2030            // The prune was durable, so reopen and rewind to B.
2031            let merkle = crate::merkle::compact::Merkle::new(Sequential);
2032            let db: TestDb<mmr::Family> = Db::init_from_merkle(
2033                merkle,
2034                context.child("witness").with_attribute("index", 2),
2035                witness_cfg,
2036                (),
2037            )
2038            .await
2039            .unwrap();
2040            let db = db.rewind(sizes[1]).await.unwrap();
2041            assert_eq!(db.size(), sizes[1]);
2042            assert_eq!(db.get_metadata(), Some(Sha256::fill(2)));
2043
2044            db.destroy().await.unwrap();
2045        });
2046    }
2047
2048    #[test_traced("INFO")]
2049    fn test_compact_prune_past_tip_keeps_tip() {
2050        deterministic::Runner::default().start(|context| async move {
2051            let partition = "immutable-prune-past-tip";
2052            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2053            let batch = db
2054                .new_batch()
2055                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
2056                .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
2057                .await;
2058            let (db, _) = db.apply_batch(batch).await.unwrap();
2059            let db = db.sync().await.unwrap();
2060            let target = db.target();
2061
2062            // Prune with a boundary beyond the tip: the tip entry must survive.
2063            let boundary = db.size() + 100;
2064            let db = db.prune(boundary).await.unwrap();
2065            assert_eq!(db.target(), target);
2066            drop(db);
2067
2068            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
2069            assert_eq!(reopened.target(), target);
2070            reopened.destroy().await.unwrap();
2071        });
2072    }
2073
2074    #[test_traced("INFO")]
2075    fn test_compact_rewind_preserves_pre_advance_batch() {
2076        deterministic::Runner::default().start(|context| async move {
2077            let db = open_db::<mmr::Family>(
2078                context.child("db"),
2079                "immutable-rewind-preserves-pre-advance",
2080            )
2081            .await;
2082
2083            let batch = db
2084                .new_batch()
2085                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
2086                .merkleize(&db, None, Location::new(0))
2087                .await;
2088            let (db, _) = db.apply_batch(batch).await.unwrap();
2089            let db = db.sync().await.unwrap();
2090            let size_after_first = db.size();
2091
2092            // Merkleize a batch against the post-commit-A state.
2093            let held = db
2094                .new_batch()
2095                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
2096                .merkleize(&db, None, Location::new(0))
2097                .await;
2098
2099            // Advance past that state and commit, then rewind back to it.
2100            let batch = db
2101                .new_batch()
2102                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
2103                .merkleize(&db, None, Location::new(0))
2104                .await;
2105            let (db, _) = db.apply_batch(batch).await.unwrap();
2106            let db = db.sync().await.unwrap();
2107            let db = db.rewind(size_after_first).await.unwrap();
2108
2109            // The rewind restored the state that `held` was merkleized against, so it still
2110            // matches the Merkle size and applies cleanly.
2111            let (db, _) = db.apply_batch(held).await.unwrap();
2112
2113            db.destroy().await.unwrap();
2114        });
2115    }
2116
2117    #[test_traced("INFO")]
2118    fn test_compact_noop_commit_after_commit() {
2119        deterministic::Runner::default().start(|context| async move {
2120            let db =
2121                open_db::<mmr::Family>(context.child("db"), "immutable-noop-after-commit").await;
2122
2123            let k1 = Sha256::hash(&[&[1]]);
2124            let v1 = Sha256::fill(11u8);
2125            let k2 = Sha256::hash(&[&[2]]);
2126            let v2 = Sha256::fill(22u8);
2127            let batch = db
2128                .new_batch()
2129                .set(k1, v1)
2130                .set(k2, v2)
2131                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
2132                .await;
2133            let (db, _) = db.apply_batch(batch).await.unwrap();
2134            let db = db.sync().await.unwrap();
2135            let root_after_first = db.root();
2136            let size_after_first = db.size();
2137
2138            let db = db.sync().await.unwrap();
2139            assert_eq!(db.size(), size_after_first);
2140            assert_eq!(db.root(), root_after_first);
2141            assert_eq!(db.target().root, db.root());
2142
2143            db.destroy().await.unwrap();
2144        });
2145    }
2146
2147    #[test_traced("INFO")]
2148    fn test_compact_noop_commit_after_reopen() {
2149        deterministic::Runner::default().start(|context| async move {
2150            let partition = "immutable-noop-after-reopen";
2151
2152            let (root_before_drop, size_before_drop) = {
2153                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
2154                let k1 = Sha256::hash(&[&[1]]);
2155                let v1 = Sha256::fill(11u8);
2156                let k2 = Sha256::hash(&[&[2]]);
2157                let v2 = Sha256::fill(22u8);
2158                let batch = db
2159                    .new_batch()
2160                    .set(k1, v1)
2161                    .set(k2, v2)
2162                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
2163                    .await;
2164                let (db, _) = db.apply_batch(batch).await.unwrap();
2165                let db = db.sync().await.unwrap();
2166                (db.root(), db.size())
2167            };
2168
2169            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
2170            assert_eq!(db.root(), root_before_drop);
2171            assert_eq!(db.size(), size_before_drop);
2172
2173            let db = db.sync().await.unwrap();
2174            assert_eq!(db.size(), size_before_drop);
2175            assert_eq!(db.root(), root_before_drop);
2176            assert_eq!(db.target().root, db.root());
2177
2178            db.destroy().await.unwrap();
2179        });
2180    }
2181
2182    #[test_traced("INFO")]
2183    fn test_compact_noop_commit_after_rewind() {
2184        deterministic::Runner::default().start(|context| async move {
2185            let db =
2186                open_db::<mmr::Family>(context.child("db"), "immutable-noop-after-rewind").await;
2187
2188            let k1 = Sha256::hash(&[&[1]]);
2189            let v1 = Sha256::fill(11u8);
2190            let k2 = Sha256::hash(&[&[2]]);
2191            let v2 = Sha256::fill(22u8);
2192            let batch = db
2193                .new_batch()
2194                .set(k1, v1)
2195                .set(k2, v2)
2196                .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
2197                .await;
2198            let (db, _) = db.apply_batch(batch).await.unwrap();
2199            let db = db.sync().await.unwrap();
2200            let root_after_first = db.root();
2201            let size_after_first = db.size();
2202
2203            let k3 = Sha256::hash(&[&[3]]);
2204            let v3 = Sha256::fill(33u8);
2205            let batch = db
2206                .new_batch()
2207                .set(k3, v3)
2208                .merkleize(&db, Some(Sha256::fill(0xbb)), Location::new(1))
2209                .await;
2210            let (db, _) = db.apply_batch(batch).await.unwrap();
2211            let db = db.sync().await.unwrap();
2212
2213            let db = db.rewind(size_after_first).await.unwrap();
2214            assert_eq!(db.size(), size_after_first);
2215            assert_eq!(db.root(), root_after_first);
2216
2217            let db = db.sync().await.unwrap();
2218            assert_eq!(db.size(), size_after_first);
2219            assert_eq!(db.root(), root_after_first);
2220            assert_eq!(db.target().root, db.root());
2221
2222            db.destroy().await.unwrap();
2223        });
2224    }
2225
2226    #[test_traced("INFO")]
2227    fn test_compact_rewind_makes_post_advance_batch_stale() {
2228        deterministic::Runner::default().start(|context| async move {
2229            let db =
2230                open_db::<mmr::Family>(context.child("db"), "immutable-rewind-makes-stale").await;
2231
2232            let batch = db
2233                .new_batch()
2234                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
2235                .merkleize(&db, None, Location::new(0))
2236                .await;
2237            let (db, _) = db.apply_batch(batch).await.unwrap();
2238            let db = db.sync().await.unwrap();
2239            let size_after_first = db.size();
2240
2241            let batch = db
2242                .new_batch()
2243                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
2244                .merkleize(&db, None, Location::new(0))
2245                .await;
2246            let (db, _) = db.apply_batch(batch).await.unwrap();
2247            let db = db.sync().await.unwrap();
2248
2249            // Merkleize a batch against the post-commit-B state, which the rewind will discard.
2250            let held = db
2251                .new_batch()
2252                .set(Sha256::hash(&[&[3]]), Sha256::fill(3u8))
2253                .merkleize(&db, None, Location::new(0))
2254                .await;
2255
2256            let db = db.rewind(size_after_first).await.unwrap();
2257
2258            // After rewind, mem.size reflects post-commit-A, but the held batch starts after
2259            // post-commit-B. Apply must be rejected with StaleBatch.
2260            assert!(matches!(db.apply_batch(held).await, Err(Error::StaleBatch)));
2261        });
2262    }
2263
2264    #[test_traced("INFO")]
2265    fn test_compact_floor_beyond_size() {
2266        deterministic::Runner::default().start(|context| async move {
2267            let db = open_db::<mmr::Family>(context.child("db"), "immutable-floor-beyond").await;
2268
2269            let batch = db.new_batch().merkleize(&db, None, Location::new(2)).await;
2270
2271            assert!(matches!(
2272                db.apply_batch(batch).await,
2273                Err(Error::FloorBeyondSize(floor, tip))
2274                    if floor == Location::new(2) && tip == Location::new(1)
2275            ));
2276        });
2277    }
2278
2279    // A chained batch whose ancestor's floor exceeds that ancestor's own commit location
2280    // must be rejected, identifying the ancestor's bound rather than the tip's.
2281    #[test_traced("INFO")]
2282    fn test_compact_ancestor_floor_beyond_size() {
2283        deterministic::Runner::default().start(|context| async move {
2284            let db = open_db::<mmr::Family>(context.child("db"), "immutable-ancestor-floor-beyond")
2285                .await;
2286
2287            // parent: set + commit at loc 2, floor=3 (one past parent's commit).
2288            let parent = db
2289                .new_batch()
2290                .set(Sha256::hash(&[&[1]]), Sha256::fill(1u8))
2291                .merkleize(&db, None, Location::new(3))
2292                .await;
2293            // child: valid on its own (floor=0), but parent's floor is bad.
2294            let child = parent
2295                .new_batch::<Sha256>()
2296                .set(Sha256::hash(&[&[2]]), Sha256::fill(2u8))
2297                .merkleize(&db, None, Location::new(0))
2298                .await;
2299
2300            assert!(matches!(
2301                db.apply_batch(child).await,
2302                Err(Error::FloorBeyondSize(floor, commit))
2303                    if floor == Location::new(3) && commit == Location::new(2)
2304            ));
2305        });
2306    }
2307}