Skip to main content

commonware_storage/qmdb/immutable/
compact.rs

1//! An immutable authenticated db that discards historical operations, retaining only a witness
2//! for each synced commit.
3//!
4//! Mirrors the API of [`crate::qmdb::immutable::Immutable`] (`new_batch -> merkleize ->
5//! apply_batch -> sync`, pipelined batch chains, `StaleBatch` validation) but is backed by
6//! the peak-only [`crate::merkle::compact`]. Because history is discarded, there are no
7//! `get` / `proof` / `bounds` methods; use the full variant if you need them.
8//!
9//! # Witness journal
10//!
11//! The witness journal holds a complete snapshot of every synced commit, so [`Db::rewind`] can
12//! restore any commit still retained there (history is bounded only by [`Db::prune`]). Reopen
13//! and rewind re-verify the persisted snapshot; corruption surfaces as [`Error::DataCorrupted`].
14//! The witness (the last-commit operation plus its inclusion proof) is also what lets compact
15//! nodes serve compact sync without retaining historical operations.
16//!
17//! # Inactivity floor
18//!
19//! Commits carry an inactivity floor for wire-format compatibility with
20//! [`crate::qmdb::immutable::Immutable`]: the root is computed over the encoded operation
21//! sequence, and that sequence must include the same floor to produce the same root as the
22//! full variant. The floor has no effect on pruning or snapshot rebuilding here; all
23//! historical in-memory state is discarded on every sync.
24
25use super::operation::Operation;
26use crate::{
27    journal::contiguous::variable::{self, Config as JournalConfig},
28    merkle::{batch, compact as compact_merkle, Family, Location},
29    qmdb::{
30        self,
31        any::value::ValueEncoding,
32        batch_chain::{self, Bounds},
33        compact::{
34            batch as compact_batch,
35            witness::{self, VerifiedWitness, Witness},
36        },
37        operation::Key,
38        sync::compact as compact_sync,
39        Error,
40    },
41    Context,
42};
43use commonware_codec::{Decode as _, Encode, EncodeShared, Read};
44use commonware_cryptography::{Digest, Hasher};
45use commonware_macros::boxed;
46use commonware_parallel::Strategy;
47use core::marker::PhantomData;
48use std::{
49    collections::BTreeMap,
50    sync::{Arc, Weak},
51};
52
53/// Configuration for a compact immutable authenticated db.
54#[derive(Clone)]
55pub struct Config<C, S: Strategy> {
56    /// Strategy used to parallelize merkleization.
57    pub strategy: S,
58
59    /// Configuration for the journal that persists the witness.
60    pub witness: JournalConfig<()>,
61
62    /// Codec config used to decode the persisted last commit operation on reopen.
63    pub commit_codec_config: C,
64}
65
66/// An immutable authenticated db that discards historical operations, retaining only a witness
67/// for each synced commit.
68pub struct Db<F, E, K, V, H, C, S: Strategy>
69where
70    F: Family,
71    E: Context,
72    K: Key,
73    V: ValueEncoding,
74    H: Hasher,
75    Operation<F, K, V>: EncodeShared,
76    Operation<F, K, V>: Read<Cfg = C>,
77    C: Clone + Send + Sync + 'static,
78{
79    merkle: compact_merkle::Merkle<F, H::Digest, S>,
80    last_commit_loc: Location<F>,
81    last_commit_metadata: Option<V::Value>,
82    inactivity_floor_loc: Location<F>,
83    commit_codec_config: C,
84    witness: witness::Store<E, F, H::Digest>,
85    _key: PhantomData<K>,
86}
87
88type CompactStateResult<F, K, V, D> =
89    Result<compact_sync::State<F, Operation<F, K, V>, D>, compact_sync::ServeError<F, D>>;
90
91/// A speculative batch for a compact immutable db.
92#[allow(clippy::type_complexity)]
93pub struct UnmerkleizedBatch<F, H, K, V, S: Strategy>
94where
95    F: Family,
96    K: Key,
97    V: ValueEncoding,
98    H: Hasher,
99    Operation<F, K, V>: EncodeShared,
100{
101    merkle_batch: compact_merkle::UnmerkleizedBatch<F, H::Digest, S>,
102    mutations: BTreeMap<K, V::Value>,
103    parent: Option<Arc<MerkleizedBatch<F, H::Digest, K, V, S>>>,
104    base_size: u64,
105    db_size: u64,
106}
107
108/// A speculative batch whose root digest has been computed.
109#[derive(Clone)]
110pub struct MerkleizedBatch<F: Family, D: Digest, K: Key, V: ValueEncoding, S: Strategy>
111where
112    Operation<F, K, V>: EncodeShared,
113{
114    pub(super) merkle_batch: Arc<batch::MerkleizedBatch<F, D, S>>,
115    pub(super) root: D,
116    pub(super) commit_metadata: Option<V::Value>,
117    pub(super) parent: Option<Weak<Self>>,
118    pub(super) bounds: batch_chain::Bounds<F>,
119    pub(super) _key: PhantomData<K>,
120}
121
122impl<F: Family, D: Digest, K: Key, V: ValueEncoding, S: Strategy> MerkleizedBatch<F, D, K, V, S>
123where
124    Operation<F, K, V>: EncodeShared,
125{
126    pub(super) fn ancestors(&self) -> impl Iterator<Item = Arc<Self>> {
127        batch_chain::ancestors(self.parent.clone(), |batch| batch.parent.as_ref())
128    }
129
130    /// Return the root digest after this batch is applied.
131    pub const fn root(&self) -> D {
132        self.root
133    }
134
135    /// Return the [`Bounds`] of the batch.
136    pub const fn bounds(&self) -> &Bounds<F> {
137        &self.bounds
138    }
139
140    /// Create a new speculative batch with this one as its parent.
141    pub fn new_batch<H>(self: &Arc<Self>) -> UnmerkleizedBatch<F, H, K, V, S>
142    where
143        H: Hasher<Digest = D>,
144    {
145        UnmerkleizedBatch {
146            merkle_batch: compact_merkle::UnmerkleizedBatch::wrap(self.merkle_batch.new_batch()),
147            mutations: BTreeMap::new(),
148            parent: Some(Arc::clone(self)),
149            base_size: self.bounds.total_size,
150            db_size: self.bounds.db_size,
151        }
152    }
153}
154
155impl<F, H, K, V, S> UnmerkleizedBatch<F, H, K, V, S>
156where
157    F: Family,
158    K: Key,
159    V: ValueEncoding,
160    H: Hasher,
161    S: Strategy,
162    Operation<F, K, V>: EncodeShared,
163{
164    pub(super) fn new<E, C>(db: &Db<F, E, K, V, H, C, S>, committed_size: u64) -> Self
165    where
166        E: Context,
167        C: Clone + Send + Sync + 'static,
168        Operation<F, K, V>: Read<Cfg = C>,
169    {
170        Self {
171            merkle_batch: db.merkle.new_batch(),
172            mutations: BTreeMap::new(),
173            parent: None,
174            base_size: committed_size,
175            db_size: committed_size,
176        }
177    }
178
179    pub fn set(mut self, key: K, value: V::Value) -> Self {
180        self.mutations.insert(key, value);
181        self
182    }
183
184    /// Resolve mutations into operations, merkleize, and return an `Arc<MerkleizedBatch>`.
185    ///
186    /// `inactivity_floor` is threaded through the commit operation for wire-format parity with
187    /// [`crate::qmdb::immutable::Immutable`]. It must be >= the database's current floor
188    /// (monotonically non-decreasing) and at most the batch's commit location
189    /// (`total_size - 1`); these bounds are validated, but the floor does not drive any local
190    /// pruning or retention in this variant.
191    #[tracing::instrument(
192        name = "qmdb.immutable.compact.batch.merkleize",
193        level = "info",
194        skip_all
195    )]
196    pub async fn merkleize<E, C>(
197        self,
198        db: &Db<F, E, K, V, H, C, S>,
199        metadata: Option<V::Value>,
200        inactivity_floor: Location<F>,
201    ) -> Arc<MerkleizedBatch<F, H::Digest, K, V, S>>
202    where
203        F: Family,
204        E: Context,
205        C: Clone + Send + Sync + 'static,
206        Operation<F, K, V>: Read<Cfg = C>,
207    {
208        let mut ops: Vec<Operation<F, K, V>> = Vec::with_capacity(self.mutations.len() + 1);
209        for (key, value) in self.mutations {
210            ops.push(Operation::Set(key, value));
211        }
212        ops.push(Operation::Commit(metadata.clone(), inactivity_floor));
213
214        let total_size = self.base_size + ops.len() as u64;
215        let inactive_peaks = F::inactive_peaks(
216            F::location_to_position(Location::new(total_size)),
217            inactivity_floor,
218        );
219        let (merkle, root) = compact_batch::merkleize_ops::<F, H, S, _>(
220            &db.merkle,
221            self.merkle_batch,
222            ops,
223            inactive_peaks,
224        )
225        .await
226        .expect("inactive_peaks computed from batch size");
227
228        let ancestors =
229            batch_chain::parent_and_ancestors(self.parent.as_ref(), |parent| parent.ancestors());
230        let ancestors = batch_chain::collect_ancestor_bounds(
231            ancestors,
232            |batch| batch.bounds.inactivity_floor,
233            |batch| batch.bounds.total_size,
234        );
235
236        Arc::new(MerkleizedBatch {
237            merkle_batch: merkle,
238            root,
239            commit_metadata: metadata,
240            parent: self.parent.as_ref().map(Arc::downgrade),
241            bounds: batch_chain::Bounds {
242                base_size: self.base_size,
243                db_size: self.db_size,
244                total_size,
245                ancestors,
246                inactivity_floor,
247            },
248            _key: PhantomData,
249        })
250    }
251}
252
253impl<F, E, K, V, H, C, S> Db<F, E, K, V, H, C, S>
254where
255    F: Family,
256    E: Context,
257    K: Key,
258    V: ValueEncoding,
259    H: Hasher,
260    S: Strategy,
261    Operation<F, K, V>: EncodeShared,
262    Operation<F, K, V>: Read<Cfg = C>,
263    C: Clone + Send + Sync + 'static,
264{
265    fn encode_commit_op(metadata: Option<V::Value>, inactivity_floor_loc: Location<F>) -> Vec<u8> {
266        Operation::<F, K, V>::Commit(metadata, inactivity_floor_loc)
267            .encode()
268            .to_vec()
269    }
270
271    /// Build a compact db handle from already-validated compact state.
272    ///
273    /// The caller has reconstructed the compact Merkle in memory and already authenticated the
274    /// supplied witness/root pair. The import lives only in memory until the first [`Self::commit`]
275    /// or [`Self::sync`], which replaces the journal's contents with it. Until then, dropping the
276    /// handle leaves the previous on-disk state untouched, and rewind/prune are rejected.
277    pub(crate) fn init_from_validated_state(
278        strategy: S,
279        journal: witness::Journal<E, F, H::Digest>,
280        commit_codec_config: C,
281        validated: compact_sync::ValidatedState<F, Operation<F, K, V>, H::Digest>,
282    ) -> Result<Self, Error<F>> {
283        let compact_sync::ValidatedState {
284            state:
285                compact_sync::State {
286                    leaf_count,
287                    pinned_nodes,
288                    last_commit_op,
289                    last_commit_proof,
290                },
291            root,
292        } = validated;
293        let last_commit_loc = Location::new(*leaf_count - 1);
294        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
295            return Err(Error::UnexpectedData(last_commit_loc));
296        };
297
298        let merkle =
299            compact_merkle::Merkle::from_compact_state(strategy, leaf_count, pinned_nodes.clone())?;
300        let imported = VerifiedWitness {
301            witness: Witness {
302                op_bytes: Self::encode_commit_op(
303                    last_commit_metadata.clone(),
304                    inactivity_floor_loc,
305                ),
306                proof: last_commit_proof,
307                pinned_nodes,
308            },
309            root,
310        };
311
312        let witness = witness::Store::from_import(journal, imported);
313        Ok(Self {
314            merkle,
315            last_commit_loc,
316            last_commit_metadata,
317            inactivity_floor_loc,
318            commit_codec_config,
319            witness,
320            _key: PhantomData,
321        })
322    }
323
324    /// Open a compact db from persisted compact state and rebuild its witness store.
325    ///
326    /// On first open, this bootstraps the initial commit and its witness so every later reopen and
327    /// rewind can assume the journal tip is a complete compact witness.
328    #[boxed]
329    pub(crate) async fn init_from_merkle(
330        mut merkle: compact_merkle::Merkle<F, H::Digest, S>,
331        witness_context: E,
332        witness_config: JournalConfig<()>,
333        commit_codec_config: C,
334    ) -> Result<Self, Error<F>>
335    where
336        F: Family,
337        Operation<F, K, V>: Read<Cfg = C>,
338    {
339        // Bootstrap: append an initial Commit(None, 0) on first open.
340        let journal: witness::Journal<E, F, H::Digest> =
341            variable::Journal::init(witness_context, witness_config).await?;
342        let (witness, last_commit_op) = witness::init::<E, F, H, S, Operation<F, K, V>>(
343            journal,
344            &mut merkle,
345            &commit_codec_config,
346            Operation::<F, K, V>::Commit(None, Location::new(0))
347                .encode()
348                .to_vec(),
349            Operation::has_floor,
350        )
351        .await?;
352        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
353            return Err(Error::DataCorrupted("last operation was not a commit"));
354        };
355        let last_commit_loc = Location::new(*witness.with(|w| w.leaf_count()) - 1);
356
357        Ok(Self {
358            merkle,
359            last_commit_loc,
360            last_commit_metadata,
361            inactivity_floor_loc,
362            commit_codec_config,
363            witness,
364            _key: PhantomData,
365        })
366    }
367
368    /// Return the root of the db.
369    pub fn root(&self) -> H::Digest
370    where
371        F: Family,
372    {
373        let hasher = qmdb::hasher::<H>();
374        let inactive_peaks = F::inactive_peaks(
375            F::location_to_position(Location::new(*self.last_commit_loc + 1)),
376            self.inactivity_floor_loc,
377        );
378        self.merkle
379            .root(&hasher, inactive_peaks)
380            .expect("compact Merkle root should not fail")
381    }
382
383    /// Return a reference to the merkleization strategy.
384    pub const fn strategy(&self) -> &S {
385        self.merkle.strategy()
386    }
387
388    /// Return the location of the last commit.
389    pub const fn last_commit_loc(&self) -> Location<F> {
390        self.last_commit_loc
391    }
392
393    /// Return the inactivity floor declared by the last committed batch.
394    pub const fn inactivity_floor_loc(&self) -> Location<F> {
395        self.inactivity_floor_loc
396    }
397
398    /// Return the location of the next operation appended to this db.
399    pub fn size(&self) -> Location<F> {
400        Location::new(*self.last_commit_loc + 1)
401    }
402
403    /// Get the metadata associated with the last commit.
404    pub fn get_metadata(&self) -> Option<V::Value> {
405        self.last_commit_metadata.clone()
406    }
407
408    /// Return the compact-sync target described by the current witness.
409    ///
410    /// This reflects the last durably persisted commit, which may lag behind live in-memory
411    /// mutations until [`Self::commit`] or [`Self::sync`] is called.
412    pub fn target(&self) -> compact_sync::Target<F, H::Digest> {
413        self.witness.with(VerifiedWitness::target)
414    }
415
416    /// Return the compact-sync state for `target`, or a stale-target error if the source's
417    /// current witness no longer matches.
418    pub(crate) fn compact_state(
419        &self,
420        target: compact_sync::Target<F, H::Digest>,
421    ) -> CompactStateResult<F, K, V, H::Digest>
422    where
423        Operation<F, K, V>: Read<Cfg = C>,
424    {
425        // Hold the witness lock only long enough to verify the requested target and snapshot the
426        // entry; decode outside it so concurrent readers do not contend.
427        let (entry, leaf_count) = self.witness.with(|w| {
428            if target.root != w.root || target.leaf_count != w.leaf_count() {
429                return Err(compact_sync::ServeError::StaleTarget {
430                    requested: target.clone(),
431                    current: w.target(),
432                });
433            }
434            Ok((w.witness.clone(), w.leaf_count()))
435        })?;
436        let Witness {
437            op_bytes,
438            proof: last_commit_proof,
439            pinned_nodes,
440        } = entry;
441        let op = Operation::<F, K, V>::decode_cfg(op_bytes.as_ref(), &self.commit_codec_config)
442            .map_err(|_| {
443                compact_sync::ServeError::Database(Error::DataCorrupted("invalid commit operation"))
444            })?;
445        Ok(compact_sync::State {
446            leaf_count,
447            pinned_nodes,
448            last_commit_op: op,
449            last_commit_proof,
450        })
451    }
452
453    /// Create a new speculative batch of operations with this database as its parent.
454    pub fn new_batch(&self) -> UnmerkleizedBatch<F, H, K, V, S> {
455        let committed_size = *self.last_commit_loc + 1;
456        UnmerkleizedBatch::new(self, committed_size)
457    }
458
459    /// Create an owned merkleized batch representing the current applied state.
460    pub fn to_batch(&self) -> Arc<MerkleizedBatch<F, H::Digest, K, V, S>>
461    where
462        F: Family,
463    {
464        let committed_size = *self.last_commit_loc + 1;
465        Arc::new(MerkleizedBatch {
466            merkle_batch: self.merkle.to_batch(),
467            root: self.root(),
468            commit_metadata: self.last_commit_metadata.clone(),
469            parent: None,
470            bounds: batch_chain::Bounds {
471                base_size: committed_size,
472                db_size: committed_size,
473                total_size: committed_size,
474                ancestors: Vec::new(),
475                inactivity_floor: self.inactivity_floor_loc,
476            },
477            _key: PhantomData,
478        })
479    }
480
481    /// Apply a merkleized batch to the database.
482    ///
483    /// Returns the range of locations written. The state is updated in memory only; call
484    /// [`Self::commit`] or [`Self::sync`] to persist.
485    ///
486    /// # Errors
487    ///
488    /// - [`Error::StaleBatch`] if the batch is detected as stale (see
489    ///   [`crate::qmdb::batch_chain`] for more details).
490    /// - [`Error::FloorRegressed`] if any commit in the chain declares a floor below the
491    ///   previous commit's floor.
492    /// - [`Error::FloorBeyondSize`] if any commit in the chain declares a floor beyond its own
493    ///   commit location.
494    #[tracing::instrument(
495        name = "qmdb.immutable.compact.db.apply_batch",
496        level = "info",
497        skip_all
498    )]
499    pub fn apply_batch(
500        &mut self,
501        batch: Arc<MerkleizedBatch<F, H::Digest, K, V, S>>,
502    ) -> Result<core::ops::Range<Location<F>>, Error<F>> {
503        let db_size = *self.last_commit_loc + 1;
504        batch
505            .bounds
506            .validate_apply_to(db_size, self.inactivity_floor_loc)?;
507
508        let start_loc = self.last_commit_loc + 1;
509        self.merkle.apply_batch(&batch.merkle_batch)?;
510        self.last_commit_loc = Location::new(batch.bounds.total_size - 1);
511        self.last_commit_metadata = batch.commit_metadata.clone();
512        self.inactivity_floor_loc = batch.bounds.inactivity_floor;
513        Ok(start_loc..Location::new(batch.bounds.total_size))
514    }
515
516    /// Durably persist the current db state to disk. This is faster than [`Self::sync`] but
517    /// reopen may need to replay the witness journal's tail to recover.
518    #[tracing::instrument(name = "qmdb.immutable.compact.db.commit", level = "info", skip_all)]
519    pub async fn commit(&mut self) -> Result<(), Error<F>> {
520        self.witness
521            .commit::<H, S>(&self.merkle, self.inactivity_floor_loc, || {
522                Self::encode_commit_op(self.last_commit_metadata.clone(), self.inactivity_floor_loc)
523            })
524            .await
525    }
526
527    /// Durably persist the current db state to disk, also persisting journal metadata to
528    /// minimize recovery work on reopen.
529    #[tracing::instrument(name = "qmdb.immutable.compact.db.sync", level = "info", skip_all)]
530    pub async fn sync(&mut self) -> Result<(), Error<F>> {
531        self.witness
532            .sync::<H, S>(&self.merkle, self.inactivity_floor_loc, || {
533                Self::encode_commit_op(self.last_commit_metadata.clone(), self.inactivity_floor_loc)
534            })
535            .await
536    }
537
538    /// Rewind the db to the synced commit with exactly `target` operations, discarding any
539    /// uncommitted batches and any later commits. The rewind is made durable before this
540    /// method returns.
541    ///
542    /// # Errors
543    ///
544    /// Returns [`crate::merkle::Error::RewindBeyondHistory`] (wrapped as [`Error::Merkle`]) if
545    /// no retained commit has exactly `target` operations (never synced, or pruned).
546    #[tracing::instrument(name = "qmdb.immutable.compact.db.rewind", level = "info", skip_all)]
547    pub async fn rewind(&mut self, target: Location<F>) -> Result<(), Error<F>>
548    where
549        F: Family,
550    {
551        // Fast path: already durably at `target` with no uncommitted state.
552        if self.size() == target
553            && self.witness.with(|w| w.leaf_count()) == target
554            && !self.witness.import_pending()
555        {
556            return Ok(());
557        }
558
559        let last_commit_op = self
560            .witness
561            .rewind::<H, S, Operation<F, K, V>>(
562                &self.merkle,
563                target,
564                &self.commit_codec_config,
565                Operation::has_floor,
566            )
567            .await?;
568        let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
569            return Err(Error::DataCorrupted("last operation was not a commit"));
570        };
571        self.last_commit_metadata = last_commit_metadata;
572        self.inactivity_floor_loc = inactivity_floor_loc;
573        self.last_commit_loc = Location::new(*target - 1);
574        Ok(())
575    }
576
577    /// Drop witnesses for commits with fewer than `pruning_boundary` operations. Some witness below
578    /// the boundary may survive.
579    ///
580    /// Pruning bounds how far back [`Self::rewind`] can reach; the current commit's witness always
581    /// survives. The prune is made durable before this method returns.
582    ///
583    /// # Errors
584    ///
585    /// Fails if a compact-sync import has not yet been persisted by [`Self::commit`] or
586    /// [`Self::sync`].
587    pub async fn prune(&mut self, pruning_boundary: Location<F>) -> Result<(), Error<F>> {
588        self.witness.prune(pruning_boundary).await
589    }
590
591    /// Destroy all persisted state associated with this database.
592    #[boxed]
593    pub async fn destroy(self) -> Result<(), Error<F>> {
594        self.witness.destroy().await?;
595        Ok(())
596    }
597}
598
599#[cfg(test)]
600mod tests {
601    use super::*;
602    use crate::{
603        merkle::mmr,
604        qmdb::{any::value::FixedEncoding, compact::witness},
605    };
606    use commonware_cryptography::{sha256::Digest, Sha256};
607    use commonware_macros::test_traced;
608    use commonware_parallel::Sequential;
609    use commonware_runtime::{
610        buffer::paged::CacheRef, deterministic, BufferPooler, Runner as _, Supervisor as _,
611    };
612    use commonware_utils::{NZUsize, NZU16, NZU64};
613    use std::num::{NonZeroU16, NonZeroUsize};
614
615    type TestDb<F> =
616        Db<F, deterministic::Context, Digest, FixedEncoding<Digest>, Sha256, (), Sequential>;
617
618    const WITNESS_PAGE_SIZE: NonZeroU16 = NZU16!(77);
619    const WITNESS_PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9);
620
621    fn witness_config(partition: &str, pooler: &impl BufferPooler) -> variable::Config<()> {
622        variable::Config {
623            partition: format!("{partition}-witness"),
624            items_per_section: NZU64!(64),
625            compression: None,
626            codec_config: (),
627            page_cache: CacheRef::from_pooler(pooler, WITNESS_PAGE_SIZE, WITNESS_PAGE_CACHE_SIZE),
628            write_buffer: NZUsize!(1024),
629        }
630    }
631
632    async fn open_db<F: Family>(context: deterministic::Context, partition: &str) -> TestDb<F> {
633        let witness_cfg = witness_config(partition, &context);
634        let merkle = crate::merkle::compact::Merkle::new(Sequential);
635        Db::init_from_merkle(merkle, context.child("witness"), witness_cfg, ())
636            .await
637            .unwrap()
638    }
639
640    /// Open the persisted witness journal directly so tests can corrupt the tip entry.
641    async fn open_witness_journal(
642        context: deterministic::Context,
643        partition: &str,
644    ) -> witness::Journal<deterministic::Context, mmr::Family, Digest> {
645        let cfg = witness_config(partition, &context);
646        variable::Journal::init(context, cfg).await.unwrap()
647    }
648
649    #[test_traced("INFO")]
650    fn test_compact_stale_batch_rejected() {
651        deterministic::Runner::default().start(|context| async move {
652            let mut db = open_db::<mmr::Family>(context.child("db"), "immutable-stale").await;
653
654            let key1 = Sha256::hash(&[1]);
655            let key2 = Sha256::hash(&[2]);
656            let value1 = Sha256::fill(10u8);
657            let value2 = Sha256::fill(20u8);
658
659            let batch_a = db
660                .new_batch()
661                .set(key1, value1)
662                .merkleize(&db, None, Location::new(0))
663                .await;
664            let batch_b = db
665                .new_batch()
666                .set(key2, value2)
667                .merkleize(&db, None, Location::new(0))
668                .await;
669
670            let expected_root = batch_a.root();
671            db.apply_batch(batch_a).unwrap();
672            assert_eq!(db.root(), expected_root);
673            assert!(matches!(
674                db.apply_batch(batch_b),
675                Err(Error::StaleBatch { .. })
676            ));
677
678            db.destroy().await.unwrap();
679        });
680    }
681
682    /// Regression: `to_batch()` must snapshot the live in-memory state, not the lagging witness
683    /// cache.
684    #[test_traced("INFO")]
685    fn test_compact_to_batch_reflects_live_state() {
686        deterministic::Runner::default().start(|context| async move {
687            let mut db =
688                open_db::<mmr::Family>(context.child("db"), "immutable-to-batch-live").await;
689
690            let pre_apply_root = db.root();
691            let pre_snapshot = db.to_batch();
692            assert_eq!(
693                pre_snapshot.root(),
694                pre_apply_root,
695                "snapshot before any mutation should match the live root"
696            );
697
698            let key = Sha256::hash(&[1]);
699            let value = Sha256::fill(10u8);
700            db.apply_batch(
701                db.new_batch()
702                    .set(key, value)
703                    .merkleize(&db, None, Location::new(0))
704                    .await,
705            )
706            .unwrap();
707
708            // Leave the witness cache behind the live Merkle state.
709            let live_root = db.root();
710            assert_ne!(
711                live_root, pre_apply_root,
712                "applying a non-empty batch must change the live root"
713            );
714
715            let snapshot = db.to_batch();
716            assert_eq!(
717                snapshot.root(),
718                live_root,
719                "to_batch().root() must match the live db.root() even before sync"
720            );
721
722            db.destroy().await.unwrap();
723        });
724    }
725
726    #[test_traced("INFO")]
727    fn test_compact_stale_batch_chained() {
728        deterministic::Runner::default().start(|context| async move {
729            let mut db =
730                open_db::<mmr::Family>(context.child("db"), "immutable-chained-stale").await;
731
732            let parent = db
733                .new_batch()
734                .set(Sha256::hash(&[1]), Sha256::fill(1u8))
735                .merkleize(&db, None, Location::new(0))
736                .await;
737            let child_a = parent
738                .new_batch::<Sha256>()
739                .set(Sha256::hash(&[2]), Sha256::fill(2u8))
740                .merkleize(&db, None, Location::new(0))
741                .await;
742            let child_b = parent
743                .new_batch::<Sha256>()
744                .set(Sha256::hash(&[3]), Sha256::fill(3u8))
745                .merkleize(&db, None, Location::new(0))
746                .await;
747
748            db.apply_batch(child_a).unwrap();
749            assert!(matches!(
750                db.apply_batch(child_b),
751                Err(Error::StaleBatch { .. })
752            ));
753
754            db.destroy().await.unwrap();
755        });
756    }
757
758    #[test_traced("INFO")]
759    fn test_compact_stale_parent_after_child_applied() {
760        deterministic::Runner::default().start(|context| async move {
761            let mut db =
762                open_db::<mmr::Family>(context.child("db"), "immutable-child-before-parent").await;
763
764            let parent = db
765                .new_batch()
766                .set(Sha256::hash(&[1]), Sha256::fill(1u8))
767                .merkleize(&db, None, Location::new(0))
768                .await;
769            let child = parent
770                .new_batch::<Sha256>()
771                .set(Sha256::hash(&[2]), Sha256::fill(2u8))
772                .merkleize(&db, None, Location::new(0))
773                .await;
774
775            db.apply_batch(child).unwrap();
776            assert!(matches!(
777                db.apply_batch(parent),
778                Err(Error::StaleBatch { .. })
779            ));
780
781            db.destroy().await.unwrap();
782        });
783    }
784
785    #[test_traced("INFO")]
786    fn test_compact_sequential_commit_parent_then_child() {
787        deterministic::Runner::default().start(|context| async move {
788            let mut db =
789                open_db::<mmr::Family>(context.child("db"), "immutable-parent-child").await;
790
791            let parent = db
792                .new_batch()
793                .set(Sha256::hash(&[1]), Sha256::fill(1u8))
794                .merkleize(&db, None, Location::new(0))
795                .await;
796            let child = parent
797                .new_batch::<Sha256>()
798                .set(Sha256::hash(&[2]), Sha256::fill(2u8))
799                .merkleize(&db, None, Location::new(0))
800                .await;
801            let expected_root = child.root();
802
803            db.apply_batch(parent).unwrap();
804            db.apply_batch(child).unwrap();
805            db.sync().await.unwrap();
806
807            assert_eq!(db.root(), expected_root);
808
809            db.destroy().await.unwrap();
810        });
811    }
812
813    #[test_traced("INFO")]
814    fn test_compact_floor_regressed() {
815        deterministic::Runner::default().start(|context| async move {
816            let mut db =
817                open_db::<mmr::Family>(context.child("db"), "immutable-floor-regressed").await;
818
819            let advance_floor = db.new_batch().set(Sha256::hash(&[1]), Sha256::fill(1u8));
820            let advance_floor = advance_floor.merkleize(&db, None, Location::new(1)).await;
821            db.apply_batch(advance_floor).unwrap();
822
823            let regressed = db
824                .new_batch()
825                .set(Sha256::hash(&[2]), Sha256::fill(2u8))
826                .merkleize(&db, None, Location::new(0))
827                .await;
828
829            assert!(matches!(
830                db.apply_batch(regressed),
831                Err(Error::FloorRegressed(new, current))
832                    if new == Location::new(0) && current == Location::new(1)
833            ));
834
835            db.destroy().await.unwrap();
836        });
837    }
838
839    // A chained batch whose tip floor is below its parent's floor must be rejected:
840    // the parent's Commit participates in the per-commit monotonicity invariant even
841    // before it is applied.
842    #[test_traced("INFO")]
843    fn test_compact_ancestor_floor_regressed() {
844        deterministic::Runner::default().start(|context| async move {
845            let mut db =
846                open_db::<mmr::Family>(context.child("db"), "immutable-regressed-ancestor-floor")
847                    .await;
848
849            let parent = db
850                .new_batch()
851                .set(Sha256::hash(&[1]), Sha256::fill(1u8))
852                .merkleize(&db, None, Location::new(1))
853                .await;
854            let child = parent
855                .new_batch::<Sha256>()
856                .set(Sha256::hash(&[2]), Sha256::fill(2u8))
857                .merkleize(&db, None, Location::new(0))
858                .await;
859
860            assert!(matches!(
861                db.apply_batch(child),
862                Err(Error::FloorRegressed(new, prev))
863                    if new == Location::new(0) && prev == Location::new(1)
864            ));
865
866            db.destroy().await.unwrap();
867        });
868    }
869
870    #[test_traced("INFO")]
871    fn test_compact_rewind_restores_commit_metadata_and_floor() {
872        deterministic::Runner::default().start(|context| async move {
873            let mut db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-meta").await;
874
875            let k1 = Sha256::hash(&[1]);
876            let v1 = Sha256::fill(11u8);
877            let meta1 = Sha256::fill(0xaa);
878            let floor1 = Location::new(0);
879            db.apply_batch(
880                db.new_batch()
881                    .set(k1, v1)
882                    .merkleize(&db, Some(meta1), floor1)
883                    .await,
884            )
885            .unwrap();
886            db.sync().await.unwrap();
887            let root_after_first = db.root();
888            let size_after_first = db.size();
889
890            let k2 = Sha256::hash(&[2]);
891            let v2 = Sha256::fill(22u8);
892            let meta2 = Sha256::fill(0xbb);
893            // Advance the floor to the commit of the first batch (loc 1).
894            let floor2 = Location::new(1);
895            db.apply_batch(
896                db.new_batch()
897                    .set(k2, v2)
898                    .merkleize(&db, Some(meta2), floor2)
899                    .await,
900            )
901            .unwrap();
902            db.sync().await.unwrap();
903            assert_eq!(db.get_metadata(), Some(meta2));
904            assert_eq!(db.inactivity_floor_loc(), floor2);
905
906            db.rewind(size_after_first).await.unwrap();
907            assert_eq!(db.root(), root_after_first);
908            assert_eq!(db.get_metadata(), Some(meta1));
909            assert_eq!(db.inactivity_floor_loc(), floor1);
910
911            db.destroy().await.unwrap();
912        });
913    }
914
915    #[test_traced("INFO")]
916    fn test_compact_rewind_persists_across_reopen() {
917        deterministic::Runner::default().start(|context| async move {
918            let partition = "immutable-rewind-reopen";
919            let meta1 = Sha256::fill(0xaa);
920            let floor1 = Location::new(0);
921            let meta2 = Sha256::fill(0xbb);
922            let floor2 = Location::new(1);
923
924            let root_after_first = {
925                let mut db = open_db::<mmr::Family>(context.child("first"), partition).await;
926                db.apply_batch(
927                    db.new_batch()
928                        .set(Sha256::hash(&[1]), Sha256::fill(11u8))
929                        .merkleize(&db, Some(meta1), floor1)
930                        .await,
931                )
932                .unwrap();
933                db.sync().await.unwrap();
934                let root = db.root();
935                let size_after_first = db.size();
936
937                db.apply_batch(
938                    db.new_batch()
939                        .set(Sha256::hash(&[2]), Sha256::fill(22u8))
940                        .merkleize(&db, Some(meta2), floor2)
941                        .await,
942                )
943                .unwrap();
944                db.sync().await.unwrap();
945
946                db.rewind(size_after_first).await.unwrap();
947                root
948            };
949
950            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
951            assert_eq!(db.root(), root_after_first);
952            assert_eq!(db.get_metadata(), Some(meta1));
953            assert_eq!(db.inactivity_floor_loc(), floor1);
954
955            db.destroy().await.unwrap();
956        });
957    }
958
959    #[test_traced("INFO")]
960    fn test_compact_commit_persists_across_reopen() {
961        deterministic::Runner::default().start(|context| async move {
962            let partition = "immutable-commit-reopen";
963            let meta1 = Sha256::fill(0xaa);
964            let meta2 = Sha256::fill(0xbb);
965
966            let root_after_second = {
967                let mut db = open_db::<mmr::Family>(context.child("first"), partition).await;
968                db.apply_batch(
969                    db.new_batch()
970                        .set(Sha256::hash(&[1]), Sha256::fill(11u8))
971                        .merkleize(&db, Some(meta1), Location::new(0))
972                        .await,
973                )
974                .unwrap();
975                db.commit().await.unwrap();
976
977                db.apply_batch(
978                    db.new_batch()
979                        .set(Sha256::hash(&[2]), Sha256::fill(22u8))
980                        .merkleize(&db, Some(meta2), Location::new(1))
981                        .await,
982                )
983                .unwrap();
984                db.commit().await.unwrap();
985                db.root()
986            };
987
988            // Reopen recovers the committed tip even though the journal was never synced.
989            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
990            assert_eq!(db.root(), root_after_second);
991            assert_eq!(db.get_metadata(), Some(meta2));
992            assert_eq!(db.inactivity_floor_loc(), Location::new(1));
993            db.destroy().await.unwrap();
994        });
995    }
996
997    #[test_traced("INFO")]
998    fn test_compact_rewind_to_committed_entry_after_reopen() {
999        deterministic::Runner::default().start(|context| async move {
1000            let partition = "immutable-commit-rewind-reopen";
1001            let meta1 = Sha256::fill(0xaa);
1002            let meta2 = Sha256::fill(0xbb);
1003
1004            let (root_a, size_a) = {
1005                let mut db = open_db::<mmr::Family>(context.child("first"), partition).await;
1006                db.apply_batch(
1007                    db.new_batch()
1008                        .set(Sha256::hash(&[1]), Sha256::fill(11u8))
1009                        .merkleize(&db, Some(meta1), Location::new(0))
1010                        .await,
1011                )
1012                .unwrap();
1013                db.commit().await.unwrap();
1014                let root_a = db.root();
1015                let size_a = db.size();
1016
1017                db.apply_batch(
1018                    db.new_batch()
1019                        .set(Sha256::hash(&[2]), Sha256::fill(22u8))
1020                        .merkleize(&db, Some(meta2), Location::new(1))
1021                        .await,
1022                )
1023                .unwrap();
1024                db.commit().await.unwrap();
1025                (root_a, size_a)
1026            };
1027
1028            // Both committed witnesses survive the crash: reopen recovers the tip, and the
1029            // earlier commit remains a valid rewind target.
1030            let mut db = open_db::<mmr::Family>(context.child("second"), partition).await;
1031            db.rewind(size_a).await.unwrap();
1032            assert_eq!(db.root(), root_a);
1033            assert_eq!(db.get_metadata(), Some(meta1));
1034            db.destroy().await.unwrap();
1035        });
1036    }
1037
1038    #[test_traced("INFO")]
1039    fn test_compact_sync_after_commit() {
1040        deterministic::Runner::default().start(|context| async move {
1041            let partition = "immutable-sync-after-commit";
1042            let meta = Sha256::fill(0xaa);
1043
1044            let root = {
1045                let mut db = open_db::<mmr::Family>(context.child("first"), partition).await;
1046                db.apply_batch(
1047                    db.new_batch()
1048                        .set(Sha256::hash(&[1]), Sha256::fill(11u8))
1049                        .merkleize(&db, Some(meta), Location::new(0))
1050                        .await,
1051                )
1052                .unwrap();
1053                db.commit().await.unwrap();
1054                // The commit already made the state durable, so this is a no-op.
1055                db.sync().await.unwrap();
1056                db.root()
1057            };
1058
1059            let db = open_db::<mmr::Family>(context.child("second"), partition).await;
1060            assert_eq!(db.root(), root);
1061            assert_eq!(db.get_metadata(), Some(meta));
1062            db.destroy().await.unwrap();
1063        });
1064    }
1065
1066    #[test_traced("INFO")]
1067    fn test_compact_reopen_rejects_tampered_witness() {
1068        deterministic::Runner::default().start(|context| async move {
1069            let partition = "immutable-witness-tamper";
1070            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1071            db.apply_batch(
1072                db.new_batch()
1073                    .set(Sha256::hash(&[7]), Sha256::fill(7u8))
1074                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1075                    .await,
1076            )
1077            .unwrap();
1078            db.sync().await.unwrap();
1079            drop(db);
1080
1081            // Corrupt the persisted proof so it no longer verifies against the stored root.
1082            let mut journal = open_witness_journal(context.child("tamper"), partition).await;
1083            let (op_bytes, mut proof, pinned_nodes) = witness::tests::tip(&journal).await;
1084            if let Some(digest) = proof.digests.first_mut() {
1085                *digest = Sha256::fill(0xff);
1086            } else {
1087                proof.leaves = Location::new(*proof.leaves + 1);
1088            }
1089            witness::tests::overwrite_tip(&mut journal, op_bytes, proof, pinned_nodes).await;
1090            drop(journal);
1091
1092            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1093            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1094                merkle,
1095                context.child("reopen_witness"),
1096                witness_config(partition, &context),
1097                (),
1098            )
1099            .await;
1100            assert!(matches!(reopened, Err(Error::DataCorrupted(_))));
1101        });
1102    }
1103
1104    #[test_traced("INFO")]
1105    fn test_compact_rewind_rejects_corrupt_target_entry() {
1106        deterministic::Runner::default().start(|context| async move {
1107            let partition = "immutable-corrupt-rewind-target";
1108            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1109            db.apply_batch(
1110                db.new_batch()
1111                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1112                    .merkleize(&db, None, Location::new(1))
1113                    .await,
1114            )
1115            .unwrap();
1116            db.sync().await.unwrap();
1117            let rewind_target = db.target().leaf_count;
1118            db.apply_batch(
1119                db.new_batch()
1120                    .set(Sha256::hash(&[2]), Sha256::fill(2u8))
1121                    .merkleize(&db, None, Location::new(1))
1122                    .await,
1123            )
1124            .unwrap();
1125            db.sync().await.unwrap();
1126            let tip_target = db.target();
1127            drop(db);
1128
1129            // Corrupt the rewind target's entry (the journal holds bootstrap, target, tip).
1130            let mut journal = open_witness_journal(context.child("corrupt"), partition).await;
1131            witness::tests::corrupt_entry(&mut journal, 1, |entry| {
1132                entry.pinned_nodes[0] = Sha256::fill(0xff);
1133            })
1134            .await;
1135            drop(journal);
1136
1137            // The tip entry is intact, so reopen succeeds.
1138            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1139            let mut reopened = TestDb::<mmr::Family>::init_from_merkle(
1140                merkle,
1141                context.child("reopen"),
1142                witness_config(partition, &context),
1143                (),
1144            )
1145            .await
1146            .unwrap();
1147            assert_eq!(reopened.target(), tip_target);
1148
1149            // The corrupt entry fails the rewind before any truncation.
1150            assert!(matches!(
1151                reopened.rewind(rewind_target).await,
1152                Err(Error::DataCorrupted(_))
1153            ));
1154            drop(reopened);
1155
1156            // The newer history survives: reopen still lands on the original tip.
1157            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1158            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1159                merkle,
1160                context.child("reopen2"),
1161                witness_config(partition, &context),
1162                (),
1163            )
1164            .await
1165            .unwrap();
1166            assert_eq!(reopened.target(), tip_target);
1167            reopened.destroy().await.unwrap();
1168        });
1169    }
1170
1171    #[test_traced("INFO")]
1172    fn test_compact_reopen_rejects_interrupted_import() {
1173        deterministic::Runner::default().start(|context| async move {
1174            let partition = "immutable-interrupted-import";
1175            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1176            db.apply_batch(
1177                db.new_batch()
1178                    .set(Sha256::hash(&[7]), Sha256::fill(7u8))
1179                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1180                    .await,
1181            )
1182            .unwrap();
1183            db.sync().await.unwrap();
1184            drop(db);
1185
1186            // Simulate a crash between an import's journal clear and its entry append: the
1187            // journal is empty but its size is nonzero.
1188            let mut journal = open_witness_journal(context.child("clear"), partition).await;
1189            let size = journal.size();
1190            journal.clear_to_size(size.max(1)).await.unwrap();
1191            drop(journal);
1192
1193            // Reopen must fail rather than bootstrap a fresh db.
1194            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1195            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1196                merkle,
1197                context.child("reopen_witness"),
1198                witness_config(partition, &context),
1199                (),
1200            )
1201            .await;
1202            assert!(matches!(reopened, Err(Error::Journal(_))));
1203        });
1204    }
1205
1206    #[test_traced("INFO")]
1207    fn test_compact_reopen_rejects_commit_floor_beyond_tip() {
1208        deterministic::Runner::default().start(|context| async move {
1209            let partition = "immutable-invalid-persisted-floor";
1210            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1211            db.apply_batch(
1212                db.new_batch()
1213                    .set(Sha256::hash(&[7]), Sha256::fill(7u8))
1214                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1215                    .await,
1216            )
1217            .unwrap();
1218            db.sync().await.unwrap();
1219            drop(db);
1220            let oversized_floor = Location::new(10);
1221
1222            // Overwrite the persisted commit op with a floor beyond its own commit location.
1223            let mut journal = open_witness_journal(context.child("tamper"), partition).await;
1224            let (_, proof, pinned_nodes) = witness::tests::tip(&journal).await;
1225            let bad_op = Operation::<mmr::Family, Digest, FixedEncoding<Digest>>::Commit(
1226                Some(Sha256::fill(0xaa)),
1227                oversized_floor,
1228            )
1229            .encode()
1230            .to_vec();
1231            witness::tests::overwrite_tip(&mut journal, bad_op, proof, pinned_nodes).await;
1232            drop(journal);
1233
1234            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1235            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1236                merkle,
1237                context.child("reopen_witness"),
1238                witness_config(partition, &context),
1239                (),
1240            )
1241            .await;
1242            assert!(matches!(
1243                reopened,
1244                Err(Error::DataCorrupted("invalid compact witness"))
1245            ));
1246        });
1247    }
1248
1249    #[test_traced("INFO")]
1250    fn test_compact_reopen_rejects_tampered_pinned_nodes() {
1251        deterministic::Runner::default().start(|context| async move {
1252            let partition = "immutable-pins-tamper";
1253            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1254            db.apply_batch(
1255                db.new_batch()
1256                    .set(Sha256::hash(&[7]), Sha256::fill(7u8))
1257                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(1))
1258                    .await,
1259            )
1260            .unwrap();
1261            db.sync().await.unwrap();
1262            drop(db);
1263
1264            // Corrupt one pinned frontier node: the root recomputed from the rebuilt Merkle no
1265            // longer matches the proof stored in the same entry.
1266            let mut journal = open_witness_journal(context.child("tamper"), partition).await;
1267            let (op_bytes, proof, mut pinned_nodes) = witness::tests::tip(&journal).await;
1268            pinned_nodes[0] = Sha256::fill(0xff);
1269            witness::tests::overwrite_tip(&mut journal, op_bytes, proof, pinned_nodes).await;
1270            drop(journal);
1271
1272            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1273            let reopened = TestDb::<mmr::Family>::init_from_merkle(
1274                merkle,
1275                context.child("reopen_witness"),
1276                witness_config(partition, &context),
1277                (),
1278            )
1279            .await;
1280            assert!(matches!(reopened, Err(Error::DataCorrupted(_))));
1281        });
1282    }
1283
1284    /// A witness entry appended but not synced (a commit interrupted before its journal sync)
1285    /// must be dropped on reopen, recovering the last synced commit.
1286    #[test_traced("INFO")]
1287    fn test_compact_reopen_drops_unsynced_witness() {
1288        deterministic::Runner::default().start(|context| async move {
1289            let partition = "immutable-witness-unsynced";
1290            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1291
1292            // Commit state A.
1293            db.apply_batch(
1294                db.new_batch()
1295                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1296                    .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(1))
1297                    .await,
1298            )
1299            .unwrap();
1300            db.sync().await.unwrap();
1301            let target_a = db.target();
1302            drop(db);
1303
1304            // Simulate the crash window: append an entry ahead of the tip without syncing it,
1305            // then drop the journal. The unsynced tail must not survive reopen.
1306            let mut journal = open_witness_journal(context.child("crash"), partition).await;
1307            let (op_bytes, mut proof, pinned_nodes) = witness::tests::tip(&journal).await;
1308            proof.leaves = Location::new(*proof.leaves + 2);
1309            witness::tests::append_unsynced(&mut journal, op_bytes, proof, pinned_nodes).await;
1310            drop(journal);
1311
1312            // Reopen must drop the unsynced entry and recover state A.
1313            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
1314            assert_eq!(reopened.target(), target_a);
1315            reopened.destroy().await.unwrap();
1316        });
1317    }
1318
1319    #[test_traced("INFO")]
1320    fn test_compact_rewind_beyond_history() {
1321        deterministic::Runner::default().start(|context| async move {
1322            let mut db =
1323                open_db::<mmr::Family>(context.child("db"), "immutable-rewind-beyond").await;
1324            // The bootstrap commit is the oldest retained state (one leaf); no commit with zero
1325            // operations exists to rewind to.
1326            assert!(matches!(
1327                db.rewind(Location::new(0)).await,
1328                Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
1329            ));
1330            // A target past the tip is not a commit either.
1331            assert!(matches!(
1332                db.rewind(Location::new(*db.size() + 100)).await,
1333                Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
1334            ));
1335            db.destroy().await.unwrap();
1336        });
1337    }
1338
1339    #[test_traced("INFO")]
1340    fn test_compact_rewind_between_commits() {
1341        deterministic::Runner::default().start(|context| async move {
1342            let mut db =
1343                open_db::<mmr::Family>(context.child("db"), "immutable-rewind-between").await;
1344
1345            // A multi-op commit jumps the committed size from 1 (bootstrap) to 4.
1346            db.apply_batch(
1347                db.new_batch()
1348                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1349                    .set(Sha256::hash(&[2]), Sha256::fill(2u8))
1350                    .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
1351                    .await,
1352            )
1353            .unwrap();
1354            db.sync().await.unwrap();
1355            let root_a = db.root();
1356            let size_a = db.size();
1357            assert_eq!(size_a, Location::new(4));
1358
1359            // A second commit moves the size to 6.
1360            db.apply_batch(
1361                db.new_batch()
1362                    .set(Sha256::hash(&[3]), Sha256::fill(3u8))
1363                    .merkleize(&db, Some(Sha256::fill(0xb1)), Location::new(0))
1364                    .await,
1365            )
1366            .unwrap();
1367            db.sync().await.unwrap();
1368            let root_b = db.root();
1369
1370            // Targets inside a commit's span match no entry, even though entries exist on
1371            // both sides.
1372            for target in [2u64, 3, 5] {
1373                assert!(matches!(
1374                    db.rewind(Location::new(target)).await,
1375                    Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
1376                ));
1377            }
1378            assert_eq!(db.root(), root_b);
1379
1380            // The exact commit boundary remains a valid target.
1381            db.rewind(size_a).await.unwrap();
1382            assert_eq!(db.root(), root_a);
1383            assert_eq!(db.get_metadata(), Some(Sha256::fill(0xa1)));
1384            db.destroy().await.unwrap();
1385        });
1386    }
1387
1388    #[test_traced("INFO")]
1389    fn test_compact_rewind_multiple_commits() {
1390        deterministic::Runner::default().start(|context| async move {
1391            let partition = "immutable-rewind-multi";
1392            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1393
1394            // Commit A, B, C, recording the state after A.
1395            db.apply_batch(
1396                db.new_batch()
1397                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1398                    .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
1399                    .await,
1400            )
1401            .unwrap();
1402            db.sync().await.unwrap();
1403            let root_a = db.root();
1404            let size_a = db.size();
1405            let target_a = db.target();
1406
1407            for i in [2u8, 3] {
1408                db.apply_batch(
1409                    db.new_batch()
1410                        .set(Sha256::hash(&[i]), Sha256::fill(i))
1411                        .merkleize(&db, Some(Sha256::fill(i)), Location::new(0))
1412                        .await,
1413                )
1414                .unwrap();
1415                db.sync().await.unwrap();
1416            }
1417            assert_ne!(db.root(), root_a);
1418
1419            // Rewind two commits in one call.
1420            db.rewind(size_a).await.unwrap();
1421            assert_eq!(db.root(), root_a);
1422            assert_eq!(db.size(), size_a);
1423            assert_eq!(db.get_metadata(), Some(Sha256::fill(0xa1)));
1424            assert_eq!(db.target(), target_a);
1425            drop(db);
1426
1427            // The rewind is durable: reopen recovers state A.
1428            let db = open_db::<mmr::Family>(context.child("reopen"), partition).await;
1429            assert_eq!(db.root(), root_a);
1430            assert_eq!(db.target(), target_a);
1431            db.destroy().await.unwrap();
1432        });
1433    }
1434
1435    #[test_traced("INFO")]
1436    fn test_compact_rewind_to_current_is_noop() {
1437        deterministic::Runner::default().start(|context| async move {
1438            let mut db = open_db::<mmr::Family>(context.child("db"), "immutable-rewind-noop").await;
1439            db.apply_batch(
1440                db.new_batch()
1441                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1442                    .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
1443                    .await,
1444            )
1445            .unwrap();
1446            db.sync().await.unwrap();
1447            let root = db.root();
1448            let size = db.size();
1449
1450            db.rewind(size).await.unwrap();
1451            assert_eq!(db.root(), root);
1452            assert_eq!(db.size(), size);
1453            db.destroy().await.unwrap();
1454        });
1455    }
1456
1457    #[test_traced("INFO")]
1458    fn test_compact_prune_then_rewind() {
1459        deterministic::Runner::default().start(|context| async move {
1460            // One entry per section so pruning takes effect at entry granularity (pruning is
1461            // section-aligned and never drops a partial section).
1462            let mut witness_cfg = witness_config("immutable-prune-rewind", &context);
1463            witness_cfg.items_per_section = NZU64!(1);
1464            let merkle = crate::merkle::compact::Merkle::new(Sequential);
1465            let mut db: TestDb<mmr::Family> =
1466                Db::init_from_merkle(merkle, context.child("witness"), witness_cfg, ())
1467                    .await
1468                    .unwrap();
1469
1470            // Commit A, B, C.
1471            let mut sizes = Vec::new();
1472            for i in [1u8, 2, 3] {
1473                db.apply_batch(
1474                    db.new_batch()
1475                        .set(Sha256::hash(&[i]), Sha256::fill(i))
1476                        .merkleize(&db, Some(Sha256::fill(i)), Location::new(0))
1477                        .await,
1478                )
1479                .unwrap();
1480                db.sync().await.unwrap();
1481                sizes.push(db.size());
1482            }
1483
1484            // Prune history below B: rewinding to B still works, rewinding to A does not.
1485            db.prune(sizes[1]).await.unwrap();
1486            assert!(matches!(
1487                db.rewind(sizes[0]).await,
1488                Err(Error::Merkle(crate::merkle::Error::RewindBeyondHistory))
1489            ));
1490            db.rewind(sizes[1]).await.unwrap();
1491            assert_eq!(db.size(), sizes[1]);
1492            assert_eq!(db.get_metadata(), Some(Sha256::fill(2)));
1493
1494            db.destroy().await.unwrap();
1495        });
1496    }
1497
1498    #[test_traced("INFO")]
1499    fn test_compact_prune_past_tip_keeps_tip() {
1500        deterministic::Runner::default().start(|context| async move {
1501            let partition = "immutable-prune-past-tip";
1502            let mut db = open_db::<mmr::Family>(context.child("db"), partition).await;
1503            db.apply_batch(
1504                db.new_batch()
1505                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1506                    .merkleize(&db, Some(Sha256::fill(0xa1)), Location::new(0))
1507                    .await,
1508            )
1509            .unwrap();
1510            db.sync().await.unwrap();
1511            let target = db.target();
1512
1513            // Prune with a boundary beyond the tip: the tip entry must survive.
1514            db.prune(Location::new(*db.size() + 100)).await.unwrap();
1515            assert_eq!(db.target(), target);
1516            drop(db);
1517
1518            let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
1519            assert_eq!(reopened.target(), target);
1520            reopened.destroy().await.unwrap();
1521        });
1522    }
1523
1524    #[test_traced("INFO")]
1525    fn test_compact_rewind_preserves_pre_advance_batch() {
1526        deterministic::Runner::default().start(|context| async move {
1527            let mut db = open_db::<mmr::Family>(
1528                context.child("db"),
1529                "immutable-rewind-preserves-pre-advance",
1530            )
1531            .await;
1532
1533            db.apply_batch(
1534                db.new_batch()
1535                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1536                    .merkleize(&db, None, Location::new(0))
1537                    .await,
1538            )
1539            .unwrap();
1540            db.sync().await.unwrap();
1541            let size_after_first = db.size();
1542
1543            // Merkleize a batch against the post-commit-A state.
1544            let held = db
1545                .new_batch()
1546                .set(Sha256::hash(&[2]), Sha256::fill(2u8))
1547                .merkleize(&db, None, Location::new(0))
1548                .await;
1549
1550            // Advance past that state and commit, then rewind back to it.
1551            db.apply_batch(
1552                db.new_batch()
1553                    .set(Sha256::hash(&[3]), Sha256::fill(3u8))
1554                    .merkleize(&db, None, Location::new(0))
1555                    .await,
1556            )
1557            .unwrap();
1558            db.sync().await.unwrap();
1559            db.rewind(size_after_first).await.unwrap();
1560
1561            // The rewind restored the state that `held` was merkleized against, so it still
1562            // matches the Merkle size and applies cleanly.
1563            db.apply_batch(held).unwrap();
1564
1565            db.destroy().await.unwrap();
1566        });
1567    }
1568
1569    #[test_traced("INFO")]
1570    fn test_compact_noop_commit_after_commit() {
1571        deterministic::Runner::default().start(|context| async move {
1572            let mut db =
1573                open_db::<mmr::Family>(context.child("db"), "immutable-noop-after-commit").await;
1574
1575            let k1 = Sha256::hash(&[1]);
1576            let v1 = Sha256::fill(11u8);
1577            let k2 = Sha256::hash(&[2]);
1578            let v2 = Sha256::fill(22u8);
1579            db.apply_batch(
1580                db.new_batch()
1581                    .set(k1, v1)
1582                    .set(k2, v2)
1583                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
1584                    .await,
1585            )
1586            .unwrap();
1587            db.sync().await.unwrap();
1588            let root_after_first = db.root();
1589            let size_after_first = db.size();
1590
1591            db.sync().await.unwrap();
1592            assert_eq!(db.size(), size_after_first);
1593            assert_eq!(db.root(), root_after_first);
1594            assert_eq!(db.target().root, db.root());
1595
1596            db.destroy().await.unwrap();
1597        });
1598    }
1599
1600    #[test_traced("INFO")]
1601    fn test_compact_noop_commit_after_reopen() {
1602        deterministic::Runner::default().start(|context| async move {
1603            let partition = "immutable-noop-after-reopen";
1604
1605            let (root_before_drop, size_before_drop) = {
1606                let mut db = open_db::<mmr::Family>(context.child("first"), partition).await;
1607                let k1 = Sha256::hash(&[1]);
1608                let v1 = Sha256::fill(11u8);
1609                let k2 = Sha256::hash(&[2]);
1610                let v2 = Sha256::fill(22u8);
1611                db.apply_batch(
1612                    db.new_batch()
1613                        .set(k1, v1)
1614                        .set(k2, v2)
1615                        .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
1616                        .await,
1617                )
1618                .unwrap();
1619                db.sync().await.unwrap();
1620                (db.root(), db.size())
1621            };
1622
1623            let mut db = open_db::<mmr::Family>(context.child("second"), partition).await;
1624            assert_eq!(db.root(), root_before_drop);
1625            assert_eq!(db.size(), size_before_drop);
1626
1627            db.sync().await.unwrap();
1628            assert_eq!(db.size(), size_before_drop);
1629            assert_eq!(db.root(), root_before_drop);
1630            assert_eq!(db.target().root, db.root());
1631
1632            db.destroy().await.unwrap();
1633        });
1634    }
1635
1636    #[test_traced("INFO")]
1637    fn test_compact_noop_commit_after_rewind() {
1638        deterministic::Runner::default().start(|context| async move {
1639            let mut db =
1640                open_db::<mmr::Family>(context.child("db"), "immutable-noop-after-rewind").await;
1641
1642            let k1 = Sha256::hash(&[1]);
1643            let v1 = Sha256::fill(11u8);
1644            let k2 = Sha256::hash(&[2]);
1645            let v2 = Sha256::fill(22u8);
1646            db.apply_batch(
1647                db.new_batch()
1648                    .set(k1, v1)
1649                    .set(k2, v2)
1650                    .merkleize(&db, Some(Sha256::fill(0xaa)), Location::new(0))
1651                    .await,
1652            )
1653            .unwrap();
1654            db.sync().await.unwrap();
1655            let root_after_first = db.root();
1656            let size_after_first = db.size();
1657
1658            let k3 = Sha256::hash(&[3]);
1659            let v3 = Sha256::fill(33u8);
1660            db.apply_batch(
1661                db.new_batch()
1662                    .set(k3, v3)
1663                    .merkleize(&db, Some(Sha256::fill(0xbb)), Location::new(1))
1664                    .await,
1665            )
1666            .unwrap();
1667            db.sync().await.unwrap();
1668
1669            db.rewind(size_after_first).await.unwrap();
1670            assert_eq!(db.size(), size_after_first);
1671            assert_eq!(db.root(), root_after_first);
1672
1673            db.sync().await.unwrap();
1674            assert_eq!(db.size(), size_after_first);
1675            assert_eq!(db.root(), root_after_first);
1676            assert_eq!(db.target().root, db.root());
1677
1678            db.destroy().await.unwrap();
1679        });
1680    }
1681
1682    #[test_traced("INFO")]
1683    fn test_compact_rewind_makes_post_advance_batch_stale() {
1684        deterministic::Runner::default().start(|context| async move {
1685            let mut db =
1686                open_db::<mmr::Family>(context.child("db"), "immutable-rewind-makes-stale").await;
1687
1688            db.apply_batch(
1689                db.new_batch()
1690                    .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1691                    .merkleize(&db, None, Location::new(0))
1692                    .await,
1693            )
1694            .unwrap();
1695            db.sync().await.unwrap();
1696            let size_after_first = db.size();
1697
1698            db.apply_batch(
1699                db.new_batch()
1700                    .set(Sha256::hash(&[2]), Sha256::fill(2u8))
1701                    .merkleize(&db, None, Location::new(0))
1702                    .await,
1703            )
1704            .unwrap();
1705            db.sync().await.unwrap();
1706
1707            // Merkleize a batch against the post-commit-B state, which the rewind will discard.
1708            let held = db
1709                .new_batch()
1710                .set(Sha256::hash(&[3]), Sha256::fill(3u8))
1711                .merkleize(&db, None, Location::new(0))
1712                .await;
1713
1714            db.rewind(size_after_first).await.unwrap();
1715
1716            // After rewind, mem.size reflects post-commit-A, but the held batch starts after
1717            // post-commit-B. Apply must be rejected with StaleBatch.
1718            assert!(matches!(
1719                db.apply_batch(held),
1720                Err(Error::StaleBatch { .. })
1721            ));
1722
1723            db.destroy().await.unwrap();
1724        });
1725    }
1726
1727    #[test_traced("INFO")]
1728    fn test_compact_floor_beyond_size() {
1729        deterministic::Runner::default().start(|context| async move {
1730            let mut db =
1731                open_db::<mmr::Family>(context.child("db"), "immutable-floor-beyond").await;
1732
1733            let batch = db.new_batch().merkleize(&db, None, Location::new(2)).await;
1734
1735            assert!(matches!(
1736                db.apply_batch(batch),
1737                Err(Error::FloorBeyondSize(floor, tip))
1738                    if floor == Location::new(2) && tip == Location::new(1)
1739            ));
1740
1741            db.destroy().await.unwrap();
1742        });
1743    }
1744
1745    // A chained batch whose ancestor's floor exceeds that ancestor's own commit location
1746    // must be rejected, identifying the ancestor's bound rather than the tip's.
1747    #[test_traced("INFO")]
1748    fn test_compact_ancestor_floor_beyond_size() {
1749        deterministic::Runner::default().start(|context| async move {
1750            let mut db =
1751                open_db::<mmr::Family>(context.child("db"), "immutable-ancestor-floor-beyond")
1752                    .await;
1753
1754            // parent: set + commit at loc 2, floor=3 (one past parent's commit).
1755            let parent = db
1756                .new_batch()
1757                .set(Sha256::hash(&[1]), Sha256::fill(1u8))
1758                .merkleize(&db, None, Location::new(3))
1759                .await;
1760            // child: valid on its own (floor=0), but parent's floor is bad.
1761            let child = parent
1762                .new_batch::<Sha256>()
1763                .set(Sha256::hash(&[2]), Sha256::fill(2u8))
1764                .merkleize(&db, None, Location::new(0))
1765                .await;
1766
1767            assert!(matches!(
1768                db.apply_batch(child),
1769                Err(Error::FloorBeyondSize(floor, commit))
1770                    if floor == Location::new(3) && commit == Location::new(2)
1771            ));
1772
1773            db.destroy().await.unwrap();
1774        });
1775    }
1776}