Skip to main content

commonware_storage/merkle/persisted/
full.rs

1//! A Merkle structure backed by a fixed-item-length journal.
2//!
3//! A [crate::journal] is used to store all unpruned nodes, and a [crate::metadata] store is
4//! used to preserve digests required for root and proof generation that would have otherwise been
5//! pruned.
6//!
7//! This module is generic over [`Family`], so it works for both MMR and MMB.
8//!
9//! # Ownership
10//!
11//! Mutating methods take the structure by value and return it on success. If a mutating
12//! method returns an error, or its future is dropped before it finishes, the structure is
13//! gone: state that was not yet durable is discarded, but everything already on disk stays
14//! recoverable.
15
16use crate::{
17    Context,
18    journal::{
19        Error as JError,
20        contiguous::{
21            Contiguous, Many,
22            fixed::{Config as JConfig, Journal},
23        },
24    },
25    merkle::{
26        Error, Family, Location, Position, Proof, Readable, batch,
27        hasher::Hasher,
28        mem::{Config as MemConfig, Mem},
29    },
30    metadata::{Config as MConfig, Metadata},
31};
32use commonware_codec::{DecodeExt, Write};
33use commonware_cryptography::Digest;
34use commonware_parallel::Strategy;
35use commonware_runtime::{Handle, buffer::paged::CacheRef};
36use commonware_utils::{range::NonEmptyRange, sequence::prefixed_u64::U64};
37use std::{
38    collections::{BTreeMap, BTreeSet},
39    num::{NonZeroU64, NonZeroUsize},
40    sync::Arc,
41};
42use tracing::{debug, error, warn};
43
44/// Append-only wrapper around [`batch::UnmerkleizedBatch`].
45///
46/// The full Merkle structure's [`Merkle::sync`] only persists *appended* nodes
47/// (positions in `[journal_size, state.size())`). Overwrites to existing positions are stored in
48/// the in-memory layer but never flushed, so they would be silently lost on crash recovery. This
49/// wrapper prevents that by exposing only append and merkleize operations, hiding `update_leaf*`
50/// at compile time.
51pub struct UnmerkleizedBatch<F: Family, D: Digest, S: Strategy> {
52    inner: batch::UnmerkleizedBatch<F, D, S>,
53}
54
55impl<F: Family, D: Digest, S: Strategy> UnmerkleizedBatch<F, D, S> {
56    /// Hash `element` and add it as a leaf.
57    pub fn add(self, hasher: &impl Hasher<F, Digest = D>, element: &[u8]) -> Self {
58        Self {
59            inner: self.inner.add(hasher, element),
60        }
61    }
62
63    /// Add a pre-computed leaf digest.
64    pub fn add_leaf_digest(self, digest: D) -> Self {
65        Self {
66            inner: self.inner.add_leaf_digest(digest),
67        }
68    }
69
70    /// Encode and hash `items` across the strategy, adding their leaf digests in order.
71    pub(crate) fn add_many<Item: Write + Send + Sync>(
72        self,
73        hasher: &impl Hasher<F, Digest = D>,
74        items: &[Item],
75    ) -> Self {
76        Self {
77            inner: self.inner.add_many(hasher, items),
78        }
79    }
80
81    /// The number of leaves visible through this batch.
82    pub fn leaves(&self) -> Location<F> {
83        self.inner.leaves()
84    }
85
86    /// Return a reference to the batch's strategy.
87    pub fn strategy(&self) -> &S {
88        self.inner.strategy()
89    }
90
91    /// Consume this batch and produce an immutable [`batch::MerkleizedBatch`] with computed nodes.
92    /// `base` provides committed node data as fallback during hash computation.
93    pub fn merkleize(
94        self,
95        base: &Mem<F, D>,
96        hasher: &impl Hasher<F, Digest = D>,
97    ) -> Arc<batch::MerkleizedBatch<F, D, S>> {
98        self.inner.merkleize(base, hasher)
99    }
100}
101
102/// Configuration for a journal-backed Merkle structure.
103#[derive(Clone)]
104pub struct Config<S: Strategy> {
105    /// The name of the `commonware-runtime::Storage` storage partition used for the journal storing
106    /// the nodes.
107    pub journal_partition: String,
108
109    /// The name of the `commonware-runtime::Storage` storage partition used for the metadata
110    /// containing pruned nodes that are still required to calculate the root and generate
111    /// proofs.
112    pub metadata_partition: String,
113
114    /// The maximum number of items to store in each blob in the backing journal.
115    pub items_per_blob: NonZeroU64,
116
117    /// The size of the write buffer to use for each blob in the backing journal.
118    pub write_buffer: NonZeroUsize,
119
120    /// Buffer size for sequential reads during recovery.
121    pub replay_buffer: NonZeroUsize,
122
123    /// Strategy used to parallelize batch operations.
124    pub strategy: S,
125
126    /// The page cache to use for caching data.
127    pub page_cache: CacheRef,
128}
129
130/// Configuration for initializing a full Merkle structure for synchronization.
131///
132/// Determines how to handle existing persistent data based on sync boundaries:
133/// - **Fresh Start**: Existing data < range start -> discard and start fresh
134/// - **Prune and Reuse**: range contains existing data -> prune and reuse
135/// - **Ahead**: retained data extends beyond range end -> rewind to range end
136/// - **Incompatible**: retained data starts after range start -> discard and start fresh
137pub struct SyncConfig<F: Family, D: Digest, S: Strategy> {
138    /// Base configuration (journal, metadata, etc.)
139    pub config: Config<S>,
140
141    /// Sync range expressed as leaf-aligned bounds.
142    pub range: NonEmptyRange<Location<F>>,
143
144    /// The pinned nodes the structure needs at the pruning boundary (range start), in the order
145    /// specified by `Family::nodes_to_pin`. If `None`, the pinned nodes are expected to already be
146    /// in the structure's metadata/journal.
147    pub pinned_nodes: Option<Vec<D>>,
148}
149
150/// A Merkle structure backed by a fixed-item-length journal.
151pub struct Merkle<F: Family, E: Context, D: Digest, S: Strategy> {
152    /// A memory resident Merkle structure used to build the structure and cache updates. It caches
153    /// all un-synced nodes, and the pinned node set as derived from both its own pruning boundary
154    /// and the full structure's pruning boundary.
155    ///
156    /// Held in an [`Arc`] so [`Merkle::snapshot`] can hand a zero-copy, immutable view to jobs
157    /// running off the calling task. Mutations go through [`Arc::make_mut`]: they are in-place
158    /// while no snapshot is alive and copy-on-write otherwise, so a snapshot never observes
159    /// later mutations.
160    pub(crate) mem: Arc<Mem<F, D>>,
161
162    /// The highest position for which this structure has been pruned, or 0 if it has never been
163    /// pruned.
164    pub(crate) pruned_to_pos: Position<F>,
165
166    /// Stores all unpruned nodes.
167    pub(crate) journal: Journal<E, D>,
168
169    /// Stores the pinned nodes for the current pruning boundary, and the corresponding pruning
170    /// boundary used to generate them. The metadata remains empty until pruning is invoked, and its
171    /// contents change only when the pruning boundary moves.
172    pub(crate) metadata: Metadata<E, U64, Vec<u8>>,
173
174    /// True while the journal may contain flushed nodes that have not yet been made durable.
175    pub(crate) journal_dirty: bool,
176
177    /// The strategy to use for parallelization.
178    pub(crate) strategy: S,
179}
180
181impl<F: Family, E: Context, D: Digest, S: Strategy> std::fmt::Debug for Merkle<F, E, D, S> {
182    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
183        f.debug_struct("Merkle")
184            .field("size", &self.size())
185            .field("leaves", &self.leaves())
186            .finish_non_exhaustive()
187    }
188}
189
190/// Prefix used for nodes in the metadata prefixed U8 key.
191const NODE_PREFIX: u8 = 0;
192
193/// Prefix used for the key storing the pruning boundary (as a leaf index) in the metadata.
194pub(crate) const PRUNED_TO_PREFIX: u8 = 1;
195
196impl<F: Family, E: Context, D: Digest, S: Strategy> Merkle<F, E, D, S> {
197    /// Return the total number of nodes in the structure, irrespective of any pruning. The next
198    /// added element's position will have this value.
199    pub fn size(&self) -> Position<F> {
200        self.mem.size()
201    }
202
203    /// Return the total number of leaves in the structure.
204    pub fn leaves(&self) -> Location<F> {
205        self.mem.leaves()
206    }
207
208    /// Attempt to get a node from the metadata, with fallback to journal lookup if it fails.
209    /// Assumes the node should exist in at least one of these sources and returns a `MissingNode`
210    /// error otherwise.
211    async fn get_from_metadata_or_journal(
212        metadata: &Metadata<E, U64, Vec<u8>>,
213        journal: &Journal<E, D>,
214        pos: Position<F>,
215    ) -> Result<D, Error<F>> {
216        if let Some(bytes) = metadata.get(&U64::new(NODE_PREFIX, *pos)) {
217            debug!(?pos, "read node from metadata");
218            let digest = D::decode(bytes.as_ref());
219            let Ok(digest) = digest else {
220                error!(
221                    ?pos,
222                    err = %digest.expect_err("digest is Err in else branch"),
223                    "could not convert node from metadata bytes to digest"
224                );
225                return Err(Error::DataCorrupted(
226                    "could not read digest at requested pos",
227                ));
228            };
229            return Ok(digest);
230        }
231
232        // If a node isn't found in the metadata, it might still be in the journal.
233        debug!(?pos, "reading node from journal");
234        let node = journal.read(*pos).await;
235        match node {
236            Ok(node) => Ok(node),
237            Err(JError::ItemPruned(_)) => {
238                error!(?pos, "node is missing from metadata and journal");
239                Err(Error::MissingNode(pos))
240            }
241            Err(e) => Err(Error::Journal(e)),
242        }
243    }
244
245    /// Returns [start, end) where `start` is the oldest retained leaf and `end` is the total leaf
246    /// count.
247    pub fn bounds(&self) -> std::ops::Range<Location<F>> {
248        Location::try_from(self.pruned_to_pos).expect("valid pruned_to_pos")..self.mem.leaves()
249    }
250
251    /// Adds the pinned nodes based on `prune_pos` to `mem`.
252    async fn add_extra_pinned_nodes(
253        mem: &mut Mem<F, D>,
254        metadata: &Metadata<E, U64, Vec<u8>>,
255        journal: &Journal<E, D>,
256        prune_pos: Position<F>,
257    ) -> Result<(), Error<F>> {
258        let prune_loc = Location::try_from(prune_pos).expect("valid prune_pos");
259        let mut pinned_nodes = BTreeMap::new();
260        for pos in F::nodes_to_pin(prune_loc) {
261            let digest = Self::get_from_metadata_or_journal(metadata, journal, pos).await?;
262            pinned_nodes.insert(pos, digest);
263        }
264        mem.add_pinned_nodes(pinned_nodes);
265
266        Ok(())
267    }
268
269    /// Initialize a new `Merkle` instance.
270    pub async fn init(
271        context: E,
272        hasher: &impl Hasher<F, Digest = D>,
273        cfg: Config<S>,
274    ) -> Result<Self, Error<F>> {
275        let journal_cfg = JConfig {
276            partition: cfg.journal_partition,
277            items_per_blob: cfg.items_per_blob,
278            page_cache: cfg.page_cache,
279            write_buffer: cfg.write_buffer,
280            replay_buffer: cfg.replay_buffer,
281        };
282        let mut journal =
283            Journal::<E, D>::init(context.child("merkle_journal"), journal_cfg).await?;
284        let mut journal_size = Position::<F>::new(journal.size());
285
286        let metadata_cfg = MConfig {
287            partition: cfg.metadata_partition,
288            codec_config: ((0..).into(), ()),
289        };
290        let metadata =
291            Metadata::<_, U64, Vec<u8>>::init(context.child("merkle_metadata"), metadata_cfg)
292                .await?;
293
294        if journal_size == 0 {
295            let mem = Mem::init(MemConfig {
296                nodes: vec![],
297                pruning_boundary: Location::new(0),
298                pinned_nodes: vec![],
299            })?;
300            return Ok(Self {
301                mem: Arc::new(mem),
302                pruned_to_pos: Position::new(0),
303                journal,
304                metadata,
305                journal_dirty: false,
306                strategy: cfg.strategy,
307            });
308        }
309
310        // Metadata stores the pruning boundary as a leaf index. Journal recovery compares node
311        // positions.
312        let key: U64 = U64::new(PRUNED_TO_PREFIX, 0);
313        let metadata_pruned_to = Location::<F>::new(metadata.get(&key).map_or(0, |bytes| {
314            u64::from_be_bytes(
315                bytes
316                    .as_slice()
317                    .try_into()
318                    .expect("metadata pruned_to is not 8 bytes"),
319            )
320        }));
321        let metadata_prune_pos = Position::try_from(metadata_pruned_to)?;
322        let journal_bounds_start = journal.bounds().start;
323
324        // Use the more restrictive (higher) pruning boundary between metadata and journal.
325        // This handles both cases: metadata ahead (crash during prune) and metadata stale.
326        //
327        // The journal boundary may not be leaf-aligned (it's blob-aligned), so round up to the
328        // position of the first leaf after the boundary.
329        let journal_boundary_pos = Position::<F>::new(journal_bounds_start);
330        let journal_boundary_floor = F::to_nearest_size(journal_boundary_pos);
331        let journal_boundary_leaf_aligned_pos = if journal_boundary_floor == journal_boundary_pos {
332            // `to_nearest_size` rounds down, so equality means the boundary is already
333            // leaf-aligned.
334            journal_boundary_floor
335        } else {
336            // If flooring backed up over the boundary, round up to the next leaf position, which
337            // is guaranteed to be above it.
338            Position::try_from(Location::try_from(journal_boundary_floor)? + 1)?
339        };
340        let effective_prune_pos =
341            std::cmp::max(metadata_prune_pos, journal_boundary_leaf_aligned_pos);
342
343        let last_valid_size = F::to_nearest_size(journal_size);
344        if effective_prune_pos > last_valid_size {
345            error!(
346                ?effective_prune_pos,
347                ?last_valid_size,
348                "pruning boundary exceeds recovered journal size"
349            );
350            return Err(Error::MissingNode(effective_prune_pos));
351        }
352
353        // Make sure the journal's oldest retained node is as expected based on the last pruning
354        // boundary stored in metadata. If they don't match, prune the journal to the appropriate
355        // location.
356        if *metadata_prune_pos > journal_bounds_start {
357            // Metadata is ahead of journal (crashed before completing journal prune).
358            // Prune the journal to match metadata.
359            (journal, _) = journal.prune(*metadata_prune_pos).await?;
360            if journal.bounds().start != journal_bounds_start {
361                // This should only happen in the event of some failure during the last attempt to
362                // prune the journal.
363                warn!(
364                    journal_bounds_start,
365                    ?metadata_prune_pos,
366                    "journal pruned to match metadata"
367                );
368            }
369        } else if *metadata_prune_pos < journal_bounds_start {
370            // Metadata is stale (e.g., missing/corrupted while journal has valid state).
371            // Use the journal's state as authoritative.
372            warn!(
373                ?metadata_prune_pos,
374                journal_bounds_start, "metadata stale, using journal pruning boundary"
375            );
376        }
377
378        let mut orphaned_leaf: Option<D> = None;
379        if last_valid_size != journal_size {
380            warn!(
381                ?last_valid_size,
382                "encountered invalid structure, recovering from last valid size"
383            );
384            // Check if there is an intact leaf following the last valid size, from which we can
385            // recover its missing parents.
386            let recovered_item = journal.read(*last_valid_size).await;
387            if let Ok(item) = recovered_item {
388                orphaned_leaf = Some(item);
389            }
390            journal = journal.rewind(*last_valid_size).await?.sync().await?;
391            journal_size = last_valid_size
392        }
393
394        // Initialize the mem in the "prune_all" state.
395        let journal_leaves = Location::try_from(journal_size)?;
396        let mut pinned_nodes = Vec::new();
397        for pos in F::nodes_to_pin(journal_leaves) {
398            let digest = Self::get_from_metadata_or_journal(&metadata, &journal, pos).await?;
399            pinned_nodes.push(digest);
400        }
401        let mut mem = Mem::init(MemConfig {
402            nodes: vec![],
403            pruning_boundary: journal_leaves,
404            pinned_nodes,
405        })?;
406        Self::add_extra_pinned_nodes(&mut mem, &metadata, &journal, effective_prune_pos).await?;
407
408        if let Some(leaf) = orphaned_leaf {
409            // Recover the orphaned leaf and any missing parents.
410            let pos = mem.size();
411            warn!(?pos, "recovering orphaned leaf");
412            let batch = mem
413                .new_batch()
414                .add_leaf_digest(leaf)
415                .merkleize(&mem, hasher);
416            mem.apply_batch(&batch)?;
417            assert_eq!(pos, journal_size);
418
419            // Inline sync: flush recovered nodes to journal.
420            for p in journal.size()..*mem.size() {
421                let p = Position::new(p);
422                let node = *mem.get_node_unchecked(p);
423                (journal, _) = journal.append(&node).await?;
424            }
425            journal = journal.sync().await?;
426            assert_eq!(mem.size(), journal.size());
427
428            // Prune mem and reinstate pinned nodes.
429            let effective_prune_loc =
430                Location::try_from(effective_prune_pos).expect("valid effective_prune_pos");
431            let mut pn = BTreeMap::new();
432            for p in F::nodes_to_pin(effective_prune_loc) {
433                let d = mem.get_node_unchecked(p);
434                pn.insert(p, *d);
435            }
436            mem.prune_all();
437            mem.add_pinned_nodes(pn);
438        }
439
440        Ok(Self {
441            mem: Arc::new(mem),
442            pruned_to_pos: effective_prune_pos,
443            journal,
444            metadata,
445            journal_dirty: false,
446            strategy: cfg.strategy,
447        })
448    }
449
450    /// Initialize a structure for synchronization, reusing existing data if possible.
451    ///
452    /// Handles sync scenarios based on existing journal data vs. the given sync range:
453    ///
454    /// 1. **Fresh Start**: existing_size <= range.start
455    ///    - Deletes existing data (if any)
456    ///    - Creates new [Journal] with pruning boundary and size at `range.start`
457    ///
458    /// 2. **Reuse**: range.start < existing_size <= range.end
459    ///    - Keeps existing journal data
460    ///    - Prunes the journal toward `range.start` (section-aligned)
461    ///
462    /// 3. **Ahead**: retained data covers range.start but ends after range.end
463    ///    - Rewinds the journal to `range.end`
464    ///
465    /// 4. **Incompatible**: retained data starts after range.start
466    ///    - Discards existing data and creates a new [Journal] at `range.start`
467    pub async fn init_sync(context: E, cfg: SyncConfig<F, D, S>) -> Result<Self, Error<F>> {
468        let prune_pos = Position::try_from(cfg.range.start())?;
469        let end_pos = Position::try_from(cfg.range.end())?;
470        let journal_cfg = JConfig {
471            partition: cfg.config.journal_partition.clone(),
472            items_per_blob: cfg.config.items_per_blob,
473            write_buffer: cfg.config.write_buffer,
474            replay_buffer: cfg.config.replay_buffer,
475            page_cache: cfg.config.page_cache.clone(),
476        };
477
478        // Open the journal, performing a rewind if necessary for crash recovery.
479        let mut journal: Journal<E, D> =
480            Journal::init(context.child("merkle_journal"), journal_cfg).await?;
481        let mut journal_size = Position::<F>::new(journal.size());
482
483        // If a crash left the journal at an invalid size (e.g., a leaf was written
484        // but its parent nodes were not), rewind to the last valid size.
485        let last_valid_size = F::to_nearest_size(journal_size);
486        if last_valid_size != journal_size {
487            warn!(
488                ?last_valid_size,
489                "init_sync: encountered invalid structure, recovering from last valid size"
490            );
491            journal = journal.rewind(*last_valid_size).await?.sync().await?;
492            journal_size = last_valid_size;
493        }
494
495        // A pruned start cannot be reconstructed from the retained suffix.
496        let journal_bounds = journal.bounds();
497        let missing_start = journal_bounds.start > *prune_pos;
498        let ahead = journal_size > *end_pos;
499        let reinitialized = missing_start || ahead || journal_size <= *prune_pos;
500        if missing_start {
501            debug!(
502                journal_size = *journal_size,
503                journal_start = journal_bounds.start,
504                range_start = *prune_pos,
505                range_end = *end_pos,
506                "existing Merkle journal is incompatible with sync range, resetting"
507            );
508            journal = journal.clear_to_size(*prune_pos).await?;
509            journal_size = Position::new(journal.size());
510        } else if ahead {
511            // Sync targets describe the same append-only tree, so a later target retains every
512            // node through this target's end.
513            journal = journal.rewind(*end_pos).await?.sync().await?;
514            journal_size = Position::new(journal.size());
515        } else if journal_size <= *prune_pos && *prune_pos != 0 {
516            journal = journal.clear_to_size(*prune_pos).await?;
517            journal_size = Position::new(journal.size());
518        }
519
520        // Open the metadata.
521        let metadata_cfg = MConfig {
522            partition: cfg.config.metadata_partition,
523            codec_config: ((0..).into(), ()),
524        };
525        let mut metadata = Metadata::init(context.child("merkle_metadata"), metadata_cfg).await?;
526
527        let prune_loc = Location::try_from(prune_pos)?;
528        let nodes_to_pin_persisted: Vec<_> = F::nodes_to_pin(prune_loc).collect();
529        if reinitialized {
530            let retained_node_keys: BTreeSet<_> = nodes_to_pin_persisted
531                .iter()
532                .map(|pos| U64::new(NODE_PREFIX, **pos))
533                .collect();
534            // Reinitializing the journal invalidates pins from an abandoned target. Retain only
535            // boundary pins so supplied values can replace them, and so a retry after a crash
536            // cannot prefer stale metadata over rebuilt journal nodes.
537            metadata.retain(|key: &U64, _| {
538                key.prefix() != NODE_PREFIX || retained_node_keys.contains(key)
539            });
540        }
541
542        // Write the pruning boundary.
543        let pruning_boundary_key = U64::new(PRUNED_TO_PREFIX, 0);
544        metadata.put(
545            pruning_boundary_key,
546            cfg.range.start().as_u64().to_be_bytes().into(),
547        );
548
549        // Write the required pinned nodes to metadata.
550        // The set of pinned nodes depends only on the prune boundary, not on the total
551        // structure size, so we validate against `nodes_to_pin(prune_loc)` alone.
552        let journal_leaves = Location::try_from(journal_size)?;
553        if let Some(pinned_nodes) = cfg.pinned_nodes {
554            // Use caller-provided pinned nodes.
555            if pinned_nodes.len() != nodes_to_pin_persisted.len() {
556                return Err(Error::<F>::InvalidPinnedNodes);
557            }
558            for (pos, digest) in nodes_to_pin_persisted.into_iter().zip(pinned_nodes.iter()) {
559                metadata.put(U64::new(NODE_PREFIX, *pos), digest.to_vec());
560            }
561        }
562
563        // Create the in-memory structure with the pinned nodes required for its size. This must be
564        // performed *before* pruning the journal to range.start to ensure all pinned nodes are
565        // present.
566        let nodes_to_pin_mem = F::nodes_to_pin(journal_leaves);
567        let mut mem_pinned_nodes = Vec::new();
568        for pos in nodes_to_pin_mem {
569            let digest = Self::get_from_metadata_or_journal(&metadata, &journal, pos).await?;
570            mem_pinned_nodes.push(digest);
571        }
572        let mut mem = Mem::init(MemConfig {
573            nodes: vec![],
574            pruning_boundary: Location::try_from(journal_size)?,
575            pinned_nodes: mem_pinned_nodes,
576        })?;
577
578        // Add the additional pinned nodes required for the pruning boundary, if applicable.
579        // This must also be done before pruning.
580        if prune_pos < journal_size {
581            Self::add_extra_pinned_nodes(&mut mem, &metadata, &journal, prune_pos).await?;
582        }
583
584        // Sync metadata before pruning so pinned nodes are persisted for crash recovery.
585        let metadata = metadata.sync().await?;
586
587        // Prune the journal to range.start.
588        (journal, _) = journal.prune(*prune_pos).await?;
589
590        Ok(Self {
591            mem: Arc::new(mem),
592            pruned_to_pos: prune_pos,
593            journal,
594            metadata,
595            journal_dirty: false,
596            strategy: cfg.config.strategy,
597        })
598    }
599
600    /// Compute and add required nodes for the given pruning point to the metadata, and write it to
601    /// disk. Return the computed set of required nodes.
602    async fn update_metadata(
603        mut self,
604        prune_to_pos: Position<F>,
605    ) -> Result<(Self, BTreeMap<Position<F>, D>), Error<F>> {
606        assert!(prune_to_pos >= self.pruned_to_pos);
607
608        let prune_loc = Location::try_from(prune_to_pos).expect("valid prune_to_pos");
609        let mut pinned_nodes = BTreeMap::new();
610        for pos in F::nodes_to_pin(prune_loc) {
611            let digest = self.get_node(pos).await?.expect(
612                "pinned node should exist if prune_to_pos is no less than self.pruned_to_pos",
613            );
614            self.metadata
615                .put(U64::new(NODE_PREFIX, *pos), digest.to_vec());
616            pinned_nodes.insert(pos, digest);
617        }
618
619        let key: U64 = U64::new(PRUNED_TO_PREFIX, 0);
620        self.metadata = self
621            .metadata
622            .put_sync(
623                key,
624                Location::try_from(prune_to_pos)?
625                    .as_u64()
626                    .to_be_bytes()
627                    .into(),
628            )
629            .await
630            .map_err(Error::Metadata)?;
631
632        Ok((self, pinned_nodes))
633    }
634
635    pub async fn get_node(&self, position: Position<F>) -> Result<Option<D>, Error<F>> {
636        if let Some(node) = self.mem.get_node(position) {
637            return Ok(Some(node));
638        }
639
640        match self.journal.read(*position).await {
641            Ok(item) => Ok(Some(item)),
642            Err(JError::ItemPruned(_)) => Ok(None),
643            Err(e) => Err(Error::Journal(e)),
644        }
645    }
646
647    /// Batched [`Self::get_node`]: `positions` must be strictly increasing. Memory-resident
648    /// nodes are served directly; the rest go through the journal's batched read, which
649    /// serves page-cache hits in bulk and fetches misses concurrently.
650    ///
651    /// # Errors
652    ///
653    /// Returns [`Error::ElementPruned`] for the first of `positions` that falls below the
654    /// journal's pruning boundary.
655    pub async fn get_nodes(&self, positions: &[Position<F>]) -> Result<Vec<D>, Error<F>> {
656        assert!(
657            positions.is_sorted_by(|a, b| a < b),
658            "positions must be strictly increasing"
659        );
660        let bounds = self.journal.bounds();
661        let mut nodes = vec![None; positions.len()];
662        let mut journal_positions = Vec::with_capacity(positions.len());
663        for (slot, &position) in nodes.iter_mut().zip(positions) {
664            if let Some(node) = self.mem.get_node(position) {
665                *slot = Some(node);
666            } else if *position >= bounds.start {
667                // In-subsequence order is preserved, so this stays strictly increasing.
668                journal_positions.push(*position);
669            } else {
670                return Err(Error::ElementPruned(position));
671            }
672        }
673
674        // Within-bounds reads are guaranteed not to return `ItemPruned` (see
675        // [`crate::journal::contiguous::Contiguous::read`]).
676        let items = if journal_positions.is_empty() {
677            Vec::new()
678        } else {
679            self.journal
680                .read_many(&journal_positions)
681                .await
682                .map_err(Error::Journal)?
683        };
684
685        // The unfilled slots are exactly the journal subsequence, in the order it was built.
686        let mut items = items.into_iter();
687        Ok(nodes
688            .into_iter()
689            .map(|node| node.unwrap_or_else(|| items.next().expect("one item per journal read")))
690            .collect())
691    }
692
693    /// Return the pinned nodes needed to authenticate a lower leaf boundary at `loc`.
694    pub async fn pinned_nodes_at(&self, loc: Location<F>) -> Result<Vec<D>, Error<F>> {
695        if !loc.is_valid() {
696            return Err(Error::LocationOverflow(loc));
697        }
698        let futs = F::nodes_to_pin(loc)
699            .map(|p| async move { self.get_node(p).await?.ok_or(Error::ElementPruned(p)) })
700            .collect::<Vec<_>>();
701        futures::future::try_join_all(futs).await
702    }
703
704    /// Flush all nodes cached in the in-memory structure to the journal without forcing them to
705    /// disk. Flushed nodes are pruned from the in-memory structure and remain readable through the
706    /// journal, but they are not guaranteed to survive a crash until [Self::sync] is called.
707    pub async fn flush(self) -> Result<Self, Error<F>> {
708        self.flush_internal().await
709    }
710
711    /// Flush all nodes cached in the in-memory structure to the journal and make them durable.
712    pub async fn sync(mut self) -> Result<Self, Error<F>> {
713        self = self.flush_internal().await?;
714
715        // Sync the journal to ensure durability before returning. This covers nodes appended by
716        // the flush above as well as nodes left non-durable by earlier [Self::flush] calls.
717        if self.journal_dirty {
718            self.journal = self.journal.sync().await?;
719            self.journal_dirty = false;
720        }
721
722        Ok(self)
723    }
724
725    /// Flush all nodes cached in the in-memory structure to the journal and begin making them
726    /// durable, returning a completion handle.
727    ///
728    /// The handle covers only nodes flushed so far. A later [Self::sync] still performs a full
729    /// durable sync.
730    pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
731        // `journal_dirty` is deliberately not cleared: the started sync covers only nodes
732        // flushed so far, and sync() remains the durability authority.
733        self = self.flush_internal().await?;
734        let (journal, handle) = self.journal.start_sync().await?;
735        self.journal = journal;
736        Ok((self, handle))
737    }
738
739    /// Append nodes cached in the in-memory structure that are missing from the journal, then
740    /// prune them from the in-memory structure. Sets [Self::journal_dirty] when nodes are
741    /// appended.
742    async fn flush_internal(mut self) -> Result<Self, Error<F>> {
743        let journal_size = Position::<F>::new(self.journal.size());
744
745        // Encode the nodes missing from the journal directly to bytes and snapshot the pinned
746        // node set for the current pruning boundary.
747        let (sync_target_leaves, encoded, pinned_nodes) = {
748            let size = self.mem.size();
749            let sync_target_leaves = self.mem.leaves();
750
751            assert!(
752                journal_size <= size,
753                "journal size should never exceed in-memory structure size"
754            );
755            if journal_size == size {
756                return Ok(self);
757            }
758
759            // Encode the un-journaled tail to an owned buffer before the journal I/O below.
760            let (head, tail) = self.mem.nodes_from(journal_size);
761            let encoded = self.journal.prepare_append(Many::Nested(&[head, tail]));
762
763            // Recompute pinned nodes since we'll need to repopulate the cache after it is cleared
764            // by pruning the mem.
765            let prune_loc = Location::try_from(self.pruned_to_pos).expect("valid pruned_to_pos");
766            let mut pinned_nodes = BTreeMap::new();
767            for pos in F::nodes_to_pin(prune_loc) {
768                let digest = self.mem.get_node_unchecked(pos);
769                pinned_nodes.insert(pos, *digest);
770            }
771
772            (sync_target_leaves, encoded, pinned_nodes)
773        };
774
775        // Append missing nodes to the journal.
776        (self.journal, _) = self.journal.append_prepared(encoded).await?;
777        self.journal_dirty = true;
778
779        // Now that the missing nodes are readable from the journal, it's safe to prune them from
780        // the mem. We prune to the previously captured leaf count.
781        let mem = Arc::make_mut(&mut self.mem);
782        mem.prune(sync_target_leaves)
783            .expect("captured leaves is in bounds");
784        mem.add_pinned_nodes(pinned_nodes);
785
786        Ok(self)
787    }
788
789    /// Prune all nodes up to but not including the given leaf location and update the pinned nodes.
790    ///
791    /// This implementation ensures that no failure can leave the structure in an unrecoverable
792    /// state, requiring it sync the structure to write any potential unsynced updates.
793    ///
794    /// Returns [Error::LocationOverflow] if `loc` exceeds [Family::MAX_LEAVES].
795    /// Returns [Error::LeafOutOfBounds] if `loc` exceeds the current leaf count.
796    pub async fn prune(mut self, loc: Location<F>) -> Result<Self, Error<F>> {
797        let pos = Position::try_from(loc)?;
798        if loc > self.mem.leaves() {
799            return Err(Error::LeafOutOfBounds(loc));
800        }
801        if pos <= self.pruned_to_pos {
802            return Ok(self);
803        }
804
805        // Flush items cached in the mem to disk to ensure the current state is recoverable.
806        self = self.sync().await?;
807
808        // Update metadata to reflect the desired pruning boundary, allowing for recovery in the
809        // event of a pruning failure.
810        let pinned_nodes;
811        (self, pinned_nodes) = self.update_metadata(pos).await?;
812
813        (self.journal, _) = self.journal.prune(*pos).await?;
814        Arc::make_mut(&mut self.mem).add_pinned_nodes(pinned_nodes);
815        self.pruned_to_pos = pos;
816
817        Ok(self)
818    }
819
820    /// Compute the root of the structure using `inactive_peaks` and the bagging carried by `hasher`.
821    pub fn root(
822        &self,
823        hasher: &impl Hasher<F, Digest = D>,
824        inactive_peaks: usize,
825    ) -> Result<D, Error<F>> {
826        self.mem.root(hasher, inactive_peaks)
827    }
828
829    /// Prune as many nodes as possible, leaving behind at most items_per_blob nodes in the current
830    /// blob.
831    pub async fn prune_all(mut self) -> Result<Self, Error<F>> {
832        let leaves = self.mem.leaves();
833        if leaves != 0 {
834            self = self.prune(leaves).await?;
835        }
836        Ok(self)
837    }
838
839    /// Close and permanently remove any disk resources.
840    pub async fn destroy(self) -> Result<(), Error<F>> {
841        self.journal.destroy().await?;
842        self.metadata.destroy().await?;
843
844        Ok(())
845    }
846
847    #[cfg(any(test, feature = "fuzzing"))]
848    /// Sync elements to disk until `write_limit` elements have been written, then abort to simulate
849    /// a partial write for testing failure scenarios.
850    pub async fn simulate_partial_sync(mut self, write_limit: usize) -> Result<(), Error<F>> {
851        if write_limit == 0 {
852            return Ok(());
853        }
854
855        let journal_size = Position::<F>::new(self.journal.size());
856
857        // Write the nodes cached in the memory-resident structure to the journal, aborting after
858        // write_count nodes have been written.
859        let mut written_count = 0usize;
860        for i in *journal_size..*self.mem.size() {
861            let node = *self.mem.get_node_unchecked(Position::new(i));
862            (self.journal, _) = self.journal.append(&node).await?;
863            written_count += 1;
864            if written_count >= write_limit {
865                break;
866            }
867        }
868        self.journal.sync().await?;
869
870        Ok(())
871    }
872
873    #[cfg(test)]
874    /// Return a copy of the currently pinned nodes for recovery tests.
875    pub fn get_pinned_nodes(&self) -> BTreeMap<Position<F>, D> {
876        self.mem.pinned_nodes()
877    }
878
879    #[cfg(test)]
880    /// Simulate a crash after pruning metadata is written but before the journal is pruned.
881    pub async fn simulate_pruning_failure(mut self, prune_to: Location<F>) -> Result<(), Error<F>> {
882        let prune_to_pos = Position::try_from(prune_to)?;
883        assert!(prune_to_pos <= self.mem.size());
884
885        // Flush items cached in the mem to disk to ensure the current state is recoverable.
886        self = self.sync().await?;
887
888        // Update metadata to reflect the desired pruning boundary, allowing for recovery in the
889        // event of a pruning failure.
890        self.update_metadata(prune_to_pos).await?;
891
892        // Don't actually prune the journal to simulate failure
893        Ok(())
894    }
895
896    /// Apply a merkleized batch to the structure.
897    ///
898    /// A batch is valid if the structure has not been modified since the batch
899    /// chain was created, or if only ancestors of this batch have been applied.
900    /// Already-committed ancestors are skipped automatically.
901    /// Applying a batch from a different fork returns [`Error::StaleBatch`].
902    pub fn apply_batch(
903        mut self,
904        batch: &batch::MerkleizedBatch<F, D, S>,
905    ) -> Result<Self, Error<F>> {
906        Arc::make_mut(&mut self.mem).apply_batch(batch)?;
907        Ok(self)
908    }
909
910    /// Create an owned [`batch::MerkleizedBatch`] representing the current committed state.
911    ///
912    /// The batch has no data (the committed items are on disk, not in memory).
913    /// This is the starting point for building owned batch chains.
914    pub(crate) fn to_batch(&self) -> Arc<batch::MerkleizedBatch<F, D, S>> {
915        batch::MerkleizedBatch::from_mem_with_strategy(&self.mem, self.strategy.clone())
916    }
917
918    /// Borrow the committed Mem for the duration of the closure.
919    pub fn with_mem<R>(&self, f: impl FnOnce(&Mem<F, D>) -> R) -> R {
920        f(&self.mem)
921    }
922
923    /// Return a zero-copy, immutable snapshot of the committed Mem.
924    ///
925    /// The snapshot never observes later mutations: mutators copy-on-write while a snapshot is
926    /// alive. Use this to move committed node fallback into a job running off the calling task
927    /// (see [`Merkle::mem`]); prefer [`Merkle::with_mem`] when a borrow suffices.
928    pub(crate) fn snapshot(&self) -> Arc<Mem<F, D>> {
929        Arc::clone(&self.mem)
930    }
931
932    /// Create a new speculative batch with this structure as its parent.
933    pub fn new_batch(&self) -> UnmerkleizedBatch<F, D, S> {
934        UnmerkleizedBatch {
935            inner: self.mem.new_batch_with_strategy(self.strategy.clone()),
936        }
937    }
938
939    /// Return a reference to the merkleization strategy.
940    pub const fn strategy(&self) -> &S {
941        &self.strategy
942    }
943
944    /// Rewind the structure by the given number of leaves.
945    ///
946    /// Adds go through the batch API ([`Self::new_batch`] / [`Self::apply_batch`]), but removing
947    /// leaves requires `rewind`. After `init` or `sync`, the in-memory structure is pruned to O(log
948    /// n) pinned nodes. A batch pop would expose new peaks that are not in memory, and `merkleize`
949    /// cannot load them because [`Readable::get_node`] is synchronous. `rewind` performs async
950    /// journal I/O to rebuild state at the target position.
951    pub(crate) async fn rewind(mut self, leaves_to_remove: usize) -> Result<Self, Error<F>> {
952        if leaves_to_remove == 0 {
953            return Ok(self);
954        }
955
956        let current_leaves = *self.leaves();
957        let destination_leaf = match current_leaves.checked_sub(leaves_to_remove as u64) {
958            Some(dest) => dest,
959            None => {
960                let pruned_to_pos = self.pruned_to_pos;
961                return Err(if pruned_to_pos == 0 {
962                    Error::Empty
963                } else {
964                    Error::ElementPruned(pruned_to_pos - 1)
965                });
966            }
967        };
968
969        let destination_loc = Location::new(destination_leaf);
970        let new_size = Position::try_from(destination_loc).expect("valid leaf");
971
972        if new_size < self.pruned_to_pos {
973            return Err(Error::ElementPruned(new_size));
974        }
975
976        // Rewind the journal if needed.
977        let journal_size = Position::<F>::new(self.journal.size());
978        if new_size < journal_size {
979            self.journal = self.journal.rewind(*new_size).await?.sync().await?;
980        }
981
982        // Truncate the in-memory structure to the target size.
983        // If the in-memory structure has been pruned past the target (e.g. after sync),
984        // rebuild from the journal/metadata instead.
985        if new_size >= Position::try_from(self.mem.bounds().start).expect("valid mem bounds start")
986        {
987            Arc::make_mut(&mut self.mem).truncate(new_size);
988        } else {
989            let mut pinned_nodes = Vec::new();
990            for pos in F::nodes_to_pin(destination_loc) {
991                pinned_nodes.push(
992                    Self::get_from_metadata_or_journal(&self.metadata, &self.journal, pos).await?,
993                );
994            }
995            let mut mem = Mem::init(MemConfig {
996                nodes: vec![],
997                pruning_boundary: destination_loc,
998                pinned_nodes,
999            })?;
1000            Self::add_extra_pinned_nodes(
1001                &mut mem,
1002                &self.metadata,
1003                &self.journal,
1004                self.pruned_to_pos,
1005            )
1006            .await?;
1007            self.mem = Arc::new(mem);
1008        }
1009
1010        Ok(self)
1011    }
1012}
1013
1014/// The [`Readable`] implementation for the full structure operates only on the in-memory
1015/// portion. After [`Merkle::sync`], nodes flushed to the journal are no longer accessible
1016/// through this interface, even though [`Merkle::bounds`] still reports them as retained.
1017impl<F: Family, E: Context, D: Digest, S: Strategy> Readable for Merkle<F, E, D, S> {
1018    type Family = F;
1019    type Digest = D;
1020
1021    fn size(&self) -> Position<F> {
1022        self.size()
1023    }
1024
1025    fn get_node(&self, pos: Position<F>) -> Option<D> {
1026        self.mem.get_node(pos)
1027    }
1028}
1029
1030impl<F: Family, E: Context, D: Digest, S: Strategy> crate::merkle::storage::Storage<F>
1031    for Merkle<F, E, D, S>
1032{
1033    type Digest = D;
1034
1035    fn size(&self) -> Position<F> {
1036        self.size()
1037    }
1038
1039    async fn get_node(&self, position: Position<F>) -> Result<Option<D>, Error<F>> {
1040        Self::get_node(self, position).await
1041    }
1042
1043    async fn get_nodes(&self, positions: &[Position<F>]) -> Result<Vec<D>, Error<F>> {
1044        Self::get_nodes(self, positions).await
1045    }
1046}
1047
1048impl<F: Family, E: Context, D: Digest, S: Strategy> Merkle<F, E, D, S> {
1049    /// Return an inclusion proof for the element at the location `loc` against a historical
1050    /// state with `leaves` leaves.
1051    ///
1052    /// The proof commits to `inactive_peaks`; peak bagging is selected by `hasher`.
1053    ///
1054    /// # Errors
1055    ///
1056    /// - Returns [Error::RangeOutOfBounds] if `leaves` is greater than `self.leaves()` or if `loc`
1057    ///   is not provable at that historical size.
1058    /// - Returns [Error::LocationOverflow] if `loc` exceeds [Family::MAX_LEAVES].
1059    /// - Returns [Error::ElementPruned] if some element needed to generate the proof has been
1060    ///   pruned.
1061    pub async fn historical_proof(
1062        &self,
1063        hasher: &impl Hasher<F, Digest = D>,
1064        leaves: Location<F>,
1065        loc: Location<F>,
1066        inactive_peaks: usize,
1067    ) -> Result<Proof<F, D>, Error<F>> {
1068        if !loc.is_valid_index() {
1069            return Err(Error::LocationOverflow(loc));
1070        }
1071        // loc is valid so it won't overflow from + 1
1072        self.historical_range_proof(hasher, leaves, loc..loc + 1, inactive_peaks)
1073            .await
1074    }
1075
1076    /// Return an inclusion proof for the elements in `range` against a historical state with
1077    /// `leaves` leaves.
1078    ///
1079    /// The proof commits to `inactive_peaks`; peak bagging is selected by `hasher`.
1080    ///
1081    /// # Errors
1082    ///
1083    /// - Returns [Error::RangeOutOfBounds] if `leaves` is greater than `self.leaves()` or if
1084    ///   `range` is not provable at that historical size.
1085    /// - Returns [Error::LocationOverflow] if any location in `range` exceeds [Family::MAX_LEAVES].
1086    /// - Returns [Error::ElementPruned] if some element needed to generate the proof has been
1087    ///   pruned.
1088    /// - Returns [Error::Empty] if the range is empty.
1089    pub async fn historical_range_proof(
1090        &self,
1091        hasher: &impl Hasher<F, Digest = D>,
1092        leaves: Location<F>,
1093        range: core::ops::Range<Location<F>>,
1094        inactive_peaks: usize,
1095    ) -> Result<Proof<F, D>, Error<F>> {
1096        if leaves > self.leaves() {
1097            return Err(Error::RangeOutOfBounds(leaves));
1098        }
1099        crate::merkle::verification::historical_range_proof(
1100            hasher,
1101            self,
1102            leaves,
1103            range,
1104            inactive_peaks,
1105        )
1106        .await
1107    }
1108
1109    /// Return an inclusion proof for the element at the location `loc` that can be verified against
1110    /// the current root.
1111    ///
1112    /// The proof commits to `inactive_peaks`; peak bagging is selected by `hasher`.
1113    ///
1114    /// Unlike the in-memory `Mem::proof`, this async method can read from the backing journal for
1115    /// nodes that have been synced out of memory.
1116    ///
1117    /// # Errors
1118    ///
1119    /// - Returns [Error::LocationOverflow] if `loc` exceeds [Family::MAX_LEAVES].
1120    /// - Returns [Error::ElementPruned] if some element needed to generate the proof has been
1121    ///   pruned.
1122    /// - Returns [Error::Empty] if the range is empty.
1123    pub async fn proof(
1124        &self,
1125        hasher: &impl Hasher<F, Digest = D>,
1126        loc: Location<F>,
1127        inactive_peaks: usize,
1128    ) -> Result<Proof<F, D>, Error<F>> {
1129        if !loc.is_valid_index() {
1130            return Err(Error::LocationOverflow(loc));
1131        }
1132        // loc is valid so it won't overflow from + 1
1133        self.range_proof(hasher, loc..loc + 1, inactive_peaks).await
1134    }
1135
1136    /// Return an inclusion proof for the elements within the specified location range.
1137    ///
1138    /// The proof commits to `inactive_peaks`; peak bagging is selected by `hasher`.
1139    ///
1140    /// Unlike the in-memory `Mem::range_proof`, this async method can read from the backing
1141    /// journal for nodes that have been synced out of memory.
1142    ///
1143    /// # Errors
1144    ///
1145    /// - Returns [Error::LocationOverflow] if any location in `range` exceeds [Family::MAX_LEAVES].
1146    /// - Returns [Error::ElementPruned] if some element needed to generate the proof has been
1147    ///   pruned.
1148    /// - Returns [Error::Empty] if the range is empty.
1149    pub async fn range_proof(
1150        &self,
1151        hasher: &impl Hasher<F, Digest = D>,
1152        range: core::ops::Range<Location<F>>,
1153        inactive_peaks: usize,
1154    ) -> Result<Proof<F, D>, Error<F>> {
1155        self.historical_range_proof(hasher, self.leaves(), range, inactive_peaks)
1156            .await
1157    }
1158}
1159
1160#[cfg(test)]
1161mod tests {
1162    use super::*;
1163    use crate::{
1164        journal::contiguous::fixed::{Config as JConfig, Journal},
1165        merkle::{
1166            Bagging::ForwardFold, Location, LocationRangeExt as _, Position, Proof,
1167            hasher::Standard, mmb, mmr,
1168        },
1169        metadata::{Config as MConfig, Metadata},
1170    };
1171    use commonware_cryptography::{
1172        Hasher as _, Sha256,
1173        sha256::{self, Digest},
1174    };
1175    use commonware_macros::test_traced;
1176    use commonware_parallel::Sequential;
1177    use commonware_runtime::{
1178        BufferPooler, Runner, Supervisor as _, buffer::paged::CacheRef, deterministic,
1179    };
1180    use commonware_utils::{NZU16, NZU64, NZUsize, non_empty_range, sequence::prefixed_u64::U64};
1181    use std::{
1182        collections::BTreeMap,
1183        num::{NonZeroU16, NonZeroUsize},
1184    };
1185
1186    fn test_digest(v: usize) -> Digest {
1187        Sha256::hash(&[&v.to_be_bytes()])
1188    }
1189
1190    const PAGE_SIZE: NonZeroU16 = NZU16!(111);
1191    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(5);
1192
1193    fn test_config(pooler: &impl BufferPooler) -> Config<Sequential> {
1194        Config {
1195            journal_partition: "journal-partition".into(),
1196            metadata_partition: "metadata-partition".into(),
1197            items_per_blob: NZU64!(7),
1198            write_buffer: NZUsize!(1024),
1199            replay_buffer: NZUsize!(1024),
1200            strategy: Sequential,
1201            page_cache: CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE),
1202        }
1203    }
1204
1205    async fn full_empty_inner<F: Family>(context: deterministic::Context) {
1206        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1207        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1208            context.child("first"),
1209            &hasher,
1210            test_config(&context),
1211        )
1212        .await
1213        .unwrap();
1214        assert_eq!(mmr.size(), 0);
1215        assert!(mmr.get_node(Position::<F>::new(0)).await.is_err());
1216        let bounds = mmr.bounds();
1217        assert!(bounds.is_empty());
1218        mmr = mmr.prune_all().await.unwrap();
1219        assert_eq!(bounds.start, 0);
1220        mmr = mmr.prune(Location::<F>::new(0)).await.unwrap();
1221        mmr = mmr.sync().await.unwrap();
1222        assert!(matches!(mmr.rewind(1).await, Err(Error::Empty)));
1223
1224        // Reopen the same partitions.
1225        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1226            context.child("reopen"),
1227            &hasher,
1228            test_config(&context),
1229        )
1230        .await
1231        .unwrap();
1232        let batch = mmr.new_batch().add(&hasher, &test_digest(0));
1233        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1234        mmr = mmr.apply_batch(&batch).unwrap();
1235        assert_eq!(mmr.size(), 1);
1236        mmr = mmr.sync().await.unwrap();
1237        assert!(mmr.get_node(Position::<F>::new(0)).await.is_ok());
1238        mmr = mmr.rewind(1).await.unwrap();
1239        assert_eq!(mmr.size(), 0);
1240        mmr.sync().await.unwrap();
1241
1242        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1243            context.child("second"),
1244            &hasher,
1245            test_config(&context),
1246        )
1247        .await
1248        .unwrap();
1249        assert_eq!(mmr.size(), 0);
1250
1251        let empty_proof = Proof::<F, Digest>::default();
1252        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1253        let root = mmr.root(&hasher, 0).unwrap();
1254        assert!(empty_proof.verify_range_inclusion(
1255            &hasher,
1256            &[] as &[Digest],
1257            Location::<F>::new(0),
1258            &root
1259        ));
1260        assert!(empty_proof.verify_multi_inclusion(
1261            &hasher,
1262            &[] as &[(Digest, Location<F>)],
1263            &root
1264        ));
1265
1266        // Confirm empty proof no longer verifies after adding an element.
1267        let batch = mmr.new_batch().add(&hasher, &test_digest(0));
1268        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1269        mmr = mmr.apply_batch(&batch).unwrap();
1270        let root = mmr.root(&hasher, 0).unwrap();
1271        assert!(!empty_proof.verify_range_inclusion(
1272            &hasher,
1273            &[] as &[Digest],
1274            Location::<F>::new(0),
1275            &root
1276        ));
1277        assert!(!empty_proof.verify_multi_inclusion(
1278            &hasher,
1279            &[] as &[(Digest, Location<F>)],
1280            &root
1281        ));
1282
1283        mmr.destroy().await.unwrap();
1284    }
1285
1286    #[test_traced]
1287    fn test_full_empty_mmr() {
1288        let executor = deterministic::Runner::default();
1289        executor.start(full_empty_inner::<mmr::Family>);
1290    }
1291
1292    #[test_traced]
1293    fn test_full_empty_mmb() {
1294        let executor = deterministic::Runner::default();
1295        executor.start(full_empty_inner::<mmb::Family>);
1296    }
1297
1298    async fn full_prune_out_of_bounds_returns_error_inner<F: Family>(
1299        context: deterministic::Context,
1300    ) {
1301        let hasher = Standard::<Sha256>::new(ForwardFold);
1302        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1303            context.child("oob_prune"),
1304            &hasher,
1305            test_config(&context),
1306        )
1307        .await
1308        .unwrap();
1309
1310        let batch = mmr.new_batch().add(&hasher, &test_digest(0));
1311        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1312        mmr = mmr.apply_batch(&batch).unwrap();
1313
1314        assert!(matches!(
1315            mmr.prune(Location::<F>::new(2)).await,
1316            Err(Error::LeafOutOfBounds(loc)) if loc == Location::<F>::new(2)
1317        ));
1318    }
1319
1320    #[test_traced]
1321    fn test_full_prune_out_of_bounds_returns_error_mmr() {
1322        let executor = deterministic::Runner::default();
1323        executor.start(full_prune_out_of_bounds_returns_error_inner::<mmr::Family>);
1324    }
1325
1326    #[test_traced]
1327    fn test_full_prune_out_of_bounds_returns_error_mmb() {
1328        let executor = deterministic::Runner::default();
1329        executor.start(full_prune_out_of_bounds_returns_error_inner::<mmb::Family>);
1330    }
1331
1332    async fn full_rewind_error_leaves_valid_state_inner<F: Family>(
1333        context: deterministic::Context,
1334    ) {
1335        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1336
1337        // Case 1: rewind partially succeeds, then returns ElementPruned.
1338        let element_pruned_context = context.child("element_pruned_case");
1339        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1340            element_pruned_context.child("element_pruned"),
1341            &hasher,
1342            test_config(&element_pruned_context),
1343        )
1344        .await
1345        .unwrap();
1346        let mut batch = mmr.new_batch();
1347        for i in 0u64..32 {
1348            batch = batch.add(&hasher, &i.to_be_bytes());
1349        }
1350        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1351        mmr = mmr.apply_batch(&batch).unwrap();
1352        let mmr = mmr.prune(Location::<F>::new(8)).await.unwrap();
1353        let leaves_before = mmr.leaves();
1354        assert!(matches!(
1355            mmr.rewind(128).await,
1356            Err(Error::ElementPruned(_))
1357        ));
1358
1359        // The failed rewind mutated nothing durable; reopening recovers the synced state.
1360        let mmr = Merkle::<F, _, Digest, Sequential>::init(
1361            element_pruned_context.child("element_pruned_reopen"),
1362            &hasher,
1363            test_config(&element_pruned_context),
1364        )
1365        .await
1366        .unwrap();
1367        assert_eq!(mmr.leaves(), leaves_before);
1368        mmr.destroy().await.unwrap();
1369
1370        // Case 2: rewind underflows and returns Empty without removing any leaves.
1371        let empty_context = context.child("empty_case");
1372        let cfg = Config {
1373            journal_partition: "empty-journal-partition".into(),
1374            metadata_partition: "empty-metadata-partition".into(),
1375            ..test_config(&empty_context)
1376        };
1377        let mut mmr =
1378            Merkle::<F, _, Digest, Sequential>::init(empty_context.child("open"), &hasher, cfg)
1379                .await
1380                .unwrap();
1381        let mut batch = mmr.new_batch();
1382        for i in 0u64..8 {
1383            batch = batch.add(&hasher, &i.to_be_bytes());
1384        }
1385        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1386        mmr = mmr.apply_batch(&batch).unwrap();
1387        mmr = mmr.sync().await.unwrap();
1388        assert!(matches!(mmr.rewind(9).await, Err(Error::Empty)));
1389
1390        // Reopen: the underflowing rewind persisted nothing.
1391        let cfg = Config {
1392            journal_partition: "empty-journal-partition".into(),
1393            metadata_partition: "empty-metadata-partition".into(),
1394            ..test_config(&empty_context)
1395        };
1396        let mmr =
1397            Merkle::<F, _, Digest, Sequential>::init(empty_context.child("reopen"), &hasher, cfg)
1398                .await
1399                .unwrap();
1400        assert_eq!(mmr.leaves(), Location::<F>::new(8));
1401        mmr.destroy().await.unwrap();
1402    }
1403
1404    #[test_traced]
1405    fn test_full_rewind_error_leaves_valid_state_mmr() {
1406        let executor = deterministic::Runner::default();
1407        executor.start(full_rewind_error_leaves_valid_state_inner::<mmr::Family>);
1408    }
1409
1410    #[test_traced]
1411    fn test_full_rewind_error_leaves_valid_state_mmb() {
1412        let executor = deterministic::Runner::default();
1413        executor.start(full_rewind_error_leaves_valid_state_inner::<mmb::Family>);
1414    }
1415
1416    async fn full_basic_inner<F: Family>(context: deterministic::Context) {
1417        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1418        let cfg = test_config(&context);
1419        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(context, &hasher, cfg)
1420            .await
1421            .unwrap();
1422        // Build a test structure with 255 leaves
1423        const LEAF_COUNT: usize = 255;
1424        let mut leaves = Vec::with_capacity(LEAF_COUNT);
1425        for i in 0..LEAF_COUNT {
1426            leaves.push(test_digest(i));
1427        }
1428        let mut batch = mmr.new_batch();
1429        for leaf in &leaves {
1430            batch = batch.add(&hasher, leaf);
1431        }
1432        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1433        mmr = mmr.apply_batch(&batch).unwrap();
1434        let expected_size = Position::<F>::try_from(Location::<F>::new(LEAF_COUNT as u64)).unwrap();
1435        assert_eq!(mmr.size(), expected_size);
1436
1437        // Generate & verify proof from element that is not yet flushed to the journal.
1438        const TEST_ELEMENT: usize = 133;
1439        let test_element_loc: Location<F> = Location::new(TEST_ELEMENT as u64);
1440
1441        let proof = mmr.proof(&hasher, test_element_loc, 0).await.unwrap();
1442        let root = mmr.root(&hasher, 0).unwrap();
1443        assert!(proof.verify_element_inclusion(
1444            &hasher,
1445            &leaves[TEST_ELEMENT],
1446            test_element_loc,
1447            &root
1448        ));
1449
1450        // Sync the structure, make sure it flushes the in-mem structure as expected.
1451        mmr = mmr.sync().await.unwrap();
1452
1453        // Now that the element is flushed from the in-mem structure, confirm its proof is still
1454        // generated correctly.
1455        let proof2 = mmr.proof(&hasher, test_element_loc, 0).await.unwrap();
1456        assert_eq!(proof, proof2);
1457
1458        // Generate & verify a proof that spans flushed elements and the last element.
1459        let range = Location::<F>::new(TEST_ELEMENT as u64)..Location::<F>::new(LEAF_COUNT as u64);
1460        let proof = mmr.range_proof(&hasher, range.clone(), 0).await.unwrap();
1461        assert!(proof.verify_range_inclusion(
1462            &hasher,
1463            &leaves[range.to_usize_range()],
1464            test_element_loc,
1465            &root
1466        ));
1467
1468        mmr.destroy().await.unwrap();
1469    }
1470
1471    /// `get_nodes` must agree with per-position `get_node` on every available position
1472    /// (journal-resident and memory-resident) and reject the positions `get_node` reports
1473    /// as absent.
1474    async fn full_get_nodes_matches_get_node_inner<F: Family>(context: deterministic::Context) {
1475        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1476        let cfg = test_config(&context);
1477        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(context, &hasher, cfg)
1478            .await
1479            .unwrap();
1480
1481        // Flushed leaves (journal-resident after sync), a pruned prefix, then unflushed
1482        // leaves on top (memory-resident).
1483        const LEAF_COUNT: usize = 200;
1484        let mut batch = mmr.new_batch();
1485        for i in 0..LEAF_COUNT {
1486            batch = batch.add(&hasher, &test_digest(i));
1487        }
1488        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1489        mmr = mmr.apply_batch(&batch).unwrap();
1490        mmr = mmr.sync().await.unwrap();
1491        mmr = mmr.prune(Location::<F>::new(50)).await.unwrap();
1492        let mut batch = mmr.new_batch();
1493        for i in LEAF_COUNT..LEAF_COUNT + 10 {
1494            batch = batch.add(&hasher, &test_digest(i));
1495        }
1496        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1497        mmr = mmr.apply_batch(&batch).unwrap();
1498
1499        // Partition by what `get_node` reports, so both APIs are judged against the same
1500        // notion of availability.
1501        let all: Vec<Position<F>> = (0..*mmr.size()).map(Position::new).collect();
1502        let mut absent = Vec::new();
1503        let mut available = Vec::new();
1504        for &position in &all {
1505            match mmr.get_node(position).await.unwrap() {
1506                Some(node) => available.push((position, node)),
1507                None => absent.push(position),
1508            }
1509        }
1510        assert!(!absent.is_empty(), "expected some pruned positions");
1511        assert!(!available.is_empty(), "expected some available positions");
1512
1513        // Spanning the pruning boundary is an error naming a position `get_node` calls absent.
1514        match mmr.get_nodes(&all).await {
1515            Err(Error::ElementPruned(position)) => {
1516                assert!(absent.contains(&position), "position {position}")
1517            }
1518            other => panic!("expected ElementPruned, got {other:?}"),
1519        }
1520
1521        // Every available position, then a sparse subset (slot correspondence), then empty.
1522        let positions: Vec<Position<F>> = available.iter().map(|&(pos, _)| pos).collect();
1523        let batched = mmr.get_nodes(&positions).await.unwrap();
1524        assert_eq!(batched.len(), available.len());
1525        for (slot, &(position, node)) in available.iter().enumerate() {
1526            assert_eq!(batched[slot], node, "position {position}");
1527        }
1528
1529        let sparse: Vec<Position<F>> = positions.iter().copied().step_by(7).collect();
1530        let batched = mmr.get_nodes(&sparse).await.unwrap();
1531        for (slot, &position) in sparse.iter().enumerate() {
1532            let single = mmr.get_node(position).await.unwrap().unwrap();
1533            assert_eq!(batched[slot], single, "position {position}");
1534        }
1535
1536        assert!(mmr.get_nodes(&[]).await.unwrap().is_empty());
1537        mmr.destroy().await.unwrap();
1538    }
1539
1540    #[test_traced]
1541    fn test_full_get_nodes_matches_get_node_mmr() {
1542        let executor = deterministic::Runner::default();
1543        executor.start(full_get_nodes_matches_get_node_inner::<mmr::Family>);
1544    }
1545
1546    #[test_traced]
1547    fn test_full_get_nodes_matches_get_node_mmb() {
1548        let executor = deterministic::Runner::default();
1549        executor.start(full_get_nodes_matches_get_node_inner::<mmb::Family>);
1550    }
1551
1552    #[test_traced]
1553    fn test_full_basic_mmr() {
1554        let executor = deterministic::Runner::default();
1555        executor.start(full_basic_inner::<mmr::Family>);
1556    }
1557
1558    #[test_traced]
1559    fn test_full_basic_mmb() {
1560        let executor = deterministic::Runner::default();
1561        executor.start(full_basic_inner::<mmb::Family>);
1562    }
1563
1564    /// Flush moves cached nodes into the journal and prunes them from memory, preserving reads,
1565    /// proofs, and the root.
1566    async fn full_flush_inner<F: Family>(context: deterministic::Context) {
1567        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1568        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1569            context.child("first"),
1570            &hasher,
1571            test_config(&context),
1572        )
1573        .await
1574        .unwrap();
1575
1576        const LEAF_COUNT: usize = 100;
1577        let mut leaves = Vec::with_capacity(LEAF_COUNT);
1578        for i in 0..LEAF_COUNT {
1579            leaves.push(test_digest(i));
1580        }
1581        let mut batch = mmr.new_batch();
1582        for leaf in &leaves {
1583            batch = batch.add(&hasher, leaf);
1584        }
1585        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1586        mmr = mmr.apply_batch(&batch).unwrap();
1587        let expected_size = Position::<F>::try_from(Location::<F>::new(LEAF_COUNT as u64)).unwrap();
1588        let root = mmr.root(&hasher, 0).unwrap();
1589
1590        // Flush writes all cached nodes to the journal and prunes them from the mem.
1591        mmr = mmr.flush().await.unwrap();
1592        assert_eq!(Position::<F>::new(mmr.journal.size()), expected_size);
1593        assert_eq!(mmr.size(), expected_size);
1594        assert_eq!(
1595            mmr.with_mem(|mem| mem.bounds().start),
1596            Location::<F>::new(LEAF_COUNT as u64)
1597        );
1598
1599        // Flushing again is a no-op.
1600        mmr = mmr.flush().await.unwrap();
1601        assert_eq!(Position::<F>::new(mmr.journal.size()), expected_size);
1602
1603        // Flushed nodes remain readable and provable, and the root is unchanged.
1604        assert_eq!(mmr.root(&hasher, 0).unwrap(), root);
1605        const TEST_ELEMENT: usize = 42;
1606        let loc: Location<F> = Location::new(TEST_ELEMENT as u64);
1607        let proof = mmr.proof(&hasher, loc, 0).await.unwrap();
1608        assert!(proof.verify_element_inclusion(&hasher, &leaves[TEST_ELEMENT], loc, &root));
1609
1610        // Sync after flush succeeds (and must fsync the journal even though there is nothing
1611        // left to flush).
1612        mmr = mmr.sync().await.unwrap();
1613
1614        mmr.destroy().await.unwrap();
1615    }
1616
1617    #[test_traced]
1618    fn test_full_flush_mmr() {
1619        let executor = deterministic::Runner::default();
1620        executor.start(full_flush_inner::<mmr::Family>);
1621    }
1622
1623    #[test_traced]
1624    fn test_full_flush_mmb() {
1625        let executor = deterministic::Runner::default();
1626        executor.start(full_flush_inner::<mmb::Family>);
1627    }
1628
1629    /// Flushed-but-unsynced nodes in the tail blob are lost on a crash, while synced nodes
1630    /// survive: the durability barrier comes from `sync`, not `flush`.
1631    fn full_flush_crash_inner<F: Family>() {
1632        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1633
1634        // Phase 1: durably sync a first batch of leaves, flush (but don't sync) a second batch into
1635        // the tail blob, then simulate an unclean shutdown.
1636        let executor = deterministic::Runner::default();
1637        let (synced_size, checkpoint) = executor.start_and_recover(|context| async move {
1638            let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1639            let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1640                context.child("first"),
1641                &hasher,
1642                Config {
1643                    // 512 items per blob keeps both batches in the tail blob: a rollover would
1644                    // fsync the sealed predecessor and defeat the flushed-but-unsynced scenario.
1645                    items_per_blob: NZU64!(512),
1646                    ..test_config(&context)
1647                },
1648            )
1649            .await
1650            .unwrap();
1651
1652            let mut batch = mmr.new_batch();
1653            for i in 0..50usize {
1654                batch = batch.add(&hasher, &test_digest(i));
1655            }
1656            let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1657            mmr = mmr.apply_batch(&batch).unwrap();
1658            let mut mmr = mmr.sync().await.unwrap();
1659            let synced_size = mmr.size();
1660
1661            let mut batch = mmr.new_batch();
1662            for i in 50..100usize {
1663                batch = batch.add(&hasher, &test_digest(i));
1664            }
1665            let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1666            mmr = mmr.apply_batch(&batch).unwrap();
1667            mmr = mmr.flush().await.unwrap();
1668            assert_eq!(Position::<F>::new(mmr.journal.size()), mmr.size());
1669
1670            synced_size
1671        });
1672
1673        // Phase 2: recover. Only the synced prefix survives.
1674        let executor = deterministic::Runner::from(checkpoint);
1675        executor.start(|context| async move {
1676            let mmr = Merkle::<F, _, Digest, Sequential>::init(
1677                context.child("second"),
1678                &hasher,
1679                Config {
1680                    // 512 items per blob keeps both batches in the tail blob: a rollover would
1681                    // fsync the sealed predecessor and defeat the flushed-but-unsynced scenario.
1682                    items_per_blob: NZU64!(512),
1683                    ..test_config(&context)
1684                },
1685            )
1686            .await
1687            .unwrap();
1688            assert_eq!(mmr.size(), synced_size);
1689        });
1690    }
1691
1692    #[test_traced]
1693    fn test_full_flush_crash_recovery_mmr() {
1694        full_flush_crash_inner::<mmr::Family>();
1695    }
1696
1697    #[test_traced]
1698    fn test_full_flush_crash_recovery_mmb() {
1699        full_flush_crash_inner::<mmb::Family>();
1700    }
1701
1702    /// A sync after a flush must make the flushed nodes durable, even though the flush left
1703    /// nothing further to write from memory.
1704    fn full_flush_then_sync_crash_inner<F: Family>() {
1705        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1706
1707        // Phase 1: flush a batch of leaves, then sync with nothing left to flush, then simulate
1708        // an unclean shutdown.
1709        let executor = deterministic::Runner::default();
1710        let (full_size, checkpoint) = executor.start_and_recover(|context| async move {
1711            let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1712            let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1713                context.child("first"),
1714                &hasher,
1715                test_config(&context),
1716            )
1717            .await
1718            .unwrap();
1719
1720            let mut batch = mmr.new_batch();
1721            for i in 0..100usize {
1722                batch = batch.add(&hasher, &test_digest(i));
1723            }
1724            let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1725            mmr = mmr.apply_batch(&batch).unwrap();
1726            mmr = mmr.flush().await.unwrap();
1727            let mmr = mmr.sync().await.unwrap();
1728
1729            mmr.size()
1730        });
1731
1732        // Phase 2: recover. Everything survives.
1733        let executor = deterministic::Runner::from(checkpoint);
1734        executor.start(|context| async move {
1735            let mmr = Merkle::<F, _, Digest, Sequential>::init(
1736                context.child("second"),
1737                &hasher,
1738                test_config(&context),
1739            )
1740            .await
1741            .unwrap();
1742            assert_eq!(mmr.size(), full_size);
1743        });
1744    }
1745
1746    #[test_traced]
1747    fn test_full_flush_then_sync_crash_recovery_mmr() {
1748        full_flush_then_sync_crash_inner::<mmr::Family>();
1749    }
1750
1751    #[test_traced]
1752    fn test_full_flush_then_sync_crash_recovery_mmb() {
1753        full_flush_then_sync_crash_inner::<mmb::Family>();
1754    }
1755
1756    /// Generates a stateful structure, simulates a crash that wrote a leaf but not its parent
1757    /// nodes, and confirms we appropriately recover to a valid state.
1758    async fn full_recovery_inner<F: Family>(context: deterministic::Context) {
1759        use crate::journal::contiguous::fixed::{Config as JConfig, Journal};
1760
1761        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1762        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1763            context.child("first"),
1764            &hasher,
1765            test_config(&context),
1766        )
1767        .await
1768        .unwrap();
1769        assert_eq!(mmr.size(), 0);
1770
1771        // Build a test structure with 252 leaves
1772        const LEAF_COUNT: usize = 252;
1773        let mut leaves = Vec::with_capacity(LEAF_COUNT);
1774        for i in 0..LEAF_COUNT {
1775            leaves.push(test_digest(i));
1776        }
1777        let mut batch = mmr.new_batch();
1778        for leaf in &leaves {
1779            batch = batch.add(&hasher, leaf);
1780        }
1781        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1782        mmr = mmr.apply_batch(&batch).unwrap();
1783        let expected_size = Position::<F>::try_from(Location::<F>::new(LEAF_COUNT as u64)).unwrap();
1784        assert_eq!(mmr.size(), expected_size);
1785        let mmr = mmr.sync().await.unwrap();
1786        drop(mmr);
1787
1788        // Simulate a crash that wrote a leaf but not its parent nodes by appending one
1789        // extra digest to the journal. This creates an invalid structure size.
1790        {
1791            let journal: Journal<_, Digest> = Journal::init(
1792                context.child("corrupt"),
1793                JConfig {
1794                    partition: "journal-partition".into(),
1795                    items_per_blob: NZU64!(7),
1796                    write_buffer: NZUsize!(1024),
1797                    replay_buffer: NZUsize!(1024),
1798                    page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
1799                },
1800            )
1801            .await
1802            .unwrap();
1803            assert_eq!(journal.size(), expected_size);
1804            let (journal, _) = journal.append(&Sha256::hash(&[b"orphan"])).await.unwrap();
1805            let journal = journal.sync().await.unwrap();
1806            assert_eq!(journal.size(), expected_size + 1);
1807        }
1808
1809        let mmr = Merkle::<F, _, Digest, Sequential>::init(
1810            context.child("second"),
1811            &hasher,
1812            test_config(&context),
1813        )
1814        .await
1815        .unwrap();
1816        // Since the orphaned leaf is replayed, the structure recovers to the previous valid state
1817        // plus the new leaf.
1818        let recovered_size =
1819            Position::<F>::try_from(Location::<F>::new(LEAF_COUNT as u64 + 1)).unwrap();
1820        assert_eq!(mmr.size(), recovered_size);
1821
1822        // Make sure dropping it and re-opening it persists the recovered state.
1823        drop(mmr);
1824        let mmr = Merkle::<F, _, Digest, Sequential>::init(
1825            context.child("third"),
1826            &hasher,
1827            test_config(&context),
1828        )
1829        .await
1830        .unwrap();
1831        assert_eq!(mmr.size(), recovered_size);
1832
1833        mmr.destroy().await.unwrap();
1834    }
1835
1836    #[test_traced]
1837    fn test_full_recovery_mmr() {
1838        let executor = deterministic::Runner::default();
1839        executor.start(full_recovery_inner::<mmr::Family>);
1840    }
1841
1842    #[test_traced]
1843    fn test_full_recovery_mmb() {
1844        let executor = deterministic::Runner::default();
1845        executor.start(full_recovery_inner::<mmb::Family>);
1846    }
1847
1848    async fn full_pruning_inner<F: Family>(context: deterministic::Context) {
1849        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1850        // make sure pruning doesn't break root computation, adding of new nodes, etc.
1851        const LEAF_COUNT: usize = 2000;
1852        let cfg_pruned = test_config(&context);
1853        let mut pruned_mmr = Merkle::<F, _, Digest, Sequential>::init(
1854            context.child("pruned"),
1855            &hasher,
1856            cfg_pruned.clone(),
1857        )
1858        .await
1859        .unwrap();
1860        let cfg_unpruned = Config {
1861            journal_partition: "unpruned-journal-partition".into(),
1862            metadata_partition: "unpruned-metadata-partition".into(),
1863            items_per_blob: NZU64!(7),
1864            write_buffer: NZUsize!(1024),
1865            replay_buffer: NZUsize!(1024),
1866            strategy: Sequential,
1867            page_cache: cfg_pruned.page_cache.clone(),
1868        };
1869        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
1870            context.child("unpruned"),
1871            &hasher,
1872            cfg_unpruned,
1873        )
1874        .await
1875        .unwrap();
1876        let mut leaves = Vec::with_capacity(LEAF_COUNT);
1877        for i in 0..LEAF_COUNT {
1878            leaves.push(test_digest(i));
1879        }
1880        let mut batch = mmr.new_batch();
1881        for leaf in &leaves {
1882            batch = batch.add(&hasher, leaf);
1883        }
1884        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1885        mmr = mmr.apply_batch(&batch).unwrap();
1886        let mut batch = pruned_mmr.new_batch();
1887        for leaf in &leaves {
1888            batch = batch.add(&hasher, leaf);
1889        }
1890        let batch = pruned_mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1891        pruned_mmr = pruned_mmr.apply_batch(&batch).unwrap();
1892        let expected_size = Position::<F>::try_from(Location::<F>::new(LEAF_COUNT as u64)).unwrap();
1893        assert_eq!(mmr.size(), expected_size);
1894        assert_eq!(pruned_mmr.size(), expected_size);
1895
1896        // Prune the structure in increments of 10 making sure the journal is still able to compute
1897        // roots and accept new elements.
1898        for i in 0usize..300 {
1899            let prune_loc = Location::<F>::new(std::cmp::min(i as u64 * 10, *pruned_mmr.leaves()));
1900            pruned_mmr = pruned_mmr.prune(prune_loc).await.unwrap();
1901            assert_eq!(prune_loc, pruned_mmr.bounds().start);
1902
1903            let digest = test_digest(LEAF_COUNT + i);
1904            leaves.push(digest);
1905            let last_leaf = leaves.last().unwrap();
1906            let batch = pruned_mmr.new_batch().add(&hasher, last_leaf);
1907            let batch = pruned_mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1908            pruned_mmr = pruned_mmr.apply_batch(&batch).unwrap();
1909            let batch = mmr.new_batch().add(&hasher, last_leaf);
1910            let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1911            mmr = mmr.apply_batch(&batch).unwrap();
1912            assert_eq!(
1913                pruned_mmr.root(&hasher, 0).unwrap(),
1914                mmr.root(&hasher, 0).unwrap()
1915            );
1916        }
1917
1918        // Sync the structures.
1919        pruned_mmr = pruned_mmr.sync().await.unwrap();
1920        assert_eq!(
1921            pruned_mmr.root(&hasher, 0).unwrap(),
1922            mmr.root(&hasher, 0).unwrap()
1923        );
1924
1925        // Sync the structure & reopen.
1926        pruned_mmr.sync().await.unwrap();
1927        let mut pruned_mmr = Merkle::<F, _, Digest, Sequential>::init(
1928            context.child("pruned_reopen"),
1929            &hasher,
1930            cfg_pruned.clone(),
1931        )
1932        .await
1933        .unwrap();
1934        assert_eq!(
1935            pruned_mmr.root(&hasher, 0).unwrap(),
1936            mmr.root(&hasher, 0).unwrap()
1937        );
1938
1939        // Prune everything.
1940        let size = pruned_mmr.size();
1941        pruned_mmr = pruned_mmr.prune_all().await.unwrap();
1942        assert_eq!(
1943            pruned_mmr.root(&hasher, 0).unwrap(),
1944            mmr.root(&hasher, 0).unwrap()
1945        );
1946        let bounds = pruned_mmr.bounds();
1947        assert!(bounds.is_empty());
1948        assert_eq!(bounds.start, Location::<F>::try_from(size).unwrap());
1949
1950        // Close structure after adding a new node without syncing and make sure state is as
1951        // expected on reopening.
1952        let batch = mmr.new_batch().add(&hasher, &test_digest(LEAF_COUNT));
1953        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1954        mmr = mmr.apply_batch(&batch).unwrap();
1955        let batch = pruned_mmr
1956            .new_batch()
1957            .add(&hasher, &test_digest(LEAF_COUNT));
1958        let batch = pruned_mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1959        pruned_mmr = pruned_mmr.apply_batch(&batch).unwrap();
1960        assert!(*pruned_mmr.size() % cfg_pruned.items_per_blob != 0);
1961        pruned_mmr.sync().await.unwrap();
1962        let mut pruned_mmr = Merkle::<F, _, Digest, Sequential>::init(
1963            context.child("pruned_reopen").with_attribute("index", 2),
1964            &hasher,
1965            cfg_pruned.clone(),
1966        )
1967        .await
1968        .unwrap();
1969        assert_eq!(
1970            pruned_mmr.root(&hasher, 0).unwrap(),
1971            mmr.root(&hasher, 0).unwrap()
1972        );
1973        let bounds = pruned_mmr.bounds();
1974        assert!(!bounds.is_empty());
1975        assert_eq!(bounds.start, Location::<F>::try_from(size).unwrap());
1976
1977        // Make sure pruning to older location is a no-op.
1978        pruned_mmr = pruned_mmr
1979            .prune(Location::<F>::try_from(size).unwrap() - 1)
1980            .await
1981            .unwrap();
1982        assert_eq!(
1983            pruned_mmr.bounds().start,
1984            Location::<F>::try_from(size).unwrap()
1985        );
1986
1987        // Add nodes until we are on a blob boundary, and confirm prune_all still removes all
1988        // retained nodes.
1989        while *pruned_mmr.size() % cfg_pruned.items_per_blob != 0 {
1990            let batch = pruned_mmr
1991                .new_batch()
1992                .add(&hasher, &test_digest(LEAF_COUNT));
1993            let batch = pruned_mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
1994            pruned_mmr = pruned_mmr.apply_batch(&batch).unwrap();
1995        }
1996        pruned_mmr = pruned_mmr.prune_all().await.unwrap();
1997        assert!(pruned_mmr.bounds().is_empty());
1998
1999        pruned_mmr.destroy().await.unwrap();
2000        mmr.destroy().await.unwrap();
2001    }
2002
2003    #[test_traced]
2004    fn test_full_pruning_mmr() {
2005        let executor = deterministic::Runner::default();
2006        executor.start(full_pruning_inner::<mmr::Family>);
2007    }
2008
2009    #[test_traced]
2010    fn test_full_pruning_mmb() {
2011        let executor = deterministic::Runner::default();
2012        executor.start(full_pruning_inner::<mmb::Family>);
2013    }
2014
2015    /// Simulate partial writes after pruning, making sure we recover to a valid state.
2016    async fn full_recovery_with_pruning_inner<F: Family>(context: deterministic::Context) {
2017        // Build structure with 2000 leaves.
2018        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
2019        const LEAF_COUNT: usize = 2000;
2020        let mut leaves = Vec::with_capacity(LEAF_COUNT);
2021        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2022            context.child("init"),
2023            &hasher,
2024            test_config(&context),
2025        )
2026        .await
2027        .unwrap();
2028        for i in 0..LEAF_COUNT {
2029            leaves.push(test_digest(i));
2030        }
2031        let mut batch = mmr.new_batch();
2032        for leaf in &leaves {
2033            batch = batch.add(&hasher, leaf);
2034        }
2035        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2036        mmr = mmr.apply_batch(&batch).unwrap();
2037        let expected_size = Position::<F>::try_from(Location::<F>::new(LEAF_COUNT as u64)).unwrap();
2038        assert_eq!(mmr.size(), expected_size);
2039        let mmr = mmr.sync().await.unwrap();
2040        drop(mmr);
2041
2042        // Prune the structure in increments of 50, simulating a partial write after each prune.
2043        for i in 0usize..200 {
2044            let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2045                context.child("iter").with_attribute("index", i),
2046                &hasher,
2047                test_config(&context),
2048            )
2049            .await
2050            .unwrap();
2051            let start_size = mmr.size();
2052            let start_leaves = *mmr.leaves();
2053            let prune_loc = Location::<F>::new(std::cmp::min(i as u64 * 50, start_leaves));
2054            if i % 5 == 0 {
2055                mmr.simulate_pruning_failure(prune_loc).await.unwrap();
2056                continue;
2057            }
2058            mmr = mmr.prune(prune_loc).await.unwrap();
2059
2060            // add new elements, simulating a partial write after each.
2061            for j in 0..10 {
2062                let digest = test_digest(100 * (i + 1) + j);
2063                leaves.push(digest);
2064                let batch = mmr
2065                    .new_batch()
2066                    .add(&hasher, leaves.last().unwrap())
2067                    .add(&hasher, leaves.last().unwrap());
2068                let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2069                mmr = mmr.apply_batch(&batch).unwrap();
2070                let digest = test_digest(LEAF_COUNT + i);
2071                leaves.push(digest);
2072                let batch = mmr
2073                    .new_batch()
2074                    .add(&hasher, leaves.last().unwrap())
2075                    .add(&hasher, leaves.last().unwrap());
2076                let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2077                mmr = mmr.apply_batch(&batch).unwrap();
2078            }
2079            let end_size = mmr.size();
2080            let total_to_write = (*end_size - *start_size) as usize;
2081            let partial_write_limit = i % total_to_write;
2082            mmr.simulate_partial_sync(partial_write_limit)
2083                .await
2084                .unwrap();
2085        }
2086
2087        let mmr = Merkle::<F, _, Digest, Sequential>::init(
2088            context.child("final"),
2089            &hasher,
2090            test_config(&context),
2091        )
2092        .await
2093        .unwrap();
2094        mmr.destroy().await.unwrap();
2095    }
2096
2097    #[test_traced("WARN")]
2098    fn test_full_recovery_with_pruning_mmr() {
2099        let executor = deterministic::Runner::default();
2100        executor.start(full_recovery_with_pruning_inner::<mmr::Family>);
2101    }
2102
2103    #[test_traced("WARN")]
2104    fn test_full_recovery_with_pruning_mmb() {
2105        let executor = deterministic::Runner::default();
2106        executor.start(full_recovery_with_pruning_inner::<mmb::Family>);
2107    }
2108
2109    async fn full_historical_proof_basic_inner<F: Family>(context: deterministic::Context) {
2110        // Create structure with 10 elements
2111        let hasher = Standard::<Sha256>::new(ForwardFold);
2112        let cfg = test_config(&context);
2113        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(context, &hasher, cfg)
2114            .await
2115            .unwrap();
2116        let mut elements = Vec::new();
2117        for i in 0..10 {
2118            elements.push(test_digest(i));
2119        }
2120        let mut batch = mmr.new_batch();
2121        for elt in &elements {
2122            batch = batch.add(&hasher, elt);
2123        }
2124        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2125        mmr = mmr.apply_batch(&batch).unwrap();
2126        let original_leaves = mmr.leaves();
2127
2128        // Historical proof should match "regular" proof when historical size == current database size
2129        let historical_proof = mmr
2130            .historical_range_proof(
2131                &hasher,
2132                original_leaves,
2133                Location::<F>::new(2)..Location::<F>::new(6),
2134                0,
2135            )
2136            .await
2137            .unwrap();
2138        assert_eq!(historical_proof.leaves, original_leaves);
2139        let root = mmr.root(&hasher, 0).unwrap();
2140        assert!(historical_proof.verify_range_inclusion(
2141            &hasher,
2142            &elements[2..6],
2143            Location::<F>::new(2),
2144            &root
2145        ));
2146        let regular_proof = mmr
2147            .range_proof(&hasher, Location::<F>::new(2)..Location::<F>::new(6), 0)
2148            .await
2149            .unwrap();
2150        assert_eq!(regular_proof.leaves, historical_proof.leaves);
2151        assert_eq!(regular_proof.digests, historical_proof.digests);
2152
2153        // Add more elements to the structure
2154        for i in 10..20 {
2155            elements.push(test_digest(i));
2156        }
2157        let mut batch = mmr.new_batch();
2158        for elt in &elements[10..20] {
2159            batch = batch.add(&hasher, elt);
2160        }
2161        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2162        mmr = mmr.apply_batch(&batch).unwrap();
2163        let new_historical_proof = mmr
2164            .historical_range_proof(
2165                &hasher,
2166                original_leaves,
2167                Location::<F>::new(2)..Location::<F>::new(6),
2168                0,
2169            )
2170            .await
2171            .unwrap();
2172        assert_eq!(new_historical_proof.leaves, historical_proof.leaves);
2173        assert_eq!(new_historical_proof.digests, historical_proof.digests);
2174
2175        mmr.destroy().await.unwrap();
2176    }
2177
2178    #[test_traced]
2179    fn test_full_historical_proof_basic_mmr() {
2180        let executor = deterministic::Runner::default();
2181        executor.start(full_historical_proof_basic_inner::<mmr::Family>);
2182    }
2183
2184    #[test_traced]
2185    fn test_full_historical_proof_basic_mmb() {
2186        let executor = deterministic::Runner::default();
2187        executor.start(full_historical_proof_basic_inner::<mmb::Family>);
2188    }
2189
2190    async fn full_historical_proof_with_pruning_inner<F: Family>(context: deterministic::Context) {
2191        let hasher = Standard::<Sha256>::new(ForwardFold);
2192        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2193            context.child("main"),
2194            &hasher,
2195            test_config(&context),
2196        )
2197        .await
2198        .unwrap();
2199
2200        // Add many elements
2201        let mut elements = Vec::new();
2202        for i in 0..50 {
2203            elements.push(test_digest(i));
2204        }
2205        let mut batch = mmr.new_batch();
2206        for elt in &elements {
2207            batch = batch.add(&hasher, elt);
2208        }
2209        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2210        mmr = mmr.apply_batch(&batch).unwrap();
2211
2212        // Prune to leaf 16 (position 30)
2213        let prune_loc = Location::<F>::new(16);
2214        let mmr = mmr.prune(prune_loc).await.unwrap();
2215
2216        // Create reference structure for verification to get correct size
2217        let mut ref_mmr = Merkle::<F, _, Digest, Sequential>::init(
2218            context.child("ref"),
2219            &hasher,
2220            Config {
2221                journal_partition: "ref-journal-pruned".into(),
2222                metadata_partition: "ref-metadata-pruned".into(),
2223                items_per_blob: NZU64!(7),
2224                write_buffer: NZUsize!(1024),
2225                replay_buffer: NZUsize!(1024),
2226                strategy: Sequential,
2227                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
2228            },
2229        )
2230        .await
2231        .unwrap();
2232
2233        let mut batch = ref_mmr.new_batch();
2234        for elt in elements.iter().take(41) {
2235            batch = batch.add(&hasher, elt);
2236        }
2237        let batch = ref_mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2238        ref_mmr = ref_mmr.apply_batch(&batch).unwrap();
2239        let historical_leaves = ref_mmr.leaves();
2240        let historical_root = ref_mmr.root(&hasher, 0).unwrap();
2241
2242        // Test proof at historical position after pruning
2243        let historical_proof = mmr
2244            .historical_range_proof(
2245                &hasher,
2246                historical_leaves,
2247                Location::<F>::new(35)..Location::<F>::new(39),
2248                0,
2249            )
2250            .await
2251            .unwrap();
2252
2253        assert_eq!(historical_proof.leaves, historical_leaves);
2254
2255        // Verify proof works despite pruning
2256        assert!(historical_proof.verify_range_inclusion(
2257            &hasher,
2258            &elements[35..39],
2259            Location::<F>::new(35),
2260            &historical_root
2261        ));
2262
2263        ref_mmr.destroy().await.unwrap();
2264        mmr.destroy().await.unwrap();
2265    }
2266
2267    #[test_traced]
2268    fn test_full_historical_proof_with_pruning_mmr() {
2269        let executor = deterministic::Runner::default();
2270        executor.start(full_historical_proof_with_pruning_inner::<mmr::Family>);
2271    }
2272
2273    #[test_traced]
2274    fn test_full_historical_proof_with_pruning_mmb() {
2275        let executor = deterministic::Runner::default();
2276        executor.start(full_historical_proof_with_pruning_inner::<mmb::Family>);
2277    }
2278
2279    async fn full_historical_proof_large_inner<F: Family>(context: deterministic::Context) {
2280        let hasher = Standard::<Sha256>::new(ForwardFold);
2281
2282        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2283            context.child("server"),
2284            &hasher,
2285            Config {
2286                journal_partition: "server-journal".into(),
2287                metadata_partition: "server-metadata".into(),
2288                items_per_blob: NZU64!(7),
2289                write_buffer: NZUsize!(1024),
2290                replay_buffer: NZUsize!(1024),
2291                strategy: Sequential,
2292                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
2293            },
2294        )
2295        .await
2296        .unwrap();
2297
2298        let mut elements = Vec::new();
2299        for i in 0..100 {
2300            elements.push(test_digest(i));
2301        }
2302        let mut batch = mmr.new_batch();
2303        for elt in &elements {
2304            batch = batch.add(&hasher, elt);
2305        }
2306        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2307        mmr = mmr.apply_batch(&batch).unwrap();
2308
2309        let range = Location::<F>::new(30)..Location::<F>::new(61);
2310
2311        // Only apply elements up to end_loc to the reference structure.
2312        let mut ref_mmr = Merkle::<F, _, Digest, Sequential>::init(
2313            context.child("client"),
2314            &hasher,
2315            Config {
2316                journal_partition: "client-journal".into(),
2317                metadata_partition: "client-metadata".into(),
2318                items_per_blob: NZU64!(7),
2319                write_buffer: NZUsize!(1024),
2320                replay_buffer: NZUsize!(1024),
2321                strategy: Sequential,
2322                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
2323            },
2324        )
2325        .await
2326        .unwrap();
2327
2328        // Add elements up to the end of the range to verify historical root
2329        let mut batch = ref_mmr.new_batch();
2330        for elt in elements.iter().take(*range.end as usize) {
2331            batch = batch.add(&hasher, elt);
2332        }
2333        let batch = ref_mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2334        ref_mmr = ref_mmr.apply_batch(&batch).unwrap();
2335        let historical_leaves = ref_mmr.leaves();
2336        let expected_root = ref_mmr.root(&hasher, 0).unwrap();
2337
2338        // Generate proof from full structure
2339        let proof = mmr
2340            .historical_range_proof(&hasher, historical_leaves, range.clone(), 0)
2341            .await
2342            .unwrap();
2343
2344        assert!(proof.verify_range_inclusion(
2345            &hasher,
2346            &elements[range.to_usize_range()],
2347            range.start,
2348            &expected_root, // Compare to historical (reference) root
2349        ));
2350
2351        ref_mmr.destroy().await.unwrap();
2352        mmr.destroy().await.unwrap();
2353    }
2354
2355    #[test_traced]
2356    fn test_full_historical_proof_large_mmr() {
2357        let executor = deterministic::Runner::default();
2358        executor.start(full_historical_proof_large_inner::<mmr::Family>);
2359    }
2360
2361    #[test_traced]
2362    fn test_full_historical_proof_large_mmb() {
2363        let executor = deterministic::Runner::default();
2364        executor.start(full_historical_proof_large_inner::<mmb::Family>);
2365    }
2366
2367    async fn full_historical_proof_singleton_inner<F: Family>(context: deterministic::Context) {
2368        let hasher = Standard::<Sha256>::new(ForwardFold);
2369        let cfg = test_config(&context);
2370        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(context, &hasher, cfg)
2371            .await
2372            .unwrap();
2373
2374        let element = test_digest(0);
2375        let batch = mmr.new_batch().add(&hasher, &element);
2376        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2377        mmr = mmr.apply_batch(&batch).unwrap();
2378
2379        // Test single element proof at historical position
2380        let single_proof = mmr
2381            .historical_range_proof(
2382                &hasher,
2383                Location::<F>::new(1),
2384                Location::<F>::new(0)..Location::<F>::new(1),
2385                0,
2386            )
2387            .await
2388            .unwrap();
2389
2390        let root = mmr.root(&hasher, 0).unwrap();
2391        assert!(single_proof.verify_range_inclusion(
2392            &hasher,
2393            &[element],
2394            Location::<F>::new(0),
2395            &root
2396        ));
2397
2398        mmr.destroy().await.unwrap();
2399    }
2400
2401    #[test_traced]
2402    fn test_full_historical_proof_singleton_mmr() {
2403        let executor = deterministic::Runner::default();
2404        executor.start(full_historical_proof_singleton_inner::<mmr::Family>);
2405    }
2406
2407    #[test_traced]
2408    fn test_full_historical_proof_singleton_mmb() {
2409        let executor = deterministic::Runner::default();
2410        executor.start(full_historical_proof_singleton_inner::<mmb::Family>);
2411    }
2412
2413    // Test `init_sync` when there is no persisted data.
2414    async fn full_init_sync_empty_inner<F: Family>(context: deterministic::Context) {
2415        let hasher = Standard::<Sha256>::new(ForwardFold);
2416
2417        // Test fresh start scenario with completely new structure (no existing data)
2418        let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
2419            config: test_config(&context),
2420            range: non_empty_range!(Location::<F>::new(0), Location::<F>::new(52)),
2421            pinned_nodes: None,
2422        };
2423
2424        let mut sync_mmr =
2425            Merkle::<F, _, Digest, Sequential>::init_sync(context.child("storage"), sync_cfg)
2426                .await
2427                .unwrap();
2428
2429        // Should be fresh structure starting empty
2430        assert_eq!(sync_mmr.size(), 0);
2431        let bounds = sync_mmr.bounds();
2432        assert_eq!(bounds.start, 0);
2433        assert!(bounds.is_empty());
2434
2435        // Should be able to add new elements
2436        let new_element = test_digest(999);
2437        let batch = sync_mmr.new_batch().add(&hasher, &new_element);
2438        let batch = sync_mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2439        sync_mmr = sync_mmr.apply_batch(&batch).unwrap();
2440
2441        // Root should be computable
2442        let _root = sync_mmr.root(&hasher, 0).unwrap();
2443
2444        sync_mmr.destroy().await.unwrap();
2445    }
2446
2447    #[test_traced]
2448    fn test_full_init_sync_empty_mmr() {
2449        let executor = deterministic::Runner::default();
2450        executor.start(full_init_sync_empty_inner::<mmr::Family>);
2451    }
2452
2453    #[test_traced]
2454    fn test_full_init_sync_empty_mmb() {
2455        let executor = deterministic::Runner::default();
2456        executor.start(full_init_sync_empty_inner::<mmb::Family>);
2457    }
2458
2459    // Test `init_sync` where the persisted structure's persisted nodes match the sync boundaries.
2460    async fn full_init_sync_nonempty_exact_match_inner<F: Family>(context: deterministic::Context) {
2461        let hasher = Standard::<Sha256>::new(ForwardFold);
2462
2463        // Create initial structure with elements.
2464        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2465            context.child("init"),
2466            &hasher,
2467            test_config(&context),
2468        )
2469        .await
2470        .unwrap();
2471        let mut batch = mmr.new_batch();
2472        for i in 0..50 {
2473            batch = batch.add(&hasher, &test_digest(i));
2474        }
2475        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2476        mmr = mmr.apply_batch(&batch).unwrap();
2477        let mmr = mmr.sync().await.unwrap();
2478        let original_size = mmr.size();
2479        let original_leaves = mmr.leaves();
2480        let original_root = mmr.root(&hasher, 0).unwrap();
2481
2482        // Sync with range.start <= existing_size <= range.end should reuse data
2483        let lower_bound_loc = mmr.bounds().start;
2484        let upper_bound_loc = mmr.leaves();
2485        let lower_bound_pos = Position::<F>::try_from(lower_bound_loc).unwrap();
2486        let upper_bound_pos = mmr.size();
2487        let mut expected_nodes = BTreeMap::new();
2488        for i in *lower_bound_pos..*upper_bound_pos {
2489            expected_nodes.insert(
2490                Position::<F>::new(i),
2491                mmr.get_node(Position::<F>::new(i)).await.unwrap().unwrap(),
2492            );
2493        }
2494        let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
2495            config: test_config(&context),
2496            range: non_empty_range!(lower_bound_loc, upper_bound_loc),
2497            pinned_nodes: None,
2498        };
2499
2500        let mmr = mmr.sync().await.unwrap();
2501        drop(mmr);
2502
2503        let sync_mmr =
2504            Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
2505                .await
2506                .unwrap();
2507
2508        // Should have existing data in the sync range.
2509        assert_eq!(sync_mmr.size(), original_size);
2510        assert_eq!(sync_mmr.leaves(), original_leaves);
2511        let bounds = sync_mmr.bounds();
2512        assert_eq!(bounds.start, lower_bound_loc);
2513        assert!(!bounds.is_empty());
2514        assert_eq!(sync_mmr.root(&hasher, 0).unwrap(), original_root);
2515        for pos in *lower_bound_pos..*upper_bound_pos {
2516            let pos = Position::<F>::new(pos);
2517            assert_eq!(
2518                sync_mmr.get_node(pos).await.unwrap(),
2519                expected_nodes.get(&pos).cloned()
2520            );
2521        }
2522
2523        sync_mmr.destroy().await.unwrap();
2524    }
2525
2526    #[test_traced]
2527    fn test_full_init_sync_nonempty_exact_match_mmr() {
2528        let executor = deterministic::Runner::default();
2529        executor.start(full_init_sync_nonempty_exact_match_inner::<mmr::Family>);
2530    }
2531
2532    #[test_traced]
2533    fn test_full_init_sync_nonempty_exact_match_mmb() {
2534        let executor = deterministic::Runner::default();
2535        executor.start(full_init_sync_nonempty_exact_match_inner::<mmb::Family>);
2536    }
2537
2538    // Test `init_sync` where the persisted structure's data partially overlaps with the sync
2539    // boundaries.
2540    async fn full_init_sync_partial_overlap_inner<F: Family>(context: deterministic::Context) {
2541        let hasher = Standard::<Sha256>::new(ForwardFold);
2542
2543        // Create initial structure with elements.
2544        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2545            context.child("init"),
2546            &hasher,
2547            test_config(&context),
2548        )
2549        .await
2550        .unwrap();
2551        let mut batch = mmr.new_batch();
2552        for i in 0..30 {
2553            batch = batch.add(&hasher, &test_digest(i));
2554        }
2555        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2556        mmr = mmr.apply_batch(&batch).unwrap();
2557        let mmr = mmr.sync().await.unwrap();
2558        let mmr = mmr.prune(Location::<F>::new(6)).await.unwrap();
2559
2560        let original_size = mmr.size();
2561        let original_leaves = mmr.leaves();
2562        let original_root = mmr.root(&hasher, 0).unwrap();
2563        let original_pruning_boundary = mmr.bounds().start;
2564        let original_pruning_pos = Position::<F>::try_from(original_pruning_boundary).unwrap();
2565
2566        // Sync with boundaries that extend beyond existing data (partial overlap).
2567        let lower_bound_loc = original_pruning_boundary;
2568        let upper_bound_loc = original_leaves + 6; // Extend beyond existing data
2569
2570        let mut expected_nodes = BTreeMap::new();
2571        for i in *original_pruning_pos..*original_size {
2572            let pos = Position::<F>::new(i);
2573            expected_nodes.insert(pos, mmr.get_node(pos).await.unwrap().unwrap());
2574        }
2575
2576        let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
2577            config: test_config(&context),
2578            range: non_empty_range!(lower_bound_loc, upper_bound_loc),
2579            pinned_nodes: None,
2580        };
2581
2582        let mmr = mmr.sync().await.unwrap();
2583        drop(mmr);
2584
2585        let sync_mmr =
2586            Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
2587                .await
2588                .unwrap();
2589
2590        // Should have existing data in the overlapping range.
2591        assert_eq!(sync_mmr.size(), original_size);
2592        let bounds = sync_mmr.bounds();
2593        assert_eq!(bounds.start, lower_bound_loc);
2594        assert!(!bounds.is_empty());
2595        assert_eq!(sync_mmr.root(&hasher, 0).unwrap(), original_root);
2596
2597        // Check that existing nodes are preserved in the overlapping range.
2598        for i in *original_pruning_pos..*original_size {
2599            let pos = Position::<F>::new(i);
2600            assert_eq!(
2601                sync_mmr.get_node(pos).await.unwrap(),
2602                expected_nodes.get(&pos).cloned()
2603            );
2604        }
2605
2606        sync_mmr.destroy().await.unwrap();
2607    }
2608
2609    #[test_traced]
2610    fn test_full_init_sync_partial_overlap_mmr() {
2611        let executor = deterministic::Runner::default();
2612        executor.start(full_init_sync_partial_overlap_inner::<mmr::Family>);
2613    }
2614
2615    #[test_traced]
2616    fn test_full_init_sync_partial_overlap_mmb() {
2617        let executor = deterministic::Runner::default();
2618        executor.start(full_init_sync_partial_overlap_inner::<mmb::Family>);
2619    }
2620
2621    async fn full_init_sync_rewinds_state_beyond_range_inner<F: Family>(
2622        context: deterministic::Context,
2623    ) {
2624        let hasher = Standard::<Sha256>::new(ForwardFold);
2625        let cfg = test_config(&context);
2626        let mut merkle =
2627            Merkle::<F, _, Digest, Sequential>::init(context.child("init"), &hasher, cfg.clone())
2628                .await
2629                .unwrap();
2630
2631        let target_end = Location::<F>::new(20);
2632        let mut batch = merkle.new_batch();
2633        for i in 0..20 {
2634            batch = batch.add(&hasher, &test_digest(i));
2635        }
2636        let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
2637        merkle = merkle.apply_batch(&batch).unwrap();
2638        let target_root = merkle.root(&hasher, 0).unwrap();
2639        let restart = Location::<F>::new(7);
2640        let pinned_nodes = merkle.pinned_nodes_at(restart).await.unwrap();
2641
2642        let mut batch = merkle.new_batch();
2643        for i in 20..50 {
2644            batch = batch.add(&hasher, &test_digest(i));
2645        }
2646        let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
2647        merkle = merkle.apply_batch(&batch).unwrap();
2648        let merkle = merkle.sync().await.unwrap();
2649        drop(merkle);
2650
2651        let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
2652            config: cfg.clone(),
2653            range: non_empty_range!(restart, target_end),
2654            pinned_nodes: Some(pinned_nodes),
2655        };
2656        let merkle = Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
2657            .await
2658            .unwrap();
2659
2660        assert_eq!(merkle.leaves(), target_end);
2661        assert_eq!(merkle.root(&hasher, 0).unwrap(), target_root);
2662        let merkle = merkle.sync().await.unwrap();
2663        drop(merkle);
2664
2665        let merkle =
2666            Merkle::<F, _, Digest, Sequential>::init(context.child("reopen"), &hasher, cfg)
2667                .await
2668                .unwrap();
2669        assert_eq!(merkle.leaves(), target_end);
2670        assert_eq!(merkle.root(&hasher, 0).unwrap(), target_root);
2671        merkle.destroy().await.unwrap();
2672    }
2673
2674    #[test_traced]
2675    fn test_full_init_sync_rewinds_state_beyond_range_mmr() {
2676        deterministic::Runner::default()
2677            .start(full_init_sync_rewinds_state_beyond_range_inner::<mmr::Family>);
2678    }
2679
2680    #[test_traced]
2681    fn test_full_init_sync_rewinds_state_beyond_range_mmb() {
2682        deterministic::Runner::default()
2683            .start(full_init_sync_rewinds_state_beyond_range_inner::<mmb::Family>);
2684    }
2685
2686    async fn full_init_sync_discards_state_pruned_past_range_inner<F: Family>(
2687        context: deterministic::Context,
2688    ) {
2689        let hasher = Standard::<Sha256>::new(ForwardFold);
2690        let cfg = test_config(&context);
2691        let mut merkle =
2692            Merkle::<F, _, Digest, Sequential>::init(context.child("init"), &hasher, cfg.clone())
2693                .await
2694                .unwrap();
2695
2696        let mut batch = merkle.new_batch();
2697        for i in 0..50 {
2698            batch = batch.add(&hasher, &test_digest(i));
2699        }
2700        let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
2701        merkle = merkle.apply_batch(&batch).unwrap();
2702        let target_root = merkle.root(&hasher, 0).unwrap();
2703        let restart = Location::<F>::new(7);
2704        let pinned_nodes = merkle.pinned_nodes_at(restart).await.unwrap();
2705        let merkle = merkle.sync().await.unwrap();
2706        let merkle = merkle.prune(Location::new(30)).await.unwrap();
2707        let merkle = merkle.sync().await.unwrap();
2708        assert!(merkle.bounds().start > restart);
2709        drop(merkle);
2710
2711        let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
2712            config: cfg,
2713            range: non_empty_range!(restart, Location::<F>::new(60)),
2714            pinned_nodes: Some(pinned_nodes),
2715        };
2716        let mut merkle =
2717            Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
2718                .await
2719                .unwrap();
2720
2721        assert_eq!(merkle.bounds(), restart..restart);
2722        let mut batch = merkle.new_batch();
2723        for i in 7..50 {
2724            batch = batch.add(&hasher, &test_digest(i));
2725        }
2726        let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
2727        merkle = merkle.apply_batch(&batch).unwrap();
2728        assert_eq!(merkle.root(&hasher, 0).unwrap(), target_root);
2729        merkle.destroy().await.unwrap();
2730    }
2731
2732    #[test_traced]
2733    fn test_full_init_sync_discards_state_pruned_past_range_mmr() {
2734        deterministic::Runner::default()
2735            .start(full_init_sync_discards_state_pruned_past_range_inner::<mmr::Family>);
2736    }
2737
2738    #[test_traced]
2739    fn test_full_init_sync_discards_state_pruned_past_range_mmb() {
2740        deterministic::Runner::default()
2741            .start(full_init_sync_discards_state_pruned_past_range_inner::<mmb::Family>);
2742    }
2743
2744    #[test_traced]
2745    fn test_full_init_sync_discards_stale_pinned_metadata() {
2746        deterministic::Runner::default().start(|context| async move {
2747            type F = mmr::Family;
2748
2749            let hasher = Standard::<Sha256>::new(ForwardFold);
2750            let cfg = test_config(&context);
2751            let mut merkle = Merkle::<F, _, Digest, Sequential>::init(
2752                context.child("old"),
2753                &hasher,
2754                cfg.clone(),
2755            )
2756            .await
2757            .unwrap();
2758            let mut batch = merkle.new_batch();
2759            for i in 0..40 {
2760                batch = batch.add(&hasher, &test_digest(i));
2761            }
2762            let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
2763            merkle = merkle.apply_batch(&batch).unwrap();
2764            let merkle = merkle.sync().await.unwrap();
2765            let merkle = merkle.prune(Location::new(30)).await.unwrap();
2766            let merkle = merkle.sync().await.unwrap();
2767            drop(merkle);
2768
2769            let mut target_cfg = test_config(&context);
2770            target_cfg.journal_partition = "target-journal-partition".into();
2771            target_cfg.metadata_partition = "target-metadata-partition".into();
2772            let mut target = Merkle::<F, _, Digest, Sequential>::init(
2773                context.child("target"),
2774                &hasher,
2775                target_cfg,
2776            )
2777            .await
2778            .unwrap();
2779            let mut batch = target.new_batch();
2780            for i in 0..20 {
2781                batch = batch.add(&hasher, &test_digest(1_000 + i));
2782            }
2783            let batch = target.with_mem(|mem| batch.merkleize(mem, &hasher));
2784            target = target.apply_batch(&batch).unwrap();
2785            let target_root = target.root(&hasher, 0).unwrap();
2786            let restart = Location::new(7);
2787            let pinned_nodes = target.pinned_nodes_at(restart).await.unwrap();
2788            target.destroy().await.unwrap();
2789
2790            let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
2791                config: cfg.clone(),
2792                range: non_empty_range!(restart, Location::new(20)),
2793                pinned_nodes: Some(pinned_nodes),
2794            };
2795            let mut merkle =
2796                Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
2797                    .await
2798                    .unwrap();
2799            let mut batch = merkle.new_batch();
2800            for i in 7..20 {
2801                batch = batch.add(&hasher, &test_digest(1_000 + i));
2802            }
2803            let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
2804            merkle = merkle.apply_batch(&batch).unwrap();
2805            assert_eq!(merkle.root(&hasher, 0).unwrap(), target_root);
2806            let merkle = merkle.sync().await.unwrap();
2807            drop(merkle);
2808
2809            let merkle =
2810                Merkle::<F, _, Digest, Sequential>::init(context.child("reopen"), &hasher, cfg)
2811                    .await
2812                    .unwrap();
2813            assert_eq!(merkle.root(&hasher, 0).unwrap(), target_root);
2814            merkle.destroy().await.unwrap();
2815        });
2816    }
2817
2818    async fn full_init_sync_rejects_extra_pinned_nodes_inner<F: Family>(
2819        context: deterministic::Context,
2820    ) {
2821        let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
2822            config: test_config(&context),
2823            range: non_empty_range!(Location::<F>::new(6), Location::<F>::new(20)),
2824            pinned_nodes: Some(vec![test_digest(1), test_digest(2), test_digest(3)]),
2825        };
2826
2827        let result =
2828            Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg).await;
2829        assert!(matches!(result, Err(Error::InvalidPinnedNodes)));
2830    }
2831
2832    #[test_traced]
2833    fn test_full_init_sync_rejects_extra_pinned_nodes_mmr() {
2834        let executor = deterministic::Runner::default();
2835        executor.start(full_init_sync_rejects_extra_pinned_nodes_inner::<mmr::Family>);
2836    }
2837
2838    #[test_traced]
2839    fn test_full_init_sync_rejects_extra_pinned_nodes_mmb() {
2840        let executor = deterministic::Runner::default();
2841        executor.start(full_init_sync_rejects_extra_pinned_nodes_inner::<mmb::Family>);
2842    }
2843
2844    // Regression test that init() handles stale metadata (lower pruning boundary than journal).
2845    // Before the fix, this would panic with an assertion failure. After the fix, it returns a
2846    // MissingNode error (which is expected when metadata is corrupted and pinned nodes are lost).
2847    async fn full_init_stale_metadata_returns_error_inner<F: Family>(
2848        context: deterministic::Context,
2849    ) {
2850        let hasher = Standard::<Sha256>::new(ForwardFold);
2851
2852        // Create a structure with some data and prune it
2853        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2854            context.child("init"),
2855            &hasher,
2856            test_config(&context),
2857        )
2858        .await
2859        .unwrap();
2860
2861        // Add 50 elements
2862        let mut batch = mmr.new_batch();
2863        for i in 0..50 {
2864            batch = batch.add(&hasher, &test_digest(i));
2865        }
2866        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
2867        mmr = mmr.apply_batch(&batch).unwrap();
2868        let mmr = mmr.sync().await.unwrap();
2869
2870        // Prune enough that the journal boundary's pinned nodes span pruned blobs.
2871        let prune_loc = Location::<F>::new(25);
2872        mmr.prune(prune_loc).await.unwrap();
2873
2874        // Simulate a crash after journal prune but before metadata was updated:
2875        // clear all metadata and write only a stale pruning boundary of 0 (no pinned nodes).
2876        let meta_cfg = MConfig {
2877            partition: test_config(&context).metadata_partition,
2878            codec_config: ((0..).into(), ()),
2879        };
2880        let mut metadata =
2881            Metadata::<_, U64, Vec<u8>>::init(context.child("meta_tamper"), meta_cfg)
2882                .await
2883                .unwrap();
2884        metadata.clear();
2885        let key = U64::new(PRUNED_TO_PREFIX, 0);
2886        metadata
2887            .put_sync(key, 0u64.to_be_bytes().to_vec())
2888            .await
2889            .unwrap();
2890
2891        // Reopen the structure - before the fix, this would panic with assertion failure
2892        // After the fix, it returns MissingNode error (pinned nodes for the lower
2893        // boundary don't exist since they were pruned from journal and weren't
2894        // stored in metadata at the lower position)
2895        let result = Merkle::<F, _, Digest, Sequential>::init(
2896            context.child("reopened"),
2897            &hasher,
2898            test_config(&context),
2899        )
2900        .await;
2901
2902        match result {
2903            Err(Error::MissingNode(_)) => {} // expected
2904            Ok(_) => panic!("expected MissingNode error, got Ok"),
2905            Err(e) => panic!("expected MissingNode error, got {:?}", e),
2906        }
2907    }
2908
2909    #[test_traced("WARN")]
2910    fn test_full_init_stale_metadata_returns_error_mmr() {
2911        let executor = deterministic::Runner::default();
2912        executor.start(full_init_stale_metadata_returns_error_inner::<mmr::Family>);
2913    }
2914
2915    #[test_traced("WARN")]
2916    fn test_full_init_stale_metadata_returns_error_mmb() {
2917        let executor = deterministic::Runner::default();
2918        executor.start(full_init_stale_metadata_returns_error_inner::<mmb::Family>);
2919    }
2920
2921    async fn full_init_rejects_prune_boundary_beyond_recovered_size_inner<F: Family>(
2922        context: deterministic::Context,
2923    ) {
2924        let hasher = Standard::<Sha256>::new(ForwardFold);
2925        let cfg = test_config(&context);
2926        let mut merkle =
2927            Merkle::<F, _, Digest, Sequential>::init(context.child("init"), &hasher, cfg.clone())
2928                .await
2929                .unwrap();
2930
2931        let mut batch = merkle.new_batch();
2932        for i in 0..12 {
2933            batch = batch.add(&hasher, &test_digest(i));
2934        }
2935        let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
2936        merkle = merkle.apply_batch(&batch).unwrap();
2937        let merkle = merkle.sync().await.unwrap();
2938        let merkle = merkle.prune(Location::new(12)).await.unwrap();
2939        let merkle = merkle.sync().await.unwrap();
2940        drop(merkle);
2941
2942        let journal_cfg = JConfig {
2943            partition: cfg.journal_partition.clone(),
2944            items_per_blob: cfg.items_per_blob,
2945            page_cache: cfg.page_cache.clone(),
2946            write_buffer: cfg.write_buffer,
2947            replay_buffer: cfg.replay_buffer,
2948        };
2949        let journal = Journal::<_, Digest>::init(context.child("interrupted_reset"), journal_cfg)
2950            .await
2951            .unwrap();
2952        let recovered_size = Position::<F>::try_from(Location::<F>::new(8)).unwrap();
2953        assert!(
2954            journal.bounds().start > *recovered_size,
2955            "test reset must discard a pruned prefix"
2956        );
2957        let journal = journal.clear_to_size(*recovered_size).await.unwrap();
2958        drop(journal);
2959
2960        match Merkle::<F, _, Digest, Sequential>::init(context.child("reopen"), &hasher, cfg).await
2961        {
2962            Err(Error::MissingNode(_)) => {}
2963            Ok(_) => panic!("pruning boundary beyond recovered size must fail closed"),
2964            Err(err) => panic!("expected MissingNode error, got {err:?}"),
2965        }
2966    }
2967
2968    #[test_traced("WARN")]
2969    fn test_full_init_rejects_prune_boundary_beyond_recovered_size_mmr() {
2970        deterministic::Runner::default()
2971            .start(full_init_rejects_prune_boundary_beyond_recovered_size_inner::<mmr::Family>);
2972    }
2973
2974    #[test_traced("WARN")]
2975    fn test_full_init_rejects_prune_boundary_beyond_recovered_size_mmb() {
2976        deterministic::Runner::default()
2977            .start(full_init_rejects_prune_boundary_beyond_recovered_size_inner::<mmb::Family>);
2978    }
2979
2980    // Test that init() handles the case where metadata pruning boundary is ahead
2981    // of journal (crashed before journal prune completed). This should successfully
2982    // prune the journal to match metadata.
2983    async fn full_init_metadata_ahead_inner<F: Family>(context: deterministic::Context) {
2984        let hasher = Standard::<Sha256>::new(ForwardFold);
2985
2986        // Create a structure with some data
2987        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
2988            context.child("init"),
2989            &hasher,
2990            test_config(&context),
2991        )
2992        .await
2993        .unwrap();
2994
2995        // Add 50 elements
2996        let mut batch = mmr.new_batch();
2997        for i in 0..50 {
2998            batch = batch.add(&hasher, &test_digest(i));
2999        }
3000        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3001        mmr = mmr.apply_batch(&batch).unwrap();
3002        let mmr = mmr.sync().await.unwrap();
3003
3004        // Prune to position 30 (this stores pinned nodes and updates metadata)
3005        let prune_loc = Location::<F>::new(16);
3006        let mmr = mmr.prune(prune_loc).await.unwrap();
3007        let expected_root = mmr.root(&hasher, 0).unwrap();
3008        let expected_size = mmr.size();
3009        drop(mmr);
3010
3011        // Reopen the structure - should recover correctly with metadata ahead of
3012        // journal boundary (metadata says 30, journal is section-aligned to 28)
3013        let mmr = Merkle::<F, _, Digest, Sequential>::init(
3014            context.child("reopened"),
3015            &hasher,
3016            test_config(&context),
3017        )
3018        .await
3019        .unwrap();
3020
3021        assert_eq!(mmr.bounds().start, prune_loc);
3022        assert_eq!(mmr.size(), expected_size);
3023        assert_eq!(mmr.root(&hasher, 0).unwrap(), expected_root);
3024
3025        mmr.destroy().await.unwrap();
3026    }
3027
3028    #[test_traced("WARN")]
3029    fn test_full_init_metadata_ahead_mmr() {
3030        let executor = deterministic::Runner::default();
3031        executor.start(full_init_metadata_ahead_inner::<mmr::Family>);
3032    }
3033
3034    #[test_traced("WARN")]
3035    fn test_full_init_metadata_ahead_mmb() {
3036        let executor = deterministic::Runner::default();
3037        executor.start(full_init_metadata_ahead_inner::<mmb::Family>);
3038    }
3039
3040    // Regression test: init_sync must compute pinned nodes BEFORE pruning the journal. Previously,
3041    // init_sync would prune the journal first, then try to read pinned nodes from the pruned
3042    // positions, causing MissingNode errors.
3043    //
3044    // Key setup: We create a structure with data but DON'T prune it, so the metadata has no pinned
3045    // nodes. Then init_sync must read pinned nodes from the journal before pruning it.
3046    async fn full_init_sync_computes_pinned_nodes_before_pruning_inner<F: Family>(
3047        context: deterministic::Context,
3048    ) {
3049        let hasher = Standard::<Sha256>::new(ForwardFold);
3050
3051        // Use small items_per_blob to create many sections and trigger pruning.
3052        let cfg = Config {
3053            journal_partition: "mmr-journal".into(),
3054            metadata_partition: "mmr-metadata".into(),
3055            items_per_blob: NZU64!(7),
3056            write_buffer: NZUsize!(64),
3057            replay_buffer: NZUsize!(64),
3058            strategy: Sequential,
3059            page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
3060        };
3061
3062        // Create structure with enough elements to span multiple sections.
3063        let mut mmr =
3064            Merkle::<F, _, Digest, Sequential>::init(context.child("init"), &hasher, cfg.clone())
3065                .await
3066                .unwrap();
3067        let mut batch = mmr.new_batch();
3068        for i in 0..100 {
3069            batch = batch.add(&hasher, &test_digest(i));
3070        }
3071        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3072        mmr = mmr.apply_batch(&batch).unwrap();
3073        let mmr = mmr.sync().await.unwrap();
3074
3075        // Don't prune - this ensures metadata has no pinned nodes. init_sync will need to
3076        // read pinned nodes from the journal.
3077        let original_size = mmr.size();
3078        let original_root = mmr.root(&hasher, 0).unwrap();
3079        drop(mmr);
3080
3081        // Reopen via init_sync with range.start > 0. This will prune the journal, so
3082        // init_sync must read pinned nodes BEFORE pruning or they'll be lost.
3083        let prune_loc = Location::<F>::new(32);
3084        let sync_cfg = SyncConfig::<F, sha256::Digest, Sequential> {
3085            config: cfg,
3086            range: non_empty_range!(prune_loc, Location::<F>::new(128)),
3087            pinned_nodes: None, // Force init_sync to compute pinned nodes from journal
3088        };
3089
3090        let sync_mmr =
3091            Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
3092                .await
3093                .unwrap();
3094
3095        // Verify the structure state is correct.
3096        assert_eq!(sync_mmr.size(), original_size);
3097        assert_eq!(sync_mmr.root(&hasher, 0).unwrap(), original_root);
3098        assert_eq!(sync_mmr.bounds().start, prune_loc);
3099
3100        sync_mmr.destroy().await.unwrap();
3101    }
3102
3103    #[test_traced]
3104    fn test_full_init_sync_computes_pinned_nodes_before_pruning_mmr() {
3105        let executor = deterministic::Runner::default();
3106        executor.start(full_init_sync_computes_pinned_nodes_before_pruning_inner::<mmr::Family>);
3107    }
3108
3109    #[test_traced]
3110    fn test_full_init_sync_computes_pinned_nodes_before_pruning_mmb() {
3111        let executor = deterministic::Runner::default();
3112        executor.start(full_init_sync_computes_pinned_nodes_before_pruning_inner::<mmb::Family>);
3113    }
3114
3115    async fn full_historical_proof_pruned_elements_inner<F: Family>(
3116        context: deterministic::Context,
3117    ) {
3118        let hasher = Standard::<Sha256>::new(ForwardFold);
3119
3120        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3121            context.child("init"),
3122            &hasher,
3123            test_config(&context),
3124        )
3125        .await
3126        .unwrap();
3127
3128        let mut batch = mmr.new_batch();
3129        for i in 0..64 {
3130            batch = batch.add(&hasher, &test_digest(i));
3131        }
3132        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3133        mmr = mmr.apply_batch(&batch).unwrap();
3134
3135        let prune_loc = Location::<F>::new(16);
3136        let mmr = mmr.prune(prune_loc).await.unwrap();
3137
3138        let historical_leaves = mmr.leaves();
3139        let mut pruned_loc = None;
3140        for loc_u64 in 0..*historical_leaves {
3141            let loc = Location::<F>::new(loc_u64);
3142            let result = mmr
3143                .historical_range_proof(&hasher, historical_leaves, loc..loc + 1, 0)
3144                .await;
3145            if matches!(result, Err(Error::ElementPruned(_))) {
3146                pruned_loc = Some(loc);
3147                break;
3148            }
3149        }
3150        let pruned_loc = pruned_loc.expect("expected at least one pruned location");
3151
3152        // Add more elements and verify pruned elements still return ElementPruned.
3153        let mut batch = mmr.new_batch();
3154        for i in 0..8 {
3155            batch = batch.add(&hasher, &test_digest(10_000 + i));
3156        }
3157        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3158        let mmr = mmr.apply_batch(&batch).unwrap();
3159
3160        let requested = mmr.leaves();
3161        let result = mmr
3162            .historical_range_proof(&hasher, requested, pruned_loc..pruned_loc + 1, 0)
3163            .await;
3164        assert!(matches!(result, Err(Error::ElementPruned(_))));
3165
3166        mmr.destroy().await.unwrap();
3167    }
3168
3169    #[test_traced]
3170    fn test_full_historical_proof_pruned_elements_mmr() {
3171        let executor = deterministic::Runner::default();
3172        executor.start(full_historical_proof_pruned_elements_inner::<mmr::Family>);
3173    }
3174
3175    #[test_traced]
3176    fn test_full_historical_proof_pruned_elements_mmb() {
3177        let executor = deterministic::Runner::default();
3178        executor.start(full_historical_proof_pruned_elements_inner::<mmb::Family>);
3179    }
3180
3181    async fn full_append_while_historical_proof_is_available_inner<F: Family>(
3182        context: deterministic::Context,
3183    ) {
3184        let hasher = Standard::<Sha256>::new(ForwardFold);
3185        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3186            context.child("init"),
3187            &hasher,
3188            test_config(&context),
3189        )
3190        .await
3191        .unwrap();
3192
3193        let mut batch = mmr.new_batch();
3194        for i in 0..20 {
3195            batch = batch.add(&hasher, &test_digest(i));
3196        }
3197        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3198        mmr = mmr.apply_batch(&batch).unwrap();
3199
3200        let historical_leaves = Location::<F>::new(10);
3201        let range = Location::<F>::new(2)..Location::<F>::new(8);
3202
3203        // Appends should remain allowed while historical proofs are available.
3204        let batch = mmr
3205            .new_batch()
3206            .add(&hasher, &test_digest(100))
3207            .add(&hasher, &test_digest(101));
3208        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3209        mmr = mmr.apply_batch(&batch).unwrap();
3210
3211        let proof = mmr
3212            .historical_range_proof(&hasher, historical_leaves, range.clone(), 0)
3213            .await
3214            .unwrap();
3215
3216        let expected = mmr
3217            .historical_range_proof(&hasher, historical_leaves, range, 0)
3218            .await
3219            .unwrap();
3220        assert_eq!(proof, expected);
3221
3222        mmr.destroy().await.unwrap();
3223    }
3224
3225    #[test_traced]
3226    fn test_full_append_while_historical_proof_is_available_mmr() {
3227        let executor = deterministic::Runner::default();
3228        executor.start(full_append_while_historical_proof_is_available_inner::<mmr::Family>);
3229    }
3230
3231    #[test_traced]
3232    fn test_full_append_while_historical_proof_is_available_mmb() {
3233        let executor = deterministic::Runner::default();
3234        executor.start(full_append_while_historical_proof_is_available_inner::<mmb::Family>);
3235    }
3236
3237    async fn full_historical_proof_after_sync_reads_from_journal_inner<F: Family>(
3238        context: deterministic::Context,
3239    ) {
3240        let hasher = Standard::<Sha256>::new(ForwardFold);
3241        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3242            context.child("init"),
3243            &hasher,
3244            test_config(&context),
3245        )
3246        .await
3247        .unwrap();
3248
3249        let mut batch = mmr.new_batch();
3250        for i in 0..64 {
3251            batch = batch.add(&hasher, &test_digest(i));
3252        }
3253        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3254        mmr = mmr.apply_batch(&batch).unwrap();
3255        let mmr = mmr.sync().await.unwrap();
3256
3257        let historical_leaves = Location::<F>::new(20);
3258        let range = Location::<F>::new(5)..Location::<F>::new(15);
3259        let expected = mmr
3260            .historical_range_proof(&hasher, historical_leaves, range.clone(), 0)
3261            .await
3262            .unwrap();
3263
3264        let actual = mmr
3265            .historical_range_proof(&hasher, historical_leaves, range, 0)
3266            .await
3267            .unwrap();
3268        assert_eq!(actual, expected);
3269
3270        mmr.destroy().await.unwrap();
3271    }
3272
3273    #[test_traced]
3274    fn test_full_historical_proof_after_sync_reads_from_journal_mmr() {
3275        let executor = deterministic::Runner::default();
3276        executor.start(full_historical_proof_after_sync_reads_from_journal_inner::<mmr::Family>);
3277    }
3278
3279    #[test_traced]
3280    fn test_full_historical_proof_after_sync_reads_from_journal_mmb() {
3281        let executor = deterministic::Runner::default();
3282        executor.start(full_historical_proof_after_sync_reads_from_journal_inner::<mmb::Family>);
3283    }
3284
3285    async fn full_historical_proof_after_pruning_inner<F: Family>(context: deterministic::Context) {
3286        let hasher = Standard::<Sha256>::new(ForwardFold);
3287        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3288            context.child("init"),
3289            &hasher,
3290            test_config(&context),
3291        )
3292        .await
3293        .unwrap();
3294
3295        let mut batch = mmr.new_batch();
3296        for i in 0..30 {
3297            batch = batch.add(&hasher, &test_digest(i));
3298        }
3299        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3300        mmr = mmr.apply_batch(&batch).unwrap();
3301
3302        let prune_loc = Location::<F>::new(10);
3303        let mmr = mmr.prune(prune_loc).await.unwrap();
3304
3305        let requested = Location::<F>::new(20);
3306        let range = prune_loc..requested;
3307        let proof = mmr
3308            .historical_range_proof(&hasher, requested, range, 0)
3309            .await
3310            .unwrap();
3311        assert!(proof.leaves > Location::<F>::new(0));
3312
3313        mmr.destroy().await.unwrap();
3314    }
3315
3316    #[test_traced]
3317    fn test_full_historical_proof_after_pruning_mmr() {
3318        let executor = deterministic::Runner::default();
3319        executor.start(full_historical_proof_after_pruning_inner::<mmr::Family>);
3320    }
3321
3322    #[test_traced]
3323    fn test_full_historical_proof_after_pruning_mmb() {
3324        let executor = deterministic::Runner::default();
3325        executor.start(full_historical_proof_after_pruning_inner::<mmb::Family>);
3326    }
3327
3328    async fn full_historical_proof_edge_cases_inner<F: Family>(context: deterministic::Context) {
3329        let hasher = Standard::<Sha256>::new(ForwardFold);
3330
3331        // Case 1: Empty structure.
3332        let mmr = Merkle::<F, _, Digest, Sequential>::init(
3333            context.child("empty"),
3334            &hasher,
3335            test_config(&context),
3336        )
3337        .await
3338        .unwrap();
3339        let empty_end = Location::<F>::new(0);
3340        let empty_result = mmr
3341            .historical_range_proof(&hasher, empty_end, empty_end..empty_end, 0)
3342            .await;
3343        assert!(matches!(empty_result, Err(Error::Empty)));
3344        let oob_result = mmr
3345            .historical_range_proof(&hasher, empty_end + 1, empty_end..empty_end + 1, 0)
3346            .await;
3347        assert!(matches!(
3348            oob_result,
3349            Err(Error::RangeOutOfBounds(loc)) if loc == empty_end + 1
3350        ));
3351        mmr.destroy().await.unwrap();
3352
3353        // Case 2: Structure has nodes but is fully pruned.
3354        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3355            context.child("fully_pruned"),
3356            &hasher,
3357            test_config(&context),
3358        )
3359        .await
3360        .unwrap();
3361        let mut batch = mmr.new_batch();
3362        for i in 0..20 {
3363            batch = batch.add(&hasher, &test_digest(i));
3364        }
3365        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3366        mmr = mmr.apply_batch(&batch).unwrap();
3367        let end = mmr.leaves();
3368        mmr = mmr.prune_all().await.unwrap();
3369        assert!(mmr.bounds().is_empty());
3370        let pruned_result = mmr
3371            .historical_range_proof(&hasher, end, end - 1..end, 0)
3372            .await;
3373        assert!(matches!(pruned_result, Err(Error::ElementPruned(_))));
3374        let oob_result = mmr
3375            .historical_range_proof(&hasher, end + 1, end - 1..end, 0)
3376            .await;
3377        assert!(matches!(
3378            oob_result,
3379            Err(Error::RangeOutOfBounds(loc)) if loc == end + 1
3380        ));
3381        mmr.destroy().await.unwrap();
3382
3383        // Case 3: All nodes but one (single leaf) are pruned.
3384        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3385            context.child("single_leaf"),
3386            &hasher,
3387            test_config(&context),
3388        )
3389        .await
3390        .unwrap();
3391        let mut batch = mmr.new_batch();
3392        for i in 0..11 {
3393            batch = batch.add(&hasher, &test_digest(i));
3394        }
3395        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3396        mmr = mmr.apply_batch(&batch).unwrap();
3397        let end = mmr.leaves();
3398        let keep_loc = end - 1;
3399        mmr = mmr.prune(keep_loc).await.unwrap();
3400        let ok_result = mmr
3401            .historical_range_proof(&hasher, end, keep_loc..end, 0)
3402            .await;
3403        assert!(ok_result.is_ok());
3404        let pruned_end = keep_loc - 1;
3405        // make sure this is in a pruned range, considering blob boundaries.
3406        let start_loc = Location::<F>::new(1);
3407        let pruned_result = mmr
3408            .historical_range_proof(&hasher, end, start_loc..pruned_end + 1, 0)
3409            .await;
3410        assert!(matches!(pruned_result, Err(Error::ElementPruned(_))));
3411        let oob_result = mmr
3412            .historical_range_proof(&hasher, end + 1, keep_loc..end, 0)
3413            .await;
3414        assert!(matches!(oob_result, Err(Error::RangeOutOfBounds(_))));
3415        mmr.destroy().await.unwrap();
3416    }
3417
3418    #[test_traced]
3419    fn test_full_historical_proof_edge_cases_mmr() {
3420        let executor = deterministic::Runner::default();
3421        executor.start(full_historical_proof_edge_cases_inner::<mmr::Family>);
3422    }
3423
3424    #[test_traced]
3425    fn test_full_historical_proof_edge_cases_mmb() {
3426        let executor = deterministic::Runner::default();
3427        executor.start(full_historical_proof_edge_cases_inner::<mmb::Family>);
3428    }
3429
3430    async fn full_historical_proof_out_of_bounds_inner<F: Family>(context: deterministic::Context) {
3431        let hasher = Standard::<Sha256>::new(ForwardFold);
3432        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3433            context.child("oob"),
3434            &hasher,
3435            test_config(&context),
3436        )
3437        .await
3438        .unwrap();
3439
3440        let mut batch = mmr.new_batch();
3441        for i in 0..8 {
3442            batch = batch.add(&hasher, &test_digest(i));
3443        }
3444        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3445        mmr = mmr.apply_batch(&batch).unwrap();
3446        let requested = mmr.leaves() + 1;
3447
3448        let result = mmr
3449            .historical_range_proof(&hasher, requested, Location::<F>::new(0)..requested, 0)
3450            .await;
3451        assert!(matches!(
3452            result,
3453            Err(Error::RangeOutOfBounds(loc)) if loc == requested
3454        ));
3455
3456        mmr.destroy().await.unwrap();
3457    }
3458
3459    #[test_traced]
3460    fn test_full_historical_proof_out_of_bounds_mmr() {
3461        let executor = deterministic::Runner::default();
3462        executor.start(full_historical_proof_out_of_bounds_inner::<mmr::Family>);
3463    }
3464
3465    #[test_traced]
3466    fn test_full_historical_proof_out_of_bounds_mmb() {
3467        let executor = deterministic::Runner::default();
3468        executor.start(full_historical_proof_out_of_bounds_inner::<mmb::Family>);
3469    }
3470
3471    async fn full_historical_proof_range_validation_inner<F: Family>(
3472        context: deterministic::Context,
3473    ) {
3474        let hasher = Standard::<Sha256>::new(ForwardFold);
3475        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3476            context.child("range_validation"),
3477            &hasher,
3478            test_config(&context),
3479        )
3480        .await
3481        .unwrap();
3482
3483        let mut batch = mmr.new_batch();
3484        for i in 0..32 {
3485            batch = batch.add(&hasher, &test_digest(i));
3486        }
3487        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3488        mmr = mmr.apply_batch(&batch).unwrap();
3489
3490        let valid_range = Location::<F>::new(0)..Location::<F>::new(1);
3491
3492        // Empty range should report Empty.
3493        let requested = Location::<F>::new(5);
3494        let empty_range = requested..requested;
3495        let empty_result = mmr
3496            .historical_range_proof(&hasher, requested, empty_range, 0)
3497            .await;
3498        assert!(matches!(empty_result, Err(Error::Empty)));
3499
3500        // Requested historical size is out of bounds.
3501        let leaves_oob = mmr.leaves() + 1;
3502        let result = mmr
3503            .historical_range_proof(&hasher, leaves_oob, valid_range.clone(), 0)
3504            .await;
3505        assert!(matches!(
3506            result,
3507            Err(Error::RangeOutOfBounds(loc)) if loc == leaves_oob
3508        ));
3509
3510        // Requested range end is out of bounds for the current structure.
3511        let end_oob = mmr.leaves() + 1;
3512        let range_oob = Location::<F>::new(0)..end_oob;
3513        let result = mmr
3514            .historical_range_proof(&hasher, requested, range_oob, 0)
3515            .await;
3516        assert!(matches!(
3517            result,
3518            Err(Error::RangeOutOfBounds(loc)) if loc == end_oob
3519        ));
3520
3521        // Requested range end out of bounds for the requested historical size but within structure.
3522        let range_end_gt_requested = requested + 1;
3523        let range_oob_at_requested = Location::<F>::new(0)..range_end_gt_requested;
3524        assert!(range_end_gt_requested <= mmr.leaves());
3525        let result = mmr
3526            .historical_range_proof(&hasher, requested, range_oob_at_requested, 0)
3527            .await;
3528        assert!(matches!(
3529            result,
3530            Err(Error::RangeOutOfBounds(loc)) if loc == range_end_gt_requested
3531        ));
3532
3533        // Range location overflow is caught as out-of-bounds (the bounds check
3534        // fires before the position conversion that would detect overflow).
3535        let overflow_loc = Location::<F>::new(u64::MAX);
3536        let overflow_range = Location::<F>::new(0)..overflow_loc;
3537        let result = mmr
3538            .historical_range_proof(&hasher, requested, overflow_range, 0)
3539            .await;
3540        assert!(matches!(
3541            result,
3542            Err(Error::RangeOutOfBounds(loc)) if loc == overflow_loc
3543        ));
3544
3545        mmr.destroy().await.unwrap();
3546    }
3547
3548    #[test_traced]
3549    fn test_full_historical_proof_range_validation_mmr() {
3550        let executor = deterministic::Runner::default();
3551        executor.start(full_historical_proof_range_validation_inner::<mmr::Family>);
3552    }
3553
3554    #[test_traced]
3555    fn test_full_historical_proof_range_validation_mmb() {
3556        let executor = deterministic::Runner::default();
3557        executor.start(full_historical_proof_range_validation_inner::<mmb::Family>);
3558    }
3559
3560    async fn full_historical_proof_non_size_prune_excludes_pruned_leaves_inner<F: Family>(
3561        context: deterministic::Context,
3562    ) {
3563        let hasher = Standard::<Sha256>::new(ForwardFold);
3564        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3565            context.child("non_size_prune"),
3566            &hasher,
3567            test_config(&context),
3568        )
3569        .await
3570        .unwrap();
3571
3572        let mut batch = mmr.new_batch();
3573        for i in 0..16 {
3574            batch = batch.add(&hasher, &test_digest(i));
3575        }
3576        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3577        mmr = mmr.apply_batch(&batch).unwrap();
3578
3579        let end = mmr.leaves();
3580        let mut failures = Vec::new();
3581        for prune_leaf in 1..*end {
3582            let prune_loc = Location::<F>::new(prune_leaf);
3583            mmr = mmr.prune(prune_loc).await.unwrap();
3584            for loc_u64 in 0..*end {
3585                let loc = Location::<F>::new(loc_u64);
3586                let range_includes_pruned_leaf = loc < prune_loc;
3587                match mmr.historical_proof(&hasher, end, loc, 0).await {
3588                    Ok(_) => {}
3589                    Err(Error::ElementPruned(_)) if range_includes_pruned_leaf => {}
3590                    Err(Error::ElementPruned(_)) => failures.push(format!(
3591                        "prune_loc={prune_loc} loc={loc} returned ElementPruned without a pruned range element"
3592                    )),
3593                    Err(err) => failures
3594                        .push(format!("prune_loc={prune_loc} loc={loc} err={err}")),
3595                }
3596            }
3597        }
3598
3599        assert!(
3600            failures.is_empty(),
3601            "historical proof generation returned unexpected errors: {failures:?}"
3602        );
3603
3604        mmr.destroy().await.unwrap();
3605    }
3606
3607    #[test_traced]
3608    fn test_full_historical_proof_non_size_prune_excludes_pruned_leaves_mmr() {
3609        let executor = deterministic::Runner::default();
3610        executor.start(
3611            full_historical_proof_non_size_prune_excludes_pruned_leaves_inner::<mmr::Family>,
3612        );
3613    }
3614
3615    #[test_traced]
3616    fn test_full_historical_proof_non_size_prune_excludes_pruned_leaves_mmb() {
3617        let executor = deterministic::Runner::default();
3618        executor.start(
3619            full_historical_proof_non_size_prune_excludes_pruned_leaves_inner::<mmb::Family>,
3620        );
3621    }
3622
3623    /// Regression: init_sync must recover from a journal left at an invalid size
3624    /// (e.g., a crash wrote a leaf but not its parent nodes).
3625    async fn full_init_sync_recovers_from_invalid_journal_size_inner<F: Family>(
3626        context: deterministic::Context,
3627    ) {
3628        let hasher = Standard::<Sha256>::new(ForwardFold);
3629
3630        // Build a structure with 3 leaves, sync, and drop.
3631        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3632            context.child("init"),
3633            &hasher,
3634            test_config(&context),
3635        )
3636        .await
3637        .unwrap();
3638        let mut batch = mmr.new_batch();
3639        for i in 0..3 {
3640            batch = batch.add(&hasher, &test_digest(i));
3641        }
3642        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3643        mmr = mmr.apply_batch(&batch).unwrap();
3644        let valid_size = mmr.size();
3645        let valid_root = mmr.root(&hasher, 0).unwrap();
3646        let mmr = mmr.sync().await.unwrap();
3647        drop(mmr);
3648
3649        // Append one extra digest to the journal, simulating a crash that wrote a
3650        // leaf (for the 4th element) but not its parent nodes. This makes the
3651        // journal size invalid.
3652        {
3653            let journal: Journal<_, Digest> = Journal::init(
3654                context.child("corrupt"),
3655                JConfig {
3656                    partition: "journal-partition".into(),
3657                    items_per_blob: NZU64!(7),
3658                    write_buffer: NZUsize!(1024),
3659                    replay_buffer: NZUsize!(1024),
3660                    page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
3661                },
3662            )
3663            .await
3664            .unwrap();
3665            assert_eq!(journal.size(), valid_size);
3666            let (journal, _) = journal.append(&Sha256::hash(&[b"orphan"])).await.unwrap();
3667            let journal = journal.sync().await.unwrap();
3668            assert_eq!(journal.size(), valid_size + 1);
3669        }
3670
3671        // init_sync should recover by rewinding to the last valid size.
3672        let sync_cfg = SyncConfig::<F, Digest, Sequential> {
3673            config: test_config(&context),
3674            range: non_empty_range!(Location::<F>::new(0), Location::<F>::new(100)),
3675            pinned_nodes: None,
3676        };
3677        let sync_mmr =
3678            Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
3679                .await
3680                .unwrap();
3681
3682        assert_eq!(sync_mmr.size(), valid_size);
3683        assert_eq!(sync_mmr.root(&hasher, 0).unwrap(), valid_root);
3684
3685        sync_mmr.destroy().await.unwrap();
3686    }
3687
3688    #[test_traced]
3689    fn test_init_sync_recovers_from_invalid_journal_size_mmr() {
3690        let executor = deterministic::Runner::default();
3691        executor.start(full_init_sync_recovers_from_invalid_journal_size_inner::<mmr::Family>);
3692    }
3693
3694    #[test_traced]
3695    fn test_init_sync_recovers_from_invalid_journal_size_mmb() {
3696        let executor = deterministic::Runner::default();
3697        executor.start(full_init_sync_recovers_from_invalid_journal_size_inner::<mmb::Family>);
3698    }
3699
3700    async fn full_stale_batch_inner<F: Family>(context: deterministic::Context) {
3701        let hasher: Standard<Sha256> = Standard::new(ForwardFold);
3702        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3703            context.child("storage"),
3704            &Standard::<Sha256>::new(ForwardFold),
3705            test_config(&context),
3706        )
3707        .await
3708        .unwrap();
3709
3710        // Create two batches from the same base.
3711        let batch_a = mmr.new_batch().add(&hasher, b"leaf-a");
3712        let batch_a = mmr.with_mem(|mem| batch_a.merkleize(mem, &hasher));
3713        let batch_b = mmr.new_batch().add(&hasher, b"leaf-b");
3714        let batch_b = mmr.with_mem(|mem| batch_b.merkleize(mem, &hasher));
3715
3716        // Apply A -- should succeed.
3717        mmr = mmr.apply_batch(&batch_a).unwrap();
3718
3719        // Apply B -- should fail (stale).
3720        assert!(matches!(
3721            mmr.apply_batch(&batch_b),
3722            Err(Error::StaleBatch { .. })
3723        ));
3724    }
3725
3726    #[test]
3727    fn test_stale_batch_mmr() {
3728        let executor = deterministic::Runner::default();
3729        executor.start(full_stale_batch_inner::<mmr::Family>);
3730    }
3731
3732    #[test]
3733    fn test_stale_batch_mmb() {
3734        let executor = deterministic::Runner::default();
3735        executor.start(full_stale_batch_inner::<mmb::Family>);
3736    }
3737
3738    /// Regression: `new_batch` must return the append-only full wrapper.
3739    async fn full_new_batch_returns_append_only_wrapper_inner<F: Family>(
3740        context: deterministic::Context,
3741    ) {
3742        let hasher = Standard::<Sha256>::new(ForwardFold);
3743        let mmr = Merkle::<F, _, Digest, Sequential>::init(
3744            context.child("storage"),
3745            &hasher,
3746            test_config(&context),
3747        )
3748        .await
3749        .unwrap();
3750
3751        let _batch: UnmerkleizedBatch<F, Digest, Sequential> = mmr.new_batch();
3752
3753        mmr.destroy().await.unwrap();
3754    }
3755
3756    #[test_traced]
3757    fn test_new_batch_returns_append_only_wrapper_mmr() {
3758        let executor = deterministic::Runner::default();
3759        executor.start(full_new_batch_returns_append_only_wrapper_inner::<mmr::Family>);
3760    }
3761
3762    #[test_traced]
3763    fn test_new_batch_returns_append_only_wrapper_mmb() {
3764        let executor = deterministic::Runner::default();
3765        executor.start(full_new_batch_returns_append_only_wrapper_inner::<mmb::Family>);
3766    }
3767
3768    /// Regression: update_leaf on a synced-out leaf must return ElementPruned, not panic.
3769    /// Before the fix, the batch took its pruning boundary from the journal's prune boundary
3770    /// (which could be 0), so the batch accepted the update. During merkleize, get_node
3771    /// returned None for the synced-out sibling and hit an expect panic.
3772    async fn full_update_leaf_after_sync_returns_pruned_inner<F: Family>(
3773        context: deterministic::Context,
3774    ) {
3775        let hasher = Standard::<Sha256>::new(ForwardFold);
3776        let mut mmr = Merkle::<F, _, Digest, Sequential>::init(
3777            context.child("storage"),
3778            &hasher,
3779            test_config(&context),
3780        )
3781        .await
3782        .unwrap();
3783
3784        // Add 50 elements and sync (flushes all nodes to journal, prunes mem).
3785        let mut batch = mmr.new_batch();
3786        for i in 0..50 {
3787            batch = batch.add(&hasher, &test_digest(i));
3788        }
3789        let batch = mmr.with_mem(|mem| batch.merkleize(mem, &hasher));
3790        mmr = mmr.apply_batch(&batch).unwrap();
3791        let mmr = mmr.sync().await.unwrap();
3792
3793        // Attempt to update leaf 0 which has been synced out of memory.
3794        // Use the inner batch type directly since the full wrapper
3795        // intentionally hides update_leaf.
3796        let batch = mmr.to_batch().new_batch();
3797        let result = batch.update_leaf(&hasher, Location::<F>::new(0), b"updated");
3798        assert!(matches!(result, Err(Error::ElementPruned(_))));
3799
3800        mmr.destroy().await.unwrap();
3801    }
3802
3803    #[test_traced]
3804    fn test_update_leaf_after_sync_returns_pruned_mmr() {
3805        let executor = deterministic::Runner::default();
3806        executor.start(full_update_leaf_after_sync_returns_pruned_inner::<mmr::Family>);
3807    }
3808
3809    #[test_traced]
3810    fn test_update_leaf_after_sync_returns_pruned_mmb() {
3811        let executor = deterministic::Runner::default();
3812        executor.start(full_update_leaf_after_sync_returns_pruned_inner::<mmb::Family>);
3813    }
3814}