1use crate::{
328 Context,
329 index::Factory as IndexFactory,
330 journal::{
331 authenticated,
332 contiguous::{fixed::Config as FConfig, variable::Config as VConfig},
333 },
334 merkle::{self, Location, full::Config as MerkleConfig},
335 qmdb::{
336 any::{
337 self, Config as AnyConfig,
338 operation::{Operation, Update},
339 },
340 bitmap::Shared,
341 },
342 translator::Translator,
343};
344use commonware_codec::{Codec, FixedSize};
345use commonware_cryptography::Hasher;
346use commonware_macros::boxed;
347use commonware_parallel::Strategy;
348use commonware_runtime::Spawner;
349use commonware_utils::bitmap::Prunable as BitMap;
350use core::num::NonZeroUsize;
351use std::sync::Arc;
352
353pub mod batch;
354pub mod db;
355pub mod grafting;
356
357pub mod ordered;
358pub mod proof;
359pub(crate) mod sync;
360pub mod unordered;
361
362use self::db::Metrics;
363
364#[derive(Clone)]
366pub struct Config<T: Translator, J, S: Strategy, B = ()> {
367 pub merkle_config: MerkleConfig<S>,
369
370 pub journal_config: J,
372
373 pub grafted_metadata_partition: String,
375
376 pub translator: T,
378
379 pub init_cache_size: Option<NonZeroUsize>,
382
383 pub init_buffer: NonZeroUsize,
385
386 pub init_concurrency: B,
391}
392
393impl<T: Translator, J, S: Strategy, B> From<Config<T, J, S, B>> for AnyConfig<T, J, S, B> {
394 fn from(cfg: Config<T, J, S, B>) -> Self {
395 Self {
396 merkle_config: cfg.merkle_config,
397 journal_config: cfg.journal_config,
398 translator: cfg.translator,
399 init_cache_size: cfg.init_cache_size,
400 init_buffer: cfg.init_buffer,
401 init_concurrency: cfg.init_concurrency,
402 }
403 }
404}
405
406pub type FixedConfig<T, S, B = ()> = Config<T, FConfig, S, B>;
408
409pub type VariableConfig<T, C, S, B = ()> = Config<T, VConfig<C>, S, B>;
411
412#[boxed]
414pub(super) async fn init<F, E, U, H, I, J, const N: usize, S>(
415 context: E,
416 config: Config<I::Translator, J::Config, S, <I as crate::qmdb::SnapshotBuild<F>>::Concurrency>,
417) -> Result<db::Db<F, E, J, I, H, U, N, S>, crate::qmdb::Error<F>>
418where
419 F: merkle::Graftable,
420 E: Context + Spawner,
421 U: Update,
422 H: Hasher,
423 I: IndexFactory<Value = Location<F>> + crate::qmdb::SnapshotBuild<F>,
424 J: authenticated::Backing<E, Item = Operation<F, U>> + 'static,
425 S: Strategy,
426 Operation<F, U>: Codec,
427{
428 const {
430 assert!(
434 N.is_multiple_of(H::Digest::SIZE),
435 "chunk size must be some multiple of the digest size",
436 );
437 assert!(N.is_power_of_two(), "chunk size must be a power of 2");
440 }
441
442 let strategy = config.merkle_config.strategy.clone();
443 let metadata_partition = config.grafted_metadata_partition.clone();
444
445 let (metadata, pruned_chunks, pinned_nodes) =
447 db::init_metadata(context.child("metadata"), &metadata_partition).await?;
448
449 let bitmap = BitMap::<N>::new_with_pruned_chunks(pruned_chunks)
451 .map_err(|_| crate::qmdb::Error::<F>::DataCorrupted("pruned chunks overflow"))?;
452 let bitmap = Arc::new(Shared::<N>::new(bitmap));
453
454 let any = any::init_with_bitmap(context.child("any"), config.into(), Some(bitmap)).await?;
457
458 let (grafted_tree, root) = db::rebuild_grafted_tree::<F, H, S, N>(
460 any.bitmap.as_ref(),
461 &pinned_nodes,
462 &any.log.merkle,
463 any.inactivity_floor_loc,
464 any.root(),
465 &strategy,
466 )
467 .await?;
468
469 let metrics = Metrics::new(context);
470 let db = db::Db {
471 any,
472 grafted_tree: Arc::new(grafted_tree),
473 metadata,
474 strategy,
475 root,
476 metrics,
477 #[cfg(test)]
478 halt_before_prune_log: false,
479 };
480 db.update_metrics();
481 Ok(db)
482}
483
484#[cfg(any(test, feature = "test-traits"))]
486pub trait BitmapPrunedBits {
487 fn pruned_bits(&self) -> u64;
489
490 fn get_bit(&self, index: u64) -> bool;
492
493 fn oldest_retained(&self) -> u64;
495}
496
497#[cfg(test)]
498pub mod tests {
499 pub use super::BitmapPrunedBits;
502 use super::{
503 FConfig, FixedConfig, MerkleConfig, VConfig, VariableConfig, grafting, ordered, unordered,
504 };
505 use crate::{
506 merkle::{self, mmb, mmr, storage::Storage as _},
507 qmdb::{
508 any::{
509 test::colliding_digest,
510 traits::{DbAny, MerkleizedBatch as _, UnmerkleizedBatch as _},
511 },
512 store::tests::{TestKey, TestValue},
513 verify_proof,
514 },
515 translator::Translator,
516 };
517 use commonware_parallel::Sequential;
518 use commonware_runtime::{
519 BufferPooler, Runner as _, Supervisor as _,
520 buffer::paged::CacheRef,
521 deterministic::{self, Context},
522 };
523 use commonware_utils::{NZU16, NZU64, NZUsize, TestRng, bitmap::Readable};
524 use core::future::Future;
525 use ordered::tests::test_build_small_close_reopen as test_ordered_build_small_close_reopen;
526 use rand::Rng;
527 use std::{
528 num::{NonZeroU16, NonZeroUsize},
529 sync::Arc,
530 };
531 use tracing::warn;
532 use unordered::tests::test_build_small_close_reopen as test_unordered_build_small_close_reopen;
533
534 type Error<F> = crate::qmdb::Error<F>;
535 type Location<F> = merkle::Location<F>;
536 type WriteVec<F, C> = Vec<(<C as DbAny<F>>::Key, Option<<C as DbAny<F>>::Value>)>;
537
538 const PAGE_SIZE: NonZeroU16 = NZU16!(88);
540 const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(8);
541
542 macro_rules! staged_merkleize_parity_test {
555 ($name:ident, $open_db:path) => {
556 #[test_traced("WARN")]
557 pub fn $name() {
558 fn key(i: u64) -> Digest {
559 Sha256::hash(&[&i.to_be_bytes()])
560 }
561 fn val(i: u64) -> Digest {
562 Sha256::hash(&[&(i + 10000).to_be_bytes()])
563 }
564
565 deterministic::Runner::default().start(|ctx| async move {
566 let db = $open_db(ctx.child("current"), "staged-parity".to_string()).await;
567
568 let mut seed = db.new_batch();
569 for i in 0..2000u64 {
570 seed = seed.write(key(i), Some(val(i)));
571 }
572 let seed = seed.merkleize(&db, None).await.unwrap();
573 let (db, _) = db.apply_batch(seed).await.unwrap();
574 let db = db.commit().await.unwrap();
575
576 for depth in [0u8, 1u8, 2u8] {
577 let mut stack = Vec::new();
581 match depth {
582 0 => {}
583 1 => {
584 let mut p = db.new_batch();
585 for i in 0..50u64 {
586 p = p.write(key(i), Some(val(i + 1_000)));
587 }
588 for i in 100..110u64 {
589 p = p.write(key(i), None);
590 }
591 stack.push(p.merkleize(&db, None).await.unwrap());
592 }
593 2 => {
594 let mut grandparent = db.new_batch();
595 for i in 0..10u64 {
596 grandparent = grandparent.write(key(i), Some(val(i + 1_000)));
597 }
598 for i in 100..110u64 {
599 grandparent = grandparent.write(key(i), None);
600 }
601 let grandparent = grandparent.merkleize(&db, None).await.unwrap();
602
603 let mut p = grandparent.new_batch::<Sha256>();
604 for i in 20..30u64 {
605 p = p.write(key(i), Some(val(i + 2_000)));
606 }
607 let p = p.merkleize(&db, None).await.unwrap();
608 stack.push(grandparent);
609 stack.push(p);
610 }
611 _ => unreachable!("covered depths"),
612 };
613 let new_batch = || {
614 stack
615 .last()
616 .map_or_else(|| db.new_batch(), |p| p.new_batch::<Sha256>())
617 };
618
619 let read_keys = [
623 key(5),
624 key(6),
625 key(9000),
626 key(5),
627 key(0),
628 key(20),
629 key(60),
630 key(105),
631 ];
632 let keys: Vec<&Digest> = read_keys.iter().collect();
633 let indexed_updates = vec![
634 (0, Some(val(5_000))),
635 (2, Some(val(5_001))),
636 (3, Some(val(5_002))),
637 (4, Some(val(5_003))),
638 (5, None),
639 (6, Some(val(5_004))),
640 (7, Some(val(5_005))),
641 ];
642 let upserts = vec![
643 (key(7000), Some(val(6_000))),
644 (key(30), Some(val(6_001))),
645 (key(5), Some(val(6_002))),
646 (key(31), None),
647 ];
648
649 let mut explicit = new_batch();
650 let explicit_values = explicit.get_many(&keys, &db).await.unwrap();
651 for (slot, value) in &indexed_updates {
652 explicit = explicit.write(read_keys[*slot], *value);
653 }
654 for (k, v) in &upserts {
655 explicit = explicit.write(*k, *v);
656 }
657 let explicit_root = explicit.merkleize(&db, None).await.unwrap().root();
658
659 let (staged_values, staged) = new_batch().stage(&keys, &db).await.unwrap();
660 let staged_root = staged
661 .merkleize(indexed_updates.clone(), upserts.clone(), None, &db)
662 .await
663 .unwrap()
664 .root();
665
666 assert_eq!(
667 explicit_values, staged_values,
668 "value mismatch at depth={depth}"
669 );
670 assert_eq!(explicit_root, staged_root, "root mismatch at depth={depth}");
671
672 let split = 3;
673 let (mut expanded_values, staged) =
674 new_batch().stage(&keys[..split], &db).await.unwrap();
675 let (range, suffix_values, staged) =
676 staged.expand(&keys[split..], &db).await.unwrap();
677 assert_eq!(range, split..keys.len());
678 expanded_values.extend(suffix_values);
679 let expanded_root = staged
680 .merkleize(indexed_updates.clone(), upserts.clone(), None, &db)
681 .await
682 .unwrap()
683 .root();
684
685 assert_eq!(
686 explicit_values, expanded_values,
687 "expanded value mismatch at depth={depth}"
688 );
689 assert_eq!(
690 explicit_root, expanded_root,
691 "expanded root mismatch at depth={depth}"
692 );
693
694 let planned = val(7_000);
695 let duplicate_update = val(7_001);
696 let (first_values, staged) =
697 new_batch().stage(&keys[..1], &db).await.unwrap();
698 let (duplicate_range, duplicate_values, staged) =
699 staged.expand(&keys[..1], &db).await.unwrap();
700 assert_eq!(duplicate_range, 1..2);
701 assert_eq!(
702 first_values[0], duplicate_values[0],
703 "duplicate expansion must assign a new slot without changing the base read"
704 );
705 assert_ne!(
706 duplicate_values[0],
707 Some(planned),
708 "expand must not observe values computed for earlier staged slots"
709 );
710
711 let duplicate_root = staged
712 .merkleize(
713 vec![
714 (0, Some(planned)),
715 (duplicate_range.start, Some(duplicate_update)),
716 ],
717 Vec::new(),
718 None,
719 &db,
720 )
721 .await
722 .unwrap()
723 .root();
724 let expected_duplicate_root = new_batch()
725 .write(read_keys[0], Some(planned))
726 .write(read_keys[0], Some(duplicate_update))
727 .merkleize(&db, None)
728 .await
729 .unwrap()
730 .root();
731 assert_eq!(
732 expected_duplicate_root, duplicate_root,
733 "duplicate expanded slots should use normal update-order semantics"
734 );
735 }
736 });
737 }
738 };
739 }
740 pub(crate) use staged_merkleize_parity_test;
741
742 pub(crate) fn fixed_config<T: Translator + Default>(
744 partition_prefix: &str,
745 pooler: &impl BufferPooler,
746 ) -> FixedConfig<T, Sequential> {
747 fixed_config_full(partition_prefix, pooler, ())
748 }
749
750 pub(crate) fn fixed_config_full<T: Translator + Default, B>(
753 partition_prefix: &str,
754 pooler: &impl BufferPooler,
755 init_concurrency: B,
756 ) -> FixedConfig<T, Sequential, B> {
757 let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
758 FixedConfig {
759 merkle_config: MerkleConfig {
760 journal_partition: format!("{partition_prefix}-journal-partition"),
761 metadata_partition: format!("{partition_prefix}-metadata-partition"),
762 items_per_blob: NZU64!(11),
763 write_buffer: NZUsize!(1024),
764 replay_buffer: NZUsize!(1024),
765 strategy: Sequential,
766 page_cache: page_cache.clone(),
767 },
768 journal_config: FConfig {
769 partition: format!("{partition_prefix}-partition-prefix"),
770 items_per_blob: NZU64!(7),
771 page_cache,
772 write_buffer: NZUsize!(1024),
773 replay_buffer: NZUsize!(1024),
774 },
775 grafted_metadata_partition: format!("{partition_prefix}-grafted-metadata-partition"),
776 translator: T::default(),
777 init_cache_size: Some(NZUsize!(1024)),
778 init_buffer: NZUsize!(1 << 21),
779 init_concurrency,
780 }
781 }
782
783 pub(crate) fn variable_config<T: Translator + Default>(
785 partition_prefix: &str,
786 pooler: &impl BufferPooler,
787 ) -> VariableConfig<T, ((), ()), Sequential> {
788 variable_config_full(partition_prefix, pooler, ())
789 }
790
791 pub(crate) fn variable_config_full<T: Translator + Default, B>(
794 partition_prefix: &str,
795 pooler: &impl BufferPooler,
796 init_concurrency: B,
797 ) -> VariableConfig<T, ((), ()), Sequential, B> {
798 let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
799 VariableConfig {
800 merkle_config: MerkleConfig {
801 journal_partition: format!("{partition_prefix}-journal-partition"),
802 metadata_partition: format!("{partition_prefix}-metadata-partition"),
803 items_per_blob: NZU64!(11),
804 write_buffer: NZUsize!(1024),
805 replay_buffer: NZUsize!(1024),
806 strategy: Sequential,
807 page_cache: page_cache.clone(),
808 },
809 journal_config: VConfig {
810 partition: format!("{partition_prefix}-partition-prefix"),
811 items_per_section: NZU64!(7),
812 compression: None,
813 codec_config: ((), ()),
814 page_cache,
815 write_buffer: NZUsize!(1024),
816 replay_buffer: NZUsize!(1024),
817 },
818 grafted_metadata_partition: format!("{partition_prefix}-grafted-metadata-partition"),
819 translator: T::default(),
820 init_cache_size: Some(NZUsize!(1024)),
821 init_buffer: NZUsize!(1 << 21),
822 init_concurrency,
823 }
824 }
825
826 pub(crate) fn fixed_config_partitioned<T: Translator + Default>(
828 partition_prefix: &str,
829 pooler: &impl BufferPooler,
830 ) -> FixedConfig<T, Sequential, NonZeroUsize> {
831 fixed_config_full(partition_prefix, pooler, NZUsize!(1))
832 }
833
834 pub(crate) fn variable_config_partitioned<T: Translator + Default>(
836 partition_prefix: &str,
837 pooler: &impl BufferPooler,
838 ) -> VariableConfig<T, ((), ()), Sequential, NonZeroUsize> {
839 variable_config_full(partition_prefix, pooler, NZUsize!(1))
840 }
841
842 fn commit_writes<'a, F: merkle::Graftable, C: DbAny<F> + 'a>(
848 db: C,
849 writes: impl IntoIterator<Item = (C::Key, Option<<C as DbAny<F>>::Value>)> + 'a,
850 ) -> std::pin::Pin<Box<dyn Future<Output = Result<C, Error<F>>> + 'a>> {
851 Box::pin(async move {
852 let mut batch = db.new_batch();
853 for (k, v) in writes {
854 batch = batch.write(k, v);
855 }
856 let merkleized = batch.merkleize(&db, None).await?;
857 let (db, _) = db.apply_batch(merkleized).await?;
858 db.commit().await
859 })
860 }
861
862 #[boxed]
865 pub async fn apply_random_ops<F, C>(
866 num_elements: u64,
867 commit_changes: bool,
868 rng_seed: u64,
869 mut db: C,
870 ) -> Result<C, Error<F>>
871 where
872 F: merkle::Graftable,
873 C: DbAny<F>,
874 C::Key: TestKey,
875 <C as DbAny<F>>::Value: TestValue,
876 {
877 warn!("rng_seed={}", rng_seed);
879 let mut rng = TestRng::new(rng_seed);
880
881 let writes: Vec<_> = (0u64..num_elements)
883 .map(|i| {
884 let k = TestKey::from_seed(i);
885 let v = TestValue::from_seed(rng.next_u64());
886 (k, Some(v))
887 })
888 .collect();
889 if commit_changes {
890 db = commit_writes(db, writes).await?;
891 }
892
893 let mut pending: WriteVec<F, C> = Vec::new();
896 for _ in 0u64..num_elements * 10 {
897 let rand_key = TestKey::from_seed(rng.next_u64() % num_elements);
898 if rng.next_u32().is_multiple_of(7) {
899 pending.push((rand_key, None));
900 continue;
901 }
902 let v = TestValue::from_seed(rng.next_u64());
903 pending.push((rand_key, Some(v)));
904 if commit_changes && rng.next_u32().is_multiple_of(20) {
905 db = commit_writes(db, pending.drain(..)).await?;
906 }
907 }
908 if commit_changes {
909 db = commit_writes(db, pending).await?;
910 }
911 Ok(db)
912 }
913
914 #[boxed]
916 async fn build_random_close_reopen_round<M, C, F, Fut>(
917 mut context: Context,
918 mut open_db: F,
919 ) -> String
920 where
921 M: merkle::Graftable + 'static,
922 C: DbAny<M> + 'static,
923 C::Key: TestKey,
924 <C as DbAny<M>>::Value: TestValue,
925 F: FnMut(Context, String) -> Fut,
926 Fut: Future<Output = C>,
927 {
928 const ELEMENTS: u64 = 1000;
929
930 let partition = "build-random".to_string();
931 let rng_seed = context.next_u64();
932 let db: C = open_db(context.child("first"), partition.clone()).await;
933 let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
934 .await
935 .unwrap();
936 let merkleized = db.new_batch().merkleize(&db, None).await.unwrap();
937 let (db, _) = db.apply_batch(merkleized).await.unwrap();
938 let db = db.sync().await.unwrap();
939
940 let root = db.root();
942 drop(db);
943 let db: C = open_db(context.child("second"), partition).await;
944
945 assert_eq!(db.root(), root);
947
948 db.destroy().await.unwrap();
949 context.auditor().state()
950 }
951
952 pub async fn test_build_random_close_reopen<M, C, F, Fut>(context: Context, open_db: F)
957 where
958 M: merkle::Graftable + 'static,
959 C: DbAny<M> + 'static,
960 C::Key: TestKey,
961 <C as DbAny<M>>::Value: TestValue,
962 F: FnMut(Context, String) -> Fut + Clone,
963 Fut: Future<Output = C>,
964 {
965 let state1 =
967 build_random_close_reopen_round::<M, C, F, Fut>(context, open_db.clone()).await;
968
969 let executor = deterministic::Runner::default();
971 let state2 = executor
972 .start(|context| build_random_close_reopen_round::<M, C, F, Fut>(context, open_db));
973
974 assert_eq!(state1, state2);
975 }
976
977 pub async fn test_commit_after_sync_recovery<M, C, F, Fut>(context: Context, mut open_db: F)
979 where
980 M: merkle::Graftable + 'static,
981 C: DbAny<M> + 'static,
982 C::Key: TestKey,
983 <C as DbAny<M>>::Value: TestValue,
984 F: FnMut(Context, String) -> Fut + Clone,
985 Fut: Future<Output = C>,
986 {
987 let mut open_db_clone = open_db.clone();
988 let partition = "commit-after-sync".to_string();
989 let db: C = Box::pin(open_db_clone(context.child("first"), partition.clone())).await;
990 let key0 = <<C as DbAny<M>>::Key as TestKey>::from_seed(0);
991 let key1 = <<C as DbAny<M>>::Key as TestKey>::from_seed(1);
992 let value0 = <C as DbAny<M>>::Value::from_seed(100);
993 let value1 = <C as DbAny<M>>::Value::from_seed(200);
994
995 let db = commit_writes::<M, C>(db, [(key0, Some(value0.clone()))])
997 .await
998 .unwrap();
999 let db = db.sync().await.unwrap();
1000
1001 let db = commit_writes::<M, C>(db, [(key1, Some(value1.clone()))])
1003 .await
1004 .unwrap();
1005 let committed_root = db.root();
1006 let committed_size = db.size();
1007 drop(db);
1008
1009 let db: C = Box::pin(open_db(context.child("second"), partition)).await;
1010 assert_eq!(db.root(), committed_root);
1011 assert_eq!(db.size(), committed_size);
1012 assert_eq!(db.get(&key0).await.unwrap(), Some(value0));
1013 assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1014
1015 db.destroy().await.unwrap();
1016 }
1017
1018 pub async fn test_simulate_write_failures<M, C, F, Fut>(mut context: Context, mut open_db: F)
1023 where
1024 M: merkle::Graftable + 'static,
1025 C: DbAny<M> + 'static,
1026 C::Key: TestKey,
1027 <C as DbAny<M>>::Value: TestValue,
1028 F: FnMut(Context, String) -> Fut + Clone,
1029 Fut: Future<Output = C>,
1030 {
1031 const ELEMENTS: u64 = 1000;
1032
1033 let partition = "build-random-fail-commit".to_string();
1034 let rng_seed = context.next_u64();
1035 let db: C = Box::pin(open_db(context.child("first"), partition.clone())).await;
1036 let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
1037 .await
1038 .unwrap();
1039 let db = commit_writes(db, []).await.unwrap();
1040 let committed_root = db.root();
1041 let committed_op_count = db.bounds().end;
1042 let boundary = db.sync_boundary();
1043 let db = db.prune(boundary).await.unwrap();
1044
1045 let db = apply_random_ops::<M, C>(ELEMENTS, false, rng_seed + 1, db)
1047 .await
1048 .unwrap();
1049
1050 drop(db);
1053 let db: C = Box::pin(open_db(
1054 context.child("scenario").with_attribute("index", 1),
1055 partition.clone(),
1056 ))
1057 .await;
1058 assert_eq!(db.root(), committed_root);
1059 assert_eq!(db.bounds().end, committed_op_count);
1060
1061 let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed + 1, db)
1063 .await
1064 .unwrap();
1065
1066 let committed_op_count = db.bounds().end;
1070 drop(db);
1071
1072 let db: C = Box::pin(open_db(
1075 context.child("scenario").with_attribute("index", 2),
1076 partition.clone(),
1077 ))
1078 .await;
1079 let scenario_2_root = db.root();
1080
1081 let fresh_partition = "build-random-fail-commit-fresh".to_string();
1084 let db: C = Box::pin(open_db(context.child("fresh"), fresh_partition.clone())).await;
1085 let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
1086 .await
1087 .unwrap();
1088 let db = commit_writes(db, []).await.unwrap();
1089 let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed + 1, db)
1090 .await
1091 .unwrap();
1092 let boundary = db.sync_boundary();
1093 let db = db.prune(boundary).await.unwrap();
1094 assert_eq!(db.bounds().end, committed_op_count);
1096 assert_eq!(db.root(), scenario_2_root);
1097
1098 db.destroy().await.unwrap();
1099 }
1100
1101 pub async fn test_different_pruning_delays_same_root<M, C, F, Fut>(
1106 context: Context,
1107 mut open_db: F,
1108 ) where
1109 M: merkle::Graftable,
1110 C: DbAny<M>,
1111 C::Key: TestKey,
1112 <C as DbAny<M>>::Value: TestValue,
1113 F: FnMut(Context, String) -> Fut + Clone,
1114 Fut: Future<Output = C>,
1115 {
1116 const NUM_OPERATIONS: u64 = 1000;
1117
1118 let mut open_db_clone = open_db.clone();
1119 let mut db_no_pruning: C = Box::pin(open_db_clone(
1121 context.child("no_pruning"),
1122 "no-pruning-test".into(),
1123 ))
1124 .await;
1125 let mut db_pruning: C =
1126 Box::pin(open_db(context.child("pruning"), "pruning-test".into())).await;
1127
1128 let mut pending_no_pruning: WriteVec<M, C> = Vec::new();
1131 let mut pending_pruning: WriteVec<M, C> = Vec::new();
1132 for i in 0..NUM_OPERATIONS {
1133 let key: C::Key = TestKey::from_seed(i);
1134 let value: <C as DbAny<M>>::Value = TestValue::from_seed(i * 1000);
1135
1136 pending_no_pruning.push((key, Some(value.clone())));
1137 pending_pruning.push((key, Some(value)));
1138
1139 if i % 50 == 49 {
1141 db_no_pruning = commit_writes(db_no_pruning, pending_no_pruning.drain(..))
1142 .await
1143 .unwrap();
1144 db_pruning = commit_writes(db_pruning, pending_pruning.drain(..))
1145 .await
1146 .unwrap();
1147 db_pruning = db_pruning
1148 .prune(db_no_pruning.sync_boundary())
1149 .await
1150 .unwrap();
1151 }
1152 }
1153
1154 let db_no_pruning = commit_writes(db_no_pruning, pending_no_pruning)
1156 .await
1157 .unwrap();
1158 let db_pruning = commit_writes(db_pruning, pending_pruning).await.unwrap();
1159
1160 let root_no_pruning = db_no_pruning.root();
1162 let root_pruning = db_pruning.root();
1163 assert_eq!(root_no_pruning, root_pruning);
1164
1165 assert_eq!(
1167 db_no_pruning.inactivity_floor_loc(),
1168 db_pruning.inactivity_floor_loc()
1169 );
1170
1171 db_no_pruning.destroy().await.unwrap();
1172 db_pruning.destroy().await.unwrap();
1173 }
1174
1175 pub async fn test_sync_persists_bitmap_pruning_boundary<M, C, F, Fut>(
1181 mut context: Context,
1182 mut open_db: F,
1183 ) where
1184 M: merkle::Graftable + 'static,
1185 C: DbAny<M> + BitmapPrunedBits + 'static,
1186 C::Key: TestKey,
1187 <C as DbAny<M>>::Value: TestValue,
1188 F: FnMut(Context, String) -> Fut + Clone,
1189 Fut: Future<Output = C>,
1190 {
1191 const ELEMENTS: u64 = 500;
1192
1193 let mut open_db_clone = open_db.clone();
1194 let partition = "sync-bitmap-pruning".to_string();
1195 let rng_seed = context.next_u64();
1196 let db: C = Box::pin(open_db_clone(context.child("first"), partition.clone())).await;
1197
1198 let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
1200 .await
1201 .unwrap();
1202 let merkleized = db.new_batch().merkleize(&db, None).await.unwrap();
1203 let (db, _) = db.apply_batch(merkleized).await.unwrap();
1204
1205 let boundary = db.sync_boundary();
1207 let db = db.prune(boundary).await.unwrap();
1208
1209 let pruned_bits_before = db.pruned_bits();
1210 warn!(
1211 "pruned_bits_before={}, inactivity_floor={}, op_count={}",
1212 pruned_bits_before,
1213 *db.inactivity_floor_loc(),
1214 *db.bounds().end
1215 );
1216
1217 assert!(
1219 pruned_bits_before > 0,
1220 "Expected bitmap to have pruned bits after prune()"
1221 );
1222
1223 let db = db.sync().await.unwrap();
1225
1226 let root_before = db.root();
1228 drop(db);
1229
1230 let db: C = Box::pin(open_db(context.child("second"), partition)).await;
1232
1233 let pruned_bits_after = db.pruned_bits();
1236 warn!("pruned_bits_after={}", pruned_bits_after);
1237
1238 assert_eq!(
1239 pruned_bits_after, pruned_bits_before,
1240 "Bitmap pruned bits mismatch after reopen - sync() may not have called write_pruned()"
1241 );
1242
1243 assert_eq!(db.root(), root_before);
1245
1246 db.destroy().await.unwrap();
1247 }
1248
1249 pub async fn test_current_db_build_big<M, C, F, Fut>(context: Context, mut open_db: F)
1255 where
1256 M: merkle::Graftable,
1257 C: DbAny<M>,
1258 C::Key: TestKey,
1259 <C as DbAny<M>>::Value: TestValue,
1260 F: FnMut(Context, String) -> Fut + Clone,
1261 Fut: Future<Output = C>,
1262 {
1263 const ELEMENTS: u64 = 1000;
1264
1265 let mut open_db_clone = open_db.clone();
1266 let db: C = Box::pin(open_db_clone(context.child("first"), "build-big".into())).await;
1267
1268 let mut map = std::collections::HashMap::<C::Key, <C as DbAny<M>>::Value>::default();
1269
1270 let mut batch = db.new_batch();
1272
1273 for i in 0u64..ELEMENTS {
1275 let k: C::Key = TestKey::from_seed(i);
1276 let v: <C as DbAny<M>>::Value = TestValue::from_seed(i * 1000);
1277 batch = batch.write(k, Some(v.clone()));
1278 map.insert(k, v);
1279 }
1280
1281 for i in 0u64..ELEMENTS {
1283 if i % 3 != 0 {
1284 continue;
1285 }
1286 let k: C::Key = TestKey::from_seed(i);
1287 let v: <C as DbAny<M>>::Value = TestValue::from_seed((i + 1) * 10000);
1288 batch = batch.write(k, Some(v.clone()));
1289 map.insert(k, v);
1290 }
1291
1292 for i in 0u64..ELEMENTS {
1294 if i % 7 != 1 {
1295 continue;
1296 }
1297 let k: C::Key = TestKey::from_seed(i);
1298 batch = batch.write(k, None);
1299 map.remove(&k);
1300 }
1301
1302 let merkleized = batch.merkleize(&db, None).await.unwrap();
1303 let (db, _) = db.apply_batch(merkleized).await.unwrap();
1304
1305 let db = db.sync().await.unwrap();
1307 let boundary = db.sync_boundary();
1308 let db = db.prune(boundary).await.unwrap();
1309
1310 let root = db.root();
1312 db.sync().await.unwrap();
1313
1314 let db: C = Box::pin(open_db(context.child("second"), "build-big".into())).await;
1316 assert_eq!(root, db.root());
1317
1318 for i in 0u64..ELEMENTS {
1320 let k: C::Key = TestKey::from_seed(i);
1321 if let Some(map_value) = map.get(&k) {
1322 let Some(db_value) = db.get(&k).await.unwrap() else {
1323 panic!("key not found in db: {k}");
1324 };
1325 assert_eq!(*map_value, db_value);
1326 } else {
1327 assert!(db.get(&k).await.unwrap().is_none());
1328 }
1329 }
1330 }
1331
1332 pub async fn test_stale_batch_side_effect_free<M, C, F, Fut>(context: Context, mut open_db: F)
1336 where
1337 M: merkle::Graftable,
1338 C: DbAny<M>,
1339 C::Key: TestKey,
1340 <C as DbAny<M>>::Value: TestValue,
1341 F: FnMut(Context, String) -> Fut,
1342 Fut: Future<Output = C>,
1343 {
1344 let db: C = Box::pin(open_db(
1345 context.child("db"),
1346 "stale-side-effect-free".into(),
1347 ))
1348 .await;
1349
1350 let key1 = <C::Key as TestKey>::from_seed(1);
1351 let key2 = <C::Key as TestKey>::from_seed(2);
1352 let value1 = <<C as DbAny<M>>::Value as TestValue>::from_seed(10);
1353 let value2 = <<C as DbAny<M>>::Value as TestValue>::from_seed(20);
1354
1355 let mut batch = db.new_batch();
1356 batch = batch.write(key1, Some(value1.clone()));
1357 let batch_a = batch.merkleize(&db, None).await.unwrap();
1358 let mut batch = db.new_batch();
1359 batch = batch.write(key2, Some(value2));
1360 let batch_b = batch.merkleize(&db, None).await.unwrap();
1361
1362 let (db, _) = db.apply_batch(batch_a).await.unwrap();
1363 let db = db.commit().await.unwrap();
1364 let expected_root = db.root();
1365 let expected_bounds = db.bounds();
1366 let expected_metadata = db.get_metadata().await.unwrap();
1367 assert_eq!(db.get(&key1).await.unwrap(), Some(value1.clone()));
1368 assert_eq!(db.get(&key2).await.unwrap(), None);
1369
1370 let Err(err) = db.apply_batch(batch_b).await else {
1371 panic!("expected StaleBatch error");
1372 };
1373 assert!(
1374 matches!(err, Error::StaleBatch),
1375 "expected StaleBatch error, got {err:?}"
1376 );
1377
1378 let db: C = Box::pin(open_db(
1380 context.child("reopen"),
1381 "stale-side-effect-free".into(),
1382 ))
1383 .await;
1384 assert_eq!(db.root(), expected_root);
1385 assert_eq!(db.bounds(), expected_bounds);
1386 assert_eq!(db.get_metadata().await.unwrap(), expected_metadata);
1387 assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1388 assert_eq!(db.get(&key2).await.unwrap(), None);
1389
1390 db.destroy().await.unwrap();
1391 }
1392
1393 use crate::translator::OneCap;
1394 use commonware_cryptography::{Hasher as _, Sha256, sha256::Digest};
1395 use commonware_macros::{boxed, test_group, test_traced};
1396
1397 type OrderedFixedDb =
1398 ordered::fixed::Db<mmr::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
1399 type OrderedVariableDb =
1400 ordered::variable::Db<mmr::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
1401 type UnorderedFixedDb =
1402 unordered::fixed::Db<mmr::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
1403 type UnorderedVariableDb = unordered::variable::Db<
1404 mmr::Family,
1405 Context,
1406 Digest,
1407 Digest,
1408 Sha256,
1409 OneCap,
1410 32,
1411 Sequential,
1412 >;
1413 type OrderedFixedP1Db = ordered::fixed::partitioned::Db<
1414 mmr::Family,
1415 Context,
1416 Digest,
1417 Digest,
1418 Sha256,
1419 OneCap,
1420 1,
1421 32,
1422 Sequential,
1423 >;
1424 type OrderedVariableP1Db = ordered::variable::partitioned::Db<
1425 mmr::Family,
1426 Context,
1427 Digest,
1428 Digest,
1429 Sha256,
1430 OneCap,
1431 1,
1432 32,
1433 Sequential,
1434 >;
1435 type UnorderedFixedP1Db = unordered::fixed::partitioned::Db<
1436 mmr::Family,
1437 Context,
1438 Digest,
1439 Digest,
1440 Sha256,
1441 OneCap,
1442 1,
1443 32,
1444 Sequential,
1445 >;
1446 type UnorderedVariableP1Db = unordered::variable::partitioned::Db<
1447 mmr::Family,
1448 Context,
1449 Digest,
1450 Digest,
1451 Sha256,
1452 OneCap,
1453 1,
1454 32,
1455 Sequential,
1456 >;
1457 type OrderedFixedP2Db = ordered::fixed::partitioned::Db<
1458 mmr::Family,
1459 Context,
1460 Digest,
1461 Digest,
1462 Sha256,
1463 OneCap,
1464 2,
1465 32,
1466 Sequential,
1467 >;
1468 type OrderedVariableP2Db = ordered::variable::partitioned::Db<
1469 mmr::Family,
1470 Context,
1471 Digest,
1472 Digest,
1473 Sha256,
1474 OneCap,
1475 2,
1476 32,
1477 Sequential,
1478 >;
1479 type UnorderedFixedP2Db = unordered::fixed::partitioned::Db<
1480 mmr::Family,
1481 Context,
1482 Digest,
1483 Digest,
1484 Sha256,
1485 OneCap,
1486 2,
1487 32,
1488 Sequential,
1489 >;
1490 type UnorderedVariableP2Db = unordered::variable::partitioned::Db<
1491 mmr::Family,
1492 Context,
1493 Digest,
1494 Digest,
1495 Sha256,
1496 OneCap,
1497 2,
1498 32,
1499 Sequential,
1500 >;
1501
1502 type OrderedFixedMmbDb =
1503 ordered::fixed::Db<mmb::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
1504 type OrderedVariableMmbDb =
1505 ordered::variable::Db<mmb::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
1506 type UnorderedFixedMmbDb =
1507 unordered::fixed::Db<mmb::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
1508 type UnorderedVariableMmbDb = unordered::variable::Db<
1509 mmb::Family,
1510 Context,
1511 Digest,
1512 Digest,
1513 Sha256,
1514 OneCap,
1515 32,
1516 Sequential,
1517 >;
1518 type OrderedFixedMmbP1Db = ordered::fixed::partitioned::Db<
1519 mmb::Family,
1520 Context,
1521 Digest,
1522 Digest,
1523 Sha256,
1524 OneCap,
1525 1,
1526 32,
1527 Sequential,
1528 >;
1529 type OrderedVariableMmbP1Db = ordered::variable::partitioned::Db<
1530 mmb::Family,
1531 Context,
1532 Digest,
1533 Digest,
1534 Sha256,
1535 OneCap,
1536 1,
1537 32,
1538 Sequential,
1539 >;
1540 type UnorderedFixedMmbP1Db = unordered::fixed::partitioned::Db<
1541 mmb::Family,
1542 Context,
1543 Digest,
1544 Digest,
1545 Sha256,
1546 OneCap,
1547 1,
1548 32,
1549 Sequential,
1550 >;
1551 type UnorderedVariableMmbP1Db = unordered::variable::partitioned::Db<
1552 mmb::Family,
1553 Context,
1554 Digest,
1555 Digest,
1556 Sha256,
1557 OneCap,
1558 1,
1559 32,
1560 Sequential,
1561 >;
1562 type OrderedFixedMmbP2Db = ordered::fixed::partitioned::Db<
1563 mmb::Family,
1564 Context,
1565 Digest,
1566 Digest,
1567 Sha256,
1568 OneCap,
1569 2,
1570 32,
1571 Sequential,
1572 >;
1573 type OrderedVariableMmbP2Db = ordered::variable::partitioned::Db<
1574 mmb::Family,
1575 Context,
1576 Digest,
1577 Digest,
1578 Sha256,
1579 OneCap,
1580 2,
1581 32,
1582 Sequential,
1583 >;
1584 type UnorderedFixedMmbP2Db = unordered::fixed::partitioned::Db<
1585 mmb::Family,
1586 Context,
1587 Digest,
1588 Digest,
1589 Sha256,
1590 OneCap,
1591 2,
1592 32,
1593 Sequential,
1594 >;
1595 type UnorderedVariableMmbP2Db = unordered::variable::partitioned::Db<
1596 mmb::Family,
1597 Context,
1598 Digest,
1599 Digest,
1600 Sha256,
1601 OneCap,
1602 2,
1603 32,
1604 Sequential,
1605 >;
1606
1607 #[test_traced]
1608 fn test_reconstruction_views_expose_all_chunk_states() {
1609 let executor = deterministic::Runner::default();
1610 executor.start(|context| async move {
1611 let db = UnorderedFixedMmbDb::init(
1612 context.child("db"),
1613 fixed_config::<OneCap>("reconstruction-views", &context),
1614 )
1615 .await
1616 .unwrap();
1617
1618 let mut batch = db.new_batch();
1621 for i in 0u64..512 {
1622 let key = Sha256::hash(&[&i.to_be_bytes()]);
1623 let value = Sha256::hash(&[&(i + 1_000).to_be_bytes()]);
1624 batch = batch.write(key, Some(value));
1625 }
1626 let batch = batch.merkleize(&db, None).await.unwrap();
1627 let (db, _) = db.apply_batch(batch).await.unwrap();
1628
1629 let end = db.bounds().end;
1632 let bitmap = db.bitmap();
1633 assert_eq!(bitmap.len(), *end);
1634 assert_eq!(bitmap.pruned_chunks(), 0);
1635
1636 let first = bitmap.get_chunk(0);
1637 let chunk_bits = first.len() as u64 * 8;
1638 let grafting_height = chunk_bits.trailing_zeros();
1639 let complete = bitmap.complete_chunks() as u64;
1640 let graftable =
1641 grafting::graftable_chunks::<mmb::Family>(*end, grafting_height).min(complete);
1642 assert_eq!(complete - graftable, 1, "expected one pending chunk");
1643 assert!(first.iter().any(|byte| *byte != 0));
1644 let pending = bitmap.get_chunk(graftable as usize);
1645 assert!(pending.iter().any(|byte| *byte != 0));
1646
1647 let (partial, partial_bits) = bitmap.last_chunk();
1648 assert!(partial_bits > 0 && partial_bits < chunk_bits);
1649 assert!(partial.iter().any(|byte| *byte != 0));
1650
1651 let witness = db.ops_root_witness().await.unwrap();
1653 assert_eq!(
1654 witness.pending_chunk_digest,
1655 Some(Sha256::hash(&[pending.as_slice()]))
1656 );
1657 assert_eq!(
1658 witness.partial_chunk,
1659 Some((partial_bits, Sha256::hash(&[partial.as_slice()])))
1660 );
1661
1662 let storage = db.grafted_storage();
1665 assert_eq!(
1666 storage.size(),
1667 <mmb::Family as merkle::Family>::location_to_position(end)
1668 );
1669
1670 let ops_pos = <mmb::Family as merkle::Graftable>::subtree_root_position(
1671 Location::new(0),
1672 grafting_height,
1673 );
1674 assert!(storage.get_node(ops_pos).await.unwrap().is_some());
1675
1676 let pinned_positions =
1679 <mmb::Family as merkle::Family>::nodes_to_pin(end).collect::<Vec<_>>();
1680 let raw_pinned = db.pinned_nodes_at(end).await.unwrap();
1681 assert_eq!(pinned_positions.len(), raw_pinned.len());
1682 let mut checked_grafted_node = false;
1683 for (pos, raw_digest) in pinned_positions.into_iter().zip(raw_pinned) {
1684 let digest = storage.get_node(pos).await.unwrap().unwrap();
1685 if pos == ops_pos {
1686 assert_ne!(digest, raw_digest);
1687 checked_grafted_node = true;
1688 }
1689 }
1690 assert!(checked_grafted_node);
1691
1692 drop(storage);
1693 db.destroy().await.unwrap();
1694 });
1695 }
1696
1697 #[test_traced]
1704 fn test_partitioned_ordered_short_key_exclusion_not_forgeable() {
1705 type ForgedExclusionDb = ordered::variable::partitioned::Db<
1706 mmr::Family,
1707 Context,
1708 Vec<u8>,
1709 Vec<u8>,
1710 Sha256,
1711 OneCap,
1712 2,
1713 32,
1714 Sequential,
1715 >;
1716
1717 let executor = deterministic::Runner::default();
1718 executor.start(|context| async move {
1719 let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE);
1720 let cfg = VariableConfig {
1721 merkle_config: MerkleConfig {
1722 journal_partition: "forged-exclusion-journal".to_string(),
1723 metadata_partition: "forged-exclusion-metadata".to_string(),
1724 items_per_blob: NZU64!(11),
1725 write_buffer: NZUsize!(1024),
1726 replay_buffer: NZUsize!(1024),
1727 strategy: Sequential,
1728 page_cache: page_cache.clone(),
1729 },
1730 journal_config: VConfig {
1731 partition: "forged-exclusion-log".to_string(),
1732 items_per_section: NZU64!(7),
1733 compression: None,
1734 codec_config: (((0..=8).into(), ()), ((0..=8).into(), ())),
1735 page_cache,
1736 write_buffer: NZUsize!(1024),
1737 replay_buffer: NZUsize!(1024),
1738 },
1739 grafted_metadata_partition: "forged-exclusion-grafted".to_string(),
1740 translator: OneCap,
1741 init_cache_size: Some(NZUsize!(1024)),
1742 init_buffer: NZUsize!(1 << 21),
1743 init_concurrency: NZUsize!(1),
1744 };
1745 let db = ForgedExclusionDb::init(context.child("db"), cfg)
1746 .await
1747 .unwrap();
1748
1749 let a = vec![0x00u8, 0x80];
1753 let b = vec![0x01u8];
1754 let c = vec![0x01u8, 0x00];
1755 let (va, vb, vc) = (vec![0x0au8], vec![0x0bu8], vec![0x0cu8]);
1756
1757 let merkleized = db
1759 .new_batch()
1760 .write(a.clone(), Some(va.clone()))
1761 .write(b.clone(), Some(vb.clone()))
1762 .merkleize(&db, None)
1763 .await
1764 .unwrap();
1765 let (db, _) = db.apply_batch(merkleized).await.unwrap();
1766 let merkleized = db
1767 .new_batch()
1768 .write(c.clone(), Some(vc.clone()))
1769 .merkleize(&db, None)
1770 .await
1771 .unwrap();
1772 let (db, _) = db.apply_batch(merkleized).await.unwrap();
1773 let root = db.root();
1774
1775 assert_eq!(db.get(&a).await.unwrap(), Some(va));
1777 assert_eq!(db.get(&b).await.unwrap(), Some(vb));
1778 assert_eq!(db.get(&c).await.unwrap(), Some(vc));
1779
1780 let (_, span_b) = db.get_span(&b).await.unwrap().unwrap();
1783 assert_eq!(
1784 span_b.next_key, c,
1785 "b.next_key must be c, not the wrapped-around a"
1786 );
1787
1788 assert!(matches!(
1790 db.exclusion_proof(&c).await,
1791 Err(Error::KeyExists)
1792 ));
1793
1794 let kvp = db.key_value_proof(b.clone()).await.unwrap();
1798 let forged = ordered::ExclusionProof::KeyValue(kvp.proof, span_b);
1799 assert!(!ForgedExclusionDb::verify_exclusion_proof(
1800 &c, &forged, &root
1801 ));
1802
1803 db.destroy().await.unwrap();
1804 });
1805 }
1806
1807 macro_rules! open_db_fn {
1809 ($db:ty, $cfg:ident) => {
1810 |ctx: Context, partition: String| async move {
1811 <$db>::init(ctx.child("storage"), $cfg::<OneCap>(&partition, &ctx))
1812 .await
1813 .unwrap()
1814 }
1815 };
1816 }
1817
1818 macro_rules! with_all_variants {
1820 ($cb:ident!($($args:tt)*)) => {
1821 $cb!($($args)*, of, OrderedFixedDb, fixed_config);
1822 $cb!($($args)*, ov, OrderedVariableDb, variable_config);
1823 $cb!($($args)*, uf, UnorderedFixedDb, fixed_config);
1824 $cb!($($args)*, uv, UnorderedVariableDb, variable_config);
1825 $cb!($($args)*, ofp1, OrderedFixedP1Db, fixed_config_partitioned);
1826 $cb!($($args)*, ovp1, OrderedVariableP1Db, variable_config_partitioned);
1827 $cb!($($args)*, ufp1, UnorderedFixedP1Db, fixed_config_partitioned);
1828 $cb!($($args)*, uvp1, UnorderedVariableP1Db, variable_config_partitioned);
1829 $cb!($($args)*, ofp2, OrderedFixedP2Db, fixed_config_partitioned);
1830 $cb!($($args)*, ovp2, OrderedVariableP2Db, variable_config_partitioned);
1831 $cb!($($args)*, ufp2, UnorderedFixedP2Db, fixed_config_partitioned);
1832 $cb!($($args)*, uvp2, UnorderedVariableP2Db, variable_config_partitioned);
1833 $cb!($($args)*, of_mmb, OrderedFixedMmbDb, fixed_config);
1834 $cb!($($args)*, ov_mmb, OrderedVariableMmbDb, variable_config);
1835 $cb!($($args)*, uf_mmb, UnorderedFixedMmbDb, fixed_config);
1836 $cb!($($args)*, uv_mmb, UnorderedVariableMmbDb, variable_config);
1837 $cb!($($args)*, ofp1_mmb, OrderedFixedMmbP1Db, fixed_config_partitioned);
1838 $cb!($($args)*, ovp1_mmb, OrderedVariableMmbP1Db, variable_config_partitioned);
1839 $cb!($($args)*, ufp1_mmb, UnorderedFixedMmbP1Db, fixed_config_partitioned);
1840 $cb!($($args)*, uvp1_mmb, UnorderedVariableMmbP1Db, variable_config_partitioned);
1841 $cb!($($args)*, ofp2_mmb, OrderedFixedMmbP2Db, fixed_config_partitioned);
1842 $cb!($($args)*, ovp2_mmb, OrderedVariableMmbP2Db, variable_config_partitioned);
1843 $cb!($($args)*, ufp2_mmb, UnorderedFixedMmbP2Db, fixed_config_partitioned);
1844 $cb!($($args)*, uvp2_mmb, UnorderedVariableMmbP2Db, variable_config_partitioned);
1845 };
1846 }
1847
1848 macro_rules! with_ordered_variants {
1850 ($cb:ident!($($args:tt)*)) => {
1851 $cb!($($args)*, of, OrderedFixedDb, fixed_config);
1852 $cb!($($args)*, ov, OrderedVariableDb, variable_config);
1853 $cb!($($args)*, ofp1, OrderedFixedP1Db, fixed_config_partitioned);
1854 $cb!($($args)*, ovp1, OrderedVariableP1Db, variable_config_partitioned);
1855 $cb!($($args)*, ofp2, OrderedFixedP2Db, fixed_config_partitioned);
1856 $cb!($($args)*, ovp2, OrderedVariableP2Db, variable_config_partitioned);
1857 $cb!($($args)*, of_mmb, OrderedFixedMmbDb, fixed_config);
1858 $cb!($($args)*, ov_mmb, OrderedVariableMmbDb, variable_config);
1859 $cb!($($args)*, ofp1_mmb, OrderedFixedMmbP1Db, fixed_config_partitioned);
1860 $cb!($($args)*, ovp1_mmb, OrderedVariableMmbP1Db, variable_config_partitioned);
1861 $cb!($($args)*, ofp2_mmb, OrderedFixedMmbP2Db, fixed_config_partitioned);
1862 $cb!($($args)*, ovp2_mmb, OrderedVariableMmbP2Db, variable_config_partitioned);
1863 };
1864 }
1865
1866 macro_rules! with_unordered_variants {
1868 ($cb:ident!($($args:tt)*)) => {
1869 $cb!($($args)*, uf, UnorderedFixedDb, fixed_config);
1870 $cb!($($args)*, uv, UnorderedVariableDb, variable_config);
1871 $cb!($($args)*, ufp1, UnorderedFixedP1Db, fixed_config_partitioned);
1872 $cb!($($args)*, uvp1, UnorderedVariableP1Db, variable_config_partitioned);
1873 $cb!($($args)*, ufp2, UnorderedFixedP2Db, fixed_config_partitioned);
1874 $cb!($($args)*, uvp2, UnorderedVariableP2Db, variable_config_partitioned);
1875 $cb!($($args)*, uf_mmb, UnorderedFixedMmbDb, fixed_config);
1876 $cb!($($args)*, uv_mmb, UnorderedVariableMmbDb, variable_config);
1877 $cb!($($args)*, ufp1_mmb, UnorderedFixedMmbP1Db, fixed_config_partitioned);
1878 $cb!($($args)*, uvp1_mmb, UnorderedVariableMmbP1Db, variable_config_partitioned);
1879 $cb!($($args)*, ufp2_mmb, UnorderedFixedMmbP2Db, fixed_config_partitioned);
1880 $cb!($($args)*, uvp2_mmb, UnorderedVariableMmbP2Db, variable_config_partitioned);
1881 };
1882 }
1883
1884 macro_rules! test_for_variant {
1889 ($f:ident, $traced:literal, $label:ident, $db:ty, $cfg:ident) => {
1890 paste::paste! {
1891 #[test_group("slow")]
1892 #[test_traced($traced)]
1893 fn [<$f _ $label>]() {
1894 let executor = deterministic::Runner::default();
1895 executor.start(|context| async move {
1896 Box::pin($f(context, open_db_fn!($db, $cfg))).await
1899 });
1900 }
1901 }
1902 };
1903 }
1904
1905 macro_rules! test_for_all_variants {
1907 ($f:ident, $traced:literal) => {
1908 with_all_variants!(test_for_variant!($f, $traced));
1909 };
1910 }
1911
1912 macro_rules! test_for_ordered_variants {
1914 ($f:ident, $traced:literal) => {
1915 with_ordered_variants!(test_for_variant!($f, $traced));
1916 };
1917 }
1918
1919 macro_rules! test_for_unordered_variants {
1921 ($f:ident, $traced:literal) => {
1922 with_unordered_variants!(test_for_variant!($f, $traced));
1923 };
1924 }
1925
1926 async fn test_ordered_build_big<M, C, F, Fut>(context: Context, open_db: F)
1928 where
1929 M: merkle::Graftable,
1930 C: DbAny<M>,
1931 C::Key: TestKey,
1932 <C as DbAny<M>>::Value: TestValue,
1933 F: FnMut(Context, String) -> Fut + Clone,
1934 Fut: Future<Output = C>,
1935 {
1936 test_current_db_build_big::<M, C, F, Fut>(context, open_db).await;
1937 }
1938
1939 async fn test_unordered_build_big<M, C, F, Fut>(context: Context, open_db: F)
1940 where
1941 M: merkle::Graftable,
1942 C: DbAny<M>,
1943 C::Key: TestKey,
1944 <C as DbAny<M>>::Value: TestValue,
1945 F: FnMut(Context, String) -> Fut + Clone,
1946 Fut: Future<Output = C>,
1947 {
1948 test_current_db_build_big::<M, C, F, Fut>(context, open_db).await;
1949 }
1950
1951 test_for_all_variants!(test_build_random_close_reopen, "WARN");
1952 test_for_all_variants!(test_simulate_write_failures, "WARN");
1953 test_for_all_variants!(test_different_pruning_delays_same_root, "WARN");
1954 test_for_all_variants!(test_sync_persists_bitmap_pruning_boundary, "WARN");
1955 test_for_all_variants!(test_commit_after_sync_recovery, "WARN");
1956 test_for_all_variants!(test_stale_batch_side_effect_free, "WARN");
1957
1958 test_for_ordered_variants!(test_ordered_build_big, "WARN");
1959 test_for_ordered_variants!(test_ordered_build_small_close_reopen, "DEBUG");
1960
1961 test_for_unordered_variants!(test_unordered_build_big, "WARN");
1962 test_for_unordered_variants!(test_unordered_build_small_close_reopen, "DEBUG");
1963
1964 fn key(i: u64) -> Digest {
1971 Sha256::hash(&[&i.to_be_bytes()])
1972 }
1973
1974 fn val(i: u64) -> Digest {
1975 Sha256::hash(&[&(i + 10000).to_be_bytes()])
1976 }
1977
1978 #[boxed]
1979 async fn mmb_commit(
1980 db: UnorderedVariableMmbDb,
1981 writes: impl IntoIterator<Item = (Digest, Option<Digest>)>,
1982 ) -> UnorderedVariableMmbDb {
1983 let mut batch = db.new_batch();
1984 for (k, v) in writes {
1985 batch = batch.write(k, v);
1986 }
1987 let merkleized = batch.merkleize(&db, None).await.unwrap();
1988 let (db, _) = db.apply_batch(merkleized).await.unwrap();
1989 db.commit().await.unwrap()
1990 }
1991
1992 #[boxed]
1993 async fn commit_writes_with_metadata(
1994 db: UnorderedVariableDb,
1995 writes: impl IntoIterator<Item = (Digest, Option<Digest>)>,
1996 metadata: Option<Digest>,
1997 ) -> (UnorderedVariableDb, std::ops::Range<Location<mmr::Family>>) {
1998 let mut batch = db.new_batch();
1999 for (k, v) in writes {
2000 batch = batch.write(k, v);
2001 }
2002 let merkleized = batch.merkleize(&db, metadata).await.unwrap();
2003 let (db, range) = db.apply_batch(merkleized).await.unwrap();
2004 let db = db.commit().await.unwrap();
2005 (db, range)
2006 }
2007
2008 #[test_traced("INFO")]
2009 fn test_current_rewind_recovery() {
2010 let executor = deterministic::Runner::default();
2011 executor.start(|context| async move {
2012 let partition = "current-rewind-recovery";
2013 let ctx = context.child("db");
2014 let db: UnorderedVariableDb = UnorderedVariableDb::init(
2015 ctx.child("storage"),
2016 variable_config::<OneCap>(partition, &ctx),
2017 )
2018 .await
2019 .unwrap();
2020 let initial_size = db.bounds().end;
2021 let initial_root = db.root();
2022 let initial_ops_root = db.ops_root();
2023 let initial_floor = db.inactivity_floor_loc();
2024
2025 let metadata_a = val(900);
2026 let (db, first_range) = commit_writes_with_metadata(
2027 db,
2028 [(key(0), Some(val(0))), (key(1), Some(val(1)))],
2029 Some(metadata_a),
2030 )
2031 .await;
2032 assert_eq!(first_range.start, initial_size);
2033 let size_before = db.bounds().end;
2034 let root_before = db.root();
2035 let ops_root_before = db.ops_root();
2036 let floor_before = db.inactivity_floor_loc();
2037 assert_eq!(size_before, first_range.end);
2038
2039 let metadata_b = val(901);
2040 let (db, second_range) = commit_writes_with_metadata(
2041 db,
2042 [
2043 (key(0), Some(val(100))),
2044 (key(1), None),
2045 (key(2), Some(val(2))),
2046 ],
2047 Some(metadata_b),
2048 )
2049 .await;
2050 assert_eq!(second_range.start, size_before);
2051 assert_ne!(db.root(), root_before);
2052 assert_eq!(db.get_metadata().await.unwrap(), Some(val(901)));
2053 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(100)));
2054 assert_eq!(db.get(&key(1)).await.unwrap(), None);
2055 assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2)));
2056
2057 let db = db.rewind(size_before).await.unwrap();
2058 assert_eq!(db.bounds().end, size_before);
2059 assert_eq!(db.root(), root_before);
2060 assert_eq!(db.ops_root(), ops_root_before);
2061 assert_eq!(db.inactivity_floor_loc(), floor_before);
2062 assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
2063 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
2064 assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
2065 assert_eq!(db.get(&key(2)).await.unwrap(), None);
2066
2067 let db = db.commit().await.unwrap();
2068 drop(db);
2069
2070 let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
2071 context.child("reopen"),
2072 variable_config::<OneCap>(partition, &context),
2073 )
2074 .await
2075 .unwrap();
2076 assert_eq!(reopened.bounds().end, size_before);
2077 assert_eq!(reopened.root(), root_before);
2078 assert_eq!(reopened.ops_root(), ops_root_before);
2079 assert_eq!(reopened.inactivity_floor_loc(), floor_before);
2080 assert_eq!(reopened.get_metadata().await.unwrap(), Some(val(900)));
2081 assert_eq!(reopened.get(&key(0)).await.unwrap(), Some(val(0)));
2082 assert_eq!(reopened.get(&key(1)).await.unwrap(), Some(val(1)));
2083 assert_eq!(reopened.get(&key(2)).await.unwrap(), None);
2084
2085 let reopened = reopened.rewind(initial_size).await.unwrap();
2086 assert_eq!(reopened.bounds().end, initial_size);
2087 assert_eq!(reopened.root(), initial_root);
2088 assert_eq!(reopened.ops_root(), initial_ops_root);
2089 assert_eq!(reopened.inactivity_floor_loc(), initial_floor);
2090 assert_eq!(reopened.get_metadata().await.unwrap(), None);
2091 assert_eq!(reopened.get(&key(0)).await.unwrap(), None);
2092 assert_eq!(reopened.get(&key(1)).await.unwrap(), None);
2093 assert_eq!(reopened.get(&key(2)).await.unwrap(), None);
2094
2095 let reopened = reopened.commit().await.unwrap();
2096 drop(reopened);
2097
2098 let reopened_initial: UnorderedVariableDb = UnorderedVariableDb::init(
2099 context.child("reopen_initial"),
2100 variable_config::<OneCap>(partition, &context),
2101 )
2102 .await
2103 .unwrap();
2104 assert_eq!(reopened_initial.bounds().end, initial_size);
2105 assert_eq!(reopened_initial.root(), initial_root);
2106 assert_eq!(reopened_initial.ops_root(), initial_ops_root);
2107 assert_eq!(reopened_initial.inactivity_floor_loc(), initial_floor);
2108 assert_eq!(reopened_initial.get_metadata().await.unwrap(), None);
2109 assert_eq!(reopened_initial.get(&key(0)).await.unwrap(), None);
2110 assert_eq!(reopened_initial.get(&key(1)).await.unwrap(), None);
2111 assert_eq!(reopened_initial.get(&key(2)).await.unwrap(), None);
2112
2113 reopened_initial.destroy().await.unwrap();
2114 });
2115 }
2116
2117 #[test_traced("INFO")]
2118 fn test_current_rewind_recovery_pruned_repeated_updates() {
2119 let executor = deterministic::Runner::default();
2120 executor.start(|context| async move {
2121 const COMMITS: u64 = 96;
2122
2123 let partition = "current-rewind-pruned-recovery";
2124 let ctx = context.child("db");
2125 let mut db: UnorderedVariableDb =
2126 UnorderedVariableDb::init(ctx.child("storage"), variable_config::<OneCap>(partition, &ctx))
2127 .await
2128 .unwrap();
2129
2130 let key0 = key(0);
2131 let mut history = Vec::new();
2132 for round in 0..COMMITS {
2133 (db, _) = commit_writes_with_metadata(
2134 db,
2135 [(key0, Some(val(20_000 + round)))],
2136 None,
2137 )
2138 .await;
2139 history.push((
2140 db.bounds().end,
2141 db.inactivity_floor_loc(),
2142 db.root(),
2143 db.ops_root(),
2144 val(20_000 + round),
2145 ));
2146 }
2147
2148 let db = db.prune(Location::new(1)).await.unwrap();
2151 let pruned_bits = db.pruned_bits();
2152 assert!(pruned_bits > 0, "expected bitmap pruning for rewind test");
2153 let bounds = db.bounds();
2154
2155 let (target_size, target_root, target_ops_root, target_value) = history
2156 .iter()
2157 .enumerate()
2158 .find_map(|(idx, (size, floor, root, ops_root, value))| {
2159 let removed_commits = history.len() - idx - 1;
2160 if removed_commits >= 3 && *size > bounds.start && *floor >= pruned_bits {
2161 Some((*size, *root, *ops_root, *value))
2162 } else {
2163 None
2164 }
2165 })
2166 .unwrap_or_else(|| {
2167 panic!(
2168 "expected legal pruned rewind target with repeated updates; bounds={bounds:?}, pruned_bits={pruned_bits}, latest_floor={:?}, history={history:?}",
2169 db.inactivity_floor_loc()
2170 )
2171 });
2172
2173 let db = db.rewind(target_size).await.unwrap();
2174 assert_eq!(db.root(), target_root);
2175 assert_eq!(db.ops_root(), target_ops_root);
2176 assert_eq!(db.bounds().end, target_size);
2177 assert_eq!(db.get(&key0).await.unwrap(), Some(target_value));
2178
2179 let db = db.commit().await.unwrap();
2180 drop(db);
2181
2182 let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
2183 context.child("reopen_pruned_recovery"),
2184 variable_config::<OneCap>(partition, &context),
2185 )
2186 .await
2187 .unwrap();
2188 assert_eq!(reopened.root(), target_root);
2189 assert_eq!(reopened.ops_root(), target_ops_root);
2190 assert_eq!(reopened.bounds().end, target_size);
2191 assert_eq!(reopened.get(&key0).await.unwrap(), Some(target_value));
2192
2193 let metadata_after_rewind = val(30_000);
2194 let new_key = key(1);
2195 let new_value = val(30_001);
2196 let (reopened, new_write_range) = commit_writes_with_metadata(
2197 reopened,
2198 [(new_key, Some(new_value))],
2199 Some(metadata_after_rewind),
2200 )
2201 .await;
2202 let expected_end = new_write_range.end;
2203 let root_after_new_write = reopened.root();
2204 let ops_root_after_new_write = reopened.ops_root();
2205 assert_eq!(reopened.bounds().end, expected_end);
2206 assert_eq!(reopened.get_metadata().await.unwrap(), Some(metadata_after_rewind));
2207 assert_eq!(reopened.get(&key0).await.unwrap(), Some(target_value));
2208 assert_eq!(reopened.get(&new_key).await.unwrap(), Some(new_value));
2209
2210 drop(reopened);
2211 let reopened_after_new_write: UnorderedVariableDb = UnorderedVariableDb::init(
2212 context.child("reopen_pruned_after_new_write"),
2213 variable_config::<OneCap>(partition, &context),
2214 )
2215 .await
2216 .unwrap();
2217 assert_eq!(reopened_after_new_write.root(), root_after_new_write);
2218 assert_eq!(reopened_after_new_write.ops_root(), ops_root_after_new_write);
2219 assert_eq!(reopened_after_new_write.bounds().end, expected_end);
2220 assert_eq!(
2221 reopened_after_new_write.get_metadata().await.unwrap(),
2222 Some(metadata_after_rewind)
2223 );
2224 assert_eq!(reopened_after_new_write.get(&key0).await.unwrap(), Some(target_value));
2225 assert_eq!(
2226 reopened_after_new_write.get(&new_key).await.unwrap(),
2227 Some(new_value)
2228 );
2229
2230 reopened_after_new_write.destroy().await.unwrap();
2231 });
2232 }
2233
2234 #[test_traced("INFO")]
2237 fn test_current_mmb_settlement_guard_defers_pruning() {
2238 let executor = deterministic::Runner::default();
2239 executor.start(|context| async move {
2240 const COMMITS: u64 = 100;
2241
2242 let partition = "current-mmb-reopen-prove-after-prune";
2243 let ctx = context.child("db");
2244 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2245 ctx.child("storage"),
2246 variable_config::<OneCap>(partition, &ctx),
2247 )
2248 .await
2249 .unwrap();
2250
2251 let k = key(0);
2252 let mut expected = None;
2253 for round in 0..COMMITS {
2254 expected = Some(val(50_000 + round));
2255 let mut batch = db.new_batch();
2256 batch = batch.write(k, expected);
2257 let merkleized = batch.merkleize(&db, None).await.unwrap();
2258 (db, _) = db.apply_batch(merkleized).await.unwrap();
2259 db = db.commit().await.unwrap();
2260 }
2261
2262 let root_before = db.root();
2263 assert!(
2264 *db.inactivity_floor_loc() >= 256,
2265 "expected inactivity floor past chunk 0"
2266 );
2267 assert_eq!(
2268 *db.sync_boundary(),
2269 0,
2270 "settlement guard should hold boundary at 0 during unsettled window"
2271 );
2272
2273 let Err(err) = db.prune(Location::<mmb::Family>::new(1)).await else {
2275 panic!("expected PruneBeyondMinRequired");
2276 };
2277 assert!(
2278 matches!(err, Error::PruneBeyondMinRequired(_, _)),
2279 "expected PruneBeyondMinRequired, got {err:?}"
2280 );
2281
2282 let reopened: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2284 context.child("reopen"),
2285 variable_config::<OneCap>(partition, &context),
2286 )
2287 .await
2288 .unwrap();
2289
2290 assert_eq!(reopened.pruned_bits(), 0);
2291 assert_eq!(reopened.root(), root_before);
2292 assert_eq!(reopened.get(&k).await.unwrap(), expected);
2293
2294 let _proof = reopened.key_value_proof(k).await.unwrap();
2296
2297 reopened.destroy().await.unwrap();
2298 });
2299 }
2300
2301 #[test_traced("INFO")]
2302 fn test_current_mmb_rewind_rejects_unsettled_pruned_window() {
2303 let executor = deterministic::Runner::default();
2304 executor.start(|context| async move {
2305 const COMMITS: u64 = 320;
2306 const N: usize = 32;
2307
2308 let partition = "current-mmb-rewind-unsettled-window";
2309 let ctx = context.child("db");
2310 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2311 ctx.child("storage"),
2312 variable_config::<OneCap>(partition, &ctx),
2313 )
2314 .await
2315 .unwrap();
2316
2317 let key0 = key(0);
2318 let mut history = Vec::new();
2319 for round in 0..COMMITS {
2320 let mut batch = db.new_batch();
2321 batch = batch.write(key0, Some(val(60_000 + round)));
2322 let merkleized = batch.merkleize(&db, None).await.unwrap();
2323 (db, _) = db.apply_batch(merkleized).await.unwrap();
2324 db = db.commit().await.unwrap();
2325 history.push((db.bounds().end, db.inactivity_floor_loc()));
2326 }
2327
2328 let boundary = db.sync_boundary();
2329 let db = db.prune(boundary).await.unwrap();
2330 let pruned_bits = db.pruned_bits();
2331 assert!(pruned_bits > 0, "expected MMB bitmap pruning to be active");
2332 let db = db.sync().await.unwrap();
2333
2334 let chunk_bits = commonware_utils::bitmap::BitMap::<N>::CHUNK_SIZE_BITS;
2335 let pruned_chunks = (pruned_bits / chunk_bits) as u64;
2336 let gh = super::grafting::height::<N>();
2337 let youngest = pruned_chunks - 1;
2338 let pair_chunk = youngest & !1;
2339 let pair_start = pair_chunk << gh;
2340 let pair_pos = <mmb::Family as merkle::Graftable>::subtree_root_position(
2341 merkle::Location::<mmb::Family>::new(pair_start),
2342 gh + 1,
2343 );
2344 let absorbed_after =
2345 <mmb::Family as merkle::Graftable>::peak_birth_size(pair_pos, gh + 1);
2346
2347 let unsafe_target = history
2348 .iter()
2349 .filter_map(|(size, floor)| {
2350 let s = **size;
2351 if s >= pruned_bits && s < absorbed_after && **floor >= pruned_bits {
2352 Some(s)
2353 } else {
2354 None
2355 }
2356 })
2357 .max()
2358 .unwrap_or_else(|| {
2359 panic!(
2360 "expected rewind target in unsettled window: pruned_bits={pruned_bits}, absorbed_after={absorbed_after}, history={history:?}"
2361 )
2362 });
2363
2364 let Err(err) = db
2365 .rewind(merkle::Location::<mmb::Family>::new(unsafe_target))
2366 .await
2367 else {
2368 panic!("expected rewind rejection in unsettled delayed-merge window");
2369 };
2370 assert!(
2371 matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
2372 "unexpected rewind error for unsettled delayed-merge window: {err:?}"
2373 );
2374 });
2375 }
2376
2377 #[test_traced]
2382 fn test_current_mmb_prune_respects_sync_boundary() {
2383 let executor = deterministic::Runner::default();
2384 executor.start(|context| async move {
2385 const COMMITS: u64 = 320;
2386
2387 let ctx = context.child("db");
2388 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2389 ctx.child("storage"),
2390 variable_config::<OneCap>("prune-clip-mmb", &ctx),
2391 )
2392 .await
2393 .unwrap();
2394
2395 let k = key(0);
2396 for round in 0..COMMITS {
2397 db = mmb_commit(db, [(k, Some(val(70_000 + round)))]).await;
2398 }
2399
2400 let prune_boundary = db.sync_boundary();
2401 let db = db.prune(prune_boundary).await.unwrap();
2402
2403 let boundary = db.sync_boundary();
2404 let floor = db.inactivity_floor_loc();
2405 assert!(
2406 boundary < floor,
2407 "delayed-merge lag must be strictly active: boundary={boundary}, floor={floor}"
2408 );
2409 assert!(
2410 db.bounds().start <= boundary,
2411 "ops journal was pruned past the settled bitmap boundary: \
2412 bounds.start={}, boundary={boundary}",
2413 db.bounds().start
2414 );
2415
2416 db.destroy().await.unwrap();
2417 });
2418 }
2419
2420 #[test_traced]
2425 fn test_current_mmr_prune_boundary_lag_is_only_chunk_alignment() {
2426 let executor = deterministic::Runner::default();
2427 executor.start(|context| async move {
2428 const COMMITS: u64 = 320;
2429 const N: usize = 32;
2430
2431 let ctx = context.child("db");
2432 let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
2433 ctx.child("storage"),
2434 variable_config::<OneCap>("prune-clip-mmr", &ctx),
2435 )
2436 .await
2437 .unwrap();
2438
2439 for round in 0..COMMITS {
2440 (db, _) = commit_writes_with_metadata(
2441 db,
2442 [(key(0), Some(val(80_000 + round)))],
2443 None,
2444 )
2445 .await;
2446 }
2447
2448 let prune_boundary = db.sync_boundary();
2449 let db = db.prune(prune_boundary).await.unwrap();
2450
2451 let boundary = db.sync_boundary();
2452 let floor = db.inactivity_floor_loc();
2453 let chunk_bits = commonware_utils::bitmap::BitMap::<N>::CHUNK_SIZE_BITS;
2454 assert!(
2455 boundary <= floor && *floor - *boundary < chunk_bits,
2456 "MMR lag should be only chunk alignment: boundary={boundary}, floor={floor}, chunk_bits={chunk_bits}"
2457 );
2458 assert!(
2459 db.bounds().start <= boundary,
2460 "ops journal bounds must be <= sync_boundary: bounds.start={}, boundary={boundary}",
2461 db.bounds().start
2462 );
2463
2464 db.destroy().await.unwrap();
2465 });
2466 }
2467
2468 #[test_traced]
2471 fn test_current_prune_below_settled_boundary_is_honored() {
2472 let executor = deterministic::Runner::default();
2473 executor.start(|context| async move {
2474 const COMMITS: u64 = 100;
2475
2476 let ctx = context.child("db");
2477 let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
2478 ctx.child("storage"),
2479 variable_config::<OneCap>("prune-below-boundary", &ctx),
2480 )
2481 .await
2482 .unwrap();
2483
2484 for round in 0..COMMITS {
2485 (db, _) = commit_writes_with_metadata(db, [(key(0), Some(val(90_000 + round)))], None)
2486 .await;
2487 }
2488
2489 assert!(*db.inactivity_floor_loc() > 1);
2490 let small = Location::new(1);
2491 let db = db.prune(small).await.unwrap();
2492
2493 assert!(
2494 db.bounds().start <= small,
2495 "journal pruning exceeded the caller-supplied target: bounds.start={}, requested={small}",
2496 db.bounds().start
2497 );
2498
2499 db.destroy().await.unwrap();
2500 });
2501 }
2502
2503 #[test_traced]
2506 fn test_current_mmb_reopen_and_prove_after_prune_delayed_merge() {
2507 let executor = deterministic::Runner::default();
2508 executor.start(|context| async move {
2509 let db_ctx = context.child("db_init");
2510 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2511 db_ctx.child("db"),
2512 variable_config::<OneCap>("test_prune_delayed_merge", &db_ctx),
2513 )
2514 .await
2515 .unwrap();
2516
2517 let k = key(0);
2518
2519 for round in 0..200u64 {
2520 db = mmb_commit(db, [(k, Some(val(60_000 + round)))]).await;
2521 }
2522
2523 let boundary = db.sync_boundary();
2524 db = db.prune(boundary).await.unwrap();
2525 db = db.sync().await.unwrap();
2526
2527 for round in 200..300u64 {
2529 db = mmb_commit(db, [(key(1), Some(val(round)))]).await;
2530 }
2531
2532 let proof = db.key_value_proof(k).await.unwrap();
2533 assert!(UnorderedVariableMmbDb::verify_key_value_proof(
2534 k,
2535 val(60_000 + 199),
2536 &proof,
2537 &db.root()
2538 ));
2539
2540 let target_root = db.root();
2541 drop(db);
2542
2543 let reopen_ctx = context.child("db_reopen");
2544 let reopened: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2545 reopen_ctx.child("db"),
2546 variable_config::<OneCap>("test_prune_delayed_merge", &reopen_ctx),
2547 )
2548 .await
2549 .unwrap();
2550
2551 assert_eq!(reopened.root(), target_root);
2552
2553 let proof = reopened.key_value_proof(k).await.unwrap();
2554 assert!(UnorderedVariableMmbDb::verify_key_value_proof(
2555 k,
2556 val(60_000 + 199),
2557 &proof,
2558 &reopened.root()
2559 ));
2560
2561 reopened.destroy().await.unwrap();
2562 });
2563 }
2564
2565 #[test_traced]
2567 fn test_current_mmb_reopen_after_prune_two_chunks() {
2568 let executor = deterministic::Runner::default();
2569 executor.start(|context| async move {
2570 let db_ctx = context.child("db");
2571 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2572 db_ctx.child("db"),
2573 variable_config::<OneCap>("test_prune_two", &db_ctx),
2574 )
2575 .await
2576 .unwrap();
2577
2578 let k = key(0);
2579 let mut expected;
2581
2582 let mut round = 0u64;
2585 loop {
2586 expected = Some(val(60_000 + round));
2587 db = mmb_commit(db, [(k, expected)]).await;
2588 round += 1;
2589 let boundary = db.sync_boundary();
2590 db = db.prune(boundary).await.unwrap();
2591 if db.pruned_bits() >= 512 {
2592 break;
2593 }
2594 assert!(
2595 round < 500,
2596 "failed to reach 2 pruned chunks after {round} commits"
2597 );
2598 }
2599 let db = db.sync().await.unwrap();
2600
2601 let target_root = db.root();
2602 drop(db);
2603
2604 let reopen_ctx = context.child("db_reopen");
2605 let reopened: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2606 reopen_ctx.child("db"),
2607 variable_config::<OneCap>("test_prune_two", &reopen_ctx),
2608 )
2609 .await
2610 .unwrap();
2611
2612 assert_eq!(reopened.root(), target_root);
2613 assert_eq!(reopened.get(&k).await.unwrap(), expected);
2614 reopened.destroy().await.unwrap();
2615 });
2616 }
2617
2618 #[test_traced]
2620 fn test_current_mmb_repeated_prune() {
2621 let executor = deterministic::Runner::default();
2622 executor.start(|context| async move {
2623 let mut db_ctx = context.child("db_init");
2624 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2625 db_ctx.child("db"),
2626 variable_config::<OneCap>("test_repeated_prune", &db_ctx),
2627 )
2628 .await
2629 .unwrap();
2630
2631 for round in 0..3u64 {
2632 let k = key(round * 1000);
2633 let mut expected = None;
2634 for i in 0..90 {
2635 expected = Some(val(round * 1000 + i));
2636 db = mmb_commit(db, [(k, expected)]).await;
2637 }
2638
2639 let boundary = db.sync_boundary();
2640 db = db.prune(boundary).await.unwrap();
2641 db = db.sync().await.unwrap();
2642
2643 let root_before = db.root();
2644 db_ctx = context.child("db").with_attribute("round", round);
2645
2646 let prev_db = db;
2647 db = UnorderedVariableMmbDb::init(
2648 db_ctx.child("db"),
2649 variable_config::<OneCap>("test_repeated_prune", &db_ctx),
2650 )
2651 .await
2652 .unwrap();
2653
2654 assert_eq!(db.root(), root_before);
2655 assert_eq!(db.get(&k).await.unwrap(), expected);
2656 drop(prev_db);
2657 }
2658
2659 db.destroy().await.unwrap();
2660 });
2661 }
2662
2663 #[test_traced]
2665 fn test_current_mmb_stepwise_growth_matches_unpruned_reference() {
2666 let executor = deterministic::Runner::default();
2667 executor.start(|context| async move {
2668 let db_ctx = context.child("db_stepwise");
2669 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2670 db_ctx.child("db"),
2671 variable_config::<OneCap>("test_stepwise", &db_ctx),
2672 )
2673 .await
2674 .unwrap();
2675
2676 let ref_ctx = context.child("ref_stepwise");
2677 let mut ref_db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2678 ref_ctx.child("db"),
2679 variable_config::<OneCap>("test_stepwise_ref", &ref_ctx),
2680 )
2681 .await
2682 .unwrap();
2683
2684 let k = key(0);
2685 let mut commit_idx = 0u64;
2686
2687 while *db.inactivity_floor_loc() < 1024 {
2689 let value = Some(val(80_000 + commit_idx));
2690 db = mmb_commit(db, [(k, value)]).await;
2691 ref_db = mmb_commit(ref_db, [(k, value)]).await;
2692 commit_idx += 1;
2693 }
2694
2695 let boundary = db.sync_boundary();
2696 db = db.prune(boundary).await.unwrap();
2697 db = db.sync().await.unwrap();
2698 assert_eq!(
2699 db.root(),
2700 ref_db.root(),
2701 "root mismatch immediately after prune"
2702 );
2703
2704 loop {
2706 let db_leaves =
2707 *Location::<mmb::Family>::try_from(db.any.log.merkle.size()).unwrap();
2708 if db_leaves >= 1560 {
2709 break;
2710 }
2711
2712 let value = Some(val(80_000 + commit_idx));
2713 db = mmb_commit(db, [(k, value)]).await;
2714 ref_db = mmb_commit(ref_db, [(k, value)]).await;
2715 commit_idx += 1;
2716
2717 let db_leaves =
2718 *Location::<mmb::Family>::try_from(db.any.log.merkle.size()).unwrap();
2719 assert_eq!(
2720 db.root(),
2721 ref_db.root(),
2722 "stepwise root mismatch: leaves={db_leaves}, commit_idx={commit_idx}"
2723 );
2724 }
2725
2726 db.destroy().await.unwrap();
2727 ref_db.destroy().await.unwrap();
2728 });
2729 }
2730
2731 #[test_traced]
2733 fn test_current_mmb_large_repeated_prune_matches_unpruned_reference() {
2734 let executor = deterministic::Runner::default();
2735 executor.start(|context| async move {
2736 const ROUNDS: u64 = 8;
2737 const COMMITS_PER_ROUND: u64 = 120;
2738
2739 let mut db_ctx = context.child("db_init");
2740 let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2741 db_ctx.child("db"),
2742 variable_config::<OneCap>("test_large_prune", &db_ctx),
2743 )
2744 .await
2745 .unwrap();
2746
2747 let ref_ctx = context.child("ref");
2748 let mut ref_db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
2749 ref_ctx.child("db"),
2750 variable_config::<OneCap>("test_large_prune_ref", &ref_ctx),
2751 )
2752 .await
2753 .unwrap();
2754
2755 let k = key(0);
2756 let mut expected = None;
2757
2758 for round in 0..ROUNDS {
2759 for i in 0..COMMITS_PER_ROUND {
2760 let value = Some(val(round * 10_000 + i));
2761 expected = value;
2762 db = mmb_commit(db, [(k, value)]).await;
2763 ref_db = mmb_commit(ref_db, [(k, value)]).await;
2764 }
2765
2766 assert_eq!(
2767 db.root(),
2768 ref_db.root(),
2769 "root mismatch before prune at round {round}"
2770 );
2771
2772 let boundary = db.sync_boundary();
2773 db = db.prune(boundary).await.unwrap();
2774 db = db.sync().await.unwrap();
2775
2776 assert_eq!(
2777 db.root(),
2778 ref_db.root(),
2779 "root mismatch after prune at round {round}"
2780 );
2781
2782 let proof = db.key_value_proof(k).await.unwrap();
2783 assert!(
2784 UnorderedVariableMmbDb::verify_key_value_proof(
2785 k,
2786 expected.expect("value should exist"),
2787 &proof,
2788 &db.root()
2789 ),
2790 "proof verification failed at round {round}"
2791 );
2792
2793 db_ctx = context.child("db_reopen").with_attribute("round", round);
2794 let prev_db = db;
2795 db = UnorderedVariableMmbDb::init(
2796 db_ctx.child("db"),
2797 variable_config::<OneCap>("test_large_prune", &db_ctx),
2798 )
2799 .await
2800 .unwrap();
2801
2802 assert_eq!(
2803 db.root(),
2804 ref_db.root(),
2805 "root mismatch after reopen at round {round}"
2806 );
2807 assert_eq!(
2808 db.get(&k).await.unwrap(),
2809 expected,
2810 "value mismatch after reopen at round {round}"
2811 );
2812
2813 let proof = db.key_value_proof(k).await.unwrap();
2814 assert!(
2815 UnorderedVariableMmbDb::verify_key_value_proof(
2816 k,
2817 expected.expect("value should exist"),
2818 &proof,
2819 &db.root()
2820 ),
2821 "proof verification failed after reopen at round {round}"
2822 );
2823
2824 drop(prev_db);
2825 }
2826
2827 db.destroy().await.unwrap();
2828 ref_db.destroy().await.unwrap();
2829 });
2830 }
2831
2832 #[test_traced]
2834 fn test_current_prune_rejects_beyond_sync_boundary_without_mutation() {
2835 let executor = deterministic::Runner::default();
2836 executor.start(|context| async move {
2837 const COMMITS: u64 = 160;
2838
2839 let partition = "current-prune-beyond-boundary";
2840 let ctx = context.child("db");
2841 let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
2842 ctx.child("storage"),
2843 variable_config::<OneCap>(partition, &ctx),
2844 )
2845 .await
2846 .unwrap();
2847
2848 let key0 = key(0);
2849 for round in 0..COMMITS {
2850 (db, _) =
2851 commit_writes_with_metadata(db, [(key0, Some(val(40_000 + round)))], None)
2852 .await;
2853 }
2854
2855 let expected_root = db.root();
2856 let expected_ops_root = db.ops_root();
2857 let expected_boundary = db.sync_boundary();
2858 let expected_pruned_bits = db.pruned_bits();
2859 let expected_value = db.get(&key0).await.unwrap();
2860
2861 let invalid_prune_loc = expected_boundary + 256;
2863 let Err(err) = db.prune(invalid_prune_loc).await else {
2864 panic!("expected prune rejection above sync boundary");
2865 };
2866 assert!(
2867 matches!(err, Error::PruneBeyondMinRequired(loc, boundary)
2868 if loc == invalid_prune_loc && boundary == expected_boundary),
2869 "expected prune rejection above sync boundary, got {err:?}"
2870 );
2871
2872 let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
2873 context.child("reopen"),
2874 variable_config::<OneCap>(partition, &context),
2875 )
2876 .await
2877 .unwrap();
2878 assert_eq!(reopened.root(), expected_root);
2879 assert_eq!(reopened.ops_root(), expected_ops_root);
2880 assert_eq!(reopened.pruned_bits(), expected_pruned_bits);
2881 assert_eq!(reopened.get(&key0).await.unwrap(), expected_value);
2882
2883 reopened.destroy().await.unwrap();
2884 });
2885 }
2886
2887 #[test_traced("INFO")]
2888 fn test_current_rewind_small_delta_large_history() {
2889 let executor = deterministic::Runner::default();
2890 executor.start(|context| async move {
2891 const COMMITS: u64 = 200;
2892
2893 let partition = "current-rewind-small-delta";
2894 let ctx = context.child("db");
2895 let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
2896 ctx.child("storage"),
2897 variable_config::<OneCap>(partition, &ctx),
2898 )
2899 .await
2900 .unwrap();
2901
2902 let key0 = key(0);
2903 let key1 = key(1);
2904 let mut history = Vec::new();
2905
2906 for round in 0..COMMITS {
2907 let key0_value = val(40_000 + round);
2908 let key1_value = if round % 3 == 1 {
2909 None
2910 } else {
2911 Some(val(50_000 + round))
2912 };
2913
2914 (db, _) = commit_writes_with_metadata(
2915 db,
2916 [(key0, Some(key0_value)), (key1, key1_value)],
2917 None,
2918 )
2919 .await;
2920
2921 history.push((
2922 db.bounds().end,
2923 db.root(),
2924 db.ops_root(),
2925 key0_value,
2926 key1_value,
2927 ));
2928 }
2929
2930 let target = *history
2931 .get(history.len() - 3)
2932 .expect("history should contain at least three commits");
2933 let (target_size, target_root, target_ops_root, target_key0, target_key1) = target;
2934
2935 let db = db.rewind(target_size).await.unwrap();
2936 assert_eq!(db.bounds().end, target_size);
2937 assert_eq!(db.root(), target_root);
2938 assert_eq!(db.ops_root(), target_ops_root);
2939 assert_eq!(db.get(&key0).await.unwrap(), Some(target_key0));
2940 assert_eq!(db.get(&key1).await.unwrap(), target_key1);
2941
2942 let db = db.commit().await.unwrap();
2943 drop(db);
2944
2945 let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
2946 context.child("reopen_small_delta"),
2947 variable_config::<OneCap>(partition, &context),
2948 )
2949 .await
2950 .unwrap();
2951 assert_eq!(reopened.bounds().end, target_size);
2952 assert_eq!(reopened.root(), target_root);
2953 assert_eq!(reopened.ops_root(), target_ops_root);
2954 assert_eq!(reopened.get(&key0).await.unwrap(), Some(target_key0));
2955 assert_eq!(reopened.get(&key1).await.unwrap(), target_key1);
2956
2957 reopened.destroy().await.unwrap();
2958 });
2959 }
2960
2961 #[test_traced("INFO")]
2962 fn test_current_rewind_pruned_target_errors() {
2963 let executor = deterministic::Runner::default();
2964 executor.start(|context| async move {
2965 const KEYS: u64 = 384;
2966
2967 let partition = "current-rewind-pruned";
2968 let ctx = context.child("db");
2969 let db: UnorderedVariableDb =
2970 UnorderedVariableDb::init(ctx.child("storage"), variable_config::<OneCap>(partition, &ctx))
2971 .await
2972 .unwrap();
2973
2974 let (db, first_range) = commit_writes_with_metadata(
2975 db,
2976 (0..KEYS).map(|i| (key(i), Some(val(i)))),
2977 None,
2978 )
2979 .await;
2980 let (db, _) = commit_writes_with_metadata(
2981 db,
2982 (0..KEYS).map(|i| (key(i), Some(val(1000 + i)))),
2983 None,
2984 )
2985 .await;
2986
2987 let boundary = db.sync_boundary();
2988 let db = db.prune(boundary).await.unwrap();
2989 let pruned_bits = db.pruned_bits();
2990 assert!(
2991 pruned_bits > *first_range.start,
2992 "expected bitmap pruning boundary above rewind target: pruned_bits={pruned_bits}, target={:?}",
2993 first_range.start
2994 );
2995
2996 let oldest_retained = db.bounds().start;
2997 let Err(boundary_err) = db.rewind(oldest_retained).await else {
2998 panic!("expected rewind rejection at retained boundary");
2999 };
3000 assert!(
3001 matches!(
3002 boundary_err,
3003 Error::Journal(crate::journal::Error::ItemPruned(_))
3004 ),
3005 "unexpected rewind error at retained boundary: {boundary_err:?}"
3006 );
3007
3008 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3009 ctx.child("reopen"),
3010 variable_config::<OneCap>(partition, &ctx),
3011 )
3012 .await
3013 .unwrap();
3014 let expected_pruned_loc = *first_range.start - 1;
3015 let Err(err) = db.rewind(first_range.start).await else {
3016 panic!("expected rewind rejection at pruned target");
3017 };
3018 assert!(
3019 matches!(
3020 err,
3021 Error::Journal(crate::journal::Error::ItemPruned(loc))
3022 if loc == expected_pruned_loc
3023 ),
3024 "unexpected rewind error: {err:?}"
3025 );
3026 });
3027 }
3028
3029 #[test_traced("INFO")]
3030 fn test_current_rewind_rejects_target_below_bitmap_floor() {
3031 let executor = deterministic::Runner::default();
3032 executor.start(|context| async move {
3033 const COMMITS: u64 = 96;
3034
3035 let partition = "current-rewind-bitmap-floor";
3036 let ctx = context.child("db");
3037 let mut db: UnorderedVariableDb =
3038 UnorderedVariableDb::init(ctx.child("storage"), variable_config::<OneCap>(partition, &ctx))
3039 .await
3040 .unwrap();
3041
3042 let mut history = Vec::new();
3043 for round in 0..COMMITS {
3044 (db, _) = commit_writes_with_metadata(
3045 db,
3046 [(key(0), Some(val(10_000 + round)))],
3047 None,
3048 )
3049 .await;
3050 history.push((db.bounds().end, db.inactivity_floor_loc()));
3051 }
3052 assert!(db.inactivity_floor_loc() > Location::new(64));
3053
3054 let prune_loc = Location::new(1);
3057 let db = db.prune(prune_loc).await.unwrap();
3058 let pruned_bits = db.pruned_bits();
3059 assert!(pruned_bits > 0);
3060 let retained_start = db.bounds().start;
3061
3062 let rewind_target = history
3065 .iter()
3066 .find_map(|(size, floor)| {
3067 if *size > *retained_start
3068 && *size >= pruned_bits
3069 && *floor >= *retained_start
3070 && *floor < pruned_bits
3071 {
3072 Some(*size)
3073 } else {
3074 None
3075 }
3076 })
3077 .unwrap_or_else(|| {
3078 panic!(
3079 "expected rewind target below bitmap boundary. retained_start={retained_start:?}, pruned_bits={pruned_bits}, latest_floor={:?}, history={history:?}",
3080 db.inactivity_floor_loc()
3081 )
3082 });
3083
3084 let Err(err) = db.rewind(rewind_target).await else {
3085 panic!("expected rewind rejection below bitmap floor");
3086 };
3087 assert!(
3088 matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
3089 "unexpected rewind error: {err:?}"
3090 );
3091 });
3092 }
3093
3094 pub async fn test_speculative_root_matches_committed<M, C, F, Fut>(
3100 context: Context,
3101 mut open_db: F,
3102 ) where
3103 M: merkle::Graftable + 'static,
3104 C: DbAny<M> + 'static,
3105 C::Key: TestKey,
3106 <C as DbAny<M>>::Value: TestValue,
3107 F: FnMut(Context, String) -> Fut + Clone,
3108 Fut: Future<Output = C>,
3109 {
3110 let mut open_db_clone = open_db.clone();
3111 let partition = "speculative-root".to_string();
3112
3113 let db: C = Box::pin(open_db_clone(context.child("init"), partition.clone())).await;
3118 let mut batch = db.new_batch();
3119 for i in 0..260 {
3120 batch = batch.write(TestKey::from_seed(i), Some(TestValue::from_seed(i + 1000)));
3121 }
3122 let merkleized = batch.merkleize(&db, None).await.unwrap();
3123 let (db, _) = db.apply_batch(merkleized).await.unwrap();
3124 let speculative_root = db.root();
3125
3126 db.sync().await.unwrap();
3128
3129 let db: C = Box::pin(open_db(context.child("reopen"), partition)).await;
3130 assert_eq!(db.root(), speculative_root);
3131
3132 db.destroy().await.unwrap();
3133 }
3134
3135 test_for_all_variants!(test_speculative_root_matches_committed, "INFO");
3136
3137 #[test_traced("INFO")]
3139 fn test_current_batch_merkleized_get() {
3140 let executor = deterministic::Runner::default();
3141 executor.start(|context| async move {
3142 let ctx = context.child("db");
3143 let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
3144 ctx.child("storage"),
3145 variable_config::<OneCap>("mg", &ctx),
3146 )
3147 .await
3148 .unwrap();
3149
3150 let ka = key(0);
3151 let kb = key(1);
3152 let kc = key(2);
3153
3154 {
3156 let mut batch = db.new_batch();
3157 batch = batch.write(ka, Some(val(0)));
3158 let merkleized = batch.merkleize(&db, None).await.unwrap();
3159 (db, _) = db.apply_batch(merkleized).await.unwrap();
3160 }
3161
3162 let va2 = val(100);
3164 let vb = val(1);
3165 let mut batch = db.new_batch();
3166 batch = batch.write(ka, Some(va2));
3167 batch = batch.write(kb, Some(vb));
3168 let merkleized = batch.merkleize(&db, None).await.unwrap();
3169
3170 assert_eq!(merkleized.get(&ka, &db).await.unwrap(), Some(va2));
3171 assert_eq!(merkleized.get(&kb, &db).await.unwrap(), Some(vb));
3172 assert_eq!(merkleized.get(&kc, &db).await.unwrap(), None);
3173
3174 db.destroy().await.unwrap();
3175 });
3176 }
3177
3178 #[test_traced("INFO")]
3181 fn test_current_batch_chaining() {
3182 let executor = deterministic::Runner::default();
3183 executor.start(|context| async move {
3184 let ctx = context.child("db");
3185 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3186 ctx.child("storage"),
3187 variable_config::<OneCap>("ch", &ctx),
3188 )
3189 .await
3190 .unwrap();
3191
3192 let mut parent = db.new_batch();
3194 for i in 0..5 {
3195 parent = parent.write(key(i), Some(val(i)));
3196 }
3197 let parent_m = parent.merkleize(&db, None).await.unwrap();
3198
3199 let mut child = parent_m.new_batch::<Sha256>();
3201 for i in 5..10 {
3202 child = child.write(key(i), Some(val(i)));
3203 }
3204 child = child.write(key(0), Some(val(999)));
3205 let child_m = child.merkleize(&db, None).await.unwrap();
3206
3207 let child_root = child_m.root();
3208
3209 assert_eq!(child_m.get(&key(0), &db).await.unwrap(), Some(val(999)));
3211 assert_eq!(child_m.get(&key(3), &db).await.unwrap(), Some(val(3)));
3212 assert_eq!(child_m.get(&key(7), &db).await.unwrap(), Some(val(7)));
3213
3214 let (db, _) = db.apply_batch(child_m).await.unwrap();
3215 assert_eq!(db.root(), child_root);
3216
3217 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(999)));
3219 for i in 1..10 {
3220 assert_eq!(db.get(&key(i)).await.unwrap(), Some(val(i)));
3221 }
3222
3223 db.destroy().await.unwrap();
3224 });
3225 }
3226
3227 #[test_traced("INFO")]
3228 fn test_current_unordered_root_matches_between_pending_and_committed_paths() {
3229 let executor = deterministic::Runner::default();
3230 executor.start(|context| async move {
3231 let ctx = context.child("db");
3232 let db: UnorderedFixedDb =
3233 UnorderedFixedDb::init(ctx.child("storage"), fixed_config::<OneCap>("ucr", &ctx))
3234 .await
3235 .unwrap();
3236 let key_a = colliding_digest(0xAA, 1);
3237 let key_b = colliding_digest(0xAA, 0);
3238
3239 let mut initial = db.new_batch();
3244 for i in 0..4 {
3245 initial = initial.write(colliding_digest(0xAA, i), Some(colliding_digest(0xBB, i)));
3246 }
3247 let merkleized = initial.merkleize(&db, None).await.unwrap();
3248 let (db, _) = db.apply_batch(merkleized).await.unwrap();
3249 let db = db.commit().await.unwrap();
3250
3251 let parent = db
3255 .new_batch()
3256 .write(key_a, Some(colliding_digest(0xCC, 1)))
3257 .merkleize(&db, None)
3258 .await
3259 .unwrap();
3260
3261 let pending_child = parent
3265 .new_batch::<Sha256>()
3266 .write(key_a, Some(colliding_digest(0xDD, 1)))
3267 .write(key_b, Some(colliding_digest(0xDD, 0)))
3268 .merkleize(&db, None)
3269 .await
3270 .unwrap();
3271
3272 let pending_root = pending_child.root();
3273 let pending_ops_root = pending_child.ops_root();
3274
3275 let (db, _) = db.apply_batch(parent).await.unwrap();
3276 let db = db.commit().await.unwrap();
3277
3278 let committed_child = db
3279 .new_batch()
3280 .write(key_a, Some(colliding_digest(0xDD, 1)))
3281 .write(key_b, Some(colliding_digest(0xDD, 0)))
3282 .merkleize(&db, None)
3283 .await
3284 .unwrap();
3285
3286 assert_eq!(pending_root, committed_child.root());
3287 assert_eq!(pending_ops_root, committed_child.ops_root());
3288
3289 let (db, _) = db.apply_batch(pending_child).await.unwrap();
3292 assert_eq!(db.root(), committed_child.root());
3293 assert_eq!(db.ops_root(), committed_child.ops_root());
3294
3295 db.destroy().await.unwrap();
3296 });
3297 }
3298
3299 #[test_traced("INFO")]
3300 fn test_current_ordered_root_matches_between_pending_and_committed_paths() {
3301 let executor = deterministic::Runner::default();
3302 executor.start(|context| async move {
3303 let ctx = context.child("db");
3304 let db: OrderedFixedDb =
3305 OrderedFixedDb::init(ctx.child("storage"), fixed_config::<OneCap>("ocr", &ctx))
3306 .await
3307 .unwrap();
3308 let key_a = colliding_digest(0xAA, 1);
3309 let key_b = colliding_digest(0xAA, 0);
3310
3311 let mut initial = db.new_batch();
3314 for i in 0..4 {
3315 initial = initial.write(colliding_digest(0xAA, i), Some(colliding_digest(0xBB, i)));
3316 }
3317 let merkleized = initial.merkleize(&db, None).await.unwrap();
3318 let (db, _) = db.apply_batch(merkleized).await.unwrap();
3319 let db = db.commit().await.unwrap();
3320
3321 let parent = db
3325 .new_batch()
3326 .write(key_a, Some(colliding_digest(0xCC, 1)))
3327 .merkleize(&db, None)
3328 .await
3329 .unwrap();
3330
3331 let pending_child = parent
3334 .new_batch::<Sha256>()
3335 .write(key_a, Some(colliding_digest(0xDD, 1)))
3336 .write(key_b, Some(colliding_digest(0xDD, 0)))
3337 .merkleize(&db, None)
3338 .await
3339 .unwrap();
3340
3341 let pending_root = pending_child.root();
3342 let pending_ops_root = pending_child.ops_root();
3343
3344 let (db, _) = db.apply_batch(parent).await.unwrap();
3345 let db = db.commit().await.unwrap();
3346
3347 let committed_child = db
3348 .new_batch()
3349 .write(key_a, Some(colliding_digest(0xDD, 1)))
3350 .write(key_b, Some(colliding_digest(0xDD, 0)))
3351 .merkleize(&db, None)
3352 .await
3353 .unwrap();
3354
3355 assert_eq!(pending_root, committed_child.root());
3356 assert_eq!(pending_ops_root, committed_child.ops_root());
3357
3358 let (db, _) = db.apply_batch(pending_child).await.unwrap();
3361 assert_eq!(db.root(), committed_child.root());
3362 assert_eq!(db.ops_root(), committed_child.ops_root());
3363
3364 db.destroy().await.unwrap();
3365 });
3366 }
3367
3368 #[test_traced("INFO")]
3370 fn test_current_batch_apply_requires_commit_for_recovery() {
3371 let executor = deterministic::Runner::default();
3372 executor.start(|context| async move {
3373 let partition = "apply_requires_commit";
3374 let ctx = context.child("db");
3375 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3376 ctx.child("storage"),
3377 variable_config::<OneCap>(partition, &ctx),
3378 )
3379 .await
3380 .unwrap();
3381
3382 let committed_root = db.root();
3383
3384 let merkleized = db
3385 .new_batch()
3386 .write(key(0), Some(val(0)))
3387 .merkleize(&db, None)
3388 .await
3389 .unwrap();
3390 let (db, _) = db.apply_batch(merkleized).await.unwrap();
3391
3392 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
3393
3394 drop(db);
3395
3396 let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
3397 context.child("reopen"),
3398 variable_config::<OneCap>(partition, &context),
3399 )
3400 .await
3401 .unwrap();
3402 assert_eq!(reopened.root(), committed_root);
3403 assert_eq!(reopened.get(&key(0)).await.unwrap(), None);
3404
3405 reopened.destroy().await.unwrap();
3406 });
3407 }
3408
3409 #[test_traced("INFO")]
3411 fn test_current_batch_single_stage_pipeline() {
3412 let executor = deterministic::Runner::default();
3413 executor.start(|context| async move {
3414 let ctx = context.child("db");
3415 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3416 ctx.child("storage"),
3417 variable_config::<OneCap>("pipe", &ctx),
3418 )
3419 .await
3420 .unwrap();
3421
3422 let mut batch = db.new_batch();
3423 batch = batch.write(key(0), Some(val(0)));
3424 let parent_merkleized = batch.merkleize(&db, None).await.unwrap();
3425 let (db, _) = db.apply_batch(parent_merkleized).await.unwrap();
3426
3427 let child_merkleized = {
3428 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
3429 let mut child = db.new_batch();
3430 child = child.write(key(1), Some(val(1)));
3431 child.merkleize(&db, None).await.unwrap()
3432 };
3433 let db = db.commit().await.unwrap();
3434
3435 let (db, _) = db.apply_batch(child_merkleized).await.unwrap();
3436 let db = db.commit().await.unwrap();
3437
3438 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
3439 assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
3440
3441 db.destroy().await.unwrap();
3442 });
3443 }
3444
3445 #[test_traced("INFO")]
3448 fn test_current_sequential_commit() {
3449 let executor = deterministic::Runner::default();
3450 executor.start(|context| async move {
3451 let ctx = context.child("db");
3452 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3453 ctx.child("storage"),
3454 variable_config::<OneCap>("ff", &ctx),
3455 )
3456 .await
3457 .unwrap();
3458
3459 let parent_m = db
3461 .new_batch()
3462 .write(key(0), Some(val(0)))
3463 .merkleize(&db, None)
3464 .await
3465 .unwrap();
3466
3467 let child_m = parent_m
3469 .new_batch::<Sha256>()
3470 .write(key(1), Some(val(1)))
3471 .merkleize(&db, None)
3472 .await
3473 .unwrap();
3474
3475 let (db, _) = db.apply_batch(parent_m).await.unwrap();
3476 let (db, _) = db.apply_batch(child_m).await.unwrap();
3477
3478 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
3480 assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
3481
3482 let ctx2 = context.child("db").with_attribute("index", 2);
3485 let db2: UnorderedVariableDb = UnorderedVariableDb::init(
3486 ctx2.child("db"),
3487 variable_config::<OneCap>("ff2", &ctx2),
3488 )
3489 .await
3490 .unwrap();
3491 let m1 = db2
3492 .new_batch()
3493 .write(key(0), Some(val(0)))
3494 .merkleize(&db2, None)
3495 .await
3496 .unwrap();
3497 let (db2, _) = db2.apply_batch(m1).await.unwrap();
3498 let m2 = db2
3499 .new_batch()
3500 .write(key(1), Some(val(1)))
3501 .merkleize(&db2, None)
3502 .await
3503 .unwrap();
3504 let (db2, _) = db2.apply_batch(m2).await.unwrap();
3505
3506 assert_eq!(db.root(), db2.root());
3507
3508 db.destroy().await.unwrap();
3509 db2.destroy().await.unwrap();
3510 });
3511 }
3512
3513 #[test_traced("INFO")]
3516 fn test_current_to_batch_then_chain() {
3517 let executor = deterministic::Runner::default();
3518 executor.start(|context| async move {
3519 let ctx = context.child("db");
3520 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3521 ctx.child("storage"),
3522 variable_config::<OneCap>("tb", &ctx),
3523 )
3524 .await
3525 .unwrap();
3526
3527 let m = db
3529 .new_batch()
3530 .write(key(0), Some(val(0)))
3531 .merkleize(&db, None)
3532 .await
3533 .unwrap();
3534 let (db, _) = db.apply_batch(m).await.unwrap();
3535
3536 let snapshot = db.to_batch();
3538 assert_eq!(snapshot.root(), db.root());
3539
3540 let child = snapshot
3542 .new_batch::<Sha256>()
3543 .write(key(1), Some(val(1)))
3544 .merkleize(&db, None)
3545 .await
3546 .unwrap();
3547
3548 assert_ne!(child.root(), snapshot.root());
3550
3551 let (db, _) = db.apply_batch(child).await.unwrap();
3553 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
3554 assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
3555
3556 db.destroy().await.unwrap();
3557 });
3558 }
3559
3560 #[test_traced("INFO")]
3565 fn test_current_live_batch_safe_across_prune() {
3566 let executor = deterministic::Runner::default();
3567 executor.start(|context| async move {
3568 let ctx = context.child("db");
3569 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3570 ctx.child("storage"),
3571 variable_config::<OneCap>("prune-live", &ctx),
3572 )
3573 .await
3574 .unwrap();
3575
3576 let mut seed = db.new_batch();
3578 for i in 0u64..300 {
3579 seed = seed.write(key(i), Some(val(i)));
3580 }
3581 let seed_m = seed.merkleize(&db, None).await.unwrap();
3582 let (db, _) = db.apply_batch(seed_m).await.unwrap();
3583 let db = db.commit().await.unwrap();
3584
3585 let mut p = db.new_batch();
3587 for i in 0u64..250 {
3588 p = p.write(key(i), Some(val(i + 10_000)));
3589 }
3590 let p_m = p.merkleize(&db, None).await.unwrap();
3591 let (db, _) = db.apply_batch(Arc::clone(&p_m)).await.unwrap();
3592 let db = db.commit().await.unwrap();
3593
3594 let c = p_m
3596 .new_batch::<Sha256>()
3597 .write(key(250), Some(val(99_999)))
3598 .merkleize(&db, None)
3599 .await
3600 .unwrap();
3601
3602 let boundary = db.sync_boundary();
3604 let db = db.prune(boundary).await.unwrap();
3605
3606 assert_eq!(c.get(&key(250), &db).await.unwrap(), Some(val(99_999)));
3608
3609 let (db, _) = db.apply_batch(c).await.unwrap();
3612 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(10_000)));
3613 assert_eq!(db.get(&key(250)).await.unwrap(), Some(val(99_999)));
3614
3615 db.destroy().await.unwrap();
3616 });
3617 }
3618
3619 #[test_traced("INFO")]
3628 fn test_current_extend_applied_batch() {
3629 let executor = deterministic::Runner::default();
3630 executor.start(|context| async move {
3631 let ctx = context.child("db");
3632 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3633 ctx.child("storage"),
3634 variable_config::<OneCap>("xtend", &ctx),
3635 )
3636 .await
3637 .unwrap();
3638
3639 let a = db
3641 .new_batch()
3642 .write(key(0), Some(val(0)))
3643 .merkleize(&db, None)
3644 .await
3645 .unwrap();
3646 let (db, _) = db.apply_batch(Arc::clone(&a)).await.unwrap();
3647
3648 let b = a
3652 .new_batch::<Sha256>()
3653 .write(key(1), Some(val(1)))
3654 .merkleize(&db, None)
3655 .await
3656 .unwrap();
3657 let (db, _) = db.apply_batch(b).await.unwrap();
3658
3659 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
3660 assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
3661
3662 let c = db
3664 .new_batch()
3665 .write(key(2), Some(val(2)))
3666 .merkleize(&db, None)
3667 .await
3668 .unwrap();
3669 let (db, _) = db.apply_batch(c).await.unwrap();
3670
3671 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
3672 assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
3673 assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2)));
3674
3675 db.destroy().await.unwrap();
3676 });
3677 }
3678
3679 #[test_traced("INFO")]
3686 fn test_current_live_batch_child_after_prune() {
3687 let executor = deterministic::Runner::default();
3688 executor.start(|context| async move {
3689 let ctx = context.child("db");
3690 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3691 ctx.child("storage"),
3692 variable_config::<OneCap>("child-after-prune", &ctx),
3693 )
3694 .await
3695 .unwrap();
3696
3697 let mut seed = db.new_batch();
3699 for i in 0u64..300 {
3700 seed = seed.write(key(i), Some(val(i)));
3701 }
3702 let seed_m = seed.merkleize(&db, None).await.unwrap();
3703 let (db, _) = db.apply_batch(seed_m).await.unwrap();
3704 let db = db.commit().await.unwrap();
3705
3706 let mut a_batch = db.new_batch();
3708 for i in 0u64..250 {
3709 a_batch = a_batch.write(key(i), Some(val(i + 10_000)));
3710 }
3711 let a = a_batch.merkleize(&db, None).await.unwrap();
3712 let (db, _) = db.apply_batch(Arc::clone(&a)).await.unwrap();
3713 let db = db.commit().await.unwrap();
3714
3715 let boundary = db.sync_boundary();
3717 let db = db.prune(boundary).await.unwrap();
3718
3719 let b = a
3723 .new_batch::<Sha256>()
3724 .write(key(300), Some(val(300)))
3725 .merkleize(&db, None)
3726 .await
3727 .unwrap();
3728
3729 let (db, _) = db.apply_batch(b).await.unwrap();
3730 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(10_000)));
3731 assert_eq!(db.get(&key(249)).await.unwrap(), Some(val(10_249)));
3732 assert_eq!(db.get(&key(300)).await.unwrap(), Some(val(300)));
3733
3734 db.destroy().await.unwrap();
3735 });
3736 }
3737
3738 #[test_traced("WARN")]
3742 fn test_current_apply_after_ancestor_dropped() {
3743 let executor = deterministic::Runner::default();
3744 executor.start(|context| async move {
3745 let ctx = context.child("db");
3746 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3747 ctx.child("storage"),
3748 variable_config::<OneCap>("adrop", &ctx),
3749 )
3750 .await
3751 .unwrap();
3752
3753 let mut a = db.new_batch();
3755 for i in 0..3 {
3756 a = a.write(key(i), Some(val(i)));
3757 }
3758 let a_m = a.merkleize(&db, None).await.unwrap();
3759
3760 let mut b = a_m.new_batch::<Sha256>();
3761 for i in 3..6 {
3762 b = b.write(key(i), Some(val(i)));
3763 }
3764 let b_m = b.merkleize(&db, None).await.unwrap();
3765
3766 let mut c = b_m.new_batch::<Sha256>();
3767 for i in 6..9 {
3768 c = c.write(key(i), Some(val(i)));
3769 }
3770 let c_m = c.merkleize(&db, None).await.unwrap();
3771
3772 drop(a_m);
3774 drop(b_m);
3775
3776 let (db, _) = db.apply_batch(c_m).await.unwrap();
3778 let db = db.commit().await.unwrap();
3779
3780 for i in 0..9 {
3782 assert_eq!(
3783 db.get(&key(i)).await.unwrap(),
3784 Some(val(i)),
3785 "key({i}) missing after apply_batch with dropped ancestors"
3786 );
3787 }
3788
3789 db.destroy().await.unwrap();
3790 });
3791 }
3792
3793 #[test_traced("WARN")]
3802 fn test_current_chain_bitmap_order_matches_sequential() {
3803 let executor = deterministic::Runner::default();
3804 executor.start(|context| async move {
3805 let ctx1 = context.child("db").with_attribute("index", 1);
3807 let db1: UnorderedVariableDb = UnorderedVariableDb::init(
3808 ctx1.child("db"),
3809 variable_config::<OneCap>("ord1", &ctx1),
3810 )
3811 .await
3812 .unwrap();
3813
3814 let (db1, _) = commit_writes_with_metadata(
3816 db1,
3817 [(key(10), Some(val(10))), (key(11), Some(val(11)))],
3818 None,
3819 )
3820 .await;
3821
3822 let a = db1
3829 .new_batch()
3830 .write(key(10), Some(val(100)))
3831 .write(key(11), None) .merkleize(&db1, None)
3833 .await
3834 .unwrap();
3835
3836 let b = a
3837 .new_batch::<Sha256>()
3838 .write(key(12), Some(val(120)))
3839 .write(key(13), Some(val(130)))
3840 .merkleize(&db1, None)
3841 .await
3842 .unwrap();
3843
3844 let c = b
3845 .new_batch::<Sha256>()
3846 .write(key(14), Some(val(140)))
3847 .merkleize(&db1, None)
3848 .await
3849 .unwrap();
3850
3851 let (db1, _) = db1.apply_batch(c).await.unwrap();
3852 let db1 = db1.commit().await.unwrap();
3853
3854 let d1 = db1
3856 .new_batch()
3857 .write(key(20), Some(val(200)))
3858 .merkleize(&db1, None)
3859 .await
3860 .unwrap();
3861 let chain_then_d_root = d1.root();
3862
3863 let ctx2 = context.child("db").with_attribute("index", 2);
3865 let db2: UnorderedVariableDb = UnorderedVariableDb::init(
3866 ctx2.child("db"),
3867 variable_config::<OneCap>("ord2", &ctx2),
3868 )
3869 .await
3870 .unwrap();
3871
3872 let (db2, _) = commit_writes_with_metadata(
3873 db2,
3874 [(key(10), Some(val(10))), (key(11), Some(val(11)))],
3875 None,
3876 )
3877 .await;
3878
3879 let a2 = db2
3880 .new_batch()
3881 .write(key(10), Some(val(100)))
3882 .write(key(11), None)
3883 .merkleize(&db2, None)
3884 .await
3885 .unwrap();
3886 let (db2, _) = db2.apply_batch(a2).await.unwrap();
3887 let db2 = db2.commit().await.unwrap();
3888
3889 let b2 = db2
3890 .new_batch()
3891 .write(key(12), Some(val(120)))
3892 .write(key(13), Some(val(130)))
3893 .merkleize(&db2, None)
3894 .await
3895 .unwrap();
3896 let (db2, _) = db2.apply_batch(b2).await.unwrap();
3897 let db2 = db2.commit().await.unwrap();
3898
3899 let c2 = db2
3900 .new_batch()
3901 .write(key(14), Some(val(140)))
3902 .merkleize(&db2, None)
3903 .await
3904 .unwrap();
3905 let (db2, _) = db2.apply_batch(c2).await.unwrap();
3906 let db2 = db2.commit().await.unwrap();
3907
3908 let d2 = db2
3909 .new_batch()
3910 .write(key(20), Some(val(200)))
3911 .merkleize(&db2, None)
3912 .await
3913 .unwrap();
3914 let sequential_then_d_root = d2.root();
3915
3916 assert_eq!(
3917 chain_then_d_root, sequential_then_d_root,
3918 "batch D's root on top of chain-applied state must match sequential state"
3919 );
3920
3921 db1.destroy().await.unwrap();
3922 db2.destroy().await.unwrap();
3923 });
3924 }
3925
3926 #[test_traced("WARN")]
3938 fn test_current_stale_bitmap_clears_after_prune() {
3939 let executor = deterministic::Runner::default();
3940 executor.start(|context| async move {
3941 let ctx = context.child("db");
3942 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3943 ctx.child("storage"),
3944 variable_config::<OneCap>("stale-clears", &ctx),
3945 )
3946 .await
3947 .unwrap();
3948
3949 let mut seed = db.new_batch();
3951 for i in 0u64..255 {
3952 seed = seed.write(key(i), Some(val(i)));
3953 }
3954 let seed_m = seed.merkleize(&db, None).await.unwrap();
3955 let (db, _) = db.apply_batch(seed_m).await.unwrap();
3956 let db = db.commit().await.unwrap();
3957
3958 let mut p = db.new_batch();
3961 for i in 1u64..255 {
3962 p = p.write(key(i), Some(val(i + 10000)));
3963 }
3964 let p_m = p.merkleize(&db, None).await.unwrap();
3965
3966 let c_m = p_m
3968 .new_batch::<Sha256>()
3969 .write(key(0), Some(val(9999)))
3970 .merkleize(&db, None)
3971 .await
3972 .unwrap();
3973
3974 let (db, _) = db.apply_batch(p_m).await.unwrap();
3976 let db = db.commit().await.unwrap();
3977
3978 let floor = *db.inactivity_floor_loc();
3979 assert!(floor >= 256, "floor must be past chunk 0: floor={floor}",);
3980
3981 let boundary = db.sync_boundary();
3982 let db = db.prune(boundary).await.unwrap();
3983 let (db, _) = db.apply_batch(c_m).await.unwrap();
3984
3985 db.destroy().await.unwrap();
3986 });
3987 }
3988
3989 #[test_traced("INFO")]
3992 fn test_current_partial_ancestor_commit() {
3993 let executor = deterministic::Runner::default();
3994 executor.start(|context| async move {
3995 let ctx = context.child("db");
3996 let db: UnorderedVariableDb = UnorderedVariableDb::init(
3997 ctx.child("storage"),
3998 variable_config::<OneCap>("pac", &ctx),
3999 )
4000 .await
4001 .unwrap();
4002
4003 let a = db
4004 .new_batch()
4005 .write(key(0), Some(val(0)))
4006 .merkleize(&db, None)
4007 .await
4008 .unwrap();
4009 let b = a
4010 .new_batch::<Sha256>()
4011 .write(key(1), Some(val(1)))
4012 .merkleize(&db, None)
4013 .await
4014 .unwrap();
4015 let c = b
4016 .new_batch::<Sha256>()
4017 .write(key(2), Some(val(2)))
4018 .merkleize(&db, None)
4019 .await
4020 .unwrap();
4021
4022 let expected_root = c.root();
4023
4024 let (db, _) = db.apply_batch(a).await.unwrap();
4025 let (db, _) = db.apply_batch(c).await.unwrap();
4026
4027 assert_eq!(db.root(), expected_root);
4028 assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
4029 assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
4030 assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2)));
4031
4032 db.destroy().await.unwrap();
4033 });
4034 }
4035
4036 #[test_traced("INFO")]
4040 fn test_current_partial_ancestor_bitmap_ordering() {
4041 let executor = deterministic::Runner::default();
4042 executor.start(|context| async move {
4043 let ctx = context.child("db");
4044 let db: UnorderedVariableDb = UnorderedVariableDb::init(
4045 ctx.child("storage"),
4046 variable_config::<OneCap>("bmo", &ctx),
4047 )
4048 .await
4049 .unwrap();
4050
4051 let a = db
4053 .new_batch()
4054 .write(key(0), Some(val(0)))
4055 .merkleize(&db, None)
4056 .await
4057 .unwrap();
4058 let b = a
4059 .new_batch::<Sha256>()
4060 .write(key(1), Some(val(1)))
4061 .merkleize(&db, None)
4062 .await
4063 .unwrap();
4064 let c = b
4065 .new_batch::<Sha256>()
4066 .write(key(2), Some(val(2)))
4067 .merkleize(&db, None)
4068 .await
4069 .unwrap();
4070 let d = c
4071 .new_batch::<Sha256>()
4072 .write(key(3), Some(val(3)))
4073 .merkleize(&db, None)
4074 .await
4075 .unwrap();
4076
4077 let (db, _) = db.apply_batch(a).await.unwrap();
4081 let (db, _) = db.apply_batch(d.clone()).await.unwrap();
4082
4083 let e = db
4088 .new_batch()
4089 .write(key(4), Some(val(4)))
4090 .merkleize(&db, None)
4091 .await
4092 .unwrap();
4093 let (db, _) = db.apply_batch(e).await.unwrap();
4094
4095 let ref_ctx = context.child("ref");
4097 let mut ref_db: UnorderedVariableDb = UnorderedVariableDb::init(
4098 ref_ctx.child("db"),
4099 variable_config::<OneCap>("bmo_ref", &ref_ctx),
4100 )
4101 .await
4102 .unwrap();
4103 for i in 0..5 {
4104 let batch = ref_db
4105 .new_batch()
4106 .write(key(i), Some(val(i)))
4107 .merkleize(&ref_db, None)
4108 .await
4109 .unwrap();
4110 (ref_db, _) = ref_db.apply_batch(batch).await.unwrap();
4111 }
4112
4113 assert_eq!(
4114 db.root(),
4115 ref_db.root(),
4116 "root mismatch: bitmap ordering bug"
4117 );
4118
4119 db.destroy().await.unwrap();
4120 ref_db.destroy().await.unwrap();
4121 });
4122 }
4123
4124 #[test_traced("INFO")]
4134 fn test_current_apply_chunks_match_speculative_chunks() {
4135 const N: usize = 32;
4136 const CHUNK_SIZE_BITS: u64 = commonware_utils::bitmap::Prunable::<N>::CHUNK_SIZE_BITS;
4137 const SEED_KEYS: u64 = CHUNK_SIZE_BITS + 50;
4140
4141 let executor = deterministic::Runner::default();
4142 executor.start(|context| async move {
4143 let ctx = context.child("db");
4144 let db: UnorderedVariableDb = UnorderedVariableDb::init(
4145 ctx.child("storage"),
4146 variable_config::<OneCap>("spec_eq", &ctx),
4147 )
4148 .await
4149 .unwrap();
4150
4151 let seed = (0..SEED_KEYS).fold(db.new_batch(), |b, i| b.write(key(i), Some(val(i))));
4153 let seed = seed.merkleize(&db, None).await.unwrap();
4154 let (db, _) = db.apply_batch(seed).await.unwrap();
4155 let db = db.commit().await.unwrap();
4156
4157 assert!(
4159 Readable::<N>::len(db.any.bitmap.as_ref()) > CHUNK_SIZE_BITS,
4160 "setup must cross a chunk boundary",
4161 );
4162
4163 let parent = db
4165 .new_batch()
4166 .write(key(10), Some(val(110))) .write(key(50), None) .write(key(CHUNK_SIZE_BITS + 5), Some(val(120))) .write(key(SEED_KEYS), Some(val(130))) .write(key(SEED_KEYS + 1), Some(val(131))) .merkleize(&db, None)
4172 .await
4173 .unwrap();
4174
4175 let child = parent
4180 .new_batch::<Sha256>()
4181 .write(key(10), Some(val(210)))
4182 .write(key(SEED_KEYS), None)
4183 .write(key(75), None)
4184 .write(key(CHUNK_SIZE_BITS + 30), Some(val(220)))
4185 .merkleize(&db, None)
4186 .await
4187 .unwrap();
4188
4189 let speculative_chunks: Vec<[u8; N]> = {
4191 let len = Readable::<N>::len(&child.bitmap);
4192 let chunk_count = len.div_ceil(CHUNK_SIZE_BITS) as usize;
4193 (0..chunk_count)
4194 .map(|idx| Readable::<N>::get_chunk(&child.bitmap, idx))
4195 .collect()
4196 };
4197 assert!(speculative_chunks.len() >= 2);
4199
4200 let (db, _) = db.apply_batch(child).await.unwrap();
4204 let committed_chunks: Vec<[u8; N]> = {
4205 let len = Readable::<N>::len(db.any.bitmap.as_ref());
4206 let chunk_count = len.div_ceil(CHUNK_SIZE_BITS) as usize;
4207 (0..chunk_count)
4208 .map(|idx| Readable::<N>::get_chunk(db.any.bitmap.as_ref(), idx))
4209 .collect()
4210 };
4211
4212 assert_eq!(
4213 speculative_chunks, committed_chunks,
4214 "speculative chunks must equal post-apply committed chunks across all chunks",
4215 );
4216
4217 db.destroy().await.unwrap();
4218 });
4219 }
4220
4221 #[test_traced("INFO")]
4223 fn test_current_mmb_ops_historical_proof_verifies_with_backward_bagging() {
4224 let executor = deterministic::Runner::default();
4225 executor.start(|context| async move {
4226 let ctx = context.child("db");
4227 let db: UnorderedFixedMmbDb = UnorderedFixedMmbDb::init(
4228 ctx.child("storage"),
4229 fixed_config::<OneCap>("mmb-ops-proof", &ctx),
4230 )
4231 .await
4232 .unwrap();
4233
4234 let writes: Vec<(Digest, Option<Digest>)> =
4236 (0u64..16).map(|i| (key(i), Some(val(i)))).collect();
4237 let db = commit_writes(db, writes).await.unwrap();
4238
4239 let ops_root = db.ops_root();
4240 let historical_size = db.bounds().end;
4241 let (proof, ops) = db
4242 .ops_historical_proof(historical_size, Location::new(0), NZU64!(32))
4243 .await
4244 .unwrap();
4245
4246 assert!(verify_proof::<Sha256, _, _>(
4248 &proof,
4249 Location::new(0),
4250 &ops,
4251 &ops_root
4252 ));
4253
4254 db.destroy().await.unwrap();
4255 });
4256 }
4257}