1use 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
44pub 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 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 pub fn add_leaf_digest(self, digest: D) -> Self {
65 Self {
66 inner: self.inner.add_leaf_digest(digest),
67 }
68 }
69
70 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 pub fn leaves(&self) -> Location<F> {
83 self.inner.leaves()
84 }
85
86 pub fn strategy(&self) -> &S {
88 self.inner.strategy()
89 }
90
91 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#[derive(Clone)]
104pub struct Config<S: Strategy> {
105 pub journal_partition: String,
108
109 pub metadata_partition: String,
113
114 pub items_per_blob: NonZeroU64,
116
117 pub write_buffer: NonZeroUsize,
119
120 pub replay_buffer: NonZeroUsize,
122
123 pub strategy: S,
125
126 pub page_cache: CacheRef,
128}
129
130pub struct SyncConfig<F: Family, D: Digest, S: Strategy> {
138 pub config: Config<S>,
140
141 pub range: NonEmptyRange<Location<F>>,
143
144 pub pinned_nodes: Option<Vec<D>>,
148}
149
150pub struct Merkle<F: Family, E: Context, D: Digest, S: Strategy> {
152 pub(crate) mem: Arc<Mem<F, D>>,
161
162 pub(crate) pruned_to_pos: Position<F>,
165
166 pub(crate) journal: Journal<E, D>,
168
169 pub(crate) metadata: Metadata<E, U64, Vec<u8>>,
173
174 pub(crate) journal_dirty: bool,
176
177 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
190const NODE_PREFIX: u8 = 0;
192
193pub(crate) const PRUNED_TO_PREFIX: u8 = 1;
195
196impl<F: Family, E: Context, D: Digest, S: Strategy> Merkle<F, E, D, S> {
197 pub fn size(&self) -> Position<F> {
200 self.mem.size()
201 }
202
203 pub fn leaves(&self) -> Location<F> {
205 self.mem.leaves()
206 }
207
208 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 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 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 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 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 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 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 journal_boundary_floor
335 } else {
336 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 if *metadata_prune_pos > journal_bounds_start {
357 (journal, _) = journal.prune(*metadata_prune_pos).await?;
360 if journal.bounds().start != journal_bounds_start {
361 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 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 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 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 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 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 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 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 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 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 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 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 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 metadata.retain(|key: &U64, _| {
538 key.prefix() != NODE_PREFIX || retained_node_keys.contains(key)
539 });
540 }
541
542 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 let journal_leaves = Location::try_from(journal_size)?;
553 if let Some(pinned_nodes) = cfg.pinned_nodes {
554 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 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 if prune_pos < journal_size {
581 Self::add_extra_pinned_nodes(&mut mem, &metadata, &journal, prune_pos).await?;
582 }
583
584 let metadata = metadata.sync().await?;
586
587 (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 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 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 journal_positions.push(*position);
669 } else {
670 return Err(Error::ElementPruned(position));
671 }
672 }
673
674 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 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 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 pub async fn flush(self) -> Result<Self, Error<F>> {
708 self.flush_internal().await
709 }
710
711 pub async fn sync(mut self) -> Result<Self, Error<F>> {
713 self = self.flush_internal().await?;
714
715 if self.journal_dirty {
718 self.journal = self.journal.sync().await?;
719 self.journal_dirty = false;
720 }
721
722 Ok(self)
723 }
724
725 pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
731 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 async fn flush_internal(mut self) -> Result<Self, Error<F>> {
743 let journal_size = Position::<F>::new(self.journal.size());
744
745 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 let (head, tail) = self.mem.nodes_from(journal_size);
761 let encoded = self.journal.prepare_append(Many::Nested(&[head, tail]));
762
763 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 (self.journal, _) = self.journal.append_prepared(encoded).await?;
777 self.journal_dirty = true;
778
779 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 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 self = self.sync().await?;
807
808 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 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 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 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 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 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 pub fn get_pinned_nodes(&self) -> BTreeMap<Position<F>, D> {
876 self.mem.pinned_nodes()
877 }
878
879 #[cfg(test)]
880 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 self = self.sync().await?;
887
888 self.update_metadata(prune_to_pos).await?;
891
892 Ok(())
894 }
895
896 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 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 pub fn with_mem<R>(&self, f: impl FnOnce(&Mem<F, D>) -> R) -> R {
920 f(&self.mem)
921 }
922
923 pub(crate) fn snapshot(&self) -> Arc<Mem<F, D>> {
929 Arc::clone(&self.mem)
930 }
931
932 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 pub const fn strategy(&self) -> &S {
941 &self.strategy
942 }
943
944 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 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 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
1014impl<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 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 self.historical_range_proof(hasher, leaves, loc..loc + 1, inactive_peaks)
1073 .await
1074 }
1075
1076 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 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 self.range_proof(hasher, loc..loc + 1, inactive_peaks).await
1134 }
1135
1136 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 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 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 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 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 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 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 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 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 mmr = mmr.sync().await.unwrap();
1452
1453 let proof2 = mmr.proof(&hasher, test_element_loc, 0).await.unwrap();
1456 assert_eq!(proof, proof2);
1457
1458 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 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 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 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 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 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 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 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 mmr = mmr.flush().await.unwrap();
1601 assert_eq!(Position::<F>::new(mmr.journal.size()), expected_size);
1602
1603 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 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 fn full_flush_crash_inner<F: Family>() {
1632 let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1633
1634 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 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 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 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 fn full_flush_then_sync_crash_inner<F: Family>() {
1705 let hasher: Standard<Sha256> = Standard::new(ForwardFold);
1706
1707 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 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 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 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 {
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 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 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 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 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 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 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 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 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 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 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 async fn full_recovery_with_pruning_inner<F: Family>(context: deterministic::Context) {
2017 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 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 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 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 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 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 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 let prune_loc = Location::<F>::new(16);
2214 let mmr = mmr.prune(prune_loc).await.unwrap();
2215
2216 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 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 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 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 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 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, ));
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 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 async fn full_init_sync_empty_inner<F: Family>(context: deterministic::Context) {
2415 let hasher = Standard::<Sha256>::new(ForwardFold);
2416
2417 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 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 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 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 async fn full_init_sync_nonempty_exact_match_inner<F: Family>(context: deterministic::Context) {
2461 let hasher = Standard::<Sha256>::new(ForwardFold);
2462
2463 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 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 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 async fn full_init_sync_partial_overlap_inner<F: Family>(context: deterministic::Context) {
2541 let hasher = Standard::<Sha256>::new(ForwardFold);
2542
2543 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 let lower_bound_loc = original_pruning_boundary;
2568 let upper_bound_loc = original_leaves + 6; 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 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 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 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 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 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 let prune_loc = Location::<F>::new(25);
2872 mmr.prune(prune_loc).await.unwrap();
2873
2874 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 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(_)) => {} 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 async fn full_init_metadata_ahead_inner<F: Family>(context: deterministic::Context) {
2984 let hasher = Standard::<Sha256>::new(ForwardFold);
2985
2986 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 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 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 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 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 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 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 let original_size = mmr.size();
3078 let original_root = mmr.root(&hasher, 0).unwrap();
3079 drop(mmr);
3080
3081 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, };
3089
3090 let sync_mmr =
3091 Merkle::<F, _, Digest, Sequential>::init_sync(context.child("sync"), sync_cfg)
3092 .await
3093 .unwrap();
3094
3095 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 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 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 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 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 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 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 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 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 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 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 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 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 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 {
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 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 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 mmr = mmr.apply_batch(&batch_a).unwrap();
3718
3719 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 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 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 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 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}