Skip to main content

commonware_storage/qmdb/keyless/
compact.rs

1//! A keyless authenticated db that discards historical operations, retaining only a witness
2//! for each applied batch.
3//!
4//! Mirrors the API of [`crate::qmdb::keyless::Keyless`] (`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. [`Db::init`]
13//! restores a retained applied state within its operation cap. [`Db::prune`] bounds the retained
14//! history. Initialization restores the db's in-memory state from an entry. The Merkle is rebuilt
15//! from the stored pinned nodes and operation, and the commit fields are decoded from the
16//! operation. An entry that cannot rebuild surfaces as [`Error::DataCorrupted`]. Each retained
17//! witness also serves compact sync at its size without retaining historical operations.
18//!
19//! # Inactivity floor
20//!
21//! Commits carry an inactivity floor for wire-format compatibility with
22//! [`crate::qmdb::keyless::Keyless`]: 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    merkle::{Family, Location, Proof, batch, compact as compact_merkle},
32    qmdb::{
33        self, Error,
34        any::value::ValueEncoding,
35        chain::{self, Bounds, Commitment},
36        compact::{batch as compact_batch, witness},
37        sync::{CompactTarget, Request, Source, source},
38    },
39};
40use commonware_codec::{Encode, EncodeShared, Read};
41use commonware_cryptography::{Digest, Hasher};
42use commonware_macros::boxed;
43use commonware_parallel::Strategy;
44use commonware_runtime::Handle;
45use std::sync::{Arc, Weak};
46
47/// A keyless authenticated db that discards historical operations, retaining only a witness
48/// for each applied batch.
49pub struct Db<F, E, V, H, S: Strategy>
50where
51    F: Family,
52    E: Context,
53    V: ValueEncoding,
54    H: Hasher,
55    Operation<F, V>: EncodeShared,
56    Operation<F, V>: Read,
57{
58    merkle: compact_merkle::Merkle<F, H::Digest, S>,
59    last_commit_metadata: Option<V::Value>,
60    inactivity_floor_loc: Location<F>,
61    commit_codec_config: <Operation<F, V> as Read>::Cfg,
62    witness: witness::Store<E, F, H::Digest>,
63}
64
65impl<F, E, V, H, S: Strategy> std::fmt::Debug for Db<F, E, V, H, S>
66where
67    F: Family,
68    E: Context,
69    V: ValueEncoding,
70    H: Hasher,
71    Operation<F, V>: EncodeShared,
72    Operation<F, V>: Read,
73{
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.debug_struct("Db")
76            .field("size", &self.size())
77            .field("inactivity_floor_loc", &self.inactivity_floor_loc())
78            .finish_non_exhaustive()
79    }
80}
81
82/// Result of merkleizing a batch.
83type MerkleizeResult<F, D, V, S> = Result<Arc<MerkleizedBatch<F, D, V, S>>, Error<F>>;
84
85/// A speculative batch for a compact keyless db.
86#[allow(clippy::type_complexity)]
87pub struct UnmerkleizedBatch<F, H, V, S: Strategy>
88where
89    F: Family,
90    V: ValueEncoding,
91    H: Hasher,
92    Operation<F, V>: EncodeShared,
93{
94    merkle_batch: compact_merkle::UnmerkleizedBatch<F, H::Digest, S>,
95    appends: Vec<V::Value>,
96    parent: Option<Arc<MerkleizedBatch<F, H::Digest, V, S>>>,
97    base: chain::Commitment<F, H::Digest>,
98}
99
100/// A speculative batch whose root digest has been computed.
101#[derive(Clone)]
102pub struct MerkleizedBatch<F: Family, D: Digest, V: ValueEncoding, S: Strategy>
103where
104    Operation<F, V>: EncodeShared,
105{
106    pub(super) merkle_batch: Arc<batch::MerkleizedBatch<F, D, S>>,
107    operations: Arc<Vec<Operation<F, V>>>,
108    pub(super) commit_metadata: Option<V::Value>,
109    pub(super) parent: Option<Weak<Self>>,
110    pub(super) bounds: chain::Bounds<F, D>,
111}
112
113impl<F: Family, D: Digest, V: ValueEncoding, S: Strategy> MerkleizedBatch<F, D, V, S>
114where
115    Operation<F, V>: EncodeShared,
116{
117    pub(super) fn ancestors(&self) -> impl Iterator<Item = Arc<Self>> + use<F, D, V, S> {
118        chain::ancestors(self.parent.clone(), |batch| batch.parent.as_ref())
119    }
120
121    /// The [`Commitment`] this batch commits to.
122    pub(super) const fn commitment(&self) -> Commitment<F, D> {
123        self.bounds.tip
124    }
125
126    /// Return the root digest after this batch is applied.
127    pub const fn root(&self) -> D {
128        self.bounds.tip.root
129    }
130
131    /// Return the [`Bounds`] of the batch.
132    pub const fn bounds(&self) -> &Bounds<F, D> {
133        &self.bounds
134    }
135
136    /// Return the operations this batch appends to the log and the location of the first.
137    pub fn operations(&self) -> (Location<F>, Arc<Vec<Operation<F, V>>>) {
138        (self.bounds.base.size, Arc::clone(&self.operations))
139    }
140
141    /// Inclusion proof for the operations returned by [`Self::operations`], anchored at
142    /// this batch's tip. The pair verifies against [`Self::root`] via
143    /// [`crate::qmdb::verify_proof`]. Together with [`Self::pinned_nodes`] they verify via
144    /// [`crate::qmdb::verify_proof_and_pinned_nodes`].
145    ///
146    /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be
147    /// held. Nodes below the chain are read from `db`'s [Merkle store][crate::merkle::mem::Mem],
148    /// which retains them at least until this batch's changes are applied (applying it or a
149    /// descendant prunes the store to its frontier).
150    ///
151    /// # Errors
152    ///
153    /// Returns [`Error::StaleRead`] if `db` is not at a state of this batch's chain,
154    /// [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or belongs to a
155    /// dropped unapplied ancestor, and [`crate::merkle::Error::Empty`] if the batch has no
156    /// operations (such as a batch from [`Db::to_batch`]).
157    pub fn proof<E, H>(&self, db: &Db<F, E, V, H, S>) -> Result<Proof<F, D>, Error<F>>
158    where
159        E: Context,
160        H: Hasher<Digest = D>,
161        Operation<F, V>: Read,
162    {
163        let db = self.bounds.compatible(db, db.commitment())?;
164        let inactive_peaks = F::inactive_peaks(self.bounds.tip.size, self.bounds.inactivity_floor);
165        let hasher = qmdb::hasher::<H>();
166        self.merkle_batch
167            .range_proof(
168                db.merkle.mem(),
169                &hasher,
170                self.bounds.base.size..self.bounds.tip.size,
171                inactive_peaks,
172            )
173            .map_err(Into::into)
174    }
175
176    /// The Merkle frontier at the first operation returned by [`Self::operations`]
177    /// ([`Family::nodes_to_pin`]), which lets a consumer holding only this batch's base rebuild
178    /// compact state and replay the operations. The operations, [`Self::proof`], and pinned
179    /// nodes verify against [`Self::root`] via [`crate::qmdb::verify_proof_and_pinned_nodes`].
180    ///
181    /// Nodes of unapplied ancestors are read through the chain, so those ancestors must still be
182    /// held. Nodes below the chain are read from `db`'s [Merkle store][crate::merkle::mem::Mem],
183    /// which retains them at least until this batch's changes are applied (applying it or a
184    /// descendant prunes the store to its frontier).
185    ///
186    /// # Errors
187    ///
188    /// Returns [`Error::StaleRead`] if `db` is not at a state of this batch's chain, and
189    /// [`crate::merkle::Error::ElementPruned`] if a required node has been pruned or belongs to a
190    /// dropped unapplied ancestor.
191    pub fn pinned_nodes<E, H>(&self, db: &Db<F, E, V, H, S>) -> Result<Vec<D>, Error<F>>
192    where
193        E: Context,
194        H: Hasher<Digest = D>,
195        Operation<F, V>: Read,
196    {
197        let db = self.bounds.compatible(db, db.commitment())?;
198        let base = db.merkle.mem();
199        F::nodes_to_pin(self.bounds.base.size)
200            .map(|pos| {
201                self.merkle_batch
202                    .get_node(pos)
203                    .or_else(|| base.get_node(pos))
204                    .ok_or(crate::merkle::Error::ElementPruned(pos))
205            })
206            .collect::<Result<Vec<_>, _>>()
207            .map_err(Into::into)
208    }
209
210    /// Create a new speculative batch with this one as its parent.
211    ///
212    /// Every unapplied ancestor in the chain must be held until the child (or any descendant) is
213    /// merkleized. Otherwise, `merkleize` returns [`Error::StaleBatch`].
214    pub fn new_batch<H>(self: &Arc<Self>) -> UnmerkleizedBatch<F, H, V, S>
215    where
216        H: Hasher<Digest = D>,
217    {
218        UnmerkleizedBatch {
219            merkle_batch: compact_merkle::UnmerkleizedBatch::wrap(self.merkle_batch.new_batch()),
220            appends: Vec::new(),
221            parent: Some(Arc::clone(self)),
222            base: self.commitment(),
223        }
224    }
225}
226
227impl<F, H, V, S> UnmerkleizedBatch<F, H, V, S>
228where
229    F: Family,
230    V: ValueEncoding,
231    H: Hasher,
232    S: Strategy,
233    Operation<F, V>: EncodeShared,
234{
235    pub(super) fn new<E>(db: &Db<F, E, V, H, S>, base: chain::Commitment<F, H::Digest>) -> Self
236    where
237        E: Context,
238        Operation<F, V>: Read,
239    {
240        Self {
241            merkle_batch: db.merkle.new_batch(),
242            appends: Vec::new(),
243            parent: None,
244            base,
245        }
246    }
247
248    /// The database boundary for this batch chain.
249    ///
250    /// A batch created from the database uses its base. A child inherits its parent's `db`.
251    fn db(&self) -> Commitment<F, H::Digest> {
252        self.parent
253            .as_ref()
254            .map_or(self.base, |parent| parent.bounds.db)
255    }
256
257    pub fn append(mut self, value: V::Value) -> Self {
258        self.appends.push(value);
259        self
260    }
261
262    /// Resolve appends into operations, merkleize, and return the batch.
263    ///
264    /// `inactivity_floor` is threaded through the commit operation for wire-format parity with
265    /// [`crate::qmdb::keyless::Keyless`]. It must be at least the floor this batch builds on (its
266    /// parent's, or the database's) and at most the batch's commit location (`total_size - 1`).
267    /// These bounds are validated, but the floor does not drive any local pruning or retention in
268    /// this variant.
269    ///
270    /// # Errors
271    ///
272    /// - Returns [`Error::StaleBatch`] if `db` is not at a state of this batch's chain or an
273    ///   unapplied ancestor was dropped.
274    /// - Returns [`Error::FloorRegressed`] if `inactivity_floor` is below the floor this batch
275    ///   builds on, and [`Error::FloorBeyondSize`] if it is past the commit location.
276    #[tracing::instrument(
277        name = "qmdb.keyless.compact.batch.merkleize",
278        level = "info",
279        skip_all
280    )]
281    pub async fn merkleize<E>(
282        self,
283        db: &Db<F, E, V, H, S>,
284        metadata: Option<V::Value>,
285        inactivity_floor: Location<F>,
286    ) -> MerkleizeResult<F, H::Digest, V, S>
287    where
288        E: Context,
289        Operation<F, V>: Read,
290    {
291        let live_ancestors: Vec<_> =
292            chain::parent_and_ancestors(self.parent.as_ref(), |parent| parent.ancestors())
293                .collect();
294        let boundary = chain::effective_boundary(
295            self.db(),
296            live_ancestors.last().map(|oldest| oldest.bounds.base),
297        );
298
299        let ancestors: Vec<_> = live_ancestors
300            .iter()
301            .map(|batch| batch.commitment())
302            .collect();
303        let db = chain::merkleizable(db, db.commitment(), boundary, ancestors.iter().copied())?;
304
305        let mut ops: Vec<Operation<F, V>> = Vec::with_capacity(self.appends.len() + 1);
306        ops.extend(self.appends.into_iter().map(Operation::Append));
307        ops.push(Operation::Commit(metadata.clone(), inactivity_floor));
308
309        let operations = Arc::new(ops);
310        let total_size = self.base.size + operations.len() as u64;
311        chain::validate_merkleize_floor::<F>(
312            self.parent.as_ref().map_or_else(
313                || db.inactivity_floor_loc(),
314                |parent| parent.bounds.inactivity_floor,
315            ),
316            inactivity_floor,
317            total_size - 1,
318        )?;
319        let inactive_peaks = F::inactive_peaks(total_size, inactivity_floor);
320        let (merkle, root) = compact_batch::merkleize_ops::<F, H, S, _>(
321            &db.merkle,
322            self.merkle_batch,
323            Arc::clone(&operations),
324            inactive_peaks,
325        )
326        .await?;
327
328        // Hold ancestor batches until their operations and nodes have been captured.
329        drop(live_ancestors);
330
331        Ok(Arc::new(MerkleizedBatch {
332            merkle_batch: merkle,
333            operations,
334            commit_metadata: metadata,
335            parent: self.parent.as_ref().map(Arc::downgrade),
336            bounds: chain::Bounds {
337                base: self.base,
338                db: boundary,
339                tip: Commitment::new(total_size, root),
340                ancestors,
341                inactivity_floor,
342            },
343        }))
344    }
345}
346
347impl<F, E, V, H, S> Db<F, E, V, H, S>
348where
349    F: Family,
350    E: Context,
351    V: ValueEncoding,
352    H: Hasher,
353    S: Strategy,
354    Operation<F, V>: EncodeShared,
355    Operation<F, V>: Read,
356{
357    /// Initialize from a retained witness.
358    /// `Some(max_size)` selects the latest retained witness for at most `max_size` operations.
359    /// `None` selects the latest retained state.
360    ///
361    /// Fresh storage receives a durable bootstrap commit and witness.
362    #[boxed]
363    pub async fn init(
364        context: E,
365        cfg: Config<<Operation<F, V> as Read>::Cfg, S>,
366        max_size: Option<Location<F>>,
367    ) -> Result<Self, Error<F>> {
368        let Config {
369            strategy,
370            witness: witness_config,
371            commit_codec_config,
372        } = cfg;
373
374        // Reconstruct and verify the selected witness before exposing any state derived from it.
375        let mut merkle = compact_merkle::Merkle::new(strategy);
376        let (witness, last_commit_op) = witness::init::<E, F, H, S, Operation<F, V>>(
377            context.child("witness"),
378            witness_config,
379            max_size,
380            &mut merkle,
381            &commit_codec_config,
382            Operation::<F, V>::Commit(None, Location::new(0))
383                .encode()
384                .to_vec(),
385        )
386        .await?;
387
388        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
389            return Err(Error::DataCorrupted("last operation was not a commit"));
390        };
391
392        Ok(Self {
393            merkle,
394            last_commit_metadata,
395            inactivity_floor_loc,
396            commit_codec_config,
397            witness,
398        })
399    }
400
401    fn encode_commit_op(metadata: Option<V::Value>, inactivity_floor_loc: Location<F>) -> Vec<u8> {
402        Operation::<F, V>::Commit(metadata, inactivity_floor_loc)
403            .encode()
404            .to_vec()
405    }
406
407    /// Build a compact db from state fetched by the sync engine.
408    ///
409    /// The imported witness lives only in memory until the first [`Self::apply_batch`],
410    /// [`Self::commit`], [`Self::sync`], or [`Self::start_sync`]. Applying a batch replaces it with
411    /// the newly applied journal checkpoint. A durability method journals it directly. Until one of
412    /// those operations succeeds, prune is rejected.
413    pub(crate) fn init_from_sync(
414        strategy: S,
415        journal: witness::Journal<E, F, H::Digest>,
416        commit_codec_config: <Operation<F, V> as Read>::Cfg,
417        last_commit_loc: Location<F>,
418        pinned_nodes: Vec<H::Digest>,
419        last_commit_op: Operation<F, V>,
420    ) -> Result<Self, Error<F>> {
421        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
422            return Err(Error::UnexpectedData(last_commit_loc));
423        };
424        witness::validate_inactivity_floor(inactivity_floor_loc, last_commit_loc)?;
425
426        let op_bytes = Self::encode_commit_op(last_commit_metadata.clone(), inactivity_floor_loc);
427        let mut merkle =
428            compact_merkle::Merkle::from_compact_state(strategy, last_commit_loc, pinned_nodes)?;
429        let hasher = qmdb::hasher::<H>();
430        merkle.append_leaf(&hasher, &op_bytes)?;
431        let imported = witness::build_witness::<F, H, S>(&merkle, inactivity_floor_loc, op_bytes)?;
432        merkle.prune_to_frontier();
433
434        let witness = witness::Store::from_import(journal, imported);
435        Ok(Self {
436            merkle,
437            last_commit_metadata,
438            inactivity_floor_loc,
439            commit_codec_config,
440            witness,
441        })
442    }
443
444    /// Return the root of the db.
445    pub const fn root(&self) -> H::Digest {
446        self.witness.tip().root
447    }
448
449    /// Return the inactivity floor declared by the last committed batch.
450    pub const fn inactivity_floor_loc(&self) -> Location<F> {
451        self.inactivity_floor_loc
452    }
453
454    /// Return the location of the next operation appended to this db.
455    pub const fn size(&self) -> Location<F> {
456        self.witness.tip().size()
457    }
458
459    /// Get the metadata associated with the last commit.
460    pub fn get_metadata(&self) -> Option<V::Value> {
461        self.last_commit_metadata.clone()
462    }
463
464    /// Return the compact-sync target described by the current witness.
465    ///
466    /// This reflects the most recently applied batch. The target remains non-durable until a
467    /// covering [`Self::commit`], [`Self::sync`], or [`Self::start_sync`] completes.
468    pub const fn target(&self) -> CompactTarget<F, H::Digest> {
469        self.witness.tip().target()
470    }
471
472    /// The [`Commitment`] for the database's current state.
473    pub(crate) const fn commitment(&self) -> chain::Commitment<F, H::Digest> {
474        chain::Commitment::new(self.size(), self.root())
475    }
476
477    /// Create a new speculative batch of operations with this database as its parent.
478    pub fn new_batch(&self) -> UnmerkleizedBatch<F, H, V, S> {
479        UnmerkleizedBatch::new(self, self.commitment())
480    }
481
482    /// Create an owned merkleized batch representing the current applied state.
483    pub fn to_batch(&self) -> Arc<MerkleizedBatch<F, H::Digest, V, S>> {
484        Arc::new(MerkleizedBatch {
485            merkle_batch: self.merkle.to_batch(),
486            operations: Arc::new(Vec::new()),
487            commit_metadata: self.last_commit_metadata.clone(),
488            parent: None,
489            bounds: chain::Bounds::from_db(self.commitment(), self.inactivity_floor_loc),
490        })
491    }
492
493    /// Check that `batch` can be applied to the database in its current state, without
494    /// applying it.
495    ///
496    /// [`Self::apply_batch`] runs the same validation but consumes the database when it
497    /// fails; callers that want to reject a bad batch and keep the handle can check first.
498    pub fn validate_batch(
499        &self,
500        batch: &MerkleizedBatch<F, H::Digest, V, S>,
501    ) -> Result<(), Error<F>> {
502        batch.bounds.validate_apply_to(self.commitment())
503    }
504
505    /// Apply a merkleized batch to the database.
506    ///
507    /// The state is updated in memory and appended to the witness journal. Call [`Self::commit`]
508    /// or [`Self::sync`], or await the handle returned by [`Self::start_sync`], to make the
509    /// applied state durable.
510    ///
511    /// # Errors
512    ///
513    /// Returns [`Error::StaleBatch`] if the database is at neither the batch's recorded database
514    /// boundary nor an ancestor's tip.
515    #[tracing::instrument(name = "qmdb.keyless.compact.db.apply_batch", level = "info", skip_all)]
516    pub async fn apply_batch(
517        mut self,
518        batch: Arc<MerkleizedBatch<F, H::Digest, V, S>>,
519    ) -> Result<Self, Error<F>> {
520        self.validate_batch(&batch)?;
521        self.merkle.apply_batch(&batch.merkle_batch)?;
522        self.last_commit_metadata = batch.commit_metadata.clone();
523        self.inactivity_floor_loc = batch.bounds.inactivity_floor;
524        let last_commit_metadata = self.last_commit_metadata.clone();
525        let inactivity_floor_loc = self.inactivity_floor_loc;
526        self.witness = self
527            .witness
528            .apply::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
529                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
530            })
531            .await?;
532        assert_eq!(self.commitment(), batch.bounds.tip);
533        Ok(self)
534    }
535
536    /// Begin durably persisting the current db state to disk.
537    ///
538    /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit],
539    /// plus a best-effort attempt to bound the recovery needed on reopen. Use [Self::sync] to
540    /// guarantee none is needed. A new sync waits for the prior sync before starting. Failures
541    /// of the deferred durability work surface on the returned handle and the next durability
542    /// operation.
543    #[tracing::instrument(name = "qmdb.keyless.compact.db.start_sync", level = "info", skip_all)]
544    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
545        let last_commit_metadata = self.last_commit_metadata.clone();
546        let inactivity_floor_loc = self.inactivity_floor_loc;
547        let handle;
548        (self.witness, handle) = self
549            .witness
550            .start_sync::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
551                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
552            })
553            .await?;
554        Ok((self, handle))
555    }
556
557    /// Durably persist the current db state to disk. This is faster than [`Self::sync`] but
558    /// reopen may need to replay the witness journal's tail to recover.
559    #[tracing::instrument(name = "qmdb.keyless.compact.db.commit", level = "info", skip_all)]
560    pub async fn commit(mut self) -> Result<Self, Error<F>> {
561        let last_commit_metadata = self.last_commit_metadata.clone();
562        let inactivity_floor_loc = self.inactivity_floor_loc;
563        self.witness = self
564            .witness
565            .commit::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
566                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
567            })
568            .await?;
569        Ok(self)
570    }
571
572    /// Durably persist the current db state to disk, also persisting journal metadata to
573    /// minimize recovery work on reopen.
574    #[tracing::instrument(name = "qmdb.keyless.compact.db.sync", level = "info", skip_all)]
575    pub async fn sync(mut self) -> Result<Self, Error<F>> {
576        let last_commit_metadata = self.last_commit_metadata.clone();
577        let inactivity_floor_loc = self.inactivity_floor_loc;
578        self.witness = self
579            .witness
580            .sync::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
581                Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
582            })
583            .await?;
584        Ok(self)
585    }
586
587    /// Drop witnesses for commits with fewer than `pruning_boundary` operations. Some witness
588    /// below the boundary may survive.
589    ///
590    /// Bounded initialization and compact sync can only use witnesses that survive. The current
591    /// commit's witness always survives. The prune is made durable before this method returns.
592    ///
593    /// # Errors
594    ///
595    /// Fails if a compact-sync import has not yet been applied to the witness journal.
596    #[tracing::instrument(name = "qmdb.keyless.compact.db.prune", level = "info", skip_all)]
597    pub async fn prune(mut self, pruning_boundary: Location<F>) -> Result<Self, Error<F>> {
598        self.witness = self.witness.prune(pruning_boundary).await?;
599        Ok(self)
600    }
601
602    /// Destroy all persisted state associated with this database.
603    #[boxed]
604    pub async fn destroy(self) -> Result<(), Error<F>> {
605        self.witness.destroy().await?;
606        Ok(())
607    }
608}
609
610impl<F, E, V, H, S> Source for Db<F, E, V, H, S>
611where
612    F: Family,
613    E: Context,
614    V: ValueEncoding,
615    H: Hasher,
616    Operation<F, V>: EncodeShared + Read,
617    S: Strategy,
618{
619    type Family = F;
620    type Digest = H::Digest;
621    type Op = Operation<F, V>;
622    type Error = qmdb::Error<F>;
623
624    async fn serve(&self, request: Request<F>) -> source::Result<Self> {
625        let response = self
626            .witness
627            .compact_state::<H, S, Self::Op>(
628                self.merkle.strategy(),
629                &self.commit_codec_config,
630                request,
631            )
632            .await?;
633        Ok((response, None))
634    }
635}
636
637#[cfg(test)]
638mod tests {
639    use super::*;
640    use crate::{
641        journal::contiguous::variable::Config as JournalConfig,
642        merkle::{mmb, mmr},
643        metadata::{Config as MetadataConfig, Metadata},
644        qmdb::{
645            any::value::FixedEncoding, compact::witness, sync::Response, verify_proof,
646            verify_proof_and_pinned_nodes,
647        },
648    };
649    use commonware_cryptography::{Sha256, sha256::Digest};
650    use commonware_macros::test_traced;
651    use commonware_parallel::Sequential;
652    use commonware_runtime::{
653        Runner as _, Spawner as _, Supervisor as _,
654        buffer::paged::CacheRef,
655        deterministic,
656        mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs, fail_pending_syncs},
657        reschedule,
658    };
659    use commonware_utils::{
660        NZU16, NZU64, NZUsize,
661        sequence::{U64, VecU64},
662    };
663    use core::future::Future;
664    use futures::FutureExt as _;
665    use std::num::{NonZeroU16, NonZeroUsize};
666
667    type TestDb<F> = Db<F, deterministic::Context, FixedEncoding<U64>, Sha256, Sequential>;
668
669    const WITNESS_PAGE_SIZE: NonZeroU16 = NZU16!(77);
670    const WITNESS_PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9);
671
672    fn witness_config(partition: &str) -> JournalConfig<()> {
673        JournalConfig {
674            partition: format!("{partition}-witness"),
675            items_per_section: NZU64!(64),
676            compression: None,
677            codec_config: (),
678            page_cache: CacheRef::new(WITNESS_PAGE_SIZE, WITNESS_PAGE_CACHE_SIZE),
679            write_buffer: NZUsize!(1024),
680            replay_buffer: NZUsize!(1024),
681        }
682    }
683
684    async fn open_bounded<F: Family>(
685        context: deterministic::Context,
686        witness_cfg: JournalConfig<()>,
687        cap: Location<F>,
688    ) -> Result<TestDb<F>, Error<F>> {
689        let cfg = Config {
690            strategy: Sequential,
691            witness: witness_cfg,
692            commit_codec_config: (),
693        };
694        Db::init(context, cfg, Some(cap)).await
695    }
696
697    async fn open_db<F: Family>(context: deterministic::Context, partition: &str) -> TestDb<F> {
698        let witness_cfg = witness_config(partition);
699        let cfg = Config {
700            strategy: Sequential,
701            witness: witness_cfg,
702            commit_codec_config: (),
703        };
704        Db::init(context, cfg, None).await.unwrap()
705    }
706
707    async fn compact_merkleize_foreign_db_inner<F: Family>(context: deterministic::Context) {
708        let db = open_db::<F>(context.child("db"), "keyless-merkleize-foreign-db").await;
709        let foreign = open_db::<F>(
710            context.child("foreign"),
711            "keyless-merkleize-foreign-db-foreign",
712        )
713        .await;
714
715        let batch = db
716            .new_batch()
717            .append(U64::new(11))
718            .merkleize(&db, None, Location::new(0))
719            .await
720            .unwrap();
721        let db = db.apply_batch(batch).await.unwrap();
722        let batch = foreign
723            .new_batch()
724            .append(U64::new(99))
725            .merkleize(&foreign, None, Location::new(0))
726            .await
727            .unwrap();
728        let foreign = foreign.apply_batch(batch).await.unwrap();
729        assert_eq!(db.size(), foreign.size());
730        assert_ne!(db.root(), foreign.root());
731
732        let batch = db.new_batch().append(U64::new(22));
733        assert!(matches!(
734            batch.merkleize(&foreign, None, Location::new(0)).await,
735            Err(Error::StaleBatch)
736        ));
737
738        let parent = db
739            .new_batch()
740            .append(U64::new(33))
741            .merkleize(&db, None, Location::new(0))
742            .await
743            .unwrap();
744        let child = parent.new_batch::<Sha256>().append(U64::new(44));
745        assert!(matches!(
746            child.merkleize(&foreign, None, Location::new(0)).await,
747            Err(Error::StaleBatch)
748        ));
749        db.destroy().await.unwrap();
750        foreign.destroy().await.unwrap();
751    }
752
753    async fn compact_merkleize_stale_sibling_inner<F: Family>(context: deterministic::Context) {
754        let db = open_db::<F>(context.child("db"), "keyless-merkleize-stale-sibling").await;
755
756        let direct = db.new_batch().append(U64::new(11));
757        let sibling = db
758            .new_batch()
759            .append(U64::new(22))
760            .merkleize(&db, None, Location::new(0))
761            .await
762            .unwrap();
763        let db = db.apply_batch(sibling).await.unwrap();
764        assert!(matches!(
765            direct.merkleize(&db, None, Location::new(0)).await,
766            Err(Error::StaleBatch)
767        ));
768
769        let parent = db
770            .new_batch()
771            .append(U64::new(33))
772            .merkleize(&db, None, Location::new(0))
773            .await
774            .unwrap();
775        let child = parent.new_batch::<Sha256>().append(U64::new(44));
776        let sibling = db
777            .new_batch()
778            .append(U64::new(55))
779            .merkleize(&db, None, Location::new(0))
780            .await
781            .unwrap();
782        let db = db.apply_batch(sibling).await.unwrap();
783        assert!(matches!(
784            child.merkleize(&db, None, Location::new(0)).await,
785            Err(Error::StaleBatch)
786        ));
787
788        db.destroy().await.unwrap();
789    }
790
791    async fn compact_merkleize_ancestor_states_inner<F: Family>(context: deterministic::Context) {
792        let db = open_db::<F>(context.child("db"), "keyless-merkleize-ancestor-states").await;
793
794        let grandparent = db
795            .new_batch()
796            .append(U64::new(1))
797            .merkleize(&db, None, Location::new(0))
798            .await
799            .unwrap();
800        let parent = grandparent
801            .new_batch::<Sha256>()
802            .append(U64::new(2))
803            .merkleize(&db, None, Location::new(0))
804            .await
805            .unwrap();
806        let pending = parent
807            .new_batch::<Sha256>()
808            .append(U64::new(3))
809            .merkleize(&db, None, Location::new(0))
810            .await
811            .unwrap();
812
813        let db = db.apply_batch(Arc::clone(&grandparent)).await.unwrap();
814        let applied = parent
815            .new_batch::<Sha256>()
816            .append(U64::new(3))
817            .merkleize(&db, None, Location::new(0))
818            .await
819            .unwrap();
820        assert_eq!(pending.root(), applied.root());
821
822        drop(grandparent);
823        let retired = parent
824            .new_batch::<Sha256>()
825            .append(U64::new(3))
826            .merkleize(&db, None, Location::new(0))
827            .await
828            .unwrap();
829        assert_eq!(retired.bounds().db, db.commitment());
830        assert_eq!(pending.root(), retired.root());
831
832        let child = parent.new_batch::<Sha256>().append(U64::new(3));
833        let db = db.apply_batch(parent).await.unwrap();
834        let child = child.merkleize(&db, None, Location::new(0)).await.unwrap();
835        assert_eq!(pending.root(), child.root());
836        let expected_root = child.root();
837        let db = db.apply_batch(child).await.unwrap();
838        assert_eq!(db.root(), expected_root);
839        db.destroy().await.unwrap();
840    }
841
842    #[test_traced]
843    fn test_compact_merkleize_foreign_db_mmr() {
844        deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::<mmr::Family>);
845    }
846
847    #[test_traced]
848    fn test_compact_merkleize_foreign_db_mmb() {
849        deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::<mmb::Family>);
850    }
851
852    #[test_traced]
853    fn test_compact_merkleize_stale_sibling_mmr() {
854        deterministic::Runner::default()
855            .start(compact_merkleize_stale_sibling_inner::<mmr::Family>);
856    }
857
858    #[test_traced]
859    fn test_compact_merkleize_stale_sibling_mmb() {
860        deterministic::Runner::default()
861            .start(compact_merkleize_stale_sibling_inner::<mmb::Family>);
862    }
863
864    #[test_traced]
865    fn test_compact_merkleize_ancestor_states_mmr() {
866        deterministic::Runner::default()
867            .start(compact_merkleize_ancestor_states_inner::<mmr::Family>);
868    }
869
870    #[test_traced]
871    fn test_compact_merkleize_ancestor_states_mmb() {
872        deterministic::Runner::default()
873            .start(compact_merkleize_ancestor_states_inner::<mmb::Family>);
874    }
875
876    /// A batch's proof and pinned nodes return `StaleRead` after a bounded initialization and
877    /// another batch bring the database to the batch's base size with a different root.
878    async fn proof_refused_after_off_chain_reopen<F: Family>(context: deterministic::Context) {
879        let db = open_db::<F>(context.child("db"), "off-chain-proof").await;
880        let mut seed = db.new_batch();
881        for value in 1..=6 {
882            seed = seed.append(U64::new(value));
883        }
884        let seed = seed.merkleize(&db, None, Location::new(0)).await.unwrap();
885        let db = db.apply_batch(seed).await.unwrap();
886        let db = db.sync().await.unwrap();
887        let batch = db
888            .new_batch()
889            .append(U64::new(100))
890            .merkleize(&db, None, Location::new(0))
891            .await
892            .unwrap();
893        let (start, ops) = batch.operations();
894        let root = batch.root();
895        let original_proof = batch.proof(&db).unwrap();
896        let original_pins = batch.pinned_nodes(&db).unwrap();
897        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
898            &original_proof,
899            start,
900            &ops,
901            &original_pins,
902            &root
903        ));
904        drop(db);
905
906        let db = open_bounded::<F>(
907            context.child("reopen"),
908            witness_config("off-chain-proof"),
909            Location::new(1),
910        )
911        .await
912        .unwrap();
913        let mut other = db.new_batch();
914        for value in 11..=16 {
915            other = other.append(U64::new(value));
916        }
917        let other = other.merkleize(&db, None, Location::new(0)).await.unwrap();
918        let db = db.apply_batch(other).await.unwrap();
919        assert_eq!(db.size(), start);
920        assert!(matches!(
921            batch
922                .new_batch::<Sha256>()
923                .merkleize(&db, None, Location::new(0))
924                .await,
925            Err(Error::StaleBatch)
926        ));
927
928        let proof = batch.proof(&db);
929        let pins = batch.pinned_nodes(&db);
930        assert!(
931            matches!(proof, Err(Error::StaleRead)),
932            "proof must return StaleRead at a state outside the batch's chain"
933        );
934        assert!(
935            matches!(pins, Err(Error::StaleRead)),
936            "pinned nodes must return StaleRead at a state outside the batch's chain"
937        );
938    }
939
940    #[test]
941    fn compact_mmr_proof_refused_after_off_chain_reopen() {
942        deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::<mmr::Family>);
943    }
944
945    #[test]
946    fn compact_mmb_proof_refused_after_off_chain_reopen() {
947        deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::<mmb::Family>);
948    }
949
950    /// Batch artifacts (operations, range proof, pinned frontier) verify against the batch root,
951    /// survive applying and dropping ancestors, and return `ElementPruned` once the batch itself is
952    /// applied and the compact store is pruned past them.
953    async fn compact_operations_and_proof_inner<F: Family>(context: deterministic::Context) {
954        let db = open_db::<F>(context.child("db"), "keyless-operations-and-proof").await;
955
956        // Seed committed state so the chain below starts above a pruned frontier.
957        let mut seed = db.new_batch();
958        for value in 1u64..=6 {
959            seed = seed.append(U64::new(value));
960        }
961        let seed = seed
962            .merkleize(&db, Some(U64::new(7)), Location::new(0))
963            .await
964            .unwrap();
965        let db = db.apply_batch(seed).await.unwrap().sync().await.unwrap();
966        let floor = db.size();
967        assert_eq!(floor, Location::new(8));
968
969        // A batch from `to_batch` has no operations to prove.
970        assert!(matches!(
971            db.to_batch().proof(&db),
972            Err(Error::Merkle(crate::merkle::Error::Empty))
973        ));
974
975        // A two-deep unapplied chain. The child's artifacts read the parent's nodes through the
976        // chain.
977        let mut parent = db.new_batch();
978        for value in 8u64..=12 {
979            parent = parent.append(U64::new(value));
980        }
981        let parent = parent
982            .merkleize(&db, Some(U64::new(13)), floor)
983            .await
984            .unwrap();
985        let child = parent
986            .new_batch::<Sha256>()
987            .append(U64::new(14))
988            .merkleize(&db, Some(U64::new(15)), floor)
989            .await
990            .unwrap();
991
992        // The operation suffix is the batch's own appends plus its commit, handed out zero-copy.
993        let (child_start, child_ops) = child.operations();
994        let (_, child_ops_again) = child.operations();
995        let child_end = child.bounds().tip.size;
996        assert!(Arc::ptr_eq(&child_ops, &child_ops_again));
997        assert_eq!(child_start, parent.bounds().tip.size);
998        assert_eq!(*child_start + child_ops.len() as u64, *child_end);
999        assert_eq!(child_end, Location::new(16));
1000        assert!(matches!(
1001            child_ops.as_slice(),
1002            [Operation::Append(value), Operation::Commit(Some(metadata), operation_floor)]
1003                if value == &U64::new(14)
1004                    && metadata == &U64::new(15)
1005                    && operation_floor == &floor
1006        ));
1007
1008        // The proof is anchored at the batch tip and verifies with or without the pins.
1009        let child_root = child.root();
1010        let child_proof = child.proof(&db).unwrap();
1011        let child_pins = child.pinned_nodes(&db).unwrap();
1012        assert_eq!(child_proof.leaves, child_end);
1013        assert_eq!(
1014            child_proof.inactive_peaks,
1015            F::inactive_peaks(child_end, floor),
1016        );
1017        assert!(verify_proof::<Sha256, _, _>(
1018            &child_proof,
1019            child_start,
1020            &child_ops,
1021            &child_root
1022        ));
1023        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
1024            &child_proof,
1025            child_start,
1026            &child_ops,
1027            &child_pins,
1028            &child_root
1029        ));
1030
1031        // The pins are order-sensitive.
1032        assert!(child_pins.len() > 1);
1033        let mut reordered_child_pins = child_pins.clone();
1034        reordered_child_pins.swap(0, 1);
1035        assert!(!verify_proof_and_pinned_nodes::<Sha256, _, _>(
1036            &child_proof,
1037            child_start,
1038            &child_ops,
1039            &reordered_child_pins,
1040            &child_root
1041        ));
1042
1043        // Pipelined consumer: applying the parent prunes the store to the parent's tip, which is
1044        // exactly the frontier the child's base pins, so the artifacts survive dropping the
1045        // parent.
1046        let db = db.apply_batch(parent).await.unwrap();
1047        let child_proof_after = child.proof(&db).unwrap();
1048        assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins);
1049        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
1050            &child_proof_after,
1051            child_start,
1052            &child_ops,
1053            &child_pins,
1054            &child_root
1055        ));
1056        let applied_start = db.size();
1057        let db = db.apply_batch(child).await.unwrap();
1058        let child_range = applied_start..db.size();
1059        assert_eq!(child_range, child_start..child_end);
1060
1061        // A commit-only batch proves exactly its commit operation.
1062        let commit_floor = db.size();
1063        let commit_only = db
1064            .new_batch()
1065            .merkleize(&db, Some(U64::new(16)), commit_floor)
1066            .await
1067            .unwrap();
1068        let (commit_start, commit_ops) = commit_only.operations();
1069        let commit_end = commit_only.bounds().tip.size;
1070        let commit_root = commit_only.root();
1071        let commit_proof = commit_only.proof(&db).unwrap();
1072        let commit_pins = commit_only.pinned_nodes(&db).unwrap();
1073        assert_eq!(commit_start, commit_floor);
1074        assert!(matches!(
1075            commit_ops.as_slice(),
1076            [Operation::Commit(Some(metadata), operation_floor)]
1077                if metadata == &U64::new(16) && operation_floor == &commit_floor
1078        ));
1079        assert_eq!(*commit_start + commit_ops.len() as u64, *commit_end);
1080        assert_eq!(commit_proof.leaves, commit_end);
1081        assert_eq!(
1082            commit_proof.inactive_peaks,
1083            F::inactive_peaks(commit_end, commit_floor)
1084        );
1085        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
1086            &commit_proof,
1087            commit_start,
1088            &commit_ops,
1089            &commit_pins,
1090            &commit_root
1091        ));
1092        let applied_start = db.size();
1093        let db = db.apply_batch(commit_only).await.unwrap();
1094        let commit_range = applied_start..db.size();
1095        assert_eq!(commit_range, commit_start..commit_end);
1096        let db = db.sync().await.unwrap();
1097
1098        // Applying a batch built mid-mountain prunes its own artifacts. The accessors return
1099        // `ElementPruned` rather than a proof that fails to verify, while a child merkleized
1100        // before the apply still finds its base frontier in the store.
1101        let late_parent = db
1102            .new_batch()
1103            .append(U64::new(17))
1104            .merkleize(&db, Some(U64::new(18)), db.size())
1105            .await
1106            .unwrap();
1107        let late = late_parent
1108            .new_batch::<Sha256>()
1109            .append(U64::new(19))
1110            .merkleize(&db, Some(U64::new(20)), db.size())
1111            .await
1112            .unwrap();
1113        let db = db.apply_batch(Arc::clone(&late_parent)).await.unwrap();
1114        assert!(matches!(
1115            late_parent.proof(&db),
1116            Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
1117        ));
1118        assert!(matches!(
1119            late_parent.pinned_nodes(&db),
1120            Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
1121        ));
1122        drop(late_parent);
1123
1124        let (late_start, late_ops) = late.operations();
1125        let late_root = late.root();
1126        let late_proof = late.proof(&db).unwrap();
1127        let late_pins = late.pinned_nodes(&db).unwrap();
1128        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
1129            &late_proof,
1130            late_start,
1131            &late_ops,
1132            &late_pins,
1133            &late_root
1134        ));
1135        let db = db.apply_batch(late).await.unwrap();
1136
1137        db.destroy().await.unwrap();
1138    }
1139
1140    #[test_traced]
1141    fn test_compact_operations_and_proof_mmr() {
1142        deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmr::Family>);
1143    }
1144
1145    #[test_traced]
1146    fn test_compact_operations_and_proof_mmb() {
1147        deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmb::Family>);
1148    }
1149
1150    /// Open the persisted witness journal directly so tests can corrupt the tip entry.
1151    async fn open_witness_journal(
1152        context: deterministic::Context,
1153        partition: &str,
1154    ) -> witness::Journal<deterministic::Context, mmr::Family, Digest> {
1155        let cfg = witness_config(partition);
1156        witness::Journal::init(context, cfg).await.unwrap()
1157    }
1158
1159    /// The witness journal rejects requests that match no retained state. Each mismatch reports
1160    /// the same error a pruned operation log would.
1161    #[test_traced("INFO")]
1162    fn test_serve_refuses_requests_outside_witness() {
1163        deterministic::Runner::default().start(|context| async move {
1164            let db = open_db::<mmr::Family>(context.child("db"), "keyless-serve-refusal").await;
1165            let floor = db.inactivity_floor_loc();
1166            let batch = db
1167                .new_batch()
1168                .append(U64::new(1))
1169                .merkleize(&db, Some(U64::new(11)), floor)
1170                .await
1171                .unwrap();
1172            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
1173            let n = db.target().size;
1174            let boundary = |size: Location<mmr::Family>, start: Location<mmr::Family>| {
1175                Request::Boundary { size, start }
1176            };
1177            let operations =
1178                |size: Location<mmr::Family>, start: Location<mmr::Family>| Request::Operations {
1179                    size,
1180                    start,
1181                    max_ops: NZU64!(1),
1182                };
1183
1184            let beyond = n + 1;
1185            assert!(matches!(
1186                db.serve(boundary(beyond, n)).await,
1187                Err(Error::Merkle(crate::merkle::Error::RangeOutOfBounds(_)))
1188            ));
1189            assert!(matches!(
1190                db.serve(operations(Location::new(0), Location::new(0)))
1191                    .await,
1192                Err(Error::Merkle(crate::merkle::Error::RangeOutOfBounds(_)))
1193            ));
1194            assert!(matches!(
1195                db.serve(operations(n - 1, n - 2)).await,
1196                Err(Error::Journal(crate::journal::Error::ItemPruned(_)))
1197            ));
1198            assert!(matches!(
1199                db.serve(boundary(n, n)).await,
1200                Err(Error::Merkle(crate::merkle::Error::RangeOutOfBounds(_)))
1201            ));
1202            assert!(matches!(
1203                db.serve(boundary(n, n - 2)).await,
1204                Err(Error::Journal(crate::journal::Error::ItemPruned(_)))
1205            ));
1206
1207            // Requests without pinned nodes are also served, even when they ask for more operations
1208            // than the witness holds.
1209            let response = db
1210                .serve(Request::Operations {
1211                    size: n,
1212                    start: n - 1,
1213                    max_ops: NZU64!(5),
1214                })
1215                .await
1216                .unwrap()
1217                .0;
1218            let Response::Operations { operations, .. } = response else {
1219                panic!("operations request should get an operations response");
1220            };
1221            assert_eq!(operations.len(), 1);
1222            let response = db.serve(boundary(n, n - 1)).await.unwrap().0;
1223            assert!(matches!(response, Response::Boundary { .. }));
1224        });
1225    }
1226
1227    /// The witness journal serves each retained applied state at its size and rejects a size with
1228    /// no retained state.
1229    #[test_traced("INFO")]
1230    fn test_serve_retained_witnesses() {
1231        deterministic::Runner::default().start(|context| async move {
1232            let mut db =
1233                open_db::<mmr::Family>(context.child("db"), "keyless-serve-retained").await;
1234
1235            // Apply two batches and record the target after each.
1236            let mut targets = Vec::new();
1237            for seed in [1, 2] {
1238                let floor = db.inactivity_floor_loc();
1239                let batch = db
1240                    .new_batch()
1241                    .append(U64::new(seed))
1242                    .merkleize(&db, Some(U64::new(seed + 10)), floor)
1243                    .await
1244                    .unwrap();
1245                db = db.apply_batch(batch).await.unwrap();
1246                targets.push(db.target());
1247            }
1248            let db = db.sync().await.unwrap();
1249
1250            // The earlier state is served at its size and verifies against its root.
1251            let first = &targets[0];
1252            let start = first.size - 1;
1253            let request = Request::Boundary {
1254                size: first.size,
1255                start,
1256            };
1257            let Response::Boundary {
1258                proof,
1259                op,
1260                pinned_nodes,
1261            } = db.serve(request).await.unwrap().0
1262            else {
1263                panic!("boundary request should get a boundary response");
1264            };
1265            assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
1266                &proof,
1267                start,
1268                std::slice::from_ref(&op),
1269                &pinned_nodes,
1270                &first.root
1271            ));
1272
1273            // An operations request at the earlier size gets the same commit operation.
1274            let Response::Operations { proof, operations } = db
1275                .serve(Request::Operations {
1276                    size: first.size,
1277                    start,
1278                    max_ops: NZU64!(1),
1279                })
1280                .await
1281                .unwrap()
1282                .0
1283            else {
1284                panic!("operations request should get an operations response");
1285            };
1286            assert_eq!(operations, vec![op]);
1287            assert!(verify_proof::<Sha256, _, _>(
1288                &proof,
1289                start,
1290                &operations,
1291                &first.root
1292            ));
1293
1294            // A size between retained states is rejected.
1295            assert!(matches!(
1296                db.serve(Request::Boundary {
1297                    size: first.size + 1,
1298                    start: first.size,
1299                })
1300                .await,
1301                Err(Error::Journal(crate::journal::Error::ItemPruned(_)))
1302            ));
1303        });
1304    }
1305
1306    /// A compact db over a delayed-sync storage backend.
1307    type DelayedDb = Db<
1308        mmr::Family,
1309        DelayedSyncContext<deterministic::Context>,
1310        FixedEncoding<U64>,
1311        Sha256,
1312        Sequential,
1313    >;
1314
1315    /// Open a [DelayedDb] whose blob syncs park on `pending`.
1316    ///
1317    /// Init durably persists the bootstrap witness, so while syncs park the returned future
1318    /// must be driven with [drive_pending_syncs] (or the mock unblocked first).
1319    fn open_delayed_db(
1320        context: deterministic::Context,
1321        partition: &str,
1322        pending: &PendingSyncs,
1323    ) -> impl Future<Output = Result<DelayedDb, Error<mmr::Family>>> {
1324        let witness_cfg = witness_config(partition);
1325        let cfg = Config {
1326            strategy: Sequential,
1327            witness: witness_cfg,
1328            commit_codec_config: (),
1329        };
1330        let context = DelayedSyncContext {
1331            inner: context,
1332            pending: pending.clone(),
1333        };
1334        DelayedDb::init(context, cfg, None)
1335    }
1336
1337    /// Apply a single-append batch carrying `seed` as both value and metadata.
1338    async fn apply_append(db: DelayedDb, seed: u64) -> DelayedDb {
1339        let floor = db.inactivity_floor_loc();
1340        let batch = db
1341            .new_batch()
1342            .append(U64::new(seed))
1343            .merkleize(&db, Some(U64::new(seed)), floor)
1344            .await
1345            .unwrap();
1346        db.apply_batch(batch).await.unwrap()
1347    }
1348
1349    /// Leave a failed recovery-watermark sync retained after dropping its public handle.
1350    async fn fail_dropped_watermark_sync(mut db: DelayedDb, pending: &PendingSyncs) -> DelayedDb {
1351        // Prove the data durable so the next call only advances recovery metadata.
1352        let first;
1353        (db, first) = db.start_sync().await.unwrap();
1354        drive_pending_syncs(pending, first).await.unwrap();
1355
1356        let dropped;
1357        (db, dropped) = db.start_sync().await.unwrap();
1358        assert_eq!(
1359            pending.lock().len(),
1360            1,
1361            "expected only the recovery-watermark sync"
1362        );
1363        fail_pending_syncs(pending);
1364        drop(dropped);
1365        db
1366    }
1367
1368    /// Applying a successor does not wait for the witness sync already in flight.
1369    #[test_traced]
1370    fn test_compact_apply_overlaps_start_sync() {
1371        deterministic::Runner::default().start(|ctx| async move {
1372            let partition = "keyless-start-sync-overlap";
1373            let pending = PendingSyncs::default();
1374            let open = open_delayed_db(ctx.child("delayed"), partition, &pending);
1375            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1376            db = apply_append(db, 1).await;
1377
1378            let starts_before = pending.starts();
1379            let entered_before = pending.entered();
1380            let completions_before = pending.completions();
1381            let handle;
1382            (db, handle) = db.start_sync().await.unwrap();
1383            assert!(pending.starts() > starts_before);
1384            assert_eq!(pending.completions(), completions_before);
1385            let first_target = db.target();
1386
1387            let waiter = ctx
1388                .child("await_sync")
1389                .spawn(|_| async move { handle.await.unwrap() });
1390            while pending.entered() == entered_before {
1391                reschedule().await;
1392            }
1393
1394            db = apply_append(db, 2).await;
1395            assert_ne!(db.root(), first_target.root);
1396            let second_target = db.target();
1397            assert_ne!(second_target, first_target);
1398            assert_eq!(
1399                pending.completions(),
1400                completions_before,
1401                "the database made progress while the sync was still in flight"
1402            );
1403
1404            pending.unblock();
1405            waiter.await.unwrap();
1406
1407            // The successor becomes durable after the next start_sync handle completes.
1408            let handle;
1409            (db, handle) = db.start_sync().await.unwrap();
1410            handle.await.unwrap();
1411            drop(db);
1412
1413            let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
1414                .await
1415                .unwrap();
1416            assert_eq!(db.target(), second_target);
1417            assert_eq!(db.get_metadata(), Some(U64::new(2)));
1418            db.destroy().await.unwrap();
1419        });
1420    }
1421
1422    /// State persisted via an awaited start_sync handle is recovered on reopen.
1423    #[test_traced]
1424    fn test_compact_start_sync_recovery() {
1425        deterministic::Runner::default().start(|ctx| async move {
1426            let partition = "keyless-start-sync-recovery";
1427            let pending = PendingSyncs::default();
1428            pending.unblock();
1429            let mut db = open_delayed_db(ctx.child("delayed"), partition, &pending)
1430                .await
1431                .unwrap();
1432            db = apply_append(db, 1).await;
1433
1434            let handle;
1435            (db, handle) = db.start_sync().await.unwrap();
1436            handle.await.unwrap();
1437            let root = db.root();
1438            drop(db);
1439
1440            let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
1441                .await
1442                .unwrap();
1443            assert_eq!(db.root(), root);
1444            assert_eq!(db.get_metadata(), Some(U64::new(1)));
1445            db.destroy().await.unwrap();
1446        });
1447    }
1448
1449    /// A sync begun by `start_sync` that fails in flight surfaces the error through both the
1450    /// returned handle and the next durability operation, even when that operation has nothing
1451    /// new to persist.
1452    #[test_traced]
1453    fn test_compact_start_sync_failure_propagates() {
1454        deterministic::Runner::default().start(|ctx| async move {
1455            let pending = PendingSyncs::default();
1456            pending.unblock();
1457            let mut db = open_delayed_db(ctx.child("delayed"), "keyless-start-sync-fail", &pending)
1458                .await
1459                .unwrap();
1460            db = apply_append(db, 1).await;
1461
1462            // Arm all future syncs to resolve to an injected error.
1463            pending.arm_fail();
1464
1465            let handle;
1466            (db, handle) = db.start_sync().await.unwrap();
1467            assert!(
1468                handle.await.is_err(),
1469                "the sync handle surfaces the failure"
1470            );
1471            let starts_before = pending.starts();
1472
1473            // The witness entry was already appended, so this commit has nothing to stage.
1474            // It must still observe the retained failure rather than no-op.
1475            assert!(
1476                db.commit().await.is_err(),
1477                "the next durability op surfaces the failed in-flight sync"
1478            );
1479            assert_eq!(
1480                pending.starts(),
1481                starts_before,
1482                "the surfaced error is the retained failure, not a fresh sync's"
1483            );
1484        });
1485    }
1486
1487    /// A `sync` with nothing new to persist still drains (and proves) the sync started by a
1488    /// prior `start_sync`.
1489    #[test_traced]
1490    fn test_compact_start_sync_then_noop_sync_drains() {
1491        deterministic::Runner::default().start(|ctx| async move {
1492            let partition = "keyless-start-sync-noop-drain";
1493            let pending = PendingSyncs::default();
1494            let open = open_delayed_db(ctx.child("delayed"), partition, &pending);
1495            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1496            db = apply_append(db, 1).await;
1497
1498            let starts_before = pending.starts();
1499            let completions_before = pending.completions();
1500            let handle;
1501            (db, handle) = db.start_sync().await.unwrap();
1502            assert!(pending.starts() > starts_before);
1503            assert_eq!(pending.completions(), completions_before);
1504            let root = db.root();
1505
1506            let db = {
1507                let mut sync = std::pin::pin!(db.sync());
1508                assert!(
1509                    sync.as_mut().now_or_never().is_none(),
1510                    "sync proceeded while the started sync was pending"
1511                );
1512                pending.unblock();
1513                sync.await.unwrap()
1514            };
1515            handle.await.unwrap();
1516            assert!(pending.completions() > completions_before);
1517            drop(db);
1518
1519            let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
1520                .await
1521                .unwrap();
1522            assert_eq!(db.root(), root);
1523            db.destroy().await.unwrap();
1524        });
1525    }
1526
1527    /// A no-op `sync` returns a retained metadata failure before starting new journal work.
1528    #[test_traced]
1529    fn test_compact_start_sync_then_noop_sync_fails_without_new_work() {
1530        deterministic::Runner::default().start(|ctx| async move {
1531            let pending = PendingSyncs::default();
1532            let open = open_delayed_db(
1533                ctx.child("delayed"),
1534                "keyless-start-sync-noop-sync-fail",
1535                &pending,
1536            );
1537            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1538            db = apply_append(db, 1).await;
1539            db = fail_dropped_watermark_sync(db, &pending).await;
1540
1541            let starts_before = pending.starts();
1542            assert!(
1543                drive_pending_syncs(&pending, db.sync()).await.is_err(),
1544                "sync absorbed the retained metadata failure"
1545            );
1546            assert_eq!(
1547                pending.starts(),
1548                starts_before,
1549                "sync started new journal work before returning the retained failure"
1550            );
1551        });
1552    }
1553
1554    /// A `commit` with nothing new to persist still waits for the sync started by a prior
1555    /// `start_sync` before reporting the tip durable, and starts no journal work when that
1556    /// sync succeeds.
1557    #[test_traced]
1558    fn test_compact_start_sync_then_noop_commit_waits() {
1559        deterministic::Runner::default().start(|ctx| async move {
1560            let pending = PendingSyncs::default();
1561            let open = open_delayed_db(
1562                ctx.child("delayed"),
1563                "keyless-start-sync-noop-commit",
1564                &pending,
1565            );
1566            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1567            db = apply_append(db, 1).await;
1568
1569            let handle;
1570            (db, handle) = db.start_sync().await.unwrap();
1571            let starts_before = pending.starts();
1572
1573            let db = {
1574                let mut commit = std::pin::pin!(db.commit());
1575                assert!(
1576                    commit.as_mut().now_or_never().is_none(),
1577                    "commit proceeded while the started sync was pending"
1578                );
1579                pending.unblock();
1580                commit.await.unwrap()
1581            };
1582            handle.await.unwrap();
1583            assert_eq!(
1584                pending.starts(),
1585                starts_before,
1586                "a successful pipelined sync still triggered journal work"
1587            );
1588            db.destroy().await.unwrap();
1589        });
1590    }
1591
1592    /// A `start_sync` with nothing new to persist returns a working handle and appends no
1593    /// duplicate witness entry.
1594    #[test_traced]
1595    fn test_compact_start_sync_noop_second_call() {
1596        deterministic::Runner::default().start(|ctx| async move {
1597            let partition = "keyless-start-sync-noop-second";
1598            let pending = PendingSyncs::default();
1599            let open = open_delayed_db(ctx.child("delayed"), partition, &pending);
1600            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1601            db = apply_append(db, 1).await;
1602
1603            let h1;
1604            (db, h1) = db.start_sync().await.unwrap();
1605            // Release the parked sync: a second start_sync waits for the prior sync before
1606            // starting, so back-to-back calls under a parked mock would deadlock.
1607            pending.unblock();
1608            h1.await.unwrap();
1609
1610            let h2;
1611            (db, h2) = db.start_sync().await.unwrap();
1612            h2.await.unwrap();
1613            let root = db.root();
1614            drop(db);
1615
1616            // The journal holds exactly the bootstrap entry and the one durable witness.
1617            let journal = open_witness_journal(ctx.child("probe"), partition).await;
1618            assert_eq!(journal.size(), 2);
1619            drop(journal);
1620
1621            let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
1622                .await
1623                .unwrap();
1624            assert_eq!(db.root(), root);
1625            db.destroy().await.unwrap();
1626        });
1627    }
1628
1629    /// A metadata sync failure from `start_sync` resurfaces on the next `commit`, even when
1630    /// that commit has new state to persist.
1631    #[test_traced]
1632    fn test_compact_start_sync_metadata_failure_resurfaces_on_commit() {
1633        deterministic::Runner::default().start(|ctx| async move {
1634            let pending = PendingSyncs::default();
1635            let open = open_delayed_db(
1636                ctx.child("delayed"),
1637                "keyless-start-sync-meta-fail",
1638                &pending,
1639            );
1640            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1641            db = apply_append(db, 1).await;
1642
1643            let handle;
1644            (db, handle) = db.start_sync().await.unwrap();
1645
1646            // start_sync parks the data sync and then the offsets sync. Complete the data
1647            // sync and fail the offsets sync.
1648            {
1649                let mut parked = pending.lock();
1650                assert_eq!(
1651                    parked.len(),
1652                    2,
1653                    "expected the data and offsets syncs parked"
1654                );
1655                let offsets = parked.pop().unwrap();
1656                let data = parked.pop().unwrap();
1657                data.release.send(Ok(())).unwrap();
1658                offsets
1659                    .release
1660                    .send(Err(commonware_runtime::Error::Io(
1661                        std::io::Error::other("injected sync failure").into(),
1662                    )))
1663                    .unwrap();
1664            }
1665            assert!(
1666                handle.await.is_err(),
1667                "the sync handle surfaces the failure"
1668            );
1669
1670            // Later syncs pass; only the retained offsets failure remains.
1671            pending.unblock();
1672
1673            // Apply another batch to prove commit checks the prior sync before persisting a new
1674            // witness.
1675            let db = apply_append(db, 2).await;
1676            assert!(
1677                db.commit().await.is_err(),
1678                "commit absorbed the retained metadata failure"
1679            );
1680        });
1681    }
1682
1683    /// A later `start_sync` cannot replace an unobserved failure from the prior handle.
1684    #[test_traced]
1685    fn test_compact_start_sync_retains_dropped_metadata_failure() {
1686        deterministic::Runner::default().start(|ctx| async move {
1687            let pending = PendingSyncs::default();
1688            let open = open_delayed_db(
1689                ctx.child("delayed"),
1690                "keyless-start-sync-dropped-meta-fail",
1691                &pending,
1692            );
1693            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
1694            db = apply_append(db, 1).await;
1695            // Leave only the failed recovery-watermark completion for the next call to observe.
1696            db = fail_dropped_watermark_sync(db, &pending).await;
1697
1698            // The store retains the dropped handle's completion. Deferred failures stay on the
1699            // handle channel, so this call succeeds but its handle must fail.
1700            let next;
1701            (db, next) = db.start_sync().await.unwrap();
1702            assert!(
1703                drive_pending_syncs(&pending, next).await.is_err(),
1704                "a later start_sync masked the retained metadata failure"
1705            );
1706            drop(db);
1707        });
1708    }
1709
1710    /// Once a start_sync handle completes successfully, a commit has nothing left
1711    /// to prove and touches no storage.
1712    #[test_traced]
1713    fn test_compact_start_sync_proven_skips_journal() {
1714        deterministic::Runner::default().start(|ctx| async move {
1715            let pending = PendingSyncs::default();
1716            pending.unblock();
1717            let mut db =
1718                open_delayed_db(ctx.child("delayed"), "keyless-start-sync-proven", &pending)
1719                    .await
1720                    .unwrap();
1721            db = apply_append(db, 1).await;
1722
1723            let handle;
1724            (db, handle) = db.start_sync().await.unwrap();
1725            handle.await.unwrap();
1726
1727            let starts_before = pending.starts();
1728            let db = db.commit().await.unwrap();
1729            assert_eq!(
1730                pending.starts(),
1731                starts_before,
1732                "a proven pipelined sync still triggered journal work"
1733            );
1734            db.destroy().await.unwrap();
1735        });
1736    }
1737
1738    /// The first persist after a compact-sync import can be pipelined: awaiting the handle
1739    /// makes the imported witness durable.
1740    #[test_traced("INFO")]
1741    fn test_compact_start_sync_persists_import() {
1742        deterministic::Runner::default().start(|context| async move {
1743            let dst = "keyless-import-start-sync-dst";
1744            let src = "keyless-import-start-sync-src";
1745            let meta_a = U64::new(11);
1746            let meta_b = U64::new(22);
1747
1748            // Build state B in a separate source partition and capture its validated state.
1749            let target_b = {
1750                let source = open_db::<mmr::Family>(context.child("src"), src).await;
1751                let batch = source
1752                    .new_batch()
1753                    .append(U64::new(2))
1754                    .merkleize(&source, Some(meta_b.clone()), Location::new(0))
1755                    .await
1756                    .unwrap();
1757                let source = source.apply_batch(batch).await.unwrap();
1758                let source = source.sync().await.unwrap();
1759                source.target()
1760            };
1761            let (_, size_b, pinned_b) = {
1762                let journal = open_witness_journal(context.child("tip"), src).await;
1763                witness::tests::tip(&journal).await
1764            };
1765            assert_eq!(size_b, target_b.size);
1766
1767            // Seed the destination partition with a different committed state A.
1768            {
1769                let seeded = open_db::<mmr::Family>(context.child("seed"), dst).await;
1770                let batch = seeded
1771                    .new_batch()
1772                    .append(U64::new(1))
1773                    .merkleize(&seeded, Some(meta_a), Location::new(0))
1774                    .await
1775                    .unwrap();
1776                let seeded = seeded.apply_batch(batch).await.unwrap();
1777                let seeded = seeded.sync().await.unwrap();
1778                assert_ne!(seeded.target(), target_b);
1779            }
1780
1781            // Import state B over the destination and make it durable through a pipelined sync.
1782            {
1783                let journal = open_witness_journal(context.child("import"), dst).await;
1784                let imported = TestDb::<mmr::Family>::init_from_sync(
1785                    Sequential,
1786                    journal,
1787                    (),
1788                    size_b - 1,
1789                    pinned_b,
1790                    Operation::Commit(Some(meta_b.clone()), Location::new(0)),
1791                )
1792                .unwrap();
1793                assert_eq!(imported.target(), target_b);
1794                let (_imported, handle) = imported.start_sync().await.unwrap();
1795                handle.await.unwrap();
1796            }
1797
1798            // Reopen recovers the imported state, replacing state A.
1799            let db = open_db::<mmr::Family>(context.child("reopen"), dst).await;
1800            assert_eq!(db.target(), target_b);
1801            assert_eq!(db.root(), target_b.root);
1802            assert_eq!(db.get_metadata(), Some(meta_b));
1803            db.destroy().await.unwrap();
1804        });
1805    }
1806
1807    #[test_traced("INFO")]
1808    fn test_compact_stale_batch_rejected() {
1809        deterministic::Runner::default().start(|context| async move {
1810            let db = open_db::<mmr::Family>(context.child("db"), "keyless-stale").await;
1811            let floor = db.inactivity_floor_loc();
1812
1813            let batch_a = db
1814                .new_batch()
1815                .append(U64::new(1))
1816                .merkleize(&db, Some(U64::new(11)), floor)
1817                .await
1818                .unwrap();
1819            let batch_b = db
1820                .new_batch()
1821                .append(U64::new(2))
1822                .merkleize(&db, Some(U64::new(22)), floor)
1823                .await
1824                .unwrap();
1825
1826            let stale = db.new_batch().append(U64::new(3));
1827            let expected_root = batch_a.root();
1828            let db = db.apply_batch(batch_a).await.unwrap();
1829            assert_eq!(db.root(), expected_root);
1830
1831            // After the apply, merkleizing a batch built on the pre-apply state and applying the
1832            // merkleized sibling both return `StaleBatch`.
1833            assert!(matches!(
1834                stale.merkleize(&db, Some(U64::new(33)), floor).await,
1835                Err(Error::StaleBatch)
1836            ));
1837            assert!(matches!(
1838                db.apply_batch(batch_b).await,
1839                Err(Error::StaleBatch)
1840            ));
1841        });
1842    }
1843
1844    #[test_traced("INFO")]
1845    fn test_compact_delayed_merkleize_after_ancestor_apply() {
1846        deterministic::Runner::default().start(|context| async move {
1847            let db = open_db::<mmr::Family>(context.child("db"), "keyless-delayed-child").await;
1848            let floor = db.inactivity_floor_loc();
1849
1850            let a = db
1851                .new_batch()
1852                .append(U64::new(1))
1853                .merkleize(&db, None, floor)
1854                .await
1855                .unwrap();
1856            let b = a
1857                .new_batch::<Sha256>()
1858                .append(U64::new(2))
1859                .merkleize(&db, None, floor)
1860                .await
1861                .unwrap();
1862            let c = b.new_batch::<Sha256>().append(U64::new(3));
1863
1864            let db = db.apply_batch(a).await.unwrap();
1865            let c = c.merkleize(&db, None, floor).await.unwrap();
1866            let expected_root = c.root();
1867            let db = db.apply_batch(c).await.unwrap();
1868
1869            assert_eq!(db.root(), expected_root);
1870        });
1871    }
1872
1873    /// `to_batch()` reflects the current applied state before it becomes durable.
1874    #[test_traced("INFO")]
1875    fn test_compact_to_batch_reflects_live_state() {
1876        deterministic::Runner::default().start(|context| async move {
1877            let db = open_db::<mmr::Family>(context.child("db"), "keyless-to-batch-live").await;
1878            let floor = db.inactivity_floor_loc();
1879
1880            let pre_apply_root = db.root();
1881            let pre_snapshot = db.to_batch();
1882            assert_eq!(
1883                pre_snapshot.root(),
1884                pre_apply_root,
1885                "snapshot before any mutation should match the live root"
1886            );
1887
1888            let batch = db
1889                .new_batch()
1890                .append(U64::new(1))
1891                .merkleize(&db, Some(U64::new(11)), floor)
1892                .await
1893                .unwrap();
1894            let db = db.apply_batch(batch).await.unwrap();
1895
1896            // Observe the applied state before making it durable.
1897            let live_root = db.root();
1898            assert_ne!(
1899                live_root, pre_apply_root,
1900                "applying a non-empty batch must change the live root"
1901            );
1902
1903            let snapshot = db.to_batch();
1904            assert_eq!(
1905                snapshot.root(),
1906                live_root,
1907                "to_batch().root() must match the live db.root() even before sync"
1908            );
1909
1910            db.destroy().await.unwrap();
1911        });
1912    }
1913
1914    #[test_traced("INFO")]
1915    fn test_compact_stale_batch_chained() {
1916        deterministic::Runner::default().start(|context| async move {
1917            let db = open_db::<mmr::Family>(context.child("db"), "keyless-chained-stale").await;
1918            let floor = db.inactivity_floor_loc();
1919
1920            let common_parent = db
1921                .new_batch()
1922                .append(U64::new(10))
1923                .merkleize(&db, Some(U64::new(110)), floor)
1924                .await
1925                .unwrap();
1926            let sibling_a = common_parent
1927                .new_batch::<Sha256>()
1928                .append(U64::new(11))
1929                .merkleize(&db, Some(U64::new(111)), floor)
1930                .await
1931                .unwrap();
1932            let sibling_b = common_parent
1933                .new_batch::<Sha256>()
1934                .append(U64::new(12))
1935                .merkleize(&db, Some(U64::new(112)), floor)
1936                .await
1937                .unwrap();
1938            let db = db.apply_batch(sibling_a).await.unwrap();
1939            assert!(matches!(
1940                db.validate_batch(&sibling_b),
1941                Err(Error::StaleBatch)
1942            ));
1943
1944            let parent_a = db
1945                .new_batch()
1946                .append(U64::new(1))
1947                .merkleize(&db, Some(U64::new(11)), floor)
1948                .await
1949                .unwrap();
1950            let parent_b = db
1951                .new_batch()
1952                .append(U64::new(2))
1953                .merkleize(&db, Some(U64::new(22)), floor)
1954                .await
1955                .unwrap();
1956            let child_b = parent_b
1957                .new_batch::<Sha256>()
1958                .append(U64::new(3))
1959                .merkleize(&db, Some(U64::new(33)), floor)
1960                .await
1961                .unwrap();
1962
1963            let db = db.apply_batch(parent_a).await.unwrap();
1964            assert!(matches!(
1965                db.validate_batch(&child_b),
1966                Err(Error::StaleBatch)
1967            ));
1968            db.destroy().await.unwrap();
1969        });
1970    }
1971
1972    #[test_traced("INFO")]
1973    fn test_compact_stale_parent_after_child_applied() {
1974        deterministic::Runner::default().start(|context| async move {
1975            let db =
1976                open_db::<mmr::Family>(context.child("db"), "keyless-child-before-parent").await;
1977            let floor = db.inactivity_floor_loc();
1978
1979            let parent = db
1980                .new_batch()
1981                .append(U64::new(1))
1982                .merkleize(&db, Some(U64::new(11)), floor)
1983                .await
1984                .unwrap();
1985            let child = parent
1986                .new_batch::<Sha256>()
1987                .append(U64::new(2))
1988                .merkleize(&db, Some(U64::new(22)), floor)
1989                .await
1990                .unwrap();
1991
1992            let db = db.apply_batch(child).await.unwrap();
1993            assert!(matches!(
1994                db.apply_batch(parent).await,
1995                Err(Error::StaleBatch)
1996            ));
1997        });
1998    }
1999
2000    #[test_traced("INFO")]
2001    fn test_compact_sequential_commit_parent_then_child() {
2002        deterministic::Runner::default().start(|context| async move {
2003            let db = open_db::<mmr::Family>(context.child("db"), "keyless-parent-child").await;
2004            let floor = db.inactivity_floor_loc();
2005
2006            let parent = db
2007                .new_batch()
2008                .append(U64::new(1))
2009                .merkleize(&db, Some(U64::new(11)), floor)
2010                .await
2011                .unwrap();
2012            let child = parent
2013                .new_batch::<Sha256>()
2014                .append(U64::new(2))
2015                .merkleize(&db, Some(U64::new(22)), floor)
2016                .await
2017                .unwrap();
2018            let expected_root = child.root();
2019
2020            let db = db.apply_batch(parent).await.unwrap();
2021            let db = db.apply_batch(child).await.unwrap().sync().await.unwrap();
2022
2023            assert_eq!(db.root(), expected_root);
2024
2025            db.destroy().await.unwrap();
2026        });
2027    }
2028
2029    #[test_traced("INFO")]
2030    fn test_compact_floor_regressed() {
2031        deterministic::Runner::default().start(|context| async move {
2032            let db = open_db::<mmr::Family>(context.child("db"), "keyless-floor-regressed").await;
2033
2034            let advance_floor = db.new_batch().append(U64::new(1));
2035            let advance_floor = advance_floor
2036                .merkleize(&db, None, Location::new(1))
2037                .await
2038                .unwrap();
2039            let db = db.apply_batch(advance_floor).await.unwrap();
2040            let db = db.sync().await.unwrap();
2041
2042            let regressed = db
2043                .new_batch()
2044                .append(U64::new(2))
2045                .merkleize(&db, None, Location::new(0))
2046                .await;
2047
2048            assert!(matches!(
2049                regressed,
2050                Err(Error::FloorRegressed(new, current))
2051                    if new == Location::new(0) && current == Location::new(1)
2052            ));
2053        });
2054    }
2055
2056    // A chained batch whose tip floor is below its parent's floor must be rejected:
2057    // the parent's Commit participates in the per-commit monotonicity invariant even
2058    // before it is applied.
2059    #[test_traced("INFO")]
2060    fn test_compact_chained_floor_regression() {
2061        deterministic::Runner::default().start(|context| async move {
2062            let db =
2063                open_db::<mmr::Family>(context.child("db"), "keyless-ancestor-floor-regressed")
2064                    .await;
2065
2066            // parent: append + commit at loc 2 with floor=2.
2067            let parent = db
2068                .new_batch()
2069                .append(U64::new(1))
2070                .merkleize(&db, None, Location::new(2))
2071                .await
2072                .unwrap();
2073            // child: append + commit at loc 4 with floor=1 (regressed from parent's floor=2).
2074            let child = parent
2075                .new_batch::<Sha256>()
2076                .append(U64::new(2))
2077                .merkleize(&db, None, Location::new(1))
2078                .await;
2079
2080            assert!(matches!(
2081                child,
2082                Err(Error::FloorRegressed(new, prev))
2083                    if new == Location::new(1) && prev == Location::new(2)
2084            ));
2085        });
2086    }
2087
2088    #[test_traced("INFO")]
2089    fn test_compact_bounded_initialization_restores_commit_metadata_and_floor() {
2090        deterministic::Runner::default().start(|context| async move {
2091            let db = open_db::<mmr::Family>(context.child("db"), "keyless-rewind-meta").await;
2092
2093            let v1 = U64::new(1);
2094            let meta1 = U64::new(11);
2095            let floor1 = Location::new(0);
2096            let batch = db
2097                .new_batch()
2098                .append(v1)
2099                .merkleize(&db, Some(meta1.clone()), floor1)
2100                .await
2101                .unwrap();
2102            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2103            let root_after_first = db.root();
2104            let size_after_first = db.size();
2105
2106            let v2 = U64::new(2);
2107            let meta2 = U64::new(22);
2108            let floor2 = Location::new(1);
2109            let batch = db
2110                .new_batch()
2111                .append(v2)
2112                .merkleize(&db, Some(meta2.clone()), floor2)
2113                .await
2114                .unwrap();
2115            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2116            assert_eq!(db.get_metadata(), Some(meta2));
2117            assert_eq!(db.inactivity_floor_loc(), floor2);
2118
2119            let db = {
2120                _ = db.sync().await.unwrap();
2121                open_bounded::<mmr::Family>(
2122                    context.child("cap"),
2123                    witness_config("keyless-rewind-meta"),
2124                    size_after_first,
2125                )
2126                .await
2127            }
2128            .unwrap();
2129            assert_eq!(db.root(), root_after_first);
2130            assert_eq!(db.get_metadata(), Some(meta1));
2131            assert_eq!(db.inactivity_floor_loc(), floor1);
2132
2133            db.destroy().await.unwrap();
2134        });
2135    }
2136
2137    #[test_traced("INFO")]
2138    fn test_compact_bounded_initialization_persists_across_reopen() {
2139        deterministic::Runner::default().start(|context| async move {
2140            let partition = "keyless-rewind-reopen";
2141            let meta1 = U64::new(11);
2142            let floor1 = Location::new(0);
2143            let meta2 = U64::new(22);
2144            let floor2 = Location::new(1);
2145
2146            let root_after_first = {
2147                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
2148                let batch = db
2149                    .new_batch()
2150                    .append(U64::new(1))
2151                    .merkleize(&db, Some(meta1.clone()), floor1)
2152                    .await
2153                    .unwrap();
2154                let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2155                let root = db.root();
2156                let size_after_first = db.size();
2157
2158                let batch = db
2159                    .new_batch()
2160                    .append(U64::new(2))
2161                    .merkleize(&db, Some(meta2), floor2)
2162                    .await
2163                    .unwrap();
2164                let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2165
2166                let _db = {
2167                    _ = db.sync().await.unwrap();
2168                    open_bounded::<mmr::Family>(
2169                        context.child("cap"),
2170                        witness_config(partition),
2171                        size_after_first,
2172                    )
2173                    .await
2174                }
2175                .unwrap();
2176                root
2177            };
2178
2179            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
2180            assert_eq!(db.root(), root_after_first);
2181            assert_eq!(db.get_metadata(), Some(meta1));
2182            assert_eq!(db.inactivity_floor_loc(), floor1);
2183
2184            db.destroy().await.unwrap();
2185        });
2186    }
2187
2188    #[test_traced("INFO")]
2189    fn test_compact_commit_persists_across_reopen() {
2190        deterministic::Runner::default().start(|context| async move {
2191            let partition = "keyless-commit-reopen";
2192            let meta1 = U64::new(11);
2193            let meta2 = U64::new(22);
2194
2195            let root_after_second = {
2196                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
2197                let batch = db
2198                    .new_batch()
2199                    .append(U64::new(1))
2200                    .merkleize(&db, Some(meta1), Location::new(0))
2201                    .await
2202                    .unwrap();
2203                let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
2204
2205                let batch = db
2206                    .new_batch()
2207                    .append(U64::new(2))
2208                    .merkleize(&db, Some(meta2.clone()), Location::new(1))
2209                    .await
2210                    .unwrap();
2211                let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
2212                db.root()
2213            };
2214
2215            // Reopen recovers the committed tip even though the journal was never synced.
2216            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
2217            assert_eq!(db.root(), root_after_second);
2218            assert_eq!(db.get_metadata(), Some(meta2));
2219            assert_eq!(db.inactivity_floor_loc(), Location::new(1));
2220            db.destroy().await.unwrap();
2221        });
2222    }
2223
2224    #[test_traced("INFO")]
2225    fn test_compact_bounded_initialization_to_committed_entry_after_reopen() {
2226        deterministic::Runner::default().start(|context| async move {
2227            let partition = "keyless-commit-rewind-reopen";
2228            let meta1 = U64::new(11);
2229            let meta2 = U64::new(22);
2230
2231            let (root_a, size_a) = {
2232                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
2233                let batch = db
2234                    .new_batch()
2235                    .append(U64::new(1))
2236                    .merkleize(&db, Some(meta1.clone()), Location::new(0))
2237                    .await
2238                    .unwrap();
2239                let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
2240                let root_a = db.root();
2241                let size_a = db.size();
2242
2243                let batch = db
2244                    .new_batch()
2245                    .append(U64::new(2))
2246                    .merkleize(&db, Some(meta2), Location::new(1))
2247                    .await
2248                    .unwrap();
2249                let _db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
2250                (root_a, size_a)
2251            };
2252
2253            // Both committed witnesses survive the crash. Reopen recovers the tip, and the earlier
2254            // commit remains a valid initialization target.
2255            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
2256            let db = {
2257                _ = db.sync().await.unwrap();
2258                open_bounded::<mmr::Family>(context.child("cap"), witness_config(partition), size_a)
2259                    .await
2260            }
2261            .unwrap();
2262            assert_eq!(db.root(), root_a);
2263            assert_eq!(db.get_metadata(), Some(meta1));
2264            db.destroy().await.unwrap();
2265        });
2266    }
2267
2268    #[test_traced("INFO")]
2269    fn test_compact_sync_after_commit() {
2270        deterministic::Runner::default().start(|context| async move {
2271            let partition = "keyless-sync-after-commit";
2272            let meta = U64::new(11);
2273
2274            let root = {
2275                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
2276                let batch = db
2277                    .new_batch()
2278                    .append(U64::new(1))
2279                    .merkleize(&db, Some(meta.clone()), Location::new(0))
2280                    .await
2281                    .unwrap();
2282                let db = db.apply_batch(batch).await.unwrap();
2283
2284                // Commit persists witness data. Sync must also persist recovery metadata.
2285                let db = db.commit().await.unwrap();
2286                let db = db.sync().await.unwrap();
2287                db.root()
2288            };
2289
2290            // Check the watermark before reopening can rebuild the offsets journal.
2291            let metadata = Metadata::<_, u64, VecU64>::init(
2292                context.child("checkpoint"),
2293                MetadataConfig {
2294                    partition: format!("{partition}-witness_offsets-metadata"),
2295                    codec_config: (),
2296                },
2297            )
2298            .await
2299            .unwrap();
2300
2301            // Key 3 records the durable prefix: the bootstrap witness and the applied batch.
2302            assert_eq!(metadata.get(&3).copied().map(u64::from), Some(2));
2303            drop(metadata);
2304
2305            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
2306            assert_eq!(db.root(), root);
2307            assert_eq!(db.get_metadata(), Some(meta));
2308            db.destroy().await.unwrap();
2309        });
2310    }
2311
2312    #[test_traced("INFO")]
2313    fn test_compact_import_persists_with_commit() {
2314        deterministic::Runner::default().start(|context| async move {
2315            let dst = "keyless-import-commit-dst";
2316            let src = "keyless-import-commit-src";
2317            let meta_a = U64::new(11);
2318            let meta_b = U64::new(22);
2319
2320            // Build state B in a separate source partition and fetch its state the way a sync
2321            // client would.
2322            let (target_b, pinned_b) = {
2323                let source = open_db::<mmr::Family>(context.child("src"), src).await;
2324                let batch = source
2325                    .new_batch()
2326                    .append(U64::new(2))
2327                    .merkleize(&source, Some(meta_b.clone()), Location::new(0))
2328                    .await
2329                    .unwrap();
2330                let source = source.apply_batch(batch).await.unwrap();
2331                let source = source.sync().await.unwrap();
2332                let target = source.target();
2333                let response = source
2334                    .serve(Request::Boundary {
2335                        size: target.size,
2336                        start: target.size - 1,
2337                    })
2338                    .await
2339                    .unwrap()
2340                    .0;
2341                let Response::Boundary { pinned_nodes, .. } = response else {
2342                    panic!("boundary request should get a boundary response");
2343                };
2344                (target, pinned_nodes)
2345            };
2346
2347            // Seed the destination partition with a different committed state A.
2348            {
2349                let seeded = open_db::<mmr::Family>(context.child("seed"), dst).await;
2350                let batch = seeded
2351                    .new_batch()
2352                    .append(U64::new(1))
2353                    .merkleize(&seeded, Some(meta_a), Location::new(0))
2354                    .await
2355                    .unwrap();
2356                let seeded = seeded.apply_batch(batch).await.unwrap();
2357                let seeded = seeded.sync().await.unwrap();
2358                assert_ne!(seeded.target(), target_b);
2359            }
2360
2361            // Import state B over the destination and make it durable with commit (not sync).
2362            {
2363                let journal = open_witness_journal(context.child("import"), dst).await;
2364                let imported = TestDb::<mmr::Family>::init_from_sync(
2365                    Sequential,
2366                    journal,
2367                    (),
2368                    target_b.size - 1,
2369                    pinned_b,
2370                    Operation::Commit(Some(meta_b.clone()), Location::new(0)),
2371                )
2372                .unwrap();
2373                assert_eq!(imported.target(), target_b);
2374                let _imported = imported.commit().await.unwrap();
2375            }
2376
2377            // Reopen recovers the committed import, replacing state A even though the journal was
2378            // never synced.
2379            let db = open_db::<mmr::Family>(context.child("reopen"), dst).await;
2380            assert_eq!(db.target(), target_b);
2381            assert_eq!(db.root(), target_b.root);
2382            assert_eq!(db.get_metadata(), Some(meta_b));
2383            db.destroy().await.unwrap();
2384        });
2385    }
2386
2387    #[test_traced("INFO")]
2388    fn test_compact_reopen_rejects_tampered_witness() {
2389        deterministic::Runner::default().start(|context| async move {
2390            let partition = "keyless-witness-tamper";
2391            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2392            let batch = db
2393                .new_batch()
2394                .append(U64::new(7))
2395                .merkleize(&db, Some(U64::new(11)), Location::new(1))
2396                .await
2397                .unwrap();
2398            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2399            drop(db);
2400
2401            // Corrupt the entry structurally. An extra pinned node cannot rebuild the Merkle.
2402            let journal = open_witness_journal(context.child("tamper"), partition).await;
2403            let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
2404            pinned_nodes.push(Sha256::fill(0xff));
2405            witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;
2406
2407            let cfg = Config {
2408                strategy: Sequential,
2409                witness: witness_config(partition),
2410                commit_codec_config: (),
2411            };
2412            let reopened =
2413                TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
2414            assert!(matches!(reopened, Err(Error::DataCorrupted(_))));
2415        });
2416    }
2417
2418    #[test_traced("INFO")]
2419    fn test_compact_bounded_initialization_rejects_corrupt_target_entry() {
2420        deterministic::Runner::default().start(|context| async move {
2421            let partition = "keyless-corrupt-rewind-target";
2422            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2423            let batch = db
2424                .new_batch()
2425                .append(U64::new(1))
2426                .merkleize(&db, None, Location::new(1))
2427                .await
2428                .unwrap();
2429            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2430            let initialization_bound = db.target().size;
2431            let batch = db
2432                .new_batch()
2433                .append(U64::new(2))
2434                .merkleize(&db, None, Location::new(1))
2435                .await
2436                .unwrap();
2437            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2438            let tip_target = db.target();
2439            drop(db);
2440
2441            // Corrupt the initialization target's entry (the journal holds bootstrap, target, tip).
2442            let mut journal = open_witness_journal(context.child("corrupt"), partition).await;
2443            journal = witness::tests::corrupt_entry(journal, 1, |entry| {
2444                entry.pinned_nodes.push(Sha256::fill(0xff));
2445            })
2446            .await;
2447            drop(journal);
2448
2449            // The tip entry is intact, so reopen succeeds.
2450            let cfg = Config {
2451                strategy: Sequential,
2452                witness: witness_config(partition),
2453                commit_codec_config: (),
2454            };
2455            let reopened = TestDb::<mmr::Family>::init(context.child("reopen"), cfg, None)
2456                .await
2457                .unwrap();
2458            assert_eq!(reopened.target(), tip_target);
2459
2460            // The corrupt entry fails the recovery before any truncation.
2461            assert!(matches!(
2462                {
2463                    drop(reopened);
2464                    open_bounded::<mmr::Family>(
2465                        context.child("cap"),
2466                        witness_config(partition),
2467                        initialization_bound,
2468                    )
2469                    .await
2470                },
2471                Err(Error::DataCorrupted(_))
2472            ));
2473
2474            // The newer history survives: reopen still lands on the original tip.
2475            let cfg = Config {
2476                strategy: Sequential,
2477                witness: witness_config(partition),
2478                commit_codec_config: (),
2479            };
2480            let reopened = TestDb::<mmr::Family>::init(context.child("reopen2"), cfg, None)
2481                .await
2482                .unwrap();
2483            assert_eq!(reopened.target(), tip_target);
2484            reopened.destroy().await.unwrap();
2485        });
2486    }
2487
2488    #[test_traced("INFO")]
2489    fn test_compact_reopen_rejects_interrupted_import() {
2490        deterministic::Runner::default().start(|context| async move {
2491            let partition = "keyless-interrupted-import";
2492            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2493            let batch = db
2494                .new_batch()
2495                .append(U64::new(7))
2496                .merkleize(&db, Some(U64::new(11)), Location::new(1))
2497                .await
2498                .unwrap();
2499            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2500            drop(db);
2501
2502            // Simulate a crash between an import's journal clear and its entry append: the
2503            // journal is empty but its size is nonzero.
2504            let journal = open_witness_journal(context.child("clear"), partition).await;
2505            let size = journal.size();
2506            let journal = journal.clear_to_size(size.max(1)).await.unwrap();
2507            drop(journal);
2508
2509            // Reopen must fail rather than bootstrap a fresh db.
2510            let cfg = Config {
2511                strategy: Sequential,
2512                witness: witness_config(partition),
2513                commit_codec_config: (),
2514            };
2515            let reopened =
2516                TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
2517            assert!(matches!(
2518                reopened,
2519                Err(Error::DataCorrupted("witness journal has no tip"))
2520            ));
2521        });
2522    }
2523
2524    #[test_traced("INFO")]
2525    fn test_compact_reopen_rejects_commit_floor_beyond_tip() {
2526        deterministic::Runner::default().start(|context| async move {
2527            let partition = "keyless-invalid-persisted-floor";
2528            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2529            let batch = db
2530                .new_batch()
2531                .append(U64::new(7))
2532                .merkleize(&db, Some(U64::new(11)), Location::new(1))
2533                .await
2534                .unwrap();
2535            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2536            drop(db);
2537            let oversized_floor = Location::new(10);
2538
2539            // Overwrite the persisted commit op with a floor beyond its own commit location.
2540            let journal = open_witness_journal(context.child("tamper"), partition).await;
2541            let (_, size, pinned_nodes) = witness::tests::tip(&journal).await;
2542            let bad_op = Operation::<mmr::Family, FixedEncoding<U64>>::Commit(
2543                Some(U64::new(11)),
2544                oversized_floor,
2545            )
2546            .encode()
2547            .to_vec();
2548            witness::tests::overwrite_tip(journal, bad_op, size, pinned_nodes).await;
2549
2550            let cfg = Config {
2551                strategy: Sequential,
2552                witness: witness_config(partition),
2553                commit_codec_config: (),
2554            };
2555            let reopened =
2556                TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
2557            assert!(matches!(
2558                reopened,
2559                Err(Error::DataCorrupted("invalid compact witness"))
2560            ));
2561        });
2562    }
2563
2564    #[test_traced("INFO")]
2565    fn test_compact_reopen_rejects_tampered_pinned_nodes() {
2566        deterministic::Runner::default().start(|context| async move {
2567            let partition = "keyless-pinned-nodes-tamper";
2568            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2569            let batch = db
2570                .new_batch()
2571                .append(U64::new(7))
2572                .merkleize(&db, Some(U64::new(11)), Location::new(1))
2573                .await
2574                .unwrap();
2575            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2576            let tampered_target = db.target();
2577            drop(db);
2578
2579            // Flip one pinned-node digest. There is no stored proof to cross-check against, so the
2580            // rebuild succeeds and yields a different root, the same way a bit-flipped replay
2581            // journal reopens with a different root.
2582            let journal = open_witness_journal(context.child("tamper"), partition).await;
2583            let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
2584            pinned_nodes[0] = Sha256::fill(0xff);
2585            witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;
2586
2587            let cfg = Config {
2588                strategy: Sequential,
2589                witness: witness_config(partition),
2590                commit_codec_config: (),
2591            };
2592            let reopened = TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None)
2593                .await
2594                .unwrap();
2595            assert_ne!(reopened.target(), tampered_target);
2596            reopened.destroy().await.unwrap();
2597        });
2598    }
2599
2600    #[test_traced("INFO")]
2601    fn test_compact_bounded_initialization_preserves_current_state() {
2602        deterministic::Runner::default().start(|context| async move {
2603            let db = open_db::<mmr::Family>(context.child("db"), "keyless-rewind-noop").await;
2604            let batch = db
2605                .new_batch()
2606                .append(U64::new(1))
2607                .merkleize(&db, Some(U64::new(11)), Location::new(0))
2608                .await
2609                .unwrap();
2610            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2611            let root = db.root();
2612            let size = db.size();
2613
2614            let db = {
2615                _ = db.sync().await.unwrap();
2616                open_bounded::<mmr::Family>(
2617                    context.child("cap"),
2618                    witness_config("keyless-rewind-noop"),
2619                    size,
2620                )
2621                .await
2622            }
2623            .unwrap();
2624            assert_eq!(db.root(), root);
2625            assert_eq!(db.size(), size);
2626            db.destroy().await.unwrap();
2627        });
2628    }
2629
2630    #[test_traced("INFO")]
2631    fn test_compact_prune_past_tip_keeps_tip() {
2632        deterministic::Runner::default().start(|context| async move {
2633            let partition = "keyless-prune-past-tip";
2634            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2635            let batch = db
2636                .new_batch()
2637                .append(U64::new(1))
2638                .merkleize(&db, Some(U64::new(11)), Location::new(0))
2639                .await
2640                .unwrap();
2641            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2642            let target = db.target();
2643
2644            // Prune with a boundary beyond the tip: the tip entry must survive.
2645            let boundary = db.size() + 100;
2646            let db = db.prune(boundary).await.unwrap();
2647            assert_eq!(db.target(), target);
2648            drop(db);
2649
2650            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
2651            assert_eq!(reopened.target(), target);
2652            reopened.destroy().await.unwrap();
2653        });
2654    }
2655
2656    #[test_traced("INFO")]
2657    fn test_compact_initialization_zero_and_above_end() {
2658        deterministic::Runner::default().start(|context| async move {
2659            let cfg = witness_config("keyless-caps");
2660            assert!(matches!(
2661                open_bounded::<mmr::Family>(context.child("zero"), cfg.clone(), Location::new(0))
2662                    .await,
2663                Err(Error::InvalidInitializationBound)
2664            ));
2665            let db = open_bounded::<mmr::Family>(
2666                context.child("fresh"),
2667                cfg.clone(),
2668                Location::new(100),
2669            )
2670            .await
2671            .unwrap();
2672            let root = db.root();
2673            assert_eq!(db.size(), Location::new(1));
2674            drop(db);
2675            let db = open_bounded::<mmr::Family>(context.child("above"), cfg, Location::new(100))
2676                .await
2677                .unwrap();
2678            assert_eq!(db.root(), root);
2679            assert_eq!(db.size(), Location::new(1));
2680        });
2681    }
2682
2683    #[test_traced("INFO")]
2684    fn test_compact_bounded_initialization_between_commits() {
2685        deterministic::Runner::default().start(|context| async move {
2686            let db = open_db::<mmr::Family>(context.child("db"), "keyless-rewind-between").await;
2687            let floor = db.inactivity_floor_loc();
2688            let initial_root = db.root();
2689
2690            // A multi-op commit jumps the committed size from 1 (bootstrap) to 4.
2691            let batch = db
2692                .new_batch()
2693                .append(U64::new(1))
2694                .append(U64::new(2))
2695                .merkleize(&db, Some(U64::new(11)), floor)
2696                .await
2697                .unwrap();
2698            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2699            let root_a = db.root();
2700            let size_a = db.size();
2701            assert_eq!(size_a, Location::new(4));
2702
2703            // A second commit moves the size to 6.
2704            let batch = db
2705                .new_batch()
2706                .append(U64::new(3))
2707                .merkleize(&db, Some(U64::new(22)), floor)
2708                .await
2709                .unwrap();
2710            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2711            let cfg = witness_config("keyless-rewind-between");
2712            drop(db);
2713            for (target, size, root) in [(5, 4, root_a), (3, 1, initial_root), (2, 1, initial_root)]
2714            {
2715                let db = open_bounded::<mmr::Family>(
2716                    context.child("cap"),
2717                    cfg.clone(),
2718                    Location::new(target),
2719                )
2720                .await
2721                .unwrap();
2722                assert_eq!(db.size(), Location::new(size));
2723                assert_eq!(db.root(), root);
2724                drop(db);
2725                let db = open_db::<mmr::Family>(context.child("restart"), "keyless-rewind-between")
2726                    .await;
2727                assert_eq!(db.root(), root);
2728            }
2729        });
2730    }
2731
2732    /// A witness entry appended but not synced (a commit interrupted before its journal sync)
2733    /// must be dropped on reopen, recovering the last synced commit.
2734    #[test_traced("INFO")]
2735    fn test_compact_reopen_drops_unsynced_witness() {
2736        deterministic::Runner::default().start(|context| async move {
2737            let partition = "keyless-witness-unsynced";
2738            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2739
2740            // Commit state A.
2741            let batch = db
2742                .new_batch()
2743                .append(U64::new(1))
2744                .merkleize(&db, Some(U64::new(11)), Location::new(1))
2745                .await
2746                .unwrap();
2747            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2748            let target_a = db.target();
2749            drop(db);
2750
2751            // Simulate the crash window: append an entry ahead of the tip without syncing it,
2752            // then drop the journal. The unsynced tail must not survive reopen.
2753            let journal = open_witness_journal(context.child("crash"), partition).await;
2754            let (op_bytes, mut size, pinned_nodes) = witness::tests::tip(&journal).await;
2755            size += 2;
2756            witness::tests::append_unsynced(journal, op_bytes, size, pinned_nodes).await;
2757
2758            // Reopen must drop the unsynced entry and recover state A.
2759            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
2760            assert_eq!(reopened.target(), target_a);
2761            reopened.destroy().await.unwrap();
2762        });
2763    }
2764
2765    #[test_traced("INFO")]
2766    fn test_compact_bounded_initialization_multiple_commits() {
2767        deterministic::Runner::default().start(|context| async move {
2768            let partition = "keyless-rewind-multi";
2769            let db = open_db::<mmr::Family>(context.child("db"), partition).await;
2770
2771            // Commit A, B, C, recording the state after A.
2772            let batch = db
2773                .new_batch()
2774                .append(U64::new(1))
2775                .merkleize(&db, Some(U64::new(11)), Location::new(0))
2776                .await
2777                .unwrap();
2778            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2779            let root_a = db.root();
2780            let size_a = db.size();
2781            let target_a = db.target();
2782
2783            let mut db = db;
2784            for i in [2u64, 3] {
2785                let batch = db
2786                    .new_batch()
2787                    .append(U64::new(i))
2788                    .merkleize(&db, Some(U64::new(i * 11)), Location::new(0))
2789                    .await
2790                    .unwrap();
2791                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2792            }
2793            assert_ne!(db.root(), root_a);
2794
2795            // Reopen two commits earlier.
2796            let db = {
2797                _ = db.sync().await.unwrap();
2798                open_bounded::<mmr::Family>(context.child("cap"), witness_config(partition), size_a)
2799                    .await
2800            }
2801            .unwrap();
2802            assert_eq!(db.root(), root_a);
2803            assert_eq!(db.size(), size_a);
2804            assert_eq!(db.get_metadata(), Some(U64::new(11)));
2805            assert_eq!(db.target(), target_a);
2806            drop(db);
2807
2808            // The recovery is durable: reopen recovers state A.
2809            let db = open_db::<mmr::Family>(context.child("reopen"), partition).await;
2810            assert_eq!(db.root(), root_a);
2811            assert_eq!(db.target(), target_a);
2812            db.destroy().await.unwrap();
2813        });
2814    }
2815
2816    #[test_traced("INFO")]
2817    fn test_compact_prune_then_bounded_initialization() {
2818        deterministic::Runner::default().start(|context| async move {
2819            // One entry per section so pruning takes effect at entry granularity (pruning is
2820            // section-aligned and never drops a partial section).
2821            let mut witness_cfg = witness_config("keyless-prune-rewind");
2822            witness_cfg.items_per_section = NZU64!(1);
2823            let cfg = Config {
2824                strategy: Sequential,
2825                witness: witness_cfg.clone(),
2826                commit_codec_config: (),
2827            };
2828            let mut db: TestDb<mmr::Family> =
2829                Db::init(context.child("db"), cfg, None).await.unwrap();
2830
2831            // Commit A, B, C.
2832            let mut sizes = Vec::new();
2833            for i in [1u64, 2, 3] {
2834                let batch = db
2835                    .new_batch()
2836                    .append(U64::new(i))
2837                    .merkleize(&db, Some(U64::new(i * 11)), Location::new(0))
2838                    .await
2839                    .unwrap();
2840                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2841                sizes.push(db.size());
2842            }
2843
2844            // Reopen retained state B before the terminal check that pruned state A is unavailable.
2845            let db = db.prune(sizes[1]).await.unwrap().sync().await.unwrap();
2846            drop(db);
2847            let db = open_bounded::<mmr::Family>(
2848                context.child("retained"),
2849                witness_cfg.clone(),
2850                sizes[1],
2851            )
2852            .await
2853            .unwrap();
2854            assert_eq!(db.size(), sizes[1]);
2855            assert_eq!(db.get_metadata(), Some(U64::new(22)));
2856            drop(db);
2857
2858            assert!(matches!(
2859                open_bounded::<mmr::Family>(context.child("pruned"), witness_cfg, sizes[0]).await,
2860                Err(Error::HistoricalFloorPruned(_))
2861            ));
2862        });
2863    }
2864
2865    #[test_traced("INFO")]
2866    fn test_compact_bounded_initialization_preserves_pre_advance_batch() {
2867        deterministic::Runner::default().start(|context| async move {
2868            let db =
2869                open_db::<mmr::Family>(context.child("db"), "keyless-rewind-preserves-pre-advance")
2870                    .await;
2871
2872            let batch = db
2873                .new_batch()
2874                .append(U64::new(1))
2875                .merkleize(&db, None, Location::new(0))
2876                .await
2877                .unwrap();
2878            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2879            let size_after_first = db.size();
2880
2881            // Merkleize a batch against the post-commit-A state.
2882            let held = db
2883                .new_batch()
2884                .append(U64::new(2))
2885                .merkleize(&db, None, Location::new(0))
2886                .await
2887                .unwrap();
2888
2889            // Advance past that state and commit, then reopen at that state.
2890            let batch = db
2891                .new_batch()
2892                .append(U64::new(3))
2893                .merkleize(&db, None, Location::new(0))
2894                .await
2895                .unwrap();
2896            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2897            let db = {
2898                _ = db.sync().await.unwrap();
2899                open_bounded::<mmr::Family>(
2900                    context.child("cap"),
2901                    witness_config("keyless-rewind-preserves-pre-advance"),
2902                    size_after_first,
2903                )
2904                .await
2905            }
2906            .unwrap();
2907
2908            // The recovery restored the state that `held` was merkleized against, so it still
2909            // matches the Merkle size and applies cleanly.
2910            let db = db.apply_batch(held).await.unwrap();
2911
2912            db.destroy().await.unwrap();
2913        });
2914    }
2915
2916    #[test_traced("INFO")]
2917    fn test_compact_noop_commit_after_commit() {
2918        deterministic::Runner::default().start(|context| async move {
2919            let db = open_db::<mmr::Family>(context.child("db"), "keyless-noop-after-commit").await;
2920
2921            let batch = db
2922                .new_batch()
2923                .append(U64::new(1))
2924                .append(U64::new(2))
2925                .merkleize(&db, Some(U64::new(11)), Location::new(0))
2926                .await
2927                .unwrap();
2928            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2929            let root_after_first = db.root();
2930            assert_eq!(db.size(), Location::new(4));
2931
2932            let db = db.sync().await.unwrap();
2933            assert_eq!(db.size(), Location::new(4));
2934            assert_eq!(db.root(), root_after_first);
2935
2936            db.destroy().await.unwrap();
2937        });
2938    }
2939
2940    #[test_traced("INFO")]
2941    fn test_compact_noop_commit_after_reopen() {
2942        deterministic::Runner::default().start(|context| async move {
2943            let partition = "keyless-noop-after-reopen";
2944
2945            let root_before_drop = {
2946                let db = open_db::<mmr::Family>(context.child("first"), partition).await;
2947                let batch = db
2948                    .new_batch()
2949                    .append(U64::new(1))
2950                    .append(U64::new(2))
2951                    .merkleize(&db, Some(U64::new(11)), Location::new(0))
2952                    .await
2953                    .unwrap();
2954                let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2955                let root = db.root();
2956                assert_eq!(db.size(), Location::new(4));
2957                root
2958            };
2959
2960            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
2961            assert_eq!(db.root(), root_before_drop);
2962            assert_eq!(db.size(), Location::new(4));
2963
2964            let db = db.sync().await.unwrap();
2965            assert_eq!(db.size(), Location::new(4));
2966            assert_eq!(db.root(), root_before_drop);
2967
2968            db.destroy().await.unwrap();
2969        });
2970    }
2971
2972    #[test_traced("INFO")]
2973    fn test_compact_noop_commit_after_bounded_initialization() {
2974        deterministic::Runner::default().start(|context| async move {
2975            let db = open_db::<mmr::Family>(context.child("db"), "keyless-noop-after-rewind").await;
2976
2977            let batch = db
2978                .new_batch()
2979                .append(U64::new(1))
2980                .append(U64::new(2))
2981                .merkleize(&db, Some(U64::new(11)), Location::new(0))
2982                .await
2983                .unwrap();
2984            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2985            let root_after_first = db.root();
2986
2987            let batch = db
2988                .new_batch()
2989                .append(U64::new(3))
2990                .merkleize(&db, Some(U64::new(22)), Location::new(1))
2991                .await
2992                .unwrap();
2993            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
2994
2995            let db = {
2996                _ = db.sync().await.unwrap();
2997                open_bounded::<mmr::Family>(
2998                    context.child("cap"),
2999                    witness_config("keyless-noop-after-rewind"),
3000                    Location::new(4),
3001                )
3002                .await
3003            }
3004            .unwrap();
3005            assert_eq!(db.size(), Location::new(4));
3006            assert_eq!(db.root(), root_after_first);
3007
3008            let db = db.sync().await.unwrap();
3009            assert_eq!(db.size(), Location::new(4));
3010            assert_eq!(db.root(), root_after_first);
3011
3012            db.destroy().await.unwrap();
3013        });
3014    }
3015
3016    #[test_traced("INFO")]
3017    fn test_compact_bounded_initialization_makes_post_advance_batch_stale() {
3018        deterministic::Runner::default().start(|context| async move {
3019            let db =
3020                open_db::<mmr::Family>(context.child("db"), "keyless-rewind-makes-stale").await;
3021
3022            let batch = db
3023                .new_batch()
3024                .append(U64::new(1))
3025                .merkleize(&db, None, Location::new(0))
3026                .await
3027                .unwrap();
3028            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
3029            let size_after_first = db.size();
3030
3031            let batch = db
3032                .new_batch()
3033                .append(U64::new(2))
3034                .merkleize(&db, None, Location::new(0))
3035                .await
3036                .unwrap();
3037            let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
3038
3039            // Merkleize a batch against the post-commit-B state, which the recovery will discard.
3040            let held = db
3041                .new_batch()
3042                .append(U64::new(3))
3043                .merkleize(&db, None, Location::new(0))
3044                .await
3045                .unwrap();
3046
3047            let db = {
3048                _ = db.sync().await.unwrap();
3049                open_bounded::<mmr::Family>(
3050                    context.child("cap"),
3051                    witness_config("keyless-rewind-makes-stale"),
3052                    size_after_first,
3053                )
3054                .await
3055            }
3056            .unwrap();
3057
3058            // After recovery, mem.size reflects post-commit-A, but the held batch starts after
3059            // post-commit-B. Apply must be rejected with StaleBatch.
3060            assert!(matches!(db.apply_batch(held).await, Err(Error::StaleBatch)));
3061        });
3062    }
3063
3064    #[test_traced("INFO")]
3065    fn test_compact_floor_beyond_size() {
3066        deterministic::Runner::default().start(|context| async move {
3067            let db = open_db::<mmr::Family>(context.child("db"), "keyless-floor-beyond").await;
3068
3069            let batch = db.new_batch().merkleize(&db, None, Location::new(2)).await;
3070
3071            assert!(matches!(
3072                batch,
3073                Err(Error::FloorBeyondSize(floor, tip))
3074                    if floor == Location::new(2) && tip == Location::new(1)
3075            ));
3076        });
3077    }
3078
3079    // Merkleize returns `FloorBeyondSize` for a chained batch whose floor exceeds its own commit
3080    // location. The commit location includes the parent's operations.
3081    #[test_traced("INFO")]
3082    fn test_compact_chained_floor_beyond_commit() {
3083        deterministic::Runner::default().start(|context| async move {
3084            let db =
3085                open_db::<mmr::Family>(context.child("db"), "keyless-ancestor-floor-beyond").await;
3086
3087            // parent: append + commit at loc 2, floor=2.
3088            let parent = db
3089                .new_batch()
3090                .append(U64::new(1))
3091                .merkleize(&db, None, Location::new(2))
3092                .await
3093                .unwrap();
3094
3095            // child: append + commit at loc 4, floor=5 (one past its commit).
3096            let child = parent
3097                .new_batch::<Sha256>()
3098                .append(U64::new(2))
3099                .merkleize(&db, None, Location::new(5))
3100                .await;
3101
3102            assert!(matches!(
3103                child,
3104                Err(Error::FloorBeyondSize(floor, commit))
3105                    if floor == Location::new(5) && commit == Location::new(4)
3106            ));
3107        });
3108    }
3109}