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