Skip to main content

commonware_storage/qmdb/immutable/
mod.rs

1//! An authenticated database that only supports adding new keyed values (no updates or
2//! deletions).
3//!
4//! Two variants are available based on value encoding:
5//! - [fixed]: For fixed-size values.
6//! - [variable]: For variable-size values.
7//!
8//! # Inactivity floor
9//!
10//! Each commit carries an inactivity floor: a location before which the application
11//! declares operations are no longer needed. The floor is embedded in the operation
12//! log and included in the Merkle root, so all replicas processing the same operations
13//! arrive at the same floor.
14//!
15//! The floor controls two things:
16//! - **Pruning**: [`Immutable::prune`] only allows pruning up to the floor.
17//! - **Reconstruction**: on restart or sync, the snapshot is rebuilt from the floor
18//!   onward. Keys set before the floor are not loaded into memory.
19//!
20//! The floor must be monotonically non-decreasing across commits and must not exceed
21//! the batch's total operation count. Pass `db.inactivity_floor_loc()` to keep the
22//! floor unchanged, or a higher value to advance it.
23//!
24//! # Examples
25//!
26//! ```ignore
27//! // Simple mode: apply a batch, then durably commit it.
28//! // The third argument to merkleize is the inactivity floor -- operations
29//! // before this location are declared inactive by the application.
30//! let floor = db.inactivity_floor_loc();
31//! let merkleized = db.new_batch()
32//!     .set(key, value)
33//!     .merkleize(&db, None, floor).await;
34//! let (db, _) = db.apply_batch(merkleized).await?;
35//! let db = db.commit().await?;
36//! ```
37//!
38//! ```ignore
39//! // Batches can still fork before you apply them.
40//! let floor = db.inactivity_floor_loc();
41//! let parent = db.new_batch()
42//!     .set(key_a, value_a)
43//!     .merkleize(&db, None, floor).await;
44//!
45//! let child_a = parent.new_batch::<Sha256>()
46//!     .set(key_b, value_b)
47//!     .merkleize(&db, None, floor).await;
48//!
49//! let child_b = parent.new_batch::<Sha256>()
50//!     .set(key_c, value_c)
51//!     .merkleize(&db, None, floor).await;
52//!
53//! let (db, _) = db.apply_batch(child_a).await?;
54//! let db = db.commit().await?;
55//! ```
56//!
57//! ```ignore
58//! // Apply a parent batch, commit, then build a child batch from the newly
59//! // published state and apply it.
60//! let floor = db.inactivity_floor_loc();
61//! let parent = db.new_batch()
62//!     .set(key_a, value_a)
63//!     .merkleize(&db, None, floor).await;
64//! let (db, _) = db.apply_batch(parent).await?;
65//! let db = db.commit().await?;
66//!
67//! let child = db.new_batch()
68//!     .set(key_b, value_b)
69//!     .merkleize(&db, None, floor).await;
70//! let (db, _) = db.apply_batch(child).await?;
71//! let db = db.commit().await?;
72//! ```
73
74use crate::{
75    Context,
76    index::{Unordered as _, unordered::Index},
77    journal::{
78        authenticated,
79        contiguous::{Contiguous, Mutable},
80    },
81    merkle::{Family, Location, Proof, full::Config as MerkleConfig},
82    qmdb::{
83        Error, any::ValueEncoding, batch_chain, metrics::Metrics, operation::Key,
84        single_operation_root,
85    },
86    translator::Translator,
87};
88use ahash::AHashSet;
89use commonware_codec::EncodeShared;
90use commonware_cryptography::Hasher;
91use commonware_macros::boxed;
92use commonware_parallel::Strategy;
93use commonware_runtime::{Handle, ReadOptions};
94use core::num::{NonZeroU64, NonZeroUsize};
95use futures::{StreamExt, pin_mut};
96use std::{ops::Range, sync::Arc};
97use tracing::warn;
98
99pub mod batch;
100mod compact;
101pub mod fixed;
102mod operation;
103pub mod sync;
104pub mod variable;
105
106pub use compact::{
107    Config as CompactConfig, Db as CompactDb, MerkleizedBatch as CompactMerkleizedBatch,
108    UnmerkleizedBatch as CompactUnmerkleizedBatch,
109};
110pub use operation::Operation;
111
112/// Build the snapshot by replaying the log from `inactivity_floor_loc`, inserting the location of
113/// every retained [Operation::Set] and keeping prior locations of the same key. Assumes the log
114/// is not pruned beyond the inactivity floor.
115///
116/// Repeats of a full key all land in the snapshot, matching a snapshot maintained live, so reads
117/// of a repeated key keep returning one of its written values however the snapshot was built.
118///
119/// `init_buffer` sizes the replay read buffer (in bytes).
120async fn build_snapshot<F, K, V, C, T>(
121    inactivity_floor_loc: Location<F>,
122    log: &C,
123    snapshot: &mut Index<T, Location<F>>,
124    init_buffer: NonZeroUsize,
125) -> Result<(), Error<F>>
126where
127    F: Family,
128    K: Key,
129    V: ValueEncoding,
130    C: Contiguous<Item = Operation<F, K, V>>,
131    T: Translator,
132{
133    let stream = log
134        .replay(*inactivity_floor_loc, init_buffer, ReadOptions::default())
135        .await?;
136    pin_mut!(stream);
137    while let Some(result) = stream.next().await {
138        let (loc, op) = result?;
139        if let Operation::Set(key, _) = op {
140            snapshot.insert(&key, Location::new(loc));
141        }
142    }
143    Ok(())
144}
145
146/// Compute the authenticated root of a newly initialized database without opening storage.
147///
148/// The initial commit never carries metadata, so this root always represents `Commit(None, 0)`.
149pub fn initial_root<F, K, V, H>() -> H::Digest
150where
151    F: Family,
152    K: Key,
153    V: ValueEncoding,
154    H: Hasher,
155    Operation<F, K, V>: EncodeShared,
156{
157    single_operation_root::<F, H>(&Operation::<F, K, V>::Commit(None, Location::new(0)))
158}
159
160/// Configuration for an [Immutable] authenticated db.
161#[derive(Clone)]
162pub struct Config<T: Translator, J, S: Strategy> {
163    /// Configuration for the Merkle structure backing the authenticated journal.
164    pub merkle_config: MerkleConfig<S>,
165
166    /// Configuration for the operations log journal.
167    pub log: J,
168
169    /// The translator used by the compressed index.
170    pub translator: T,
171
172    /// Size (in bytes) of the read buffer used to replay the log during init.
173    pub init_buffer: NonZeroUsize,
174}
175
176/// An authenticated database that only supports adding new keyed values (no updates or
177/// deletions).
178///
179/// # Invariant
180///
181/// A key must be set at most once across the database history. If a key is set more than once,
182/// reads of that key may return any of its written values.
183///
184/// Use [fixed::Db] or [variable::Db] for concrete instantiations.
185pub struct Immutable<
186    F: Family,
187    E: Context,
188    K: Key,
189    V: ValueEncoding,
190    C: Mutable<Item = Operation<F, K, V>>,
191    H: Hasher,
192    T: Translator,
193    S: Strategy,
194> where
195    C::Item: EncodeShared,
196{
197    /// Authenticated journal of operations.
198    journal: authenticated::Journal<F, E, C, H, S>,
199
200    /// Cached canonical operations root.
201    pub(crate) root: H::Digest,
202
203    /// A map from each active key to the location of the operation that set its value.
204    ///
205    /// # Invariant
206    ///
207    /// Only references operations of type [Operation::Set].
208    pub(crate) snapshot: Index<T, Location<F>>,
209
210    /// The location of the last commit operation.
211    pub(crate) last_commit_loc: Location<F>,
212
213    /// The inactivity floor declared by the last committed batch.
214    /// Operations before this location are considered inactive by the application.
215    pub(crate) inactivity_floor_loc: Location<F>,
216
217    /// Metrics for this database.
218    metrics: Metrics<E>,
219}
220
221impl<F, E, K, V, C, H, T, S> std::fmt::Debug for Immutable<F, E, K, V, C, H, T, S>
222where
223    F: Family,
224    E: Context,
225    K: Key,
226    V: ValueEncoding,
227    C: Mutable<Item = Operation<F, K, V>>,
228    C::Item: EncodeShared,
229    H: Hasher,
230    T: Translator,
231    S: Strategy,
232{
233    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
234        f.debug_struct("Immutable")
235            .field("bounds", &self.bounds())
236            .field("inactivity_floor_loc", &self.inactivity_floor_loc())
237            .finish_non_exhaustive()
238    }
239}
240
241// Shared read-only functionality.
242impl<F, E, K, V, C, H, T, S> Immutable<F, E, K, V, C, H, T, S>
243where
244    F: Family,
245    E: Context,
246    K: Key,
247    V: ValueEncoding,
248    C: Mutable<Item = Operation<F, K, V>>,
249    C::Item: EncodeShared,
250    H: Hasher,
251    T: Translator,
252    S: Strategy,
253{
254    /// Initialize from a pre-constructed authenticated journal.
255    ///
256    /// Seeds an initial commit if the journal is empty, builds the in-memory snapshot,
257    /// and returns the initialized database.
258    #[boxed]
259    pub(crate) async fn init_from_journal(
260        mut journal: authenticated::Journal<F, E, C, H, S>,
261        context: E,
262        translator: T,
263        init_buffer: NonZeroUsize,
264    ) -> Result<Self, Error<F>> {
265        if journal.size() == 0 {
266            warn!("Authenticated log is empty, initialized new db.");
267            (journal, _) = journal
268                .append(&Operation::Commit(None, Location::new(0)))
269                .await?;
270            journal = journal.sync().await?;
271        }
272
273        let mut snapshot = Index::new(context.child("snapshot"), translator);
274
275        let (last_commit_loc, inactivity_floor_loc) = {
276            let bounds = journal.journal.bounds();
277            let last_commit_loc =
278                Location::new(bounds.end.checked_sub(1).expect("commit should exist"));
279
280            // Read the floor from the last commit operation.
281            let last_op = journal.journal.read(*last_commit_loc).await?;
282            let inactivity_floor_loc = last_op
283                .has_floor()
284                .expect("last operation should be a commit with floor");
285            if inactivity_floor_loc > last_commit_loc {
286                return Err(Error::DataCorrupted("inactivity floor exceeds last commit"));
287            }
288
289            // Replay the log from the inactivity floor to build the snapshot. Every retained
290            // location is inserted, mirroring the live apply path, so a repeated key keeps
291            // serving one of its written values across restarts and rewinds.
292            build_snapshot(
293                inactivity_floor_loc,
294                &journal.journal,
295                &mut snapshot,
296                init_buffer,
297            )
298            .await?;
299
300            (last_commit_loc, inactivity_floor_loc)
301        };
302        let inactive_peaks = F::inactive_peaks(last_commit_loc + 1, inactivity_floor_loc);
303        let root = journal.root(inactive_peaks)?;
304
305        let metrics = Metrics::new(context);
306        let db = Self {
307            journal,
308            root,
309            snapshot,
310            last_commit_loc,
311            inactivity_floor_loc,
312            metrics,
313        };
314        db.update_metrics();
315        Ok(db)
316    }
317
318    /// Return the inactivity floor location declared by the last committed batch.
319    pub const fn inactivity_floor_loc(&self) -> Location<F> {
320        self.inactivity_floor_loc
321    }
322
323    /// Return the Location of the next operation appended to this db.
324    pub fn size(&self) -> Location<F> {
325        self.bounds().end
326    }
327
328    /// Return [start, end) where `start` and `end - 1` are the Locations of the oldest and newest
329    /// retained operations respectively.
330    pub fn bounds(&self) -> Range<Location<F>> {
331        Location::new(self.journal.bounds().start)..Location::new(self.journal.bounds().end)
332    }
333
334    /// Update state gauges from the current database state.
335    fn update_metrics(&self) {
336        let bounds = self.journal.bounds();
337        self.metrics.update(
338            bounds.end,
339            bounds.start,
340            *self.inactivity_floor_loc,
341            *self.last_commit_loc,
342        );
343    }
344
345    /// Return the most recent location from which this database can safely be synced, and the
346    /// upper bound on [`Self::prune`]'s `loc`. For immutable databases, this equals the
347    /// inactivity floor declared by the last committed batch.
348    pub const fn sync_boundary(&self) -> Location<F> {
349        self.inactivity_floor_loc
350    }
351
352    /// Get the value of `key` in the db, or None if it has no value or its corresponding operation
353    /// has been pruned.
354    pub async fn get(&self, key: &K) -> Result<Option<V::Value>, Error<F>> {
355        let _timer = self.metrics.get_timer();
356        self.metrics.get_calls.inc();
357        self.metrics.lookups_requested.inc();
358        let iter = self.snapshot.get(key);
359        let oldest = self.journal.bounds().start;
360        let mut result = None;
361        for &loc in iter {
362            if loc < oldest {
363                continue;
364            }
365            if let Some(v) = Self::get_from_loc(&self.journal, key, loc).await? {
366                result = Some(v);
367                break;
368            }
369        }
370
371        Ok(result)
372    }
373
374    /// Batch read multiple keys.
375    ///
376    /// Returns results in the same order as the input keys.
377    pub async fn get_many(&self, keys: &[&K]) -> Result<Vec<Option<V::Value>>, Error<F>> {
378        if keys.is_empty() {
379            return Ok(Vec::new());
380        }
381
382        let _timer = self.metrics.get_many_timer();
383        self.metrics.get_many_calls.inc();
384        self.metrics.lookups_requested.inc_by(keys.len() as u64);
385        let mut candidates: Vec<(usize, u64)> = Vec::with_capacity(keys.len());
386        let mut results: Vec<Option<V::Value>> = vec![None; keys.len()];
387
388        let oldest = self.journal.bounds().start;
389
390        for (key_idx, key) in keys.iter().enumerate() {
391            for &loc in self.snapshot.get(key) {
392                if loc < oldest {
393                    continue;
394                }
395                candidates.push((key_idx, *loc));
396            }
397        }
398
399        if candidates.is_empty() {
400            return Ok(results);
401        }
402
403        candidates.sort_unstable_by_key(|&(_, pos)| pos);
404
405        let mut positions: Vec<u64> = Vec::with_capacity(candidates.len());
406        for &(_, pos) in &candidates {
407            if positions.last() != Some(&pos) {
408                positions.push(pos);
409            }
410        }
411
412        let ops = self.journal.read_many(&positions).await?;
413
414        for &(key_idx, pos) in &candidates {
415            if results[key_idx].is_some() {
416                continue;
417            }
418            let op_idx = positions
419                .binary_search(&pos)
420                .expect("position was deduped from candidates");
421            let Operation::Set(k, v) = &ops[op_idx] else {
422                return Err(Error::UnexpectedData(Location::new(pos)));
423            };
424            if k == keys[key_idx] {
425                results[key_idx] = Some(v.clone());
426            }
427        }
428
429        Ok(results)
430    }
431
432    /// Get the value of the operation with location `loc` in the db if it matches `key`. Returns
433    /// [`crate::qmdb::Error::OperationPruned`] if loc precedes the oldest retained location. The
434    /// location is otherwise assumed valid.
435    async fn get_from_loc(
436        reader: &impl Contiguous<Item = Operation<F, K, V>>,
437        key: &K,
438        loc: Location<F>,
439    ) -> Result<Option<V::Value>, Error<F>> {
440        if loc < reader.bounds().start {
441            return Err(Error::OperationPruned(loc));
442        }
443
444        let Operation::Set(k, v) = reader.read(*loc).await? else {
445            return Err(Error::UnexpectedData(loc));
446        };
447
448        if k != *key { Ok(None) } else { Ok(Some(v)) }
449    }
450
451    /// Get the metadata associated with the last commit.
452    pub async fn get_metadata(&self) -> Result<Option<V::Value>, Error<F>> {
453        let last_commit_loc = self.last_commit_loc;
454        let Operation::Commit(metadata, _floor) =
455            self.journal.journal.read(*last_commit_loc).await?
456        else {
457            unreachable!("no commit operation at location of last commit {last_commit_loc}");
458        };
459
460        Ok(metadata)
461    }
462
463    /// Analogous to proof but with respect to the state of the database when it had `op_count`
464    /// operations.
465    ///
466    /// # Contract
467    ///
468    /// `op_count` must be a commit-boundary size: the operation at `op_count - 1` must
469    /// itself be a commit op. Non-commit-boundary sizes are not supported because the
470    /// inactivity floor governing them is not directly retrievable.
471    ///
472    /// # Errors
473    ///
474    /// Returns [crate::merkle::Error::LocationOverflow] if `op_count` or `start_loc` >
475    /// [crate::merkle::Family::MAX_LEAVES].
476    /// Returns [crate::merkle::Error::RangeOutOfBounds] if `op_count` > number of operations, or
477    /// if `start_loc` >= `op_count`.
478    /// Returns [`Error::OperationPruned`] if `start_loc` has been pruned.
479    /// Returns [`Error::HistoricalFloorPruned`] if `op_count - 1` is retained but is not a
480    /// commit op, either because the caller passed a non-commit-boundary `op_count` or
481    /// because pruning removed the commit that would have governed `op_count`.
482    #[allow(clippy::type_complexity)]
483    #[tracing::instrument(
484        name = "qmdb.immutable.db.historical_proof",
485        level = "info",
486        skip_all,
487        fields(
488            op_count = *op_count,
489            start_loc = *start_loc,
490            max_ops = max_ops.get(),
491        ),
492    )]
493    pub async fn historical_proof(
494        &self,
495        op_count: Location<F>,
496        start_loc: Location<F>,
497        max_ops: NonZeroU64,
498    ) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, K, V>>), Error<F>> {
499        if op_count > self.journal.size() {
500            return Err(crate::merkle::Error::RangeOutOfBounds(op_count).into());
501        }
502
503        let inactive_peaks =
504            crate::qmdb::inactive_peaks_at::<F, _>(&self.journal, op_count).await?;
505
506        Ok(self
507            .journal
508            .historical_proof(op_count, start_loc, max_ops, inactive_peaks)
509            .await?)
510    }
511
512    /// Generate and return:
513    ///  1. a proof of all operations applied to the db in the range starting at (and including)
514    ///     location `start_loc`, and ending at the first of either:
515    ///     - the last operation performed, or
516    ///     - the operation `max_ops` from the start.
517    ///  2. the operations corresponding to the leaves in this range.
518    pub async fn proof(
519        &self,
520        start_index: Location<F>,
521        max_ops: NonZeroU64,
522    ) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, K, V>>), Error<F>> {
523        let op_count = self.bounds().end;
524        self.historical_proof(op_count, start_index, max_ops).await
525    }
526
527    /// Prune operations prior to `prune_loc`. This does not affect the db's root, but it will
528    /// affect retrieval of any keys that were set prior to `prune_loc`.
529    ///
530    /// Pruning is irreversible and requires no prior commit. After a crash, the database remains
531    /// recoverable; uncommitted operations are not guaranteed to survive.
532    ///
533    /// # Errors
534    ///
535    /// - Returns [Error::PruneBeyondMinRequired] if `prune_loc` > inactivity floor.
536    /// - Returns [crate::merkle::Error::LocationOverflow] if `prune_loc` > [crate::merkle::Family::MAX_LEAVES].
537    #[tracing::instrument(name = "qmdb.immutable.db.prune", level = "info", skip_all)]
538    #[boxed]
539    pub async fn prune(mut self, loc: Location<F>) -> Result<Self, Error<F>> {
540        let _timer = self.metrics.prune_timer();
541        self.metrics.prune_calls.inc();
542        if loc > self.inactivity_floor_loc {
543            return Err(Error::PruneBeyondMinRequired(
544                loc,
545                self.inactivity_floor_loc,
546            ));
547        }
548        (self.journal, _) = self.journal.prune(loc).await?;
549        self.update_metrics();
550        Ok(self)
551    }
552
553    /// Rewind the database to `size` operations, where `size` is the location of the next append.
554    ///
555    /// This rewinds both the operations journal and its Merkle structure to the historical
556    /// state at `size`, and removes rewound set operations from the in-memory snapshot.
557    ///
558    /// # Errors
559    ///
560    /// Returns an error when:
561    /// - `size` is not a valid rewind target
562    /// - the target's required logical range is not fully retained (for immutable, this means the
563    ///   oldest retained location is already beyond the rewind boundary)
564    /// - `size - 1` is not a commit operation
565    ///
566    /// Any error from this method is fatal for this handle. Rewind may mutate journal state
567    /// before this method finishes rebuilding in-memory rewind state. Callers must drop this
568    /// database handle after any `Err` from `rewind` and reopen from storage.
569    ///
570    /// A successful rewind is not restart-stable until a subsequent [`Immutable::commit`] or
571    /// [`Immutable::sync`] completes, or until the handle returned by a subsequent
572    /// [`Immutable::start_sync`] completes.
573    #[tracing::instrument(name = "qmdb.immutable.db.rewind", level = "info", skip_all)]
574    #[boxed]
575    pub async fn rewind(mut self, size: Location<F>) -> Result<Self, Error<F>> {
576        let rewind_size = *size;
577        let current_size = *self.last_commit_loc + 1;
578        if rewind_size == current_size {
579            return Ok(self);
580        }
581        if rewind_size == 0 || rewind_size > current_size {
582            return Err(Error::Journal(crate::journal::Error::InvalidRewind(
583                rewind_size,
584            )));
585        }
586
587        let (rewind_last_loc, rewind_floor, rewound_keys) = {
588            let bounds = self.journal.bounds();
589            let rewind_last_loc = Location::new(rewind_size - 1);
590            if rewind_size <= bounds.start {
591                return Err(Error::Journal(crate::journal::Error::ItemPruned(
592                    *rewind_last_loc,
593                )));
594            }
595            let rewind_last_op = self.journal.read(*rewind_last_loc).await?;
596            let Operation::Commit(_, rewind_floor) = &rewind_last_op else {
597                return Err(Error::UnexpectedData(rewind_last_loc));
598            };
599            let rewind_floor = *rewind_floor;
600            if *rewind_floor < bounds.start {
601                return Err(Error::Journal(crate::journal::Error::ItemPruned(
602                    *rewind_floor,
603                )));
604            }
605
606            let mut rewound_keys = Vec::new();
607            for loc in rewind_size..current_size {
608                if let Operation::Set(key, _) = self.journal.read(loc).await? {
609                    rewound_keys.push(key);
610                }
611            }
612
613            (rewind_last_loc, rewind_floor, rewound_keys)
614        };
615
616        let old_floor = self.inactivity_floor_loc;
617
618        // Journal rewind happens before in-memory snapshot updates. If a later step fails, this
619        // handle may be internally diverged and must be dropped by the caller.
620        self.journal = self.journal.rewind(rewind_size).await?;
621
622        // Remove keys that were set in the range [rewind_size, current_size) from the snapshot.
623        let rewind_loc = Location::<F>::new(rewind_size);
624        for key in &rewound_keys {
625            // Filter by location to make sure we don't also prune keys that happen to collide.
626            self.snapshot.retain(key, |loc| *loc < rewind_loc);
627        }
628
629        // If the rewind target has a lower floor than the current snapshot was
630        // built from, insert keys from the gap [rewind_floor, old_floor) that
631        // were excluded by the higher-floor reconstruction. A key written more
632        // than once may end up with multiple snapshot entries, and reads of it
633        // may return any of its written values.
634        if rewind_floor < old_floor {
635            let gap_end = core::cmp::min(*old_floor, rewind_size);
636            for loc in *rewind_floor..gap_end {
637                if let Operation::Set(key, _) = self.journal.journal.read(loc).await? {
638                    self.snapshot.insert(&key, Location::new(loc));
639                }
640            }
641        }
642
643        self.last_commit_loc = rewind_last_loc;
644        self.inactivity_floor_loc = rewind_floor;
645        let inactive_peaks = F::inactive_peaks(size, rewind_floor);
646        self.root = self.journal.root(inactive_peaks)?;
647        self.update_metrics();
648
649        Ok(self)
650    }
651
652    /// Return the canonical QMDB root of the db.
653    pub const fn root(&self) -> H::Digest {
654        self.root
655    }
656
657    /// Return a reference to the merkleization strategy.
658    pub const fn strategy(&self) -> &S {
659        self.journal.strategy()
660    }
661
662    /// Return the pinned Merkle nodes at the given location.
663    pub async fn pinned_nodes_at(&self, loc: Location<F>) -> Result<Vec<H::Digest>, Error<F>> {
664        self.journal
665            .merkle
666            .pinned_nodes_at(loc)
667            .await
668            .map_err(Into::into)
669    }
670
671    /// Sync all database state to disk. While this isn't necessary to ensure durability of
672    /// committed operations, periodic invocation may reduce memory usage and the time required to
673    /// recover the database on restart.
674    #[tracing::instrument(name = "qmdb.immutable.db.sync", level = "info", skip_all)]
675    pub async fn sync(mut self) -> Result<Self, Error<F>> {
676        let _timer = self.metrics.sync_timer();
677        self.metrics.sync_calls.inc();
678        self.journal = self.journal.sync().await?;
679        Ok(self)
680    }
681
682    /// Begin durably persisting the journal state published by prior [`Immutable::apply_batch`]
683    /// calls.
684    ///
685    /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit],
686    /// plus a best-effort attempt to bound the recovery needed on startup. Use [Self::sync] to
687    /// guarantee none is needed. A new sync waits for the prior sync before starting. Failures
688    /// of the deferred durability work surface on the returned handle. A failed data sync also
689    /// fails the next durability operation. A failed recovery-watermark sync is not observed by
690    /// [Self::commit], and a failed merkle-node sync may not be. Both resurface on the next
691    /// [Self::sync].
692    #[tracing::instrument(name = "qmdb.immutable.db.start_sync", level = "info", skip_all)]
693    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
694        self.metrics.start_sync_calls.inc();
695        let handle;
696        (self.journal, handle) = self.journal.start_sync().await?;
697        Ok((self, handle))
698    }
699
700    /// Durably commit the journal state published by prior [`Immutable::apply_batch`] calls.
701    #[tracing::instrument(name = "qmdb.immutable.db.commit", level = "info", skip_all)]
702    pub async fn commit(mut self) -> Result<Self, Error<F>> {
703        let _timer = self.metrics.commit_timer();
704        self.metrics.commit_calls.inc();
705        self.journal = self.journal.commit().await?;
706        Ok(self)
707    }
708
709    /// Destroy the db, removing all data from disk.
710    #[boxed]
711    pub async fn destroy(self) -> Result<(), Error<F>> {
712        Ok(self.journal.destroy().await?)
713    }
714
715    /// The [`Commitment`](batch_chain::Commitment) for the database's current state.
716    pub(crate) fn commitment(&self) -> batch_chain::Commitment<F, H::Digest> {
717        batch_chain::Commitment::new(self.last_commit_loc + 1, self.root)
718    }
719
720    /// Create a new speculative batch of operations with this database as its parent.
721    #[allow(clippy::type_complexity)]
722    pub fn new_batch(&self) -> batch::UnmerkleizedBatch<F, H, K, V, S> {
723        batch::UnmerkleizedBatch::new(self, self.commitment())
724    }
725
726    /// Check that `batch` can be applied to the database in its current state, without
727    /// applying it.
728    ///
729    /// [`Self::apply_batch`] runs the same validation but consumes the database when it
730    /// fails; callers that want to reject a bad batch and keep the handle can check first.
731    pub fn validate_batch(
732        &self,
733        batch: &batch::MerkleizedBatch<F, H::Digest, K, V, S>,
734    ) -> Result<(), Error<F>> {
735        batch
736            .bounds
737            .validate_apply_to(self.commitment(), self.inactivity_floor_loc)
738    }
739
740    /// Apply a [`batch::MerkleizedBatch`] to the database.
741    ///
742    /// A batch is valid only if every batch applied to the database since this batch's
743    /// ancestor chain was created is an ancestor of this batch. Applying a batch from a
744    /// different fork returns [`Error::StaleBatch`] (see [`crate::qmdb::batch_chain`] for
745    /// more details).
746    ///
747    /// # Errors
748    ///
749    /// - [`Error::StaleBatch`] if the batch is detected as stale (see
750    ///   [`crate::qmdb::batch_chain`] for more details).
751    /// - [`Error::FloorRegressed`] if any commit in the chain (the tip or any
752    ///   unapplied ancestor) declares an inactivity floor below the previous
753    ///   commit's floor (or, for the oldest unapplied commit, below the
754    ///   database's current floor).
755    /// - [`Error::FloorBeyondSize`] if any commit in the chain (the tip or any
756    ///   unapplied ancestor) declares an inactivity floor that exceeds its
757    ///   own commit operation's location. The maximum valid floor for a
758    ///   commit is its own location; a floor past the commit would permit
759    ///   pruning the commit itself.
760    ///
761    /// Floor validation happens before any journal mutation, so on floor errors the on-disk
762    /// state is unchanged and reopening recovers the database as it was.
763    ///
764    /// Returns the range of locations written.
765    ///
766    /// This publishes the batch to the in-memory database state and appends it to the journal.
767    /// Call [`Immutable::commit`] or [`Immutable::sync`], or await the handle returned by
768    /// [`Immutable::start_sync`], to make the applied state durable.
769    #[tracing::instrument(name = "qmdb.immutable.db.apply_batch", level = "info", skip_all)]
770    pub async fn apply_batch(
771        mut self,
772        batch: Arc<batch::MerkleizedBatch<F, H::Digest, K, V, S>>,
773    ) -> Result<(Self, Range<Location<F>>), Error<F>> {
774        let _timer = self.metrics.apply_batch_timer();
775        self.metrics.apply_batch_calls.inc();
776        self.validate_batch(&batch)?;
777        let db_size = self.last_commit_loc + 1;
778
779        // Apply journal.
780        self.journal = self.journal.apply_batch(&batch.journal_batch).await?;
781
782        // Apply snapshot inserts. Child first (child wins via `seen`), then
783        // uncommitted ancestor batches.
784        //
785        // `seen` is only consulted when at least one ancestor diff will be applied, so it is
786        // skipped entirely otherwise.
787        let bounds = self.journal.bounds();
788        let track_shadow = batch
789            .bounds
790            .ancestors
791            .iter()
792            .any(|a| a.state.size > db_size);
793        let seen_cap = if track_shadow {
794            batch.diff.len()
795                + batch
796                    .bounds
797                    .ancestors
798                    .iter()
799                    .zip(&batch.ancestor_diffs)
800                    .filter(|(a, _)| a.state.size > db_size)
801                    .map(|(_, d)| d.len())
802                    .sum::<usize>()
803        } else {
804            0
805        };
806        let mut seen: AHashSet<&K> = AHashSet::with_capacity(seen_cap);
807        for (key, entry) in batch.diff.iter() {
808            if track_shadow {
809                seen.insert(key);
810            }
811            self.snapshot
812                .insert_and_retain(key, entry.loc, |v| *v >= bounds.start);
813        }
814        for (i, ancestor_diff) in batch.ancestor_diffs.iter().enumerate() {
815            if batch.bounds.ancestors[i].state.size <= db_size {
816                continue;
817            }
818            for (key, entry) in ancestor_diff.iter() {
819                if seen.insert(key) {
820                    self.snapshot
821                        .insert_and_retain(key, entry.loc, |v| *v >= bounds.start);
822                }
823            }
824        }
825
826        // Update state.
827        self.last_commit_loc = batch.bounds.tip.size - 1;
828        self.inactivity_floor_loc = batch.bounds.inactivity_floor;
829        self.root = batch.root();
830        let range = db_size..batch.bounds.tip.size;
831        self.update_metrics();
832        self.metrics
833            .operations_applied
834            .inc_by(*range.end - *range.start);
835        Ok((self, range))
836    }
837}
838
839impl<F, E, K, V, C, H, T, S> crate::qmdb::sync::Source for Immutable<F, E, K, V, C, H, T, S>
840where
841    F: Family,
842    E: Context,
843    K: Key,
844    V: ValueEncoding,
845    C: Mutable<Item = Operation<F, K, V>>,
846    C::Item: EncodeShared,
847    H: Hasher,
848    T: Translator,
849    S: Strategy,
850{
851    type Family = F;
852    type Digest = H::Digest;
853    type Op = Operation<F, K, V>;
854    type Error = Error<F>;
855
856    async fn serve(
857        &self,
858        request: crate::qmdb::sync::Request<F>,
859    ) -> Result<
860        (
861            crate::qmdb::sync::Response<F, Self::Op, Self::Digest>,
862            crate::qmdb::sync::FeedbackTx,
863        ),
864        Self::Error,
865    > {
866        self.journal.serve(request).await
867    }
868}
869
870#[cfg(test)]
871pub(super) mod tests {
872    use super::*;
873    use crate::{
874        merkle::{Family, Location},
875        qmdb::{verify_proof, verify_proof_and_pinned_nodes},
876        translator::TwoCap,
877    };
878    use commonware_codec::EncodeShared;
879    use commonware_cryptography::{Sha256, sha256, sha256::Digest};
880    use commonware_runtime::{Supervisor as _, deterministic};
881    use commonware_utils::NZU64;
882    use core::{future::Future, pin::Pin};
883    use std::ops::Range;
884
885    const ITEMS_PER_SECTION: u64 = 5;
886
887    /// Emits the named test against `mmr::Family` and `mmb::Family`.
888    macro_rules! immutable_tests {
889        ($($name:ident => $scenario:ident, $open:ident;)*) => {
890            $(
891                #[test_traced]
892                fn $name() {
893                    deterministic::Runner::default().start(|ctx| async move {
894                        tests::$scenario(ctx, $open::<mmr::Family>).await;
895                    });
896                }
897            )*
898            paste::paste! {
899                $(
900                    #[test_traced]
901                    fn [<$name _mmb>]() {
902                        deterministic::Runner::default().start(|ctx| async move {
903                            tests::$scenario(ctx, $open::<mmb::Family>).await;
904                        });
905                    }
906                )*
907            }
908        };
909    }
910
911    pub(super) use immutable_tests;
912
913    type TestDb<F, V, C> = Immutable<
914        F,
915        deterministic::Context,
916        Digest,
917        V,
918        C,
919        Sha256,
920        TwoCap,
921        commonware_parallel::Sequential,
922    >;
923
924    #[boxed]
925    pub(crate) async fn run_empty<F: Family, V, C>(
926        context: deterministic::Context,
927        open_db: impl Fn(
928            deterministic::Context,
929        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
930    ) where
931        V: ValueEncoding<Value = Digest>,
932        C: Mutable<Item = Operation<F, Digest, V>>,
933        C::Item: EncodeShared,
934    {
935        let db = open_db(context.child("first")).await;
936        let bounds = db.bounds();
937        assert_eq!(bounds.end, 1);
938        assert_eq!(bounds.start, Location::new(0));
939        assert_eq!(db.inactivity_floor_loc(), Location::new(0));
940        assert!(db.get_metadata().await.unwrap().is_none());
941
942        // Make sure closing/reopening gets us back to the same state, even after adding an uncommitted op.
943        let k1 = Sha256::fill(1u8);
944        let v1 = Sha256::fill(2u8);
945        let root = db.root();
946        {
947            let _batch = db.new_batch().set(k1, v1);
948            // Don't merkleize/apply -- simulate failed commit
949        }
950        drop(db);
951        let db = open_db(context.child("second")).await;
952        assert_eq!(db.root(), root);
953        assert_eq!(db.bounds().end, 1);
954
955        // Test calling commit on an empty db which should make it (durably) non-empty.
956        let merkleized = db.new_batch().merkleize(&db, None, Location::new(0)).await;
957        let (db, _) = db.apply_batch(merkleized).await.unwrap();
958        let db = db.commit().await.unwrap();
959        assert_eq!(db.bounds().end, 2); // commit op added
960        let root = db.root();
961        drop(db);
962
963        let db = open_db(context.child("third")).await;
964        assert_eq!(db.root(), root);
965
966        db.destroy().await.unwrap();
967    }
968
969    #[boxed]
970    pub(crate) async fn run_commit_after_sync_recovery<F: Family, V, C>(
971        context: deterministic::Context,
972        open_db: impl Fn(
973            deterministic::Context,
974        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
975    ) where
976        V: ValueEncoding<Value = Digest>,
977        C: Mutable<Item = Operation<F, Digest, V>>,
978        C::Item: EncodeShared,
979    {
980        let db = open_db(context.child("first")).await;
981        let k1 = Sha256::fill(1u8);
982        let k2 = Sha256::fill(2u8);
983        let v1 = Sha256::fill(3u8);
984        let v2 = Sha256::fill(4u8);
985
986        // Commit and sync the first key so recovery has an older persisted boundary.
987        let (db, _) = commit_sets(db, [(k1, v1)], None).await;
988        let db = db.sync().await.unwrap();
989
990        // Commit a second key without syncing; reopen must replay it from journal data.
991        let (db, _) = commit_sets(db, [(k2, v2)], None).await;
992        let committed_bounds = db.bounds();
993        let committed_root = db.root();
994        drop(db);
995
996        let db = open_db(context.child("second")).await;
997        assert_eq!(db.bounds(), committed_bounds);
998        assert_eq!(db.root(), committed_root);
999        assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
1000        assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
1001
1002        db.destroy().await.unwrap();
1003    }
1004
1005    #[boxed]
1006    pub(crate) async fn run_build_basic<F: Family, V, C>(
1007        context: deterministic::Context,
1008        open_db: impl Fn(
1009            deterministic::Context,
1010        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1011    ) where
1012        V: ValueEncoding<Value = Digest>,
1013        C: Mutable<Item = Operation<F, Digest, V>>,
1014        C::Item: EncodeShared,
1015    {
1016        // Build a db with 2 keys.
1017        let db = open_db(context.child("first")).await;
1018
1019        let k1 = Sha256::fill(1u8);
1020        let k2 = Sha256::fill(2u8);
1021        let v1 = Sha256::fill(3u8);
1022        let v2 = Sha256::fill(4u8);
1023
1024        assert!(db.get(&k1).await.unwrap().is_none());
1025        assert!(db.get(&k2).await.unwrap().is_none());
1026
1027        // Set and commit the first key.
1028        let metadata = Some(Sha256::fill(99u8));
1029        let merkleized = db
1030            .new_batch()
1031            .set(k1, v1)
1032            .merkleize(&db, metadata, Location::new(0))
1033            .await;
1034        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1035        let db = db.commit().await.unwrap();
1036        assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
1037        assert!(db.get(&k2).await.unwrap().is_none());
1038        assert_eq!(db.bounds().end, 3);
1039        assert_eq!(db.get_metadata().await.unwrap(), Some(Sha256::fill(99u8)));
1040
1041        // Set and commit the second key.
1042        let merkleized = db
1043            .new_batch()
1044            .set(k2, v2)
1045            .merkleize(&db, None, Location::new(0))
1046            .await;
1047        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1048        let db = db.commit().await.unwrap();
1049        assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
1050        assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);
1051        assert_eq!(db.bounds().end, 5);
1052        assert_eq!(db.get_metadata().await.unwrap(), None);
1053
1054        // Capture state.
1055        let root = db.root();
1056
1057        // Add an uncommitted op then simulate failure.
1058        let k3 = Sha256::fill(5u8);
1059        let v3 = Sha256::fill(6u8);
1060        {
1061            let _batch = db.new_batch().set(k3, v3);
1062            // Don't merkleize/apply -- simulate failed commit
1063        }
1064
1065        // Reopen, make sure state is restored to last commit point.
1066        drop(db); // Simulate failed commit
1067        let db = open_db(context.child("second")).await;
1068        assert!(db.get(&k3).await.unwrap().is_none());
1069        assert_eq!(db.root(), root);
1070        assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
1071        assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);
1072        assert_eq!(db.bounds().end, 5);
1073        assert_eq!(db.get_metadata().await.unwrap(), None);
1074
1075        // Cleanup.
1076        db.destroy().await.unwrap();
1077    }
1078
1079    #[boxed]
1080    pub(crate) async fn run_proof_verify<F: Family, V, C>(
1081        context: deterministic::Context,
1082        open_db: impl Fn(
1083            deterministic::Context,
1084        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1085    ) where
1086        V: ValueEncoding<Value = Digest>,
1087        C: Mutable<Item = Operation<F, Digest, V>>,
1088        C::Item: EncodeShared,
1089    {
1090        let db = open_db(context.child("first")).await;
1091
1092        let k1 = Sha256::fill(1u8);
1093        let v1 = Sha256::fill(10u8);
1094        let merkleized = db
1095            .new_batch()
1096            .set(k1, v1)
1097            .merkleize(&db, None, Location::new(0))
1098            .await;
1099        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1100        let db = db.commit().await.unwrap();
1101
1102        let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap();
1103        let root = db.root();
1104        assert!(verify_proof::<Sha256, _, _>(
1105            &proof,
1106            Location::new(0),
1107            &ops,
1108            &root
1109        ));
1110
1111        db.destroy().await.unwrap();
1112    }
1113
1114    #[boxed]
1115    pub(crate) async fn run_prune<F: Family, V, C>(
1116        context: deterministic::Context,
1117        open_db: impl Fn(
1118            deterministic::Context,
1119        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1120    ) where
1121        V: ValueEncoding<Value = Digest>,
1122        C: Mutable<Item = Operation<F, Digest, V>>,
1123        C::Item: EncodeShared,
1124    {
1125        let mut db = open_db(context.child("first")).await;
1126
1127        for i in 0..20u8 {
1128            let key = Sha256::fill(i);
1129            let value = Sha256::fill(i.wrapping_add(100));
1130            let floor = db.bounds().end;
1131            let merkleized = db
1132                .new_batch()
1133                .set(key, value)
1134                .merkleize(&db, None, floor)
1135                .await;
1136            (db, _) = db.apply_batch(merkleized).await.unwrap();
1137            db = db.commit().await.unwrap();
1138        }
1139
1140        let root_before = db.root();
1141        let bounds_before = db.bounds();
1142
1143        let prune_loc = bounds_before.end - 5;
1144        let db = db.prune(prune_loc).await.unwrap();
1145
1146        assert_eq!(db.root(), root_before);
1147
1148        let key_0 = Sha256::fill(0u8);
1149        assert!(db.get(&key_0).await.unwrap().is_none());
1150
1151        let key_19 = Sha256::fill(19u8);
1152        assert_eq!(
1153            db.get(&key_19).await.unwrap(),
1154            Some(Sha256::fill(19u8.wrapping_add(100)))
1155        );
1156
1157        db.destroy().await.unwrap();
1158    }
1159
1160    /// Pruning immediately after an uncommitted batch must leave the database recoverable. Since
1161    /// prune is not a durability boundary, recovery may return either the durable baseline or the
1162    /// buffered state, but never a mixture whose floor references pruned operations.
1163    #[boxed]
1164    pub(crate) async fn run_prune_after_uncommitted_apply_batch_recovery<F: Family, V, C>(
1165        context: deterministic::Context,
1166        open_db: impl Fn(
1167            deterministic::Context,
1168        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1169    ) where
1170        V: ValueEncoding<Value = Digest>,
1171        C: Mutable<Item = Operation<F, Digest, V>>,
1172        C::Item: EncodeShared,
1173    {
1174        let db = open_db(context.child("first")).await;
1175
1176        // Fill more than one journal blob and establish a durable baseline whose floor still
1177        // requires the oldest blob.
1178        let mut batch = db.new_batch();
1179        for i in 0..6u8 {
1180            batch = batch.set(Sha256::fill(i), Sha256::fill(i.wrapping_add(10)));
1181        }
1182        let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1183        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1184        let db = db.sync().await.unwrap();
1185        let durable_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
1186
1187        // Apply, but do not commit, a batch that advances the floor far enough for prune to
1188        // remove the oldest blob.
1189        let buffered_floor = db.bounds().end;
1190        let key = Sha256::fill(100);
1191        let value = Sha256::fill(101);
1192        let merkleized = db
1193            .new_batch()
1194            .set(key, value)
1195            .merkleize(&db, None, buffered_floor)
1196            .await;
1197        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1198        let buffered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
1199        assert_ne!(buffered_state, durable_state);
1200
1201        let db = db.prune(buffered_floor).await.unwrap();
1202        assert!(db.bounds().start > Location::new(0));
1203        drop(db);
1204
1205        // Reopen must produce one coherent state. In particular, it must not recover the old
1206        // floor after the prune has removed operations that floor still needs.
1207        let db = open_db(context.child("second")).await;
1208        assert!(db.bounds().start <= db.inactivity_floor_loc());
1209        let recovered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
1210        assert!(
1211            recovered_state == durable_state || recovered_state == buffered_state,
1212            "recovered state is neither the durable baseline nor the buffered state"
1213        );
1214
1215        db.destroy().await.unwrap();
1216    }
1217
1218    /// `operations()` must cover exactly the batch's own applied range and match the
1219    /// operations a post-apply `historical_proof` recovers from the log, for a db-based
1220    /// batch and for a chained batch applied after its ancestor.
1221    #[boxed]
1222    pub(crate) async fn run_operations_match_applied_log<F: Family, V, C>(
1223        context: deterministic::Context,
1224        open_db: impl Fn(
1225            deterministic::Context,
1226        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1227    ) where
1228        V: ValueEncoding<Value = Digest>,
1229        C: Mutable<Item = Operation<F, Digest, V>>,
1230        C::Item: EncodeShared + PartialEq + core::fmt::Debug,
1231    {
1232        let db = open_db(context.child("db")).await;
1233
1234        let seed = db
1235            .new_batch()
1236            .set(Sha256::fill(1u8), Sha256::fill(11u8))
1237            .set(Sha256::fill(2u8), Sha256::fill(12u8))
1238            .merkleize(&db, None, db.inactivity_floor_loc())
1239            .await;
1240        let (seed_start, seed_ops) = seed.operations();
1241        let seed_root = seed.root();
1242        let seed_proof = seed.proof(&db).unwrap();
1243        let seed_pins = seed.pinned_nodes(&db).unwrap();
1244        let (db, seed_range) = db.apply_batch(seed).await.unwrap();
1245        assert_eq!(seed_start, seed_range.start);
1246        assert_eq!(*seed_start + seed_ops.len() as u64, *seed_range.end);
1247
1248        // A chained batch's operations are its own suffix only.
1249        let parent = db
1250            .new_batch()
1251            .set(Sha256::fill(3u8), Sha256::fill(13u8))
1252            .merkleize(&db, None, db.inactivity_floor_loc())
1253            .await;
1254        let child = parent
1255            .new_batch::<Sha256>()
1256            .set(Sha256::fill(4u8), Sha256::fill(14u8))
1257            .merkleize(&db, None, db.inactivity_floor_loc())
1258            .await;
1259        let (parent_start, parent_ops) = parent.operations();
1260        let (child_start, child_ops) = child.operations();
1261        let (parent_root, child_root) = (parent.root(), child.root());
1262        let (parent_pins, child_pins) = (
1263            parent.pinned_nodes(&db).unwrap(),
1264            child.pinned_nodes(&db).unwrap(),
1265        );
1266        let (parent_proof, child_proof) = (parent.proof(&db).unwrap(), child.proof(&db).unwrap());
1267        let (db, parent_range) = db.apply_batch(parent).await.unwrap();
1268        let (db, child_range) = db.apply_batch(child).await.unwrap();
1269        assert_eq!(parent_start, parent_range.start);
1270        assert_eq!(*parent_start + parent_ops.len() as u64, *parent_range.end);
1271        assert_eq!(child_start, child_range.start);
1272        assert_eq!(*child_start + child_ops.len() as u64, *child_range.end);
1273
1274        // A write-free batch still captures its commit-only suffix.
1275        let empty = db
1276            .new_batch()
1277            .merkleize(&db, None, db.inactivity_floor_loc())
1278            .await;
1279        let (empty_start, empty_ops) = empty.operations();
1280        let (empty_root, empty_proof) = (empty.root(), empty.proof(&db).unwrap());
1281        let empty_pins = empty.pinned_nodes(&db).unwrap();
1282        let (db, empty_range) = db.apply_batch(empty).await.unwrap();
1283        assert_eq!(empty_start, empty_range.start);
1284        assert_eq!(*empty_start + empty_ops.len() as u64, *empty_range.end);
1285
1286        // Every captured delta and proof must match what the log recovers for its
1287        // range, and verify against the batch's own root with and without the pins.
1288        for (start, ops, proof, pins, root) in [
1289            (seed_start, seed_ops, seed_proof, seed_pins, seed_root),
1290            (
1291                parent_start,
1292                parent_ops,
1293                parent_proof,
1294                parent_pins,
1295                parent_root,
1296            ),
1297            (child_start, child_ops, child_proof, child_pins, child_root),
1298            (empty_start, empty_ops, empty_proof, empty_pins, empty_root),
1299        ] {
1300            let len = core::num::NonZeroU64::new(ops.len() as u64).unwrap();
1301            let end = Location::new(*start + ops.len() as u64);
1302            let (log_proof, log_ops) = db.historical_proof(end, start, len).await.unwrap();
1303            assert_eq!(log_ops, *ops);
1304            assert_eq!(log_proof, proof);
1305            assert!(verify_proof::<Sha256, _, _>(&proof, start, &ops, &root));
1306            assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
1307                &proof, start, &ops, &pins, &root
1308            ));
1309        }
1310
1311        // Flushing the applied batch prunes the store to its peaks. The late batch's base is
1312        // mid-mountain, so its artifacts are refused rather than returned unverifiable.
1313        let late = db
1314            .new_batch()
1315            .set(Sha256::fill(5u8), Sha256::fill(15u8))
1316            .merkleize(&db, None, db.inactivity_floor_loc())
1317            .await;
1318        let (db, _) = db.apply_batch(Arc::clone(&late)).await.unwrap();
1319        let db = db.commit().await.unwrap();
1320        assert!(matches!(
1321            late.proof(&db),
1322            Err(crate::qmdb::Error::Merkle(
1323                crate::merkle::Error::ElementPruned(_)
1324            ))
1325        ));
1326        assert!(matches!(
1327            late.pinned_nodes(&db),
1328            Err(crate::qmdb::Error::Merkle(
1329                crate::merkle::Error::ElementPruned(_)
1330            ))
1331        ));
1332
1333        // A batch built on the flushed store reads every node below it from the pinned peaks.
1334        let flushed = db
1335            .new_batch()
1336            .set(Sha256::fill(6u8), Sha256::fill(16u8))
1337            .merkleize(&db, None, db.inactivity_floor_loc())
1338            .await;
1339        let (flushed_start, flushed_ops) = flushed.operations();
1340        let flushed_root = flushed.root();
1341        let flushed_proof = flushed.proof(&db).unwrap();
1342        let flushed_pins = flushed.pinned_nodes(&db).unwrap();
1343        assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
1344            &flushed_proof,
1345            flushed_start,
1346            &flushed_ops,
1347            &flushed_pins,
1348            &flushed_root
1349        ));
1350        let (db, flushed_range) = db.apply_batch(flushed).await.unwrap();
1351        assert_eq!(flushed_start, flushed_range.start);
1352        assert_eq!(
1353            *flushed_start + flushed_ops.len() as u64,
1354            *flushed_range.end
1355        );
1356
1357        db.destroy().await.unwrap();
1358    }
1359
1360    #[boxed]
1361    pub(crate) async fn run_batch_chain<F: Family, V, C>(
1362        context: deterministic::Context,
1363        open_db: impl Fn(
1364            deterministic::Context,
1365        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1366    ) where
1367        V: ValueEncoding<Value = Digest>,
1368        C: Mutable<Item = Operation<F, Digest, V>>,
1369        C::Item: EncodeShared,
1370    {
1371        let db = open_db(context.child("first")).await;
1372
1373        let k1 = Sha256::fill(1u8);
1374        let k2 = Sha256::fill(2u8);
1375        let k3 = Sha256::fill(3u8);
1376        let v1 = Sha256::fill(11u8);
1377        let v2 = Sha256::fill(12u8);
1378        let v3 = Sha256::fill(13u8);
1379
1380        let parent = db
1381            .new_batch()
1382            .set(k1, v1)
1383            .merkleize(&db, None, Location::new(0))
1384            .await;
1385        let child = parent
1386            .new_batch::<Sha256>()
1387            .set(k2, v2)
1388            .merkleize(&db, None, Location::new(0))
1389            .await;
1390
1391        assert_eq!(child.get(&k1, &db).await.unwrap(), Some(v1));
1392        assert_eq!(child.get(&k2, &db).await.unwrap(), Some(v2));
1393        assert!(child.get(&k3, &db).await.unwrap().is_none());
1394
1395        let (db, _) = db.apply_batch(child).await.unwrap();
1396        let db = db.commit().await.unwrap();
1397
1398        assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
1399        assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
1400
1401        let merkleized = db
1402            .new_batch()
1403            .set(k3, v3)
1404            .merkleize(&db, None, Location::new(0))
1405            .await;
1406        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1407        let db = db.commit().await.unwrap();
1408        assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
1409
1410        db.destroy().await.unwrap();
1411    }
1412
1413    #[boxed]
1414    pub(crate) async fn run_build_and_authenticate<F: Family, V, C>(
1415        context: deterministic::Context,
1416        open_db: impl Fn(
1417            deterministic::Context,
1418        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1419    ) where
1420        V: ValueEncoding<Value = Digest>,
1421        C: Mutable<Item = Operation<F, Digest, V>>,
1422        C::Item: EncodeShared,
1423    {
1424        // Build a db with `ELEMENTS` key/value pairs and prove ranges over them.
1425        let db = open_db(context.child("first")).await;
1426
1427        let mut batch = db.new_batch();
1428        for i in 0u64..2_000 {
1429            let k = Sha256::hash(&[&i.to_be_bytes()]);
1430            let v = Sha256::fill(i as u8);
1431            batch = batch.set(k, v);
1432        }
1433        let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1434        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1435        let db = db.commit().await.unwrap();
1436        assert_eq!(db.bounds().end, 2_000 + 2);
1437
1438        // Drop & reopen the db, making sure it has exactly the same state.
1439        let root = db.root();
1440        drop(db);
1441
1442        let db = open_db(context.child("second")).await;
1443        assert_eq!(root, db.root());
1444        assert_eq!(db.bounds().end, 2_000 + 2);
1445        for i in 0u64..2_000 {
1446            let k = Sha256::hash(&[&i.to_be_bytes()]);
1447            let v = Sha256::fill(i as u8);
1448            assert_eq!(db.get(&k).await.unwrap().unwrap(), v);
1449        }
1450
1451        // Make sure all ranges of 5 operations are provable, including truncated ranges at the
1452        // end.
1453        let max_ops = NZU64!(5);
1454        for i in 0..*db.bounds().end {
1455            let (proof, log) = db.proof(Location::new(i), max_ops).await.unwrap();
1456            assert!(verify_proof::<Sha256, _, _>(
1457                &proof,
1458                Location::new(i),
1459                &log,
1460                &root
1461            ));
1462        }
1463
1464        db.destroy().await.unwrap();
1465    }
1466
1467    #[boxed]
1468    pub(crate) async fn run_recovery_from_failed_merkle_sync<F: Family, V, C>(
1469        context: deterministic::Context,
1470        open_db: impl Fn(
1471            deterministic::Context,
1472        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1473    ) where
1474        V: ValueEncoding<Value = Digest>,
1475        C: Mutable<Item = Operation<F, Digest, V>>,
1476        C::Item: EncodeShared,
1477    {
1478        // Insert 1000 keys then sync.
1479        const ELEMENTS: u64 = 1000;
1480        let db = open_db(context.child("first")).await;
1481
1482        let mut batch = db.new_batch();
1483        for i in 0u64..ELEMENTS {
1484            let k = Sha256::hash(&[&i.to_be_bytes()]);
1485            let v = Sha256::fill(i as u8);
1486            batch = batch.set(k, v);
1487        }
1488        let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1489        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1490        let db = db.commit().await.unwrap();
1491        assert_eq!(db.bounds().end, ELEMENTS + 2);
1492        let db = db.sync().await.unwrap();
1493        let halfway_root = db.root();
1494
1495        // Insert another 1000 keys (different from the first batch) then commit.
1496        let mut batch = db.new_batch();
1497        for i in ELEMENTS..ELEMENTS * 2 {
1498            let k = Sha256::hash(&[&i.to_be_bytes()]);
1499            let v = Sha256::fill(i as u8);
1500            batch = batch.set(k, v);
1501        }
1502        let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1503        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1504        db.commit().await.unwrap(); // Drop before syncing
1505
1506        // Recovery should replay the log to regenerate the merkle structure.
1507        // op_count = 1002 (first batch + commit) + 1000 (second batch) + 1 (second commit) = 2003
1508        let db = open_db(context.child("second")).await;
1509        assert_eq!(db.bounds().end, 2003);
1510        let root = db.root();
1511        assert_ne!(root, halfway_root);
1512
1513        // Drop & reopen could preserve the final commit.
1514        drop(db);
1515        let db = open_db(context.child("third")).await;
1516        assert_eq!(db.bounds().end, 2003);
1517        assert_eq!(db.root(), root);
1518
1519        db.destroy().await.unwrap();
1520    }
1521
1522    #[boxed]
1523    pub(crate) async fn run_recovery_from_failed_log_sync<F: Family, V, C>(
1524        context: deterministic::Context,
1525        open_db: impl Fn(
1526            deterministic::Context,
1527        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1528    ) where
1529        V: ValueEncoding<Value = Digest>,
1530        C: Mutable<Item = Operation<F, Digest, V>>,
1531        C::Item: EncodeShared,
1532    {
1533        let db = open_db(context.child("first")).await;
1534
1535        // Insert a single key and then commit to create a first commit point.
1536        let k1 = Sha256::fill(1u8);
1537        let v1 = Sha256::fill(3u8);
1538        let merkleized = db
1539            .new_batch()
1540            .set(k1, v1)
1541            .merkleize(&db, None, Location::new(0))
1542            .await;
1543        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1544        let db = db.commit().await.unwrap();
1545        let first_commit_root = db.root();
1546
1547        // Simulate failure. Sets that are never merkleized/applied are lost.
1548        // Recovery should restore the last commit point.
1549        drop(db);
1550
1551        // Recovery should back up to previous commit point.
1552        let db = open_db(context.child("second")).await;
1553        assert_eq!(db.bounds().end, 3);
1554        let root = db.root();
1555        assert_eq!(root, first_commit_root);
1556
1557        db.destroy().await.unwrap();
1558    }
1559
1560    #[boxed]
1561    pub(crate) async fn run_pruning<F: Family, V, C>(
1562        context: deterministic::Context,
1563        open_db: impl Fn(
1564            deterministic::Context,
1565        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1566    ) where
1567        V: ValueEncoding<Value = Digest>,
1568        C: Mutable<Item = Operation<F, Digest, V>>,
1569        C::Item: EncodeShared,
1570    {
1571        // Build a db with `ELEMENTS` key/value pairs then prune some of them.
1572        const ELEMENTS: u64 = 2_000;
1573        let db = open_db(context.child("first")).await;
1574
1575        // Batch writes keys in BTreeMap-sorted order, so build the sorted key
1576        // list to map between journal locations and keys.
1577        let mut sorted_keys: Vec<sha256::Digest> = (1u64..ELEMENTS + 1)
1578            .map(|i| Sha256::hash(&[&i.to_be_bytes()]))
1579            .collect();
1580        sorted_keys.sort();
1581        // Location 0: initial commit; locations 1..=ELEMENTS: Set ops in sorted
1582        // key order; location ELEMENTS+1: batch commit.
1583        // key_at_loc(L) = sorted_keys[L - 1] for 1 <= L <= ELEMENTS.
1584
1585        let mut batch = db.new_batch();
1586        for i in 1u64..ELEMENTS + 1 {
1587            let k = Sha256::hash(&[&i.to_be_bytes()]);
1588            let v = Sha256::fill(i as u8);
1589            batch = batch.set(k, v);
1590        }
1591        // The inactivity floor must cover both prune targets in this test.
1592        // Second prune request is at ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1.
1593        let inactivity_floor = Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1);
1594        let merkleized = batch.merkleize(&db, None, inactivity_floor).await;
1595        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1596        assert_eq!(db.bounds().end, ELEMENTS + 2);
1597
1598        // Prune the db to the first half of the operations.
1599        let db = db.prune(Location::new((ELEMENTS + 2) / 2)).await.unwrap();
1600        let bounds = db.bounds();
1601        assert_eq!(bounds.end, ELEMENTS + 2);
1602
1603        // items_per_section is 5, so half should be exactly at a blob boundary, in which case
1604        // the actual pruning location should match the requested.
1605        let oldest_retained_loc = bounds.start;
1606        assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2));
1607
1608        // Try to fetch a pruned key (at location oldest_retained - 1).
1609        let pruned_key = sorted_keys[*oldest_retained_loc as usize - 2];
1610        assert!(db.get(&pruned_key).await.unwrap().is_none());
1611
1612        // Try to fetch unpruned key (at location oldest_retained).
1613        let unpruned_key = sorted_keys[*oldest_retained_loc as usize - 1];
1614        assert!(db.get(&unpruned_key).await.unwrap().is_some());
1615
1616        // Drop & reopen the db, making sure it has exactly the same state.
1617        let root = db.root();
1618        db.sync().await.unwrap();
1619
1620        let db = open_db(context.child("second")).await;
1621        assert_eq!(root, db.root());
1622        let bounds = db.bounds();
1623        assert_eq!(bounds.end, ELEMENTS + 2);
1624        let oldest_retained_loc = bounds.start;
1625        assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2));
1626
1627        // Prune to a non-blob boundary.
1628        let loc = Location::new(ELEMENTS / 2 + (ITEMS_PER_SECTION * 2 - 1));
1629        let db = db.prune(loc).await.unwrap();
1630        // Actual boundary should be a multiple of 5.
1631        let oldest_retained_loc = db.bounds().start;
1632        assert_eq!(
1633            oldest_retained_loc,
1634            Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION)
1635        );
1636
1637        // Confirm boundary persists across restart.
1638        db.sync().await.unwrap();
1639        let db = open_db(context.child("third")).await;
1640        let oldest_retained_loc = db.bounds().start;
1641        assert_eq!(
1642            oldest_retained_loc,
1643            Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION)
1644        );
1645
1646        // Try to fetch a key before the inactivity floor (not in snapshot after reopen).
1647        let floor_val = ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1;
1648        let inactive_key = sorted_keys[floor_val as usize - 2];
1649        assert!(db.get(&inactive_key).await.unwrap().is_none());
1650
1651        // Try to fetch a key at the inactivity floor (in snapshot after reopen).
1652        let active_key = sorted_keys[floor_val as usize - 1];
1653        assert!(db.get(&active_key).await.unwrap().is_some());
1654
1655        // Confirm behavior of trying to create a proof of pruned items is as expected.
1656        let pruned_pos = ELEMENTS / 2;
1657        let proof_result = db
1658            .proof(Location::new(pruned_pos), NZU64!(pruned_pos + 100))
1659            .await;
1660        assert!(
1661            matches!(proof_result, Err(Error::Journal(crate::journal::Error::ItemPruned(pos))) if pos == pruned_pos)
1662        );
1663
1664        db.destroy().await.unwrap();
1665    }
1666
1667    #[boxed]
1668    pub(crate) async fn run_prune_beyond_floor<F: Family, V, C>(
1669        context: deterministic::Context,
1670        open_db: impl Fn(
1671            deterministic::Context,
1672        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1673    ) where
1674        V: ValueEncoding<Value = Digest>,
1675        C: Mutable<Item = Operation<F, Digest, V>>,
1676        C::Item: EncodeShared,
1677    {
1678        let db = open_db(context.child("test")).await;
1679
1680        // Test pruning empty database (floor=0, so prune(1) fails)
1681        let result = db.prune(Location::new(1)).await;
1682        assert!(
1683            matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor))
1684                if prune_loc == Location::new(1) && floor == Location::new(0))
1685        );
1686
1687        let db = open_db(context.child("test")).await;
1688
1689        // Add key-value pairs and commit
1690        let k1 = Digest::from(*b"12345678901234567890123456789012");
1691        let k2 = Digest::from(*b"abcdefghijklmnopqrstuvwxyz123456");
1692        let k3 = Digest::from(*b"99999999999999999999999999999999");
1693        let v1 = Sha256::fill(1u8);
1694        let v2 = Sha256::fill(2u8);
1695        let v3 = Sha256::fill(3u8);
1696
1697        // First batch with floor=3 (the commit location).
1698        let merkleized = db
1699            .new_batch()
1700            .set(k1, v1)
1701            .set(k2, v2)
1702            .merkleize(&db, None, Location::new(3))
1703            .await;
1704        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1705
1706        // op_count is 4 (initial_commit, k1, k2, commit), last_commit is at location 3
1707        assert_eq!(*db.last_commit_loc, 3);
1708
1709        // Second batch with floor=5 (the new commit location).
1710        let merkleized = db
1711            .new_batch()
1712            .set(k3, v3)
1713            .merkleize(&db, None, Location::new(5))
1714            .await;
1715        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1716
1717        // Test valid prune (3 <= floor of 5)
1718        let db = db.prune(Location::new(3)).await.unwrap();
1719
1720        // Test pruning beyond inactivity floor
1721        let floor = db.inactivity_floor_loc();
1722        let beyond = floor + 1;
1723        let result = db.prune(beyond).await;
1724        assert!(
1725            matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, f))
1726                if prune_loc == beyond && f == floor)
1727        );
1728    }
1729
1730    async fn commit_sets<F: Family, V, C>(
1731        db: TestDb<F, V, C>,
1732        sets: impl IntoIterator<Item = (Digest, V::Value)>,
1733        metadata: Option<V::Value>,
1734    ) -> (TestDb<F, V, C>, Range<Location<F>>)
1735    where
1736        V: ValueEncoding<Value = Digest>,
1737        C: Mutable<Item = Operation<F, Digest, V>>,
1738        C::Item: EncodeShared,
1739    {
1740        commit_sets_with_floor(db, sets, metadata, Location::new(0)).await
1741    }
1742
1743    async fn commit_sets_with_floor<F: Family, V, C>(
1744        db: TestDb<F, V, C>,
1745        sets: impl IntoIterator<Item = (Digest, V::Value)>,
1746        metadata: Option<V::Value>,
1747        floor: Location<F>,
1748    ) -> (TestDb<F, V, C>, Range<Location<F>>)
1749    where
1750        V: ValueEncoding<Value = Digest>,
1751        C: Mutable<Item = Operation<F, Digest, V>>,
1752        C::Item: EncodeShared,
1753    {
1754        let mut batch = db.new_batch();
1755        for (key, value) in sets {
1756            batch = batch.set(key, value);
1757        }
1758        let merkleized = batch.merkleize(&db, metadata, floor).await;
1759        let (db, range) = db.apply_batch(merkleized).await.unwrap();
1760        let db = db.commit().await.unwrap();
1761        (db, range)
1762    }
1763
1764    #[boxed]
1765    pub(crate) async fn run_rewind_recovery<F: Family, V, C>(
1766        context: deterministic::Context,
1767        open_db: impl Fn(
1768            deterministic::Context,
1769        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1770    ) where
1771        V: ValueEncoding<Value = Digest>,
1772        C: Mutable<Item = Operation<F, Digest, V>>,
1773        C::Item: EncodeShared,
1774    {
1775        let db = open_db(context.child("db")).await;
1776
1777        let key1 = Sha256::hash(&[&1u64.to_be_bytes()]);
1778        let key2 = Sha256::hash(&[&2u64.to_be_bytes()]);
1779        let key3 = Sha256::hash(&[&3u64.to_be_bytes()]);
1780        let key4 = Sha256::hash(&[&4u64.to_be_bytes()]);
1781
1782        let value1 = Sha256::fill(11u8);
1783        let value2 = Sha256::fill(22u8);
1784        let value3 = Sha256::fill(33u8);
1785        let value4 = Sha256::fill(66u8);
1786
1787        let metadata_a = Sha256::fill(44u8);
1788        let (db, first_range) =
1789            commit_sets(db, [(key1, value1), (key2, value2)], Some(metadata_a)).await;
1790        let size_before = db.bounds().end;
1791        let root_before = db.root();
1792        let last_commit_before = db.last_commit_loc;
1793        assert_eq!(size_before, first_range.end);
1794
1795        let metadata_b = Sha256::fill(55u8);
1796        let (db, second_range) =
1797            commit_sets(db, [(key3, value3), (key4, value4)], Some(metadata_b)).await;
1798        assert_eq!(second_range.start, size_before);
1799        assert_ne!(db.root(), root_before);
1800        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_b));
1801        assert_eq!(db.get(&key3).await.unwrap(), Some(value3));
1802        assert_eq!(db.get(&key4).await.unwrap(), Some(value4));
1803
1804        let db = db.rewind(size_before).await.unwrap();
1805        assert_eq!(db.root(), root_before);
1806        assert_eq!(db.bounds().end, size_before);
1807        assert_eq!(db.last_commit_loc, last_commit_before);
1808        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
1809        assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1810        assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
1811        assert_eq!(db.get(&key3).await.unwrap(), None);
1812        assert_eq!(db.get(&key4).await.unwrap(), None);
1813
1814        db.commit().await.unwrap();
1815        let db = open_db(context.child("reopen")).await;
1816        assert_eq!(db.root(), root_before);
1817        assert_eq!(db.bounds().end, size_before);
1818        assert_eq!(db.last_commit_loc, last_commit_before);
1819        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
1820        assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1821        assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
1822        assert_eq!(db.get(&key3).await.unwrap(), None);
1823        assert_eq!(db.get(&key4).await.unwrap(), None);
1824
1825        db.destroy().await.unwrap();
1826    }
1827
1828    /// Regression: a key Set before the rewind boundary that translator-collides with a key in the
1829    /// rewound suffix must survive rewind. Earlier the snapshot remove pruned the entire translated
1830    /// bucket and dropped the retained key.
1831    #[boxed]
1832    pub(crate) async fn run_rewind_preserves_collision_bucket<F: Family, V, C>(
1833        context: deterministic::Context,
1834        open_db: impl Fn(
1835            deterministic::Context,
1836        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1837    ) where
1838        V: ValueEncoding<Value = Digest>,
1839        C: Mutable<Item = Operation<F, Digest, V>>,
1840        C::Item: EncodeShared,
1841    {
1842        let db = open_db(context.child("db")).await;
1843
1844        // Two keys sharing the first two bytes collide under TwoCap.
1845        let mut k1_bytes = [0u8; 32];
1846        let mut k2_bytes = [0u8; 32];
1847        k1_bytes[0] = 0xAA;
1848        k1_bytes[1] = 0xBB;
1849        k2_bytes[0] = 0xAA;
1850        k2_bytes[1] = 0xBB;
1851        k1_bytes[31] = 0x01;
1852        k2_bytes[31] = 0x02;
1853        let key1 = Digest::from(k1_bytes);
1854        let key2 = Digest::from(k2_bytes);
1855        let value1 = Sha256::fill(11u8);
1856        let value2 = Sha256::fill(22u8);
1857
1858        let (db, _) = commit_sets(db, [(key1, value1)], None).await;
1859        let size_after_first = db.bounds().end;
1860        let (db, _) = commit_sets(db, [(key2, value2)], None).await;
1861        assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1862        assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
1863
1864        let db = db.rewind(size_after_first).await.unwrap();
1865
1866        // The retained key must still be readable; pre-fix this returned None because the
1867        // translator bucket was wiped by the suffix-key remove.
1868        assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1869        assert_eq!(db.get(&key2).await.unwrap(), None);
1870
1871        db.destroy().await.unwrap();
1872    }
1873
1874    #[boxed]
1875    pub(crate) async fn run_rewind_pruned_target_errors<F: Family, V, C>(
1876        context: deterministic::Context,
1877        open_small_sections_db: impl Fn(
1878            deterministic::Context,
1879        )
1880            -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1881    ) where
1882        V: ValueEncoding<Value = Digest>,
1883        C: Mutable<Item = Operation<F, Digest, V>>,
1884        C::Item: EncodeShared,
1885    {
1886        let db = open_small_sections_db(context.child("db")).await;
1887
1888        let (mut db, first_range) = commit_sets(
1889            db,
1890            (0u64..16).map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8))),
1891            None,
1892        )
1893        .await;
1894
1895        let mut round = 0u64;
1896        loop {
1897            round += 1;
1898            assert!(
1899                round <= 64,
1900                "failed to prune enough history for rewind test"
1901            );
1902
1903            // Floor must be >= last_commit_loc for prune to succeed.
1904            // With 16 sets, commit is at current end + 16.
1905            let floor = db.bounds().end + 16;
1906            (db, _) = commit_sets_with_floor(
1907                db,
1908                (0u64..16).map(|i| {
1909                    let seed = round * 100 + i;
1910                    (
1911                        Sha256::hash(&[&seed.to_be_bytes()]),
1912                        Sha256::fill(seed as u8),
1913                    )
1914                }),
1915                None,
1916                floor,
1917            )
1918            .await;
1919            let last_commit = db.last_commit_loc;
1920            db = db.prune(last_commit).await.unwrap();
1921
1922            if db.bounds().start > first_range.start {
1923                break;
1924            }
1925        }
1926
1927        let oldest_retained = db.bounds().start;
1928        let Err(boundary_err) = db.rewind(oldest_retained).await else {
1929            panic!("expected rewind to fail");
1930        };
1931        assert!(
1932            matches!(
1933                boundary_err,
1934                Error::Journal(crate::journal::Error::ItemPruned(_))
1935            ),
1936            "unexpected rewind error at retained boundary: {boundary_err:?}"
1937        );
1938
1939        let db = open_small_sections_db(context.child("db")).await;
1940        let Err(err) = db.rewind(first_range.start).await else {
1941            panic!("expected rewind to fail");
1942        };
1943        assert!(
1944            matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
1945            "unexpected rewind error: {err:?}"
1946        );
1947    }
1948
1949    /// batch.get() reads pending mutations and falls through to base DB.
1950    #[boxed]
1951    pub(crate) async fn run_batch_get_read_through<F: Family, V, C>(
1952        context: deterministic::Context,
1953        open_db: impl Fn(
1954            deterministic::Context,
1955        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1956    ) where
1957        V: ValueEncoding<Value = Digest>,
1958        C: Mutable<Item = Operation<F, Digest, V>>,
1959        C::Item: EncodeShared,
1960    {
1961        let db = open_db(context.child("db")).await;
1962
1963        // Pre-populate with key A.
1964        let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
1965        let val_a = Sha256::fill(1u8);
1966        let merkleized = db
1967            .new_batch()
1968            .set(key_a, val_a)
1969            .merkleize(&db, None, Location::new(0))
1970            .await;
1971        let (db, _) = db.apply_batch(merkleized).await.unwrap();
1972
1973        // batch.get(&A) should return DB value.
1974        let mut batch = db.new_batch();
1975        assert_eq!(batch.get(&key_a, &db).await.unwrap(), Some(val_a));
1976
1977        // Set B in batch, batch.get(&B) returns the value.
1978        let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
1979        let val_b = Sha256::fill(2u8);
1980        batch = batch.set(key_b, val_b);
1981        assert_eq!(batch.get(&key_b, &db).await.unwrap(), Some(val_b));
1982
1983        // Nonexistent key.
1984        let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
1985        assert_eq!(batch.get(&key_c, &db).await.unwrap(), None);
1986
1987        db.destroy().await.unwrap();
1988    }
1989
1990    /// Child batch reads parent diff and adds its own mutations.
1991    #[boxed]
1992    pub(crate) async fn run_batch_stacked_get<F: Family, V, C>(
1993        context: deterministic::Context,
1994        open_db: impl Fn(
1995            deterministic::Context,
1996        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1997    ) where
1998        V: ValueEncoding<Value = Digest>,
1999        C: Mutable<Item = Operation<F, Digest, V>>,
2000        C::Item: EncodeShared,
2001    {
2002        let db = open_db(context.child("db")).await;
2003
2004        // Parent batch: set A.
2005        let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
2006        let val_a = Sha256::fill(10u8);
2007        let parent = db.new_batch().set(key_a, val_a);
2008        let parent_m = parent.merkleize(&db, None, Location::new(0)).await;
2009
2010        // Child reads parent's A.
2011        let mut child = parent_m.new_batch::<Sha256>();
2012        assert_eq!(child.get(&key_a, &db).await.unwrap(), Some(val_a));
2013
2014        // Child sets B.
2015        let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
2016        let val_b = Sha256::fill(20u8);
2017        child = child.set(key_b, val_b);
2018        assert_eq!(child.get(&key_b, &db).await.unwrap(), Some(val_b));
2019
2020        // Nonexistent key.
2021        let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
2022        assert_eq!(child.get(&key_c, &db).await.unwrap(), None);
2023
2024        db.destroy().await.unwrap();
2025    }
2026
2027    /// Two-level stacked batch apply works end-to-end.
2028    #[boxed]
2029    pub(crate) async fn run_batch_stacked_apply<F: Family, V, C>(
2030        context: deterministic::Context,
2031        open_db: impl Fn(
2032            deterministic::Context,
2033        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2034    ) where
2035        V: ValueEncoding<Value = Digest>,
2036        C: Mutable<Item = Operation<F, Digest, V>>,
2037        C::Item: EncodeShared,
2038    {
2039        let db = open_db(context.child("db")).await;
2040
2041        // Sort keys so operations are in BTreeMap order (same as merkleize writes).
2042        let mut kvs_first: Vec<(Digest, Digest)> = (0u64..5)
2043            .map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8)))
2044            .collect();
2045        kvs_first.sort_by_key(|a| a.0);
2046
2047        let mut kvs_second: Vec<(Digest, Digest)> = (5u64..10)
2048            .map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8)))
2049            .collect();
2050        kvs_second.sort_by_key(|a| a.0);
2051
2052        // Parent batch: set keys 0..5.
2053        let mut parent = db.new_batch();
2054        for (k, v) in &kvs_first {
2055            parent = parent.set(*k, *v);
2056        }
2057        let parent_m = parent.merkleize(&db, None, Location::new(0)).await;
2058
2059        // Child batch: set keys 5..10.
2060        let mut child = parent_m.new_batch::<Sha256>();
2061        for (k, v) in &kvs_second {
2062            child = child.set(*k, *v);
2063        }
2064        let child_m = child.merkleize(&db, None, Location::new(0)).await;
2065        let expected_root = child_m.root();
2066        let (db, _) = db.apply_batch(child_m).await.unwrap();
2067
2068        assert_eq!(db.root(), expected_root);
2069
2070        // All 10 keys should be accessible.
2071        for (k, v) in kvs_first.iter().chain(kvs_second.iter()) {
2072            assert_eq!(db.get(k).await.unwrap(), Some(*v));
2073        }
2074
2075        db.destroy().await.unwrap();
2076    }
2077
2078    /// MerkleizedBatch::root() matches db.root() after apply_batch().
2079    #[boxed]
2080    pub(crate) async fn run_batch_speculative_root<F: Family, V, C>(
2081        context: deterministic::Context,
2082        open_db: impl Fn(
2083            deterministic::Context,
2084        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2085    ) where
2086        V: ValueEncoding<Value = Digest>,
2087        C: Mutable<Item = Operation<F, Digest, V>>,
2088        C::Item: EncodeShared,
2089    {
2090        let db = open_db(context.child("db")).await;
2091
2092        let mut batch = db.new_batch();
2093        for i in 0u8..10 {
2094            let k = Sha256::hash(&[&[i]]);
2095            batch = batch.set(k, Sha256::fill(i));
2096        }
2097        let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
2098
2099        let speculative = merkleized.root();
2100        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2101        assert_eq!(db.root(), speculative);
2102
2103        // Second batch with metadata.
2104        let metadata = Some(Sha256::fill(55u8));
2105        let mut batch = db.new_batch();
2106        let k = Sha256::hash(&[&[0xAA]]);
2107        batch = batch.set(k, Sha256::fill(0xAA));
2108        let merkleized = batch.merkleize(&db, metadata, Location::new(0)).await;
2109        let speculative = merkleized.root();
2110        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2111        assert_eq!(db.root(), speculative);
2112
2113        db.destroy().await.unwrap();
2114    }
2115
2116    /// MerkleizedBatch::get() reads from diff and base DB.
2117    #[boxed]
2118    pub(crate) async fn run_merkleized_batch_get<F: Family, V, C>(
2119        context: deterministic::Context,
2120        open_db: impl Fn(
2121            deterministic::Context,
2122        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2123    ) where
2124        V: ValueEncoding<Value = Digest>,
2125        C: Mutable<Item = Operation<F, Digest, V>>,
2126        C::Item: EncodeShared,
2127    {
2128        let db = open_db(context.child("db")).await;
2129
2130        // Pre-populate base DB.
2131        let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
2132        let val_a = Sha256::fill(10u8);
2133        let merkleized = db
2134            .new_batch()
2135            .set(key_a, val_a)
2136            .merkleize(&db, None, Location::new(0))
2137            .await;
2138        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2139
2140        // Create a merkleized batch with a new key.
2141        let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
2142        let val_b = Sha256::fill(20u8);
2143        let merkleized = db
2144            .new_batch()
2145            .set(key_b, val_b)
2146            .merkleize(&db, None, Location::new(0))
2147            .await;
2148
2149        // Read base DB value through merkleized batch.
2150        assert_eq!(merkleized.get(&key_a, &db).await.unwrap(), Some(val_a));
2151
2152        // Read this batch's key from the diff.
2153        assert_eq!(merkleized.get(&key_b, &db).await.unwrap(), Some(val_b));
2154
2155        // Nonexistent key.
2156        let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
2157        assert_eq!(merkleized.get(&key_c, &db).await.unwrap(), None);
2158
2159        db.destroy().await.unwrap();
2160    }
2161
2162    /// Independent sequential batches applied one at a time.
2163    #[boxed]
2164    pub(crate) async fn run_batch_sequential_apply<F: Family, V, C>(
2165        context: deterministic::Context,
2166        open_db: impl Fn(
2167            deterministic::Context,
2168        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2169    ) where
2170        V: ValueEncoding<Value = Digest>,
2171        C: Mutable<Item = Operation<F, Digest, V>>,
2172        C::Item: EncodeShared,
2173    {
2174        let db = open_db(context.child("db")).await;
2175
2176        let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
2177        let val_a = Sha256::fill(1u8);
2178
2179        // First batch.
2180        let m = db
2181            .new_batch()
2182            .set(key_a, val_a)
2183            .merkleize(&db, None, Location::new(0))
2184            .await;
2185        let root1 = m.root();
2186        let (db, _) = db.apply_batch(m).await.unwrap();
2187        assert_eq!(db.root(), root1);
2188        assert_eq!(db.get(&key_a).await.unwrap(), Some(val_a));
2189
2190        // Second independent batch.
2191        let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
2192        let val_b = Sha256::fill(2u8);
2193        let m = db
2194            .new_batch()
2195            .set(key_b, val_b)
2196            .merkleize(&db, None, Location::new(0))
2197            .await;
2198        let root2 = m.root();
2199        let (db, _) = db.apply_batch(m).await.unwrap();
2200        assert_eq!(db.root(), root2);
2201        assert_eq!(db.get(&key_b).await.unwrap(), Some(val_b));
2202
2203        db.destroy().await.unwrap();
2204    }
2205
2206    /// Many sequential batches accumulate correctly.
2207    #[boxed]
2208    pub(crate) async fn run_batch_many_sequential<F: Family, V, C>(
2209        context: deterministic::Context,
2210        open_db: impl Fn(
2211            deterministic::Context,
2212        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2213    ) where
2214        V: ValueEncoding<Value = Digest>,
2215        C: Mutable<Item = Operation<F, Digest, V>>,
2216        C::Item: EncodeShared,
2217    {
2218        let mut db = open_db(context.child("db")).await;
2219
2220        const BATCHES: u64 = 20;
2221        const KEYS_PER_BATCH: u64 = 5;
2222
2223        let mut all_kvs: Vec<(Digest, Digest)> = Vec::new();
2224
2225        for batch_idx in 0..BATCHES {
2226            let mut batch = db.new_batch();
2227            for j in 0..KEYS_PER_BATCH {
2228                let seed = batch_idx * 100 + j;
2229                let k = Sha256::hash(&[&seed.to_be_bytes()]);
2230                let v = Sha256::fill(seed as u8);
2231                batch = batch.set(k, v);
2232                all_kvs.push((k, v));
2233            }
2234            let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
2235            (db, _) = db.apply_batch(merkleized).await.unwrap();
2236        }
2237
2238        // Verify all key-values are readable.
2239        for (k, v) in &all_kvs {
2240            assert_eq!(db.get(k).await.unwrap(), Some(*v));
2241        }
2242
2243        // Verify proof over the full range.
2244        let root = db.root();
2245        let (proof, ops) = db.proof(Location::new(0), NZU64!(10000)).await.unwrap();
2246        assert!(verify_proof::<Sha256, _, _>(
2247            &proof,
2248            Location::new(0),
2249            &ops,
2250            &root
2251        ));
2252
2253        // Expected: 1 initial commit + BATCHES * (KEYS_PER_BATCH + 1 commit).
2254        let expected = 1 + BATCHES * (KEYS_PER_BATCH + 1);
2255        assert_eq!(db.bounds().end, expected);
2256
2257        db.destroy().await.unwrap();
2258    }
2259
2260    /// Empty batch (zero mutations) produces correct speculative root.
2261    #[boxed]
2262    pub(crate) async fn run_batch_empty_batch<F: Family, V, C>(
2263        context: deterministic::Context,
2264        open_db: impl Fn(
2265            deterministic::Context,
2266        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2267    ) where
2268        V: ValueEncoding<Value = Digest>,
2269        C: Mutable<Item = Operation<F, Digest, V>>,
2270        C::Item: EncodeShared,
2271    {
2272        let db = open_db(context.child("db")).await;
2273
2274        // Apply a non-empty batch first.
2275        let k = Sha256::hash(&[&[1u8]]);
2276        let merkleized = db
2277            .new_batch()
2278            .set(k, Sha256::fill(1u8))
2279            .merkleize(&db, None, Location::new(0))
2280            .await;
2281        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2282        let root_before = db.root();
2283        let size_before = db.bounds().end;
2284
2285        // Empty batch with no mutations.
2286        let merkleized = db.new_batch().merkleize(&db, None, Location::new(0)).await;
2287        let speculative = merkleized.root();
2288        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2289
2290        // Root changed (a new Commit op was appended).
2291        assert_ne!(db.root(), root_before);
2292        assert_eq!(db.root(), speculative);
2293        // Size grew by exactly 1 (the Commit op).
2294        assert_eq!(db.bounds().end, size_before + 1);
2295
2296        db.destroy().await.unwrap();
2297    }
2298
2299    /// MerkleizedBatch::get() works on a chained child's merkleized batch.
2300    #[boxed]
2301    pub(crate) async fn run_batch_chained_merkleized_get<F: Family, V, C>(
2302        context: deterministic::Context,
2303        open_db: impl Fn(
2304            deterministic::Context,
2305        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2306    ) where
2307        V: ValueEncoding<Value = Digest>,
2308        C: Mutable<Item = Operation<F, Digest, V>>,
2309        C::Item: EncodeShared,
2310    {
2311        let db = open_db(context.child("db")).await;
2312
2313        // Pre-populate base DB.
2314        let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
2315        let val_a = Sha256::fill(10u8);
2316        let merkleized = db
2317            .new_batch()
2318            .set(key_a, val_a)
2319            .merkleize(&db, None, Location::new(0))
2320            .await;
2321        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2322
2323        // Parent batch sets key B.
2324        let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
2325        let val_b = Sha256::fill(1u8);
2326        let parent_m = db
2327            .new_batch()
2328            .set(key_b, val_b)
2329            .merkleize(&db, None, Location::new(0))
2330            .await;
2331
2332        // Child batch sets key C.
2333        let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
2334        let val_c = Sha256::fill(2u8);
2335        let child_m = parent_m
2336            .new_batch::<Sha256>()
2337            .set(key_c, val_c)
2338            .merkleize(&db, None, Location::new(0))
2339            .await;
2340
2341        // Child's MerkleizedBatch can read all three layers:
2342        // base DB value
2343        assert_eq!(child_m.get(&key_a, &db).await.unwrap(), Some(val_a));
2344        // parent diff value
2345        assert_eq!(child_m.get(&key_b, &db).await.unwrap(), Some(val_b));
2346        // child's own value
2347        assert_eq!(child_m.get(&key_c, &db).await.unwrap(), Some(val_c));
2348        // nonexistent key
2349        let key_d = Sha256::hash(&[&3u64.to_be_bytes()]);
2350        assert_eq!(child_m.get(&key_d, &db).await.unwrap(), None);
2351
2352        db.destroy().await.unwrap();
2353    }
2354
2355    /// Large single batch, verifying all values and proof.
2356    #[boxed]
2357    pub(crate) async fn run_batch_large<F: Family, V, C>(
2358        context: deterministic::Context,
2359        open_db: impl Fn(
2360            deterministic::Context,
2361        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2362    ) where
2363        V: ValueEncoding<Value = Digest>,
2364        C: Mutable<Item = Operation<F, Digest, V>>,
2365        C::Item: EncodeShared,
2366    {
2367        let db = open_db(context.child("db")).await;
2368
2369        const N: u64 = 500;
2370        let mut kvs: Vec<(Digest, Digest)> = Vec::new();
2371
2372        let mut batch = db.new_batch();
2373        for i in 0..N {
2374            let k = Sha256::hash(&[&i.to_be_bytes()]);
2375            let v = Sha256::fill((i % 256) as u8);
2376            batch = batch.set(k, v);
2377            kvs.push((k, v));
2378        }
2379        let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
2380        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2381
2382        // Verify every value.
2383        for (k, v) in &kvs {
2384            assert_eq!(db.get(k).await.unwrap(), Some(*v));
2385        }
2386
2387        // Verify proof over the full range.
2388        let root = db.root();
2389        let (proof, ops) = db.proof(Location::new(0), NZU64!(1000)).await.unwrap();
2390        assert!(verify_proof::<Sha256, _, _>(
2391            &proof,
2392            Location::new(0),
2393            &ops,
2394            &root
2395        ));
2396
2397        // Expected: 1 initial commit + N sets + 1 commit.
2398        assert_eq!(db.bounds().end, 1 + N + 1);
2399
2400        db.destroy().await.unwrap();
2401    }
2402
2403    /// Child batch overrides same key set by parent.
2404    #[boxed]
2405    pub(crate) async fn run_batch_chained_key_override<F: Family, V, C>(
2406        context: deterministic::Context,
2407        open_db: impl Fn(
2408            deterministic::Context,
2409        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2410    ) where
2411        V: ValueEncoding<Value = Digest>,
2412        C: Mutable<Item = Operation<F, Digest, V>>,
2413        C::Item: EncodeShared,
2414    {
2415        let db = open_db(context.child("db")).await;
2416
2417        let key = Sha256::hash(&[&0u64.to_be_bytes()]);
2418        let val_parent = Sha256::fill(1u8);
2419        let val_child = Sha256::fill(2u8);
2420
2421        // Parent sets key.
2422        let parent_m = db
2423            .new_batch()
2424            .set(key, val_parent)
2425            .merkleize(&db, None, Location::new(0))
2426            .await;
2427
2428        // Child overrides same key.
2429        let mut child = parent_m.new_batch::<Sha256>();
2430        child = child.set(key, val_child);
2431
2432        // Child's pending mutation wins over parent diff.
2433        assert_eq!(child.get(&key, &db).await.unwrap(), Some(val_child));
2434
2435        let child_m = child.merkleize(&db, None, Location::new(0)).await;
2436
2437        // After merkleize, child's diff wins.
2438        assert_eq!(child_m.get(&key, &db).await.unwrap(), Some(val_child));
2439
2440        // Apply and verify.
2441        let (db, _) = db.apply_batch(child_m).await.unwrap();
2442        assert_eq!(db.get(&key).await.unwrap(), Some(val_child));
2443
2444        db.destroy().await.unwrap();
2445    }
2446
2447    /// Same key set across two sequential applied batches. This breaks the key-uniqueness
2448    /// invariant, so reads may return any of the written values. `get()` must still return one
2449    /// of them, live and across a restart, and after pruning every other version it returns the
2450    /// survivor. The prune check runs on a never-restarted db so the snapshot still holds both
2451    /// locations and `get()` must skip the pruned one within the bucket.
2452    ///
2453    /// `open_db_small_sections` must return a DB whose log has `items_per_section=1`
2454    /// so pruning is per-item.
2455    #[boxed]
2456    pub(crate) async fn run_batch_sequential_key_override<F: Family, V, C>(
2457        context: deterministic::Context,
2458        open_db_small_sections: impl Fn(
2459            deterministic::Context,
2460        )
2461            -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2462    ) where
2463        V: ValueEncoding<Value = Digest>,
2464        C: Mutable<Item = Operation<F, Digest, V>>,
2465        C::Item: EncodeShared,
2466    {
2467        let db = open_db_small_sections(context.child("db")).await;
2468
2469        let key = Sha256::hash(&[&0u64.to_be_bytes()]);
2470        let v1 = Sha256::fill(1u8);
2471        let v2 = Sha256::fill(2u8);
2472
2473        // First batch sets key.
2474        // Layout: 0=initial commit, 1=Set(key,v1), 2=Commit
2475        let merkleized = db
2476            .new_batch()
2477            .set(key, v1)
2478            .merkleize(&db, None, Location::new(0))
2479            .await;
2480        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2481        assert_eq!(db.get(&key).await.unwrap(), Some(v1));
2482
2483        // Second batch sets same key to different value.
2484        // Layout continues: 3=Set(key,v2), 4=Commit
2485        let merkleized = db
2486            .new_batch()
2487            .set(key, v2)
2488            .merkleize(&db, None, Location::new(0))
2489            .await;
2490        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2491
2492        // Either written value may be served for the repeated key.
2493        let live = db.get(&key).await.unwrap().unwrap();
2494        assert!(live == v1 || live == v2);
2495
2496        // A restart must also serve one of the written values.
2497        db.commit().await.unwrap();
2498        let db = open_db_small_sections(context.child("reopen")).await;
2499        let reopened = db.get(&key).await.unwrap().unwrap();
2500        assert!(reopened == v1 || reopened == v2);
2501        db.destroy().await.unwrap();
2502
2503        // Rebuild the same history on a fresh db without restarting, so the
2504        // snapshot bucket holds both locations. Floor=4 permits prune(2).
2505        // Layout: 0=initial commit, 1=Set(key,v1), 2=Commit, 3=Set(key,v2),
2506        // 4=Commit(floor=4)
2507        let db = open_db_small_sections(context.child("prune")).await;
2508        let merkleized = db
2509            .new_batch()
2510            .set(key, v1)
2511            .merkleize(&db, None, Location::new(0))
2512            .await;
2513        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2514        let merkleized = db
2515            .new_batch()
2516            .set(key, v2)
2517            .merkleize(&db, None, Location::new(4))
2518            .await;
2519        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2520
2521        // Prune past the first Set (loc 1). With items_per_section=1, pruning
2522        // to loc 2 removes the blob containing loc 1. get() must skip the
2523        // pruned location within the bucket and serve the survivor.
2524        let db = db.prune(Location::new(2)).await.unwrap();
2525        assert_eq!(db.get(&key).await.unwrap(), Some(v2));
2526
2527        db.destroy().await.unwrap();
2528    }
2529
2530    /// Metadata propagates through merkleize and clears with None.
2531    #[boxed]
2532    pub(crate) async fn run_batch_metadata<F: Family, V, C>(
2533        context: deterministic::Context,
2534        open_db: impl Fn(
2535            deterministic::Context,
2536        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2537    ) where
2538        V: ValueEncoding<Value = Digest>,
2539        C: Mutable<Item = Operation<F, Digest, V>>,
2540        C::Item: EncodeShared,
2541    {
2542        let db = open_db(context.child("db")).await;
2543
2544        // Batch with metadata.
2545        let metadata = Sha256::fill(42u8);
2546        let k = Sha256::hash(&[&[1u8]]);
2547        let merkleized = db
2548            .new_batch()
2549            .set(k, Sha256::fill(1u8))
2550            .merkleize(&db, Some(metadata), Location::new(0))
2551            .await;
2552        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2553        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
2554
2555        // Second batch clears metadata.
2556        let merkleized = db.new_batch().merkleize(&db, None, Location::new(0)).await;
2557        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2558        assert_eq!(db.get_metadata().await.unwrap(), None);
2559
2560        db.destroy().await.unwrap();
2561    }
2562
2563    #[boxed]
2564    pub(crate) async fn run_stale_batch_rejected<F: Family, V, C>(
2565        context: deterministic::Context,
2566        open_db: impl Fn(
2567            deterministic::Context,
2568        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2569    ) where
2570        V: ValueEncoding<Value = Digest>,
2571        C: Mutable<Item = Operation<F, Digest, V>>,
2572        C::Item: EncodeShared,
2573    {
2574        let db = open_db(context.child("db")).await;
2575
2576        let key1 = Sha256::hash(&[&[1]]);
2577        let key2 = Sha256::hash(&[&[2]]);
2578        let v1 = Sha256::fill(10u8);
2579        let v2 = Sha256::fill(20u8);
2580
2581        // Create two batches from the same DB state.
2582        let batch_a = db
2583            .new_batch()
2584            .set(key1, v1)
2585            .merkleize(&db, None, Location::new(0))
2586            .await;
2587        let batch_b = db
2588            .new_batch()
2589            .set(key2, v2)
2590            .merkleize(&db, None, Location::new(0))
2591            .await;
2592
2593        // Apply the first and commit it -- should succeed.
2594        let (db, _) = db.apply_batch(batch_a).await.unwrap();
2595        assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2596        assert_eq!(db.get(&key2).await.unwrap(), None);
2597        assert_eq!(db.get_metadata().await.unwrap(), None);
2598        let db = db.commit().await.unwrap();
2599        let root = db.root();
2600        let size = db.size();
2601
2602        // Apply the second -- should fail because the DB was modified.
2603        let result = db.apply_batch(batch_b).await;
2604        assert!(matches!(result, Err(Error::StaleBatch)));
2605
2606        // The rejection mutated nothing: reopening recovers the committed state.
2607        let db = open_db(context.child("reopen")).await;
2608        assert_eq!(db.root(), root);
2609        assert_eq!(db.size(), size);
2610        assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2611        assert_eq!(db.get(&key2).await.unwrap(), None);
2612        db.destroy().await.unwrap();
2613    }
2614
2615    #[boxed]
2616    pub(crate) async fn run_stale_batch_chained<F: Family, V, C>(
2617        context: deterministic::Context,
2618        open_db: impl Fn(
2619            deterministic::Context,
2620        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2621    ) where
2622        V: ValueEncoding<Value = Digest>,
2623        C: Mutable<Item = Operation<F, Digest, V>>,
2624        C::Item: EncodeShared,
2625    {
2626        let db = open_db(context.child("db")).await;
2627
2628        let key1 = Sha256::hash(&[&[1]]);
2629        let key2 = Sha256::hash(&[&[2]]);
2630        let key3 = Sha256::hash(&[&[3]]);
2631
2632        let common_parent = db
2633            .new_batch()
2634            .set(Sha256::hash(&[&[10]]), Sha256::fill(10u8))
2635            .merkleize(&db, None, Location::new(0))
2636            .await;
2637        let sibling_a = common_parent
2638            .new_batch::<Sha256>()
2639            .set(Sha256::hash(&[&[11]]), Sha256::fill(11u8))
2640            .merkleize(&db, None, Location::new(0))
2641            .await;
2642        let sibling_b = common_parent
2643            .new_batch::<Sha256>()
2644            .set(Sha256::hash(&[&[12]]), Sha256::fill(12u8))
2645            .merkleize(&db, None, Location::new(0))
2646            .await;
2647        let (db, _) = db.apply_batch(sibling_a).await.unwrap();
2648        assert!(matches!(
2649            db.validate_batch(&sibling_b),
2650            Err(Error::StaleBatch)
2651        ));
2652
2653        // Build equal-size sibling parents, then extend only one sibling.
2654        let parent_a = db
2655            .new_batch()
2656            .set(key1, Sha256::fill(1u8))
2657            .merkleize(&db, None, Location::new(0))
2658            .await;
2659        let parent_b = db
2660            .new_batch()
2661            .set(key2, Sha256::fill(2u8))
2662            .merkleize(&db, None, Location::new(0))
2663            .await;
2664        let child_b = parent_b
2665            .new_batch::<Sha256>()
2666            .set(key3, Sha256::fill(3u8))
2667            .merkleize(&db, None, Location::new(0))
2668            .await;
2669
2670        let (db, _) = db.apply_batch(parent_a).await.unwrap();
2671        assert!(matches!(
2672            db.validate_batch(&child_b),
2673            Err(Error::StaleBatch)
2674        ));
2675        db.destroy().await.unwrap();
2676    }
2677
2678    #[boxed]
2679    pub(crate) async fn run_partial_ancestor_commit<F: Family, V, C>(
2680        context: deterministic::Context,
2681        open_db: impl Fn(
2682            deterministic::Context,
2683        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2684    ) where
2685        V: ValueEncoding<Value = Digest>,
2686        C: Mutable<Item = Operation<F, Digest, V>>,
2687        C::Item: EncodeShared,
2688    {
2689        let db = open_db(context.child("db")).await;
2690
2691        let key1 = Sha256::hash(&[&[1]]);
2692        let key2 = Sha256::hash(&[&[2]]);
2693        let key3 = Sha256::hash(&[&[3]]);
2694        let v1 = Sha256::fill(1u8);
2695        let v2 = Sha256::fill(2u8);
2696        let v3 = Sha256::fill(3u8);
2697
2698        // Chain: DB <- A <- B <- C
2699        let a = db
2700            .new_batch()
2701            .set(key1, v1)
2702            .merkleize(&db, None, Location::new(0))
2703            .await;
2704        let b = a
2705            .new_batch::<Sha256>()
2706            .set(key2, v2)
2707            .merkleize(&db, None, Location::new(0))
2708            .await;
2709        let c = b
2710            .new_batch::<Sha256>()
2711            .set(key3, v3)
2712            .merkleize(&db, None, Location::new(0))
2713            .await;
2714
2715        let expected_root = c.root();
2716
2717        // Apply only A, then apply C directly (B uncommitted).
2718        let (db, _) = db.apply_batch(a).await.unwrap();
2719        let (db, _) = db.apply_batch(c).await.unwrap();
2720
2721        assert_eq!(db.root(), expected_root);
2722        assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2723        assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2724        assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
2725
2726        db.destroy().await.unwrap();
2727    }
2728
2729    #[boxed]
2730    pub(crate) async fn run_delayed_merkleize_after_ancestor_apply<F: Family, V, C>(
2731        context: deterministic::Context,
2732        open_db: impl Fn(
2733            deterministic::Context,
2734        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2735    ) where
2736        V: ValueEncoding<Value = Digest>,
2737        C: Mutable<Item = Operation<F, Digest, V>>,
2738        C::Item: EncodeShared,
2739    {
2740        let db = open_db(context.child("db")).await;
2741
2742        let key1 = Sha256::hash(&[&[1]]);
2743        let key2 = Sha256::hash(&[&[2]]);
2744        let key3 = Sha256::hash(&[&[3]]);
2745        let v1 = Sha256::fill(1u8);
2746        let v2 = Sha256::fill(2u8);
2747        let v3 = Sha256::fill(3u8);
2748
2749        let a = db
2750            .new_batch()
2751            .set(key1, v1)
2752            .merkleize(&db, None, Location::new(0))
2753            .await;
2754        let b = a
2755            .new_batch::<Sha256>()
2756            .set(key2, v2)
2757            .merkleize(&db, None, Location::new(0))
2758            .await;
2759        let c = b.new_batch::<Sha256>().set(key3, v3);
2760
2761        let (db, _) = db.apply_batch(a).await.unwrap();
2762        let c = c.merkleize(&db, None, Location::new(0)).await;
2763        let expected_root = c.root();
2764        let (db, _) = db.apply_batch(c).await.unwrap();
2765
2766        assert_eq!(db.root(), expected_root);
2767        assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2768        assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2769        assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
2770
2771        db.destroy().await.unwrap();
2772    }
2773
2774    #[boxed]
2775    pub(crate) async fn run_sequential_commit_parent_then_child<F: Family, V, C>(
2776        context: deterministic::Context,
2777        open_db: impl Fn(
2778            deterministic::Context,
2779        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2780    ) where
2781        V: ValueEncoding<Value = Digest>,
2782        C: Mutable<Item = Operation<F, Digest, V>>,
2783        C::Item: EncodeShared,
2784    {
2785        let db = open_db(context.child("db")).await;
2786
2787        let key1 = Sha256::hash(&[&[1]]);
2788        let key2 = Sha256::hash(&[&[2]]);
2789        let v1 = Sha256::fill(1u8);
2790        let v2 = Sha256::fill(2u8);
2791
2792        // Parent batch.
2793        let parent_m = db
2794            .new_batch()
2795            .set(key1, v1)
2796            .merkleize(&db, None, Location::new(0))
2797            .await;
2798
2799        // Child batch built on parent.
2800        let child_m = parent_m
2801            .new_batch::<Sha256>()
2802            .set(key2, v2)
2803            .merkleize(&db, None, Location::new(0))
2804            .await;
2805
2806        // Apply parent first, then child. This is a valid sequential commit.
2807        let (db, _) = db.apply_batch(parent_m).await.unwrap();
2808        let (db, _) = db.apply_batch(child_m).await.unwrap();
2809
2810        // Both keys present.
2811        assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2812        assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2813
2814        db.destroy().await.unwrap();
2815    }
2816
2817    #[boxed]
2818    pub(crate) async fn run_child_root_matches_pending_and_committed<F: Family, V, C>(
2819        context: deterministic::Context,
2820        open_db: impl Fn(
2821            deterministic::Context,
2822        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2823    ) where
2824        V: ValueEncoding<Value = Digest>,
2825        C: Mutable<Item = Operation<F, Digest, V>>,
2826        C::Item: EncodeShared,
2827    {
2828        let db = open_db(context.child("db")).await;
2829
2830        let key1 = Sha256::hash(&[&[1]]);
2831        let key2 = Sha256::hash(&[&[2]]);
2832
2833        // Build the child while the parent is still pending.
2834        let parent = db
2835            .new_batch()
2836            .set(key1, Sha256::fill(1u8))
2837            .merkleize(&db, None, Location::new(0))
2838            .await;
2839        let pending_child = parent
2840            .new_batch::<Sha256>()
2841            .set(key2, Sha256::fill(2u8))
2842            .merkleize(&db, None, Location::new(0))
2843            .await;
2844
2845        // Commit the parent, then rebuild the same logical child from the
2846        // committed DB state and compare roots.
2847        let (db, _) = db.apply_batch(parent).await.unwrap();
2848        let db = db.commit().await.unwrap();
2849
2850        let committed_child = db
2851            .new_batch()
2852            .set(key2, Sha256::fill(2u8))
2853            .merkleize(&db, None, Location::new(0))
2854            .await;
2855
2856        assert_eq!(pending_child.root(), committed_child.root());
2857
2858        db.destroy().await.unwrap();
2859    }
2860
2861    #[boxed]
2862    pub(crate) async fn run_stale_batch_child_applied_before_parent<F: Family, V, C>(
2863        context: deterministic::Context,
2864        open_db: impl Fn(
2865            deterministic::Context,
2866        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2867    ) where
2868        V: ValueEncoding<Value = Digest>,
2869        C: Mutable<Item = Operation<F, Digest, V>>,
2870        C::Item: EncodeShared,
2871    {
2872        let db = open_db(context.child("db")).await;
2873
2874        let key1 = Sha256::hash(&[&[1]]);
2875        let key2 = Sha256::hash(&[&[2]]);
2876
2877        // Parent batch.
2878        let parent_m = db
2879            .new_batch()
2880            .set(key1, Sha256::fill(1u8))
2881            .merkleize(&db, None, Location::new(0))
2882            .await;
2883
2884        // Child batch.
2885        let child_m = parent_m
2886            .new_batch::<Sha256>()
2887            .set(key2, Sha256::fill(2u8))
2888            .merkleize(&db, None, Location::new(0))
2889            .await;
2890
2891        // Apply child first (it carries all parent ops too).
2892        let (db, _) = db.apply_batch(child_m).await.unwrap();
2893
2894        // Parent is stale.
2895        let result = db.apply_batch(parent_m).await;
2896        assert!(matches!(result, Err(Error::StaleBatch)));
2897    }
2898
2899    /// to_batch() creates an owned snapshot whose root matches the committed DB.
2900    /// A child batch chained from it can be applied.
2901    #[boxed]
2902    pub(crate) async fn run_to_batch<F: Family, V, C>(
2903        context: deterministic::Context,
2904        open_db: impl Fn(
2905            deterministic::Context,
2906        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2907    ) where
2908        V: ValueEncoding<Value = Digest>,
2909        C: Mutable<Item = Operation<F, Digest, V>>,
2910        C::Item: EncodeShared,
2911    {
2912        let db = open_db(context.child("db")).await;
2913
2914        // Populate.
2915        let key1 = Sha256::hash(&[&[1]]);
2916        let v1 = Sha256::fill(10u8);
2917        let merkleized = db
2918            .new_batch()
2919            .set(key1, v1)
2920            .merkleize(&db, None, Location::new(0))
2921            .await;
2922        let (db, _) = db.apply_batch(merkleized).await.unwrap();
2923
2924        // to_batch root matches committed root.
2925        let snapshot = db.to_batch();
2926        assert_eq!(snapshot.root(), db.root());
2927
2928        // Chain a child from the snapshot, apply it.
2929        let key2 = Sha256::hash(&[&[2]]);
2930        let v2 = Sha256::fill(20u8);
2931        let child = snapshot
2932            .new_batch::<Sha256>()
2933            .set(key2, v2)
2934            .merkleize(&db, None, Location::new(0))
2935            .await;
2936        let (db, _) = db.apply_batch(child).await.unwrap();
2937
2938        assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2939        assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2940
2941        db.destroy().await.unwrap();
2942    }
2943
2944    /// Regression: applying a batch after its ancestor Arc is dropped (without
2945    /// committing) must still apply the ancestor's snapshot diffs.
2946    #[boxed]
2947    pub(crate) async fn run_apply_after_ancestor_dropped<F: Family, V, C>(
2948        context: deterministic::Context,
2949        open_db: impl Fn(
2950            deterministic::Context,
2951        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2952    ) where
2953        V: ValueEncoding<Value = Digest>,
2954        C: Mutable<Item = Operation<F, Digest, V>>,
2955        C::Item: EncodeShared,
2956    {
2957        let db = open_db(context.child("db")).await;
2958
2959        let key1 = Sha256::hash(&[&[1]]);
2960        let key2 = Sha256::hash(&[&[2]]);
2961        let key3 = Sha256::hash(&[&[3]]);
2962        let v1 = Sha256::fill(1u8);
2963        let v2 = Sha256::fill(2u8);
2964        let v3 = Sha256::fill(3u8);
2965
2966        // Chain: DB <- A <- B <- C
2967        let a = db
2968            .new_batch()
2969            .set(key1, v1)
2970            .merkleize(&db, None, Location::new(0))
2971            .await;
2972        let b = a
2973            .new_batch::<Sha256>()
2974            .set(key2, v2)
2975            .merkleize(&db, None, Location::new(0))
2976            .await;
2977        let c = b
2978            .new_batch::<Sha256>()
2979            .set(key3, v3)
2980            .merkleize(&db, None, Location::new(0))
2981            .await;
2982
2983        // Drop A and B without committing. Their Weak refs in C are now dead.
2984        drop(a);
2985        drop(b);
2986
2987        // Apply only the tip. This is !skip_ancestors (DB hasn't changed).
2988        let (db, _) = db.apply_batch(c).await.unwrap();
2989
2990        // All three keys must be in the snapshot.
2991        assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2992        assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2993        assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
2994
2995        db.destroy().await.unwrap();
2996    }
2997
2998    /// Verify the inactivity floor is zero for a fresh empty database and is
2999    /// correctly set after applying batches with specific floor values.
3000    #[boxed]
3001    pub(crate) async fn run_inactivity_floor_tracking<F: Family, V, C>(
3002        context: deterministic::Context,
3003        open_db: impl Fn(
3004            deterministic::Context,
3005        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3006    ) where
3007        V: ValueEncoding<Value = Digest>,
3008        C: Mutable<Item = Operation<F, Digest, V>>,
3009        C::Item: EncodeShared,
3010    {
3011        let db = open_db(context.child("test")).await;
3012
3013        // Empty DB has floor=0.
3014        assert_eq!(db.inactivity_floor_loc(), Location::new(0));
3015
3016        // Apply batch with floor=0, floor stays 0.
3017        let k1 = Sha256::fill(1u8);
3018        let v1 = Sha256::fill(2u8);
3019        let merkleized = db
3020            .new_batch()
3021            .set(k1, v1)
3022            .merkleize(&db, None, Location::new(0))
3023            .await;
3024        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3025        assert_eq!(db.inactivity_floor_loc(), Location::new(0));
3026
3027        // Apply batch with floor=3, floor advances.
3028        let k2 = Sha256::fill(3u8);
3029        let v2 = Sha256::fill(4u8);
3030        let merkleized = db
3031            .new_batch()
3032            .set(k2, v2)
3033            .merkleize(&db, None, Location::new(3))
3034            .await;
3035        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3036        assert_eq!(db.inactivity_floor_loc(), Location::new(3));
3037
3038        // Floor persists across restart.
3039        let db = db.commit().await.unwrap();
3040        db.sync().await.unwrap();
3041        let db = open_db(context.child("reopen")).await;
3042        assert_eq!(db.inactivity_floor_loc(), Location::new(3));
3043
3044        db.destroy().await.unwrap();
3045    }
3046
3047    /// Verify that applying a batch with a floor equal to the current floor succeeds,
3048    /// and that a higher floor also succeeds.
3049    #[boxed]
3050    pub(crate) async fn run_floor_monotonicity<F: Family, V, C>(
3051        context: deterministic::Context,
3052        open_db: impl Fn(
3053            deterministic::Context,
3054        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3055    ) where
3056        V: ValueEncoding<Value = Digest>,
3057        C: Mutable<Item = Operation<F, Digest, V>>,
3058        C::Item: EncodeShared,
3059    {
3060        let db = open_db(context.child("test")).await;
3061
3062        // DB starts with 1 op (initial commit).
3063        // First batch: 1 set + 1 commit = total_size 3. Use floor=2 (the commit loc).
3064        let k1 = Sha256::fill(1u8);
3065        let v1 = Sha256::fill(2u8);
3066        let merkleized = db
3067            .new_batch()
3068            .set(k1, v1)
3069            .merkleize(&db, None, Location::new(2))
3070            .await;
3071        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3072        assert_eq!(db.inactivity_floor_loc(), Location::new(2));
3073
3074        // Same floor is OK. Second batch: 1 set + 1 commit = total_size 5. floor=2 < 5.
3075        let k2 = Sha256::fill(3u8);
3076        let v2 = Sha256::fill(4u8);
3077        let merkleized = db
3078            .new_batch()
3079            .set(k2, v2)
3080            .merkleize(&db, None, Location::new(2))
3081            .await;
3082        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3083        assert_eq!(db.inactivity_floor_loc(), Location::new(2));
3084
3085        // Higher floor also succeeds. Third batch: 1 set + 1 commit = total_size 7. floor=5 < 7.
3086        let k3 = Sha256::fill(5u8);
3087        let v3 = Sha256::fill(6u8);
3088        let merkleized = db
3089            .new_batch()
3090            .set(k3, v3)
3091            .merkleize(&db, None, Location::new(5))
3092            .await;
3093        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3094        assert_eq!(db.inactivity_floor_loc(), Location::new(5));
3095
3096        db.destroy().await.unwrap();
3097    }
3098
3099    /// Verify that the inactivity floor is correctly restored after a rewind.
3100    #[boxed]
3101    pub(crate) async fn run_rewind_restores_floor<F: Family, V, C>(
3102        context: deterministic::Context,
3103        open_db: impl Fn(
3104            deterministic::Context,
3105        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3106    ) where
3107        V: ValueEncoding<Value = Digest>,
3108        C: Mutable<Item = Operation<F, Digest, V>>,
3109        C::Item: EncodeShared,
3110    {
3111        let db = open_db(context.child("test")).await;
3112
3113        // Apply first batch with floor=2.
3114        let k1 = Sha256::fill(1u8);
3115        let v1 = Sha256::fill(2u8);
3116        let merkleized = db
3117            .new_batch()
3118            .set(k1, v1)
3119            .merkleize(&db, None, Location::new(2))
3120            .await;
3121        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3122        let db = db.commit().await.unwrap();
3123        let first_size = db.bounds().end;
3124        assert_eq!(db.inactivity_floor_loc(), Location::new(2));
3125
3126        // Apply second batch with floor=4 (the new commit's location).
3127        let k2 = Sha256::fill(3u8);
3128        let v2 = Sha256::fill(4u8);
3129        let merkleized = db
3130            .new_batch()
3131            .set(k2, v2)
3132            .merkleize(&db, None, Location::new(4))
3133            .await;
3134        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3135        let db = db.commit().await.unwrap();
3136        assert_eq!(db.inactivity_floor_loc(), Location::new(4));
3137
3138        // Rewind to the first batch.
3139        let db = db.rewind(first_size).await.unwrap();
3140        assert_eq!(db.inactivity_floor_loc(), Location::new(2));
3141
3142        db.destroy().await.unwrap();
3143    }
3144
3145    /// Verify that applying a batch with a floor lower than the current floor
3146    /// returns an error.
3147    #[boxed]
3148    pub(crate) async fn run_floor_monotonicity_violation<F: Family, V, C>(
3149        context: deterministic::Context,
3150        open_db: impl Fn(
3151            deterministic::Context,
3152        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3153    ) where
3154        V: ValueEncoding<Value = Digest>,
3155        C: Mutable<Item = Operation<F, Digest, V>>,
3156        C::Item: EncodeShared,
3157    {
3158        let db = open_db(context.child("test")).await;
3159
3160        // DB starts with 1 op. First batch: 1 set + 1 commit = total_size 3. floor=2.
3161        let k1 = Sha256::fill(1u8);
3162        let v1 = Sha256::fill(2u8);
3163        let merkleized = db
3164            .new_batch()
3165            .set(k1, v1)
3166            .merkleize(&db, None, Location::new(2))
3167            .await;
3168        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3169
3170        // Apply batch with floor=1 (regression). Should return an error.
3171        let k2 = Sha256::fill(3u8);
3172        let v2 = Sha256::fill(4u8);
3173        let merkleized = db
3174            .new_batch()
3175            .set(k2, v2)
3176            .merkleize(&db, None, Location::new(1))
3177            .await;
3178        let result = db.apply_batch(merkleized).await;
3179        assert!(matches!(result, Err(Error::FloorRegressed(new, current))
3180                if new == Location::new(1) && current == Location::new(2)));
3181    }
3182
3183    /// Verify that applying a batch with a floor beyond the total operation
3184    /// count returns an error.
3185    #[boxed]
3186    pub(crate) async fn run_floor_beyond_size<F: Family, V, C>(
3187        context: deterministic::Context,
3188        open_db: impl Fn(
3189            deterministic::Context,
3190        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3191    ) where
3192        V: ValueEncoding<Value = Digest>,
3193        C: Mutable<Item = Operation<F, Digest, V>>,
3194        C::Item: EncodeShared,
3195    {
3196        let db = open_db(context.child("test")).await;
3197
3198        // DB has 1 op (initial commit). A batch with 1 set + 1 commit = total_size 3.
3199        // Setting floor=100 exceeds total_size.
3200        let k1 = Sha256::fill(1u8);
3201        let v1 = Sha256::fill(2u8);
3202        let merkleized = db
3203            .new_batch()
3204            .set(k1, v1)
3205            .merkleize(&db, None, Location::new(100))
3206            .await;
3207        let result = db.apply_batch(merkleized).await;
3208        assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit))
3209                if floor == Location::new(100) && commit == Location::new(2)));
3210
3211        let db = open_db(context.child("test")).await;
3212
3213        // Boundary: floor == total_size must also be rejected. The commit op is
3214        // at total_size - 1, so a floor equal to total_size would allow a later
3215        // prune to remove the commit and leave the db unrecoverable.
3216        let k2 = Sha256::fill(3u8);
3217        let v2 = Sha256::fill(4u8);
3218        let merkleized = db
3219            .new_batch()
3220            .set(k2, v2)
3221            .merkleize(&db, None, Location::new(3))
3222            .await;
3223        let result = db.apply_batch(merkleized).await;
3224        assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit))
3225                if floor == Location::new(3) && commit == Location::new(2)));
3226
3227        // Floor == total_size - 1 (the commit location) is the maximum valid.
3228        let db = open_db(context.child("test")).await;
3229        let merkleized = db
3230            .new_batch()
3231            .set(k2, v2)
3232            .merkleize(&db, None, Location::new(2))
3233            .await;
3234        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3235
3236        db.destroy().await.unwrap();
3237    }
3238
3239    /// A chained batch where an *ancestor* declares a floor below the previous ancestor's
3240    /// floor must be rejected, even when that ancestor's floor is still at or above the
3241    /// database's current floor. This isolates the cross-batch monotonicity step (every
3242    /// commit's floor at or above the previous commit's floor) from the simpler "every
3243    /// commit's floor at or above the live floor" rule.
3244    #[boxed]
3245    pub(crate) async fn run_chained_ancestor_floor_regression<F: Family, V, C>(
3246        context: deterministic::Context,
3247        open_db: impl Fn(
3248            deterministic::Context,
3249        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3250    ) where
3251        V: ValueEncoding<Value = Digest>,
3252        C: Mutable<Item = Operation<F, Digest, V>>,
3253        C::Item: EncodeShared,
3254    {
3255        let db = open_db(context.child("test")).await;
3256
3257        // Live floor is 0 (from the seeded initial commit).
3258        // a: 1 set + commit at loc 2, floor=2 (valid: >= 0, == commit_loc).
3259        // b: 1 set + commit at loc 4, floor=1 (regresses below a's floor=2, but still >= 0).
3260        // c: 1 set + commit at loc 6, floor=2 (would be valid in isolation).
3261        // Applying c must fail at b with FloorRegressed(1, 2) -- not at b vs the live floor,
3262        // and not at c, proving the per-ancestor running floor is what catches it.
3263        let a = db
3264            .new_batch()
3265            .set(Sha256::fill(1u8), Sha256::fill(2u8))
3266            .merkleize(&db, None, Location::new(2))
3267            .await;
3268        let b = a
3269            .new_batch::<Sha256>()
3270            .set(Sha256::fill(3u8), Sha256::fill(4u8))
3271            .merkleize(&db, None, Location::new(1))
3272            .await;
3273        let c = b
3274            .new_batch::<Sha256>()
3275            .set(Sha256::fill(5u8), Sha256::fill(6u8))
3276            .merkleize(&db, None, Location::new(2))
3277            .await;
3278
3279        let root_before = db.root();
3280        let last_commit_before = db.last_commit_loc;
3281        let floor_before = db.inactivity_floor_loc();
3282
3283        let Err(err) = db.apply_batch(c).await else {
3284            panic!("expected apply_batch to fail");
3285        };
3286        assert!(
3287            matches!(err, Error::FloorRegressed(new, prev)
3288                if new == Location::new(1) && prev == Location::new(2)),
3289            "unexpected error: {err:?}"
3290        );
3291
3292        // Reopen the partition and verify the rejected chain persisted nothing.
3293        let db = open_db(context.child("test")).await;
3294        assert_eq!(db.root(), root_before);
3295        assert_eq!(db.last_commit_loc, last_commit_before);
3296        assert_eq!(db.inactivity_floor_loc(), floor_before);
3297
3298        db.destroy().await.unwrap();
3299    }
3300
3301    /// A chained batch where an *ancestor's* floor exceeds its own commit location must be
3302    /// rejected, identifying the ancestor's commit_loc (not the tip's). This is the more
3303    /// dangerous variant: monotonicity can still be satisfied while the floor poisons future
3304    /// `historical_proof` and rewind.
3305    #[boxed]
3306    pub(crate) async fn run_chained_ancestor_floor_beyond_size<F: Family, V, C>(
3307        context: deterministic::Context,
3308        open_db: impl Fn(
3309            deterministic::Context,
3310        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3311    ) where
3312        V: ValueEncoding<Value = Digest>,
3313        C: Mutable<Item = Operation<F, Digest, V>>,
3314        C::Item: EncodeShared,
3315    {
3316        let db = open_db(context.child("test")).await;
3317
3318        // a: 1 set + commit at loc 2; declare floor=3 (one past the commit -- invalid).
3319        // b: tip valid on its own (floor=0 <= b's commit_loc), but a's floor is bad.
3320        let a = db
3321            .new_batch()
3322            .set(Sha256::fill(1u8), Sha256::fill(2u8))
3323            .merkleize(&db, None, Location::new(3))
3324            .await;
3325        let b = a
3326            .new_batch::<Sha256>()
3327            .set(Sha256::fill(3u8), Sha256::fill(4u8))
3328            .merkleize(&db, None, Location::new(0))
3329            .await;
3330
3331        let root_before = db.root();
3332        let last_commit_before = db.last_commit_loc;
3333        let floor_before = db.inactivity_floor_loc();
3334
3335        let Err(err) = db.apply_batch(b).await else {
3336            panic!("expected apply_batch to fail");
3337        };
3338        // The error must identify the ancestor's commit_loc (2), not the tip's (4).
3339        assert!(
3340            matches!(err, Error::FloorBeyondSize(floor, commit)
3341                if floor == Location::new(3) && commit == Location::new(2)),
3342            "unexpected error: {err:?}"
3343        );
3344
3345        // Reopen the partition and verify the rejected chain persisted nothing.
3346        let db = open_db(context.child("test")).await;
3347        assert_eq!(db.root(), root_before);
3348        assert_eq!(db.last_commit_loc, last_commit_before);
3349        assert_eq!(db.inactivity_floor_loc(), floor_before);
3350
3351        db.destroy().await.unwrap();
3352    }
3353
3354    /// Regression test for rewind-after-reopen with floor change.
3355    ///
3356    /// After reopening a database (which rebuilds the snapshot from the latest
3357    /// floor), rewinding to an earlier commit with a lower floor must restore
3358    /// all keys that were live at the rewind target -- not just the ones that
3359    /// happened to be in the rebuilt snapshot.
3360    #[boxed]
3361    pub(crate) async fn run_rewind_after_reopen_with_floor_change<F: Family, V, C>(
3362        context: deterministic::Context,
3363        open_db: impl Fn(
3364            deterministic::Context,
3365        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3366    ) where
3367        V: ValueEncoding<Value = Digest>,
3368        C: Mutable<Item = Operation<F, Digest, V>>,
3369        C::Item: EncodeShared,
3370    {
3371        let db = open_db(context.child("first")).await;
3372
3373        let k1 = Sha256::fill(1u8);
3374        let k2 = Sha256::fill(2u8);
3375        let k3 = Sha256::fill(3u8);
3376        let v1 = Sha256::fill(11u8);
3377        let v2 = Sha256::fill(12u8);
3378        let v3 = Sha256::fill(13u8);
3379
3380        // Commit A: 3 keys with floor=0.
3381        let (db, _) = commit_sets(db, [(k1, v1), (k2, v2), (k3, v3)], None).await;
3382        let first_size = db.bounds().end;
3383        let first_root = db.root();
3384
3385        // Commit B: 3 more keys with floor=first_size (declares batch A inactive).
3386        let k4 = Sha256::fill(4u8);
3387        let k5 = Sha256::fill(5u8);
3388        let k6 = Sha256::fill(6u8);
3389        let v4 = Sha256::fill(14u8);
3390        let v5 = Sha256::fill(15u8);
3391        let v6 = Sha256::fill(16u8);
3392        let (db, _) =
3393            commit_sets_with_floor(db, [(k4, v4), (k5, v5), (k6, v6)], None, first_size).await;
3394        db.sync().await.unwrap();
3395
3396        // Reopen: snapshot rebuilt from floor=first_size, batch A keys excluded.
3397        let db = open_db(context.child("second")).await;
3398
3399        // Verify batch A keys are NOT in the reopened snapshot (expected).
3400        assert!(db.get(&k1).await.unwrap().is_none());
3401
3402        // Rewind to commit A.
3403        let db = db.rewind(first_size).await.unwrap();
3404
3405        // All batch A keys must be accessible after rewind.
3406        assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
3407        assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
3408        assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3409        assert_eq!(db.root(), first_root);
3410        assert_eq!(db.inactivity_floor_loc(), Location::new(0));
3411
3412        // Batch B keys must NOT be accessible.
3413        assert!(db.get(&k4).await.unwrap().is_none());
3414
3415        db.destroy().await.unwrap();
3416    }
3417
3418    /// Regression test: rewind-after-reopen where the rewind target is NOT the
3419    /// immediate predecessor. This ensures the snapshot gap fill only covers
3420    /// [rewind_floor, old_floor) and does not re-insert keys already present.
3421    #[boxed]
3422    pub(crate) async fn run_rewind_after_reopen_partial_floor_gap<F: Family, V, C>(
3423        context: deterministic::Context,
3424        open_db: impl Fn(
3425            deterministic::Context,
3426        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3427    ) where
3428        V: ValueEncoding<Value = Digest>,
3429        C: Mutable<Item = Operation<F, Digest, V>>,
3430        C::Item: EncodeShared,
3431    {
3432        let db = open_db(context.child("first")).await;
3433
3434        let k1 = Sha256::fill(1u8);
3435        let v1 = Sha256::fill(11u8);
3436
3437        // Commit A: 1 key, floor=0.
3438        let (db, _) = commit_sets(db, [(k1, v1)], None).await;
3439        let first_size = db.bounds().end;
3440        let first_root = db.root();
3441
3442        // Commit B: 1 key, floor=first_size.
3443        let k2 = Sha256::fill(2u8);
3444        let v2 = Sha256::fill(12u8);
3445        let (db, _) = commit_sets_with_floor(db, [(k2, v2)], None, first_size).await;
3446        let second_size = db.bounds().end;
3447
3448        // Commit C: 1 key, floor=second_size. This raises the floor
3449        // above commit B's keys, so reopen excludes both A and B keys.
3450        let k3 = Sha256::fill(3u8);
3451        let v3 = Sha256::fill(13u8);
3452        let (db, _) = commit_sets_with_floor(db, [(k3, v3)], None, second_size).await;
3453        db.sync().await.unwrap();
3454
3455        // Reopen: snapshot rebuilt from floor=second_size. Only k3 is in snapshot.
3456        let db = open_db(context.child("second")).await;
3457        assert!(db.get(&k1).await.unwrap().is_none());
3458        assert!(db.get(&k2).await.unwrap().is_none());
3459        assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3460
3461        // Rewind to commit B (not A). The gap fill should add keys from
3462        // [first_size, second_size) -- which includes k2 but not k1.
3463        // k3 is in the suffix and gets removed. k2 from the gap gets inserted.
3464        let db = db.rewind(second_size).await.unwrap();
3465        assert!(db.get(&k1).await.unwrap().is_none()); // below B's floor
3466        assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
3467        assert!(db.get(&k3).await.unwrap().is_none()); // in suffix, removed
3468
3469        // Now rewind further to commit A.
3470        let db = db.rewind(first_size).await.unwrap();
3471        assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
3472        assert!(db.get(&k2).await.unwrap().is_none()); // above first_size, truncated
3473        assert_eq!(db.root(), first_root);
3474        assert_eq!(db.inactivity_floor_loc(), Location::new(0));
3475
3476        db.destroy().await.unwrap();
3477    }
3478
3479    /// Rewind-after-reopen with a repeated key in the floor gap. The gap fill
3480    /// must restore the key, and reads may return any of its written values.
3481    #[boxed]
3482    pub(crate) async fn run_rewind_after_reopen_repeated_key_gap<F: Family, V, C>(
3483        context: deterministic::Context,
3484        open_db: impl Fn(
3485            deterministic::Context,
3486        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3487    ) where
3488        V: ValueEncoding<Value = Digest>,
3489        C: Mutable<Item = Operation<F, Digest, V>>,
3490        C::Item: EncodeShared,
3491    {
3492        let db = open_db(context.child("first")).await;
3493
3494        let key = Sha256::fill(7u8);
3495        let v1 = Sha256::fill(17u8);
3496        let v2 = Sha256::fill(18u8);
3497        let k3 = Sha256::fill(8u8);
3498        let v3 = Sha256::fill(19u8);
3499
3500        // Commit A: Set(key, v1) with floor=0.
3501        let (db, _) = commit_sets(db, [(key, v1)], None).await;
3502        let first_size = db.bounds().end;
3503
3504        // Commit B: Set(key, v2) with floor=0. Either written value may be served.
3505        let (db, _) = commit_sets(db, [(key, v2)], None).await;
3506        let second_size = db.bounds().end;
3507        let live = db.get(&key).await.unwrap().unwrap();
3508        assert!(live == v1 || live == v2);
3509
3510        // Commit C: raises floor above both earlier writes.
3511        let (db, _) = commit_sets_with_floor(db, [(k3, v3)], None, second_size).await;
3512        db.sync().await.unwrap();
3513
3514        // Reopen: snapshot rebuilt from floor=second_size, key excluded.
3515        let db = open_db(context.child("second")).await;
3516        assert!(db.get(&key).await.unwrap().is_none());
3517        assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3518
3519        // Rewind to commit B: gap fill re-inserts both Set(key,...) entries.
3520        let db = db.rewind(second_size).await.unwrap();
3521        let rewound = db.get(&key).await.unwrap().unwrap();
3522        assert!(rewound == v1 || rewound == v2);
3523
3524        // Rewind further to commit A: the v2 entry is dropped and get() must
3525        // serve v1, proving the gap fill restored the v1 location.
3526        let db = db.rewind(first_size).await.unwrap();
3527        assert_eq!(db.get(&key).await.unwrap(), Some(v1));
3528
3529        db.destroy().await.unwrap();
3530    }
3531
3532    /// After restart, the snapshot can contain only the newer write for a
3533    /// repeated key. Rewind restores the older write's snapshot entry, and
3534    /// reads may return any of the written values.
3535    #[boxed]
3536    pub(crate) async fn run_rewind_after_reopen_mixed_gap_retained<F: Family, V, C>(
3537        context: deterministic::Context,
3538        open_db: impl Fn(
3539            deterministic::Context,
3540        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3541    ) where
3542        V: ValueEncoding<Value = Digest>,
3543        C: Mutable<Item = Operation<F, Digest, V>>,
3544        C::Item: EncodeShared,
3545    {
3546        let db = open_db(context.child("first")).await;
3547
3548        let key = Sha256::fill(7u8);
3549        let v1 = Sha256::fill(17u8);
3550        let v2 = Sha256::fill(18u8);
3551        let k3 = Sha256::fill(8u8);
3552        let v3 = Sha256::fill(19u8);
3553
3554        // Commit A: Set(key, v1) at loc=0, floor=0.
3555        let (db, _) = commit_sets(db, [(key, v1)], None).await;
3556        let first_size = db.bounds().end;
3557
3558        // Commit B: Set(key, v2), floor=0. Either written value may be served.
3559        let (db, _) = commit_sets(db, [(key, v2)], None).await;
3560        let second_size = db.bounds().end;
3561        let live = db.get(&key).await.unwrap().unwrap();
3562        assert!(live == v1 || live == v2);
3563
3564        // Commit C: raises floor to first_size, so loc=0 is below floor but
3565        // loc for v2 is retained.
3566        let (db, _) = commit_sets_with_floor(db, [(k3, v3)], None, first_size).await;
3567        db.sync().await.unwrap();
3568
3569        // Reopen: snapshot rebuilt from floor=first_size. The v2 write for key
3570        // is retained; the v1 write is excluded.
3571        let db = open_db(context.child("second")).await;
3572        assert_eq!(db.get(&key).await.unwrap(), Some(v2));
3573
3574        // Rewind to commit B: gap fill re-inserts the v1 write alongside the
3575        // retained v2 entry, and get() serves one of the two.
3576        let db = db.rewind(second_size).await.unwrap();
3577        let rewound = db.get(&key).await.unwrap().unwrap();
3578        assert!(rewound == v1 || rewound == v2);
3579
3580        // Rewind further to commit A: the v2 entry is dropped and get() must
3581        // serve v1, proving the gap fill restored the v1 location.
3582        let db = db.rewind(first_size).await.unwrap();
3583        assert_eq!(db.get(&key).await.unwrap(), Some(v1));
3584
3585        db.destroy().await.unwrap();
3586    }
3587
3588    /// A live db retains every location of a repeated key, so rewinding across the newer
3589    /// write keeps serving the older retained one with no reopen involved.
3590    #[boxed]
3591    pub(crate) async fn run_rewind_repeated_key_live<F: Family, V, C>(
3592        context: deterministic::Context,
3593        open_db: impl Fn(
3594            deterministic::Context,
3595        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3596    ) where
3597        V: ValueEncoding<Value = Digest>,
3598        C: Mutable<Item = Operation<F, Digest, V>>,
3599        C::Item: EncodeShared,
3600    {
3601        let db = open_db(context.child("first")).await;
3602
3603        let key = Sha256::fill(7u8);
3604        let v1 = Sha256::fill(17u8);
3605        let v2 = Sha256::fill(18u8);
3606
3607        // Commit A: Set(key, v1) with floor=0.
3608        let (db, _) = commit_sets(db, [(key, v1)], None).await;
3609        let first_size = db.bounds().end;
3610
3611        // Commit B: Set(key, v2) with floor=0. Either written value may be served.
3612        let (db, _) = commit_sets(db, [(key, v2)], None).await;
3613        let live = db.get(&key).await.unwrap().unwrap();
3614        assert!(live == v1 || live == v2);
3615
3616        // Rewind to commit A: the v2 location is dropped and the retained v1
3617        // location keeps serving the key.
3618        let db = db.rewind(first_size).await.unwrap();
3619        assert_eq!(db.get(&key).await.unwrap(), Some(v1));
3620
3621        db.destroy().await.unwrap();
3622    }
3623
3624    /// Replay keeps only a repeated key's newest location, so a reopened db must still honor
3625    /// the repeated-key read contract after a rewind that crosses the newer write: the older
3626    /// write stays retained at an unchanged floor, and reads of the key may return any of its
3627    /// written values, never `None`.
3628    #[boxed]
3629    pub(crate) async fn run_rewind_after_reopen_repeated_key_retained<F: Family, V, C>(
3630        context: deterministic::Context,
3631        open_db: impl Fn(
3632            deterministic::Context,
3633        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3634    ) where
3635        V: ValueEncoding<Value = Digest>,
3636        C: Mutable<Item = Operation<F, Digest, V>>,
3637        C::Item: EncodeShared,
3638    {
3639        let db = open_db(context.child("first")).await;
3640
3641        let key = Sha256::fill(7u8);
3642        let v1 = Sha256::fill(17u8);
3643        let v2 = Sha256::fill(18u8);
3644
3645        // Commit A: Set(key, v1) with floor=0.
3646        let (db, _) = commit_sets(db, [(key, v1)], None).await;
3647        let first_size = db.bounds().end;
3648
3649        // Commit B: Set(key, v2) with floor=0, then persist for the reopen.
3650        let (db, _) = commit_sets(db, [(key, v2)], None).await;
3651        db.sync().await.unwrap();
3652
3653        // Reopen: replay visits both writes and keeps only the newer location.
3654        let db = open_db(context.child("second")).await;
3655        assert_eq!(db.get(&key).await.unwrap(), Some(v2));
3656
3657        // Rewind to commit A with an unchanged floor: the newer location is dropped, and the
3658        // older write, still retained in the restored journal, must keep the key readable.
3659        let db = db.rewind(first_size).await.unwrap();
3660        assert_eq!(db.get(&key).await.unwrap(), Some(v1));
3661
3662        db.destroy().await.unwrap();
3663    }
3664
3665    /// After committing with `floor = commit_loc` and pruning down to it, the live set is
3666    /// exactly one operation — the commit itself. This is the minimum non-empty live set
3667    /// achievable under the per-commit bound. The DB must remain fully usable:
3668    ///
3669    /// - `prune(commit_loc + 1)` is rejected (the floor is a hard ceiling).
3670    /// - `prune` does not affect the root (documented invariant).
3671    /// - Reopen reconstructs `inactivity_floor_loc` from the sole surviving commit op, and the
3672    ///   in-memory snapshot is empty (all Sets were below the floor).
3673    /// - A follow-on batch applies cleanly on top from the floor-at-max state.
3674    #[boxed]
3675    pub(crate) async fn run_single_commit_live_set<F: Family, V, C>(
3676        context: deterministic::Context,
3677        open_db: impl Fn(
3678            deterministic::Context,
3679        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3680    ) where
3681        V: ValueEncoding<Value = Digest>,
3682        C: Mutable<Item = Operation<F, Digest, V>>,
3683        C::Item: EncodeShared,
3684    {
3685        let db = open_db(context.child("test")).await;
3686
3687        // Initial commit is at loc 0. 3 sets + 1 commit → commit lands at loc 4.
3688        // Declare floor = 4 (= commit_loc), the tight maximum.
3689        let metadata = Sha256::fill(42u8);
3690        let commit_loc = Location::<F>::new(4);
3691        let k1 = Sha256::fill(1u8);
3692        let k2 = Sha256::fill(2u8);
3693        let k3 = Sha256::fill(3u8);
3694        let v1 = Sha256::fill(11u8);
3695        let v2 = Sha256::fill(12u8);
3696        let v3 = Sha256::fill(13u8);
3697        let merkleized = db
3698            .new_batch()
3699            .set(k1, v1)
3700            .set(k2, v2)
3701            .set(k3, v3)
3702            .merkleize(&db, Some(metadata), commit_loc)
3703            .await;
3704        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3705        let db = db.commit().await.unwrap();
3706        assert_eq!(db.last_commit_loc, commit_loc);
3707        assert_eq!(db.inactivity_floor_loc(), commit_loc);
3708        let root_after_commit = db.root();
3709
3710        // All three keys are in the in-memory snapshot pre-prune.
3711        assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
3712        assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
3713        assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3714
3715        // Prune at the floor — the maximum prune allowed.
3716        // Pruning is blob-aligned, so `bounds.start` may not physically advance all the way
3717        // to `commit_loc`; what matters semantically is that the floor authorizes pruning
3718        // of everything below the commit and that any further prune is rejected.
3719        let db = db.prune(commit_loc).await.unwrap();
3720        let bounds = db.bounds();
3721        assert!(
3722            bounds.start <= commit_loc,
3723            "prune must not advance bounds.start past the floor"
3724        );
3725        assert_eq!(bounds.end, commit_loc + 1);
3726
3727        // State preserved across the prune; root unchanged; commit metadata still readable.
3728        assert_eq!(db.last_commit_loc, commit_loc);
3729        assert_eq!(db.inactivity_floor_loc(), commit_loc);
3730        assert_eq!(db.root(), root_after_commit);
3731        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
3732
3733        // Persist, then verify pruning one past the floor is rejected — the floor is
3734        // the hard ceiling.
3735        let db = db.sync().await.unwrap();
3736        let Err(err) = db.prune(commit_loc + 1).await else {
3737            panic!("expected prune to fail");
3738        };
3739        assert!(matches!(err, Error::PruneBeyondMinRequired(p, f)
3740                if *p == *commit_loc + 1 && *f == *commit_loc));
3741
3742        // Reopen. `init_from_journal` rebuilds the snapshot by replaying from
3743        // the floor (= commit_loc). The only op at/above the floor is the commit, which
3744        // contributes no keys — so the rebuilt snapshot is empty.
3745        let db = open_db(context.child("reopened")).await;
3746        assert_eq!(db.last_commit_loc, commit_loc);
3747        assert_eq!(db.inactivity_floor_loc(), commit_loc);
3748        assert_eq!(db.root(), root_after_commit);
3749        // The commit op at `commit_loc` is the anchor that survived pruning — its metadata
3750        // must come back through `get_metadata` after the snapshot rebuild.
3751        assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
3752
3753        // Keys set below the floor are excluded from the rebuilt snapshot.
3754        assert!(db.get(&k1).await.unwrap().is_none());
3755        assert!(db.get(&k2).await.unwrap().is_none());
3756        assert!(db.get(&k3).await.unwrap().is_none());
3757
3758        // A follow-on batch applies on top. Monotonicity requires the new floor to be at
3759        // least `commit_loc` (= 4); advancing to the new tight max (= 6) exercises the
3760        // floor-at-max → new-batch transition.
3761        let k4 = Sha256::fill(4u8);
3762        let v4 = Sha256::fill(14u8);
3763        let next_commit_loc = Location::<F>::new(6);
3764        let merkleized = db
3765            .new_batch()
3766            .set(k4, v4)
3767            .merkleize(&db, None, next_commit_loc)
3768            .await;
3769        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3770        let db = db.commit().await.unwrap();
3771        assert_eq!(db.last_commit_loc, next_commit_loc);
3772        assert_eq!(db.inactivity_floor_loc(), next_commit_loc);
3773
3774        // New key readable; keys from the pre-prune batch remain excluded.
3775        assert_eq!(db.get(&k4).await.unwrap(), Some(v4));
3776        assert!(db.get(&k1).await.unwrap().is_none());
3777        // Follow-on commit replaced the anchor: its metadata was `None`, so `get_metadata`
3778        // should no longer return the original metadata.
3779        assert_eq!(db.get_metadata().await.unwrap(), None);
3780
3781        db.destroy().await.unwrap();
3782    }
3783
3784    /// `get_many` on the DB and on unmerkleized/merkleized batches returns results
3785    /// that match individual `get` calls.
3786    #[boxed]
3787    pub(crate) async fn run_get_many<F: Family, V, C>(
3788        context: deterministic::Context,
3789        open_db: impl Fn(
3790            deterministic::Context,
3791        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3792    ) where
3793        V: ValueEncoding<Value = Digest>,
3794        C: Mutable<Item = Operation<F, Digest, V>>,
3795        C::Item: EncodeShared,
3796    {
3797        let db = open_db(context.child("db")).await;
3798
3799        let k1 = Sha256::fill(1u8);
3800        let k2 = Sha256::fill(2u8);
3801        let k3 = Sha256::fill(3u8);
3802        let k_missing = Sha256::fill(99u8);
3803
3804        let v1 = Sha256::fill(11u8);
3805        let v2 = Sha256::fill(12u8);
3806        let v3 = Sha256::fill(13u8);
3807
3808        // Commit k1 and k2 to disk.
3809        let merkleized = db
3810            .new_batch()
3811            .set(k1, v1)
3812            .set(k2, v2)
3813            .merkleize(&db, None, db.inactivity_floor_loc())
3814            .await;
3815        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3816        let db = db.commit().await.unwrap();
3817
3818        // DB-level get_many.
3819        let results = db.get_many(&[&k1, &k2, &k_missing]).await.unwrap();
3820        assert_eq!(results, vec![Some(v1), Some(v2), None]);
3821
3822        // Empty input.
3823        let results = db.get_many(&([] as [&Digest; 0])).await.unwrap();
3824        assert!(results.is_empty());
3825
3826        // Unmerkleized batch: mutations + DB fallthrough.
3827        let batch = db.new_batch().set(k3, v3);
3828        let results = batch.get_many(&[&k3, &k1, &k_missing], &db).await.unwrap();
3829        assert_eq!(results, vec![Some(v3), Some(v1), None]);
3830
3831        // Merkleized batch: diff + parent chain + DB fallthrough.
3832        let parent = db
3833            .new_batch()
3834            .set(k3, v3)
3835            .merkleize(&db, None, db.inactivity_floor_loc())
3836            .await;
3837        let results = parent.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap();
3838        assert_eq!(results, vec![Some(v1), Some(v3), None]);
3839
3840        // Child of merkleized parent reads parent diff.
3841        let v3_new = Sha256::fill(30u8);
3842        let child = parent.new_batch::<Sha256>().set(k3, v3_new);
3843        let results = child.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap();
3844        assert_eq!(results, vec![Some(v1), Some(v3_new), None]);
3845
3846        db.destroy().await.unwrap();
3847    }
3848
3849    /// `get_many` reports unexpected data when the snapshot points at a non-`Set` operation.
3850    #[boxed]
3851    pub(crate) async fn run_get_many_unexpected_data<F: Family, V, C>(
3852        context: deterministic::Context,
3853        open_db: impl Fn(
3854            deterministic::Context,
3855        ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3856    ) where
3857        V: ValueEncoding<Value = Digest>,
3858        C: Mutable<Item = Operation<F, Digest, V>>,
3859        C::Item: EncodeShared,
3860    {
3861        let db = open_db(context.child("db")).await;
3862
3863        let key = Sha256::fill(1u8);
3864        let value = Sha256::fill(11u8);
3865        let merkleized = db
3866            .new_batch()
3867            .set(key, value)
3868            .merkleize(&db, None, db.inactivity_floor_loc())
3869            .await;
3870        let (db, _) = db.apply_batch(merkleized).await.unwrap();
3871        let mut db = db.commit().await.unwrap();
3872
3873        let bad_key = Sha256::fill(99u8);
3874        let bad_loc = db.last_commit_loc;
3875        db.snapshot.insert(&bad_key, bad_loc);
3876
3877        let err = db.get(&bad_key).await.unwrap_err();
3878        assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc));
3879
3880        let err = db.get_many(&[&bad_key]).await.unwrap_err();
3881        assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc));
3882
3883        db.destroy().await.unwrap();
3884    }
3885}