1use crate::{
81 index::{unordered::Index, Unordered as _},
82 journal::{
83 authenticated,
84 contiguous::{Contiguous, Mutable},
85 },
86 merkle::{full::Config as MerkleConfig, Family, Location, Proof},
87 qmdb::{
88 any::ValueEncoding, build_snapshot_from_log, metrics::Metrics, operation::Key,
89 single_operation_root, Error,
90 },
91 translator::Translator,
92 Context,
93};
94use ahash::AHashSet;
95use commonware_codec::EncodeShared;
96use commonware_cryptography::Hasher;
97use commonware_macros::boxed;
98use commonware_parallel::Strategy;
99use core::num::{NonZeroU64, NonZeroUsize};
100use std::{ops::Range, sync::Arc};
101use tracing::warn;
102
103pub mod batch;
104mod compact;
105pub mod fixed;
106mod operation;
107pub mod sync;
108pub mod variable;
109
110pub use compact::{
111 Config as CompactConfig, Db as CompactDb, MerkleizedBatch as CompactMerkleizedBatch,
112 UnmerkleizedBatch as CompactUnmerkleizedBatch,
113};
114pub use operation::Operation;
115
116pub fn initial_root<F, K, V, H>() -> H::Digest
120where
121 F: Family,
122 K: Key,
123 V: ValueEncoding,
124 H: Hasher,
125 Operation<F, K, V>: EncodeShared,
126{
127 single_operation_root::<F, H>(&Operation::<F, K, V>::Commit(None, Location::new(0)))
128}
129
130#[derive(Clone)]
132pub struct Config<T: Translator, J, S: Strategy> {
133 pub merkle_config: MerkleConfig<S>,
135
136 pub log: J,
138
139 pub translator: T,
141
142 pub init_cache_size: Option<NonZeroUsize>,
145}
146
147pub struct Immutable<
157 F: Family,
158 E: Context,
159 K: Key,
160 V: ValueEncoding,
161 C: Mutable<Item = Operation<F, K, V>>,
162 H: Hasher,
163 T: Translator,
164 S: Strategy,
165> where
166 C::Item: EncodeShared,
167{
168 pub(crate) journal: authenticated::Journal<F, E, C, H, S>,
170
171 pub(crate) root: H::Digest,
173
174 pub(crate) snapshot: Index<T, Location<F>>,
180
181 pub(crate) last_commit_loc: Location<F>,
183
184 pub(crate) inactivity_floor_loc: Location<F>,
187
188 metrics: Metrics<E>,
190}
191
192impl<F, E, K, V, C, H, T, S> Immutable<F, E, K, V, C, H, T, S>
194where
195 F: Family,
196 E: Context,
197 K: Key,
198 V: ValueEncoding,
199 C: Mutable<Item = Operation<F, K, V>>,
200 C::Item: EncodeShared,
201 H: Hasher,
202 T: Translator,
203 S: Strategy,
204{
205 #[boxed]
210 pub(crate) async fn init_from_journal(
211 mut journal: authenticated::Journal<F, E, C, H, S>,
212 context: E,
213 translator: T,
214 cache_size: Option<NonZeroUsize>,
215 ) -> Result<Self, Error<F>> {
216 if journal.size() == 0 {
217 warn!("Authenticated log is empty, initialized new db.");
218 journal
219 .append(&Operation::Commit(None, Location::new(0)))
220 .await?;
221 journal.sync().await?;
222 }
223
224 let mut snapshot = Index::new(context.child("snapshot"), translator);
225
226 let (last_commit_loc, inactivity_floor_loc) = {
227 let bounds = journal.journal.bounds();
228 let last_commit_loc =
229 Location::new(bounds.end.checked_sub(1).expect("commit should exist"));
230
231 let last_op = journal.journal.read(*last_commit_loc).await?;
233 let inactivity_floor_loc = last_op
234 .has_floor()
235 .expect("last operation should be a commit with floor");
236 if inactivity_floor_loc > last_commit_loc {
237 return Err(Error::DataCorrupted("inactivity floor exceeds last commit"));
238 }
239
240 build_snapshot_from_log::<F, _, _, _>(
242 inactivity_floor_loc,
243 &journal.journal,
244 &mut snapshot,
245 cache_size,
246 |_, _| {},
247 )
248 .await?;
249
250 (last_commit_loc, inactivity_floor_loc)
251 };
252 let inactive_peaks = F::inactive_peaks(
253 F::location_to_position(Location::new(*last_commit_loc + 1)),
254 inactivity_floor_loc,
255 );
256 let root = journal.root(inactive_peaks)?;
257
258 let metrics = Metrics::new(context);
259 let db = Self {
260 journal,
261 root,
262 snapshot,
263 last_commit_loc,
264 inactivity_floor_loc,
265 metrics,
266 };
267 db.update_metrics();
268 Ok(db)
269 }
270
271 pub const fn inactivity_floor_loc(&self) -> Location<F> {
273 self.inactivity_floor_loc
274 }
275
276 pub fn size(&self) -> Location<F> {
278 self.bounds().end
279 }
280
281 pub fn bounds(&self) -> Range<Location<F>> {
284 Location::new(self.journal.bounds().start)..Location::new(self.journal.bounds().end)
285 }
286
287 fn update_metrics(&self) {
289 let bounds = self.journal.bounds();
290 self.metrics.update(
291 bounds.end,
292 bounds.start,
293 *self.inactivity_floor_loc,
294 *self.last_commit_loc,
295 );
296 }
297
298 pub const fn sync_boundary(&self) -> Location<F> {
302 self.inactivity_floor_loc
303 }
304
305 pub async fn get(&self, key: &K) -> Result<Option<V::Value>, Error<F>> {
308 let _timer = self.metrics.get_timer();
309 self.metrics.get_calls.inc();
310 self.metrics.lookups_requested.inc();
311 let iter = self.snapshot.get(key);
312 let oldest = self.journal.bounds().start;
313 let mut result = None;
314 for &loc in iter {
315 if loc < oldest {
316 continue;
317 }
318 if let Some(v) = Self::get_from_loc(&self.journal, key, loc).await? {
319 result = Some(v);
320 break;
321 }
322 }
323
324 Ok(result)
325 }
326
327 pub async fn get_many(&self, keys: &[&K]) -> Result<Vec<Option<V::Value>>, Error<F>> {
331 if keys.is_empty() {
332 return Ok(Vec::new());
333 }
334
335 let _timer = self.metrics.get_many_timer();
336 self.metrics.get_many_calls.inc();
337 self.metrics.lookups_requested.inc_by(keys.len() as u64);
338 let mut candidates: Vec<(usize, u64)> = Vec::with_capacity(keys.len());
339 let mut results: Vec<Option<V::Value>> = vec![None; keys.len()];
340
341 let oldest = self.journal.bounds().start;
342
343 for (key_idx, key) in keys.iter().enumerate() {
344 for &loc in self.snapshot.get(key) {
345 if loc < oldest {
346 continue;
347 }
348 candidates.push((key_idx, *loc));
349 }
350 }
351
352 if candidates.is_empty() {
353 return Ok(results);
354 }
355
356 candidates.sort_unstable_by_key(|&(_, pos)| pos);
357
358 let mut positions: Vec<u64> = Vec::with_capacity(candidates.len());
359 for &(_, pos) in &candidates {
360 if positions.last() != Some(&pos) {
361 positions.push(pos);
362 }
363 }
364
365 let ops = self.journal.read_many(&positions).await?;
366
367 for &(key_idx, pos) in &candidates {
368 if results[key_idx].is_some() {
369 continue;
370 }
371 let op_idx = positions
372 .binary_search(&pos)
373 .expect("position was deduped from candidates");
374 let Operation::Set(k, v) = &ops[op_idx] else {
375 return Err(Error::UnexpectedData(Location::new(pos)));
376 };
377 if k == keys[key_idx] {
378 results[key_idx] = Some(v.clone());
379 }
380 }
381
382 Ok(results)
383 }
384
385 async fn get_from_loc(
389 reader: &impl Contiguous<Item = Operation<F, K, V>>,
390 key: &K,
391 loc: Location<F>,
392 ) -> Result<Option<V::Value>, Error<F>> {
393 if loc < reader.bounds().start {
394 return Err(Error::OperationPruned(loc));
395 }
396
397 let Operation::Set(k, v) = reader.read(*loc).await? else {
398 return Err(Error::UnexpectedData(loc));
399 };
400
401 if k != *key {
402 Ok(None)
403 } else {
404 Ok(Some(v))
405 }
406 }
407
408 pub async fn get_metadata(&self) -> Result<Option<V::Value>, Error<F>> {
410 let last_commit_loc = self.last_commit_loc;
411 let Operation::Commit(metadata, _floor) =
412 self.journal.journal.read(*last_commit_loc).await?
413 else {
414 unreachable!("no commit operation at location of last commit {last_commit_loc}");
415 };
416
417 Ok(metadata)
418 }
419
420 #[allow(clippy::type_complexity)]
440 #[tracing::instrument(
441 name = "qmdb.immutable.db.historical_proof",
442 level = "info",
443 skip_all,
444 fields(
445 op_count = *op_count,
446 start_loc = *start_loc,
447 max_ops = max_ops.get(),
448 ),
449 )]
450 pub async fn historical_proof(
451 &self,
452 op_count: Location<F>,
453 start_loc: Location<F>,
454 max_ops: NonZeroU64,
455 ) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, K, V>>), Error<F>> {
456 if op_count > self.journal.size() {
457 return Err(crate::merkle::Error::RangeOutOfBounds(op_count).into());
458 }
459
460 let inactive_peaks =
461 crate::qmdb::inactive_peaks_at::<F, _>(&self.journal, op_count, |op| op.has_floor())
462 .await?;
463
464 Ok(self
465 .journal
466 .historical_proof(op_count, start_loc, max_ops, inactive_peaks)
467 .await?)
468 }
469
470 pub async fn proof(
477 &self,
478 start_index: Location<F>,
479 max_ops: NonZeroU64,
480 ) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, K, V>>), Error<F>> {
481 let op_count = self.bounds().end;
482 self.historical_proof(op_count, start_index, max_ops).await
483 }
484
485 #[tracing::instrument(name = "qmdb.immutable.db.prune", level = "info", skip_all)]
496 pub async fn prune(&mut self, loc: Location<F>) -> Result<(), Error<F>> {
497 let _timer = self.metrics.prune_timer();
498 self.metrics.prune_calls.inc();
499 if loc > self.inactivity_floor_loc {
500 return Err(Error::PruneBeyondMinRequired(
501 loc,
502 self.inactivity_floor_loc,
503 ));
504 }
505 self.journal.prune(loc).await?;
506 self.update_metrics();
507 Ok(())
508 }
509
510 #[tracing::instrument(name = "qmdb.immutable.db.rewind", level = "info", skip_all)]
530 pub async fn rewind(&mut self, size: Location<F>) -> Result<(), Error<F>> {
531 let rewind_size = *size;
532 let current_size = *self.last_commit_loc + 1;
533 if rewind_size == current_size {
534 return Ok(());
535 }
536 if rewind_size == 0 || rewind_size > current_size {
537 return Err(Error::Journal(crate::journal::Error::InvalidRewind(
538 rewind_size,
539 )));
540 }
541
542 let (rewind_last_loc, rewind_floor, rewound_keys) = {
543 let bounds = self.journal.bounds();
544 let rewind_last_loc = Location::new(rewind_size - 1);
545 if rewind_size <= bounds.start {
546 return Err(Error::Journal(crate::journal::Error::ItemPruned(
547 *rewind_last_loc,
548 )));
549 }
550 let rewind_last_op = self.journal.read(*rewind_last_loc).await?;
551 let Operation::Commit(_, rewind_floor) = &rewind_last_op else {
552 return Err(Error::UnexpectedData(rewind_last_loc));
553 };
554 let rewind_floor = *rewind_floor;
555 if *rewind_floor < bounds.start {
556 return Err(Error::Journal(crate::journal::Error::ItemPruned(
557 *rewind_floor,
558 )));
559 }
560
561 let mut rewound_keys = Vec::new();
562 for loc in rewind_size..current_size {
563 if let Operation::Set(key, _) = self.journal.read(loc).await? {
564 rewound_keys.push(key);
565 }
566 }
567
568 (rewind_last_loc, rewind_floor, rewound_keys)
569 };
570
571 let old_floor = self.inactivity_floor_loc;
572
573 self.journal.rewind(rewind_size).await?;
576
577 let rewind_loc = Location::<F>::new(rewind_size);
579 for key in &rewound_keys {
580 self.snapshot.retain(key, |loc| *loc < rewind_loc);
582 }
583
584 if rewind_floor < old_floor {
590 let gap_end = core::cmp::min(*old_floor, rewind_size);
591 for loc in *rewind_floor..gap_end {
592 if let Operation::Set(key, _) = self.journal.journal.read(loc).await? {
593 self.snapshot.insert(&key, Location::new(loc));
594 }
595 }
596 }
597
598 self.last_commit_loc = rewind_last_loc;
599 self.inactivity_floor_loc = rewind_floor;
600 let inactive_peaks = F::inactive_peaks(F::location_to_position(size), rewind_floor);
601 self.root = self.journal.root(inactive_peaks)?;
602 self.update_metrics();
603
604 Ok(())
605 }
606
607 pub const fn root(&self) -> H::Digest {
609 self.root
610 }
611
612 pub const fn strategy(&self) -> &S {
614 self.journal.strategy()
615 }
616
617 pub async fn pinned_nodes_at(&self, loc: Location<F>) -> Result<Vec<H::Digest>, Error<F>> {
619 self.journal
620 .merkle
621 .pinned_nodes_at(loc)
622 .await
623 .map_err(Into::into)
624 }
625
626 #[tracing::instrument(name = "qmdb.immutable.db.sync", level = "info", skip_all)]
630 pub async fn sync(&mut self) -> Result<(), Error<F>> {
631 let _timer = self.metrics.sync_timer();
632 self.metrics.sync_calls.inc();
633 self.journal.sync().await?;
634 Ok(())
635 }
636
637 #[tracing::instrument(name = "qmdb.immutable.db.commit", level = "info", skip_all)]
639 pub async fn commit(&mut self) -> Result<(), Error<F>> {
640 let _timer = self.metrics.commit_timer();
641 self.metrics.commit_calls.inc();
642 self.journal.commit().await?;
643 Ok(())
644 }
645
646 #[boxed]
648 pub async fn destroy(self) -> Result<(), Error<F>> {
649 Ok(self.journal.destroy().await?)
650 }
651
652 #[allow(clippy::type_complexity)]
654 pub fn new_batch(&self) -> batch::UnmerkleizedBatch<F, H, K, V, S> {
655 let journal_size = *self.last_commit_loc + 1;
656 batch::UnmerkleizedBatch::new(self, journal_size)
657 }
658
659 #[tracing::instrument(name = "qmdb.immutable.db.apply_batch", level = "info", skip_all)]
688 pub async fn apply_batch(
689 &mut self,
690 batch: Arc<batch::MerkleizedBatch<F, H::Digest, K, V, S>>,
691 ) -> Result<Range<Location<F>>, Error<F>> {
692 let _timer = self.metrics.apply_batch_timer();
693 self.metrics.apply_batch_calls.inc();
694 let db_size = *self.last_commit_loc + 1;
695 batch
696 .bounds
697 .validate_apply_to(db_size, self.inactivity_floor_loc)?;
698 let start_loc = Location::new(db_size);
699
700 self.journal.apply_batch(&batch.journal_batch).await?;
702
703 let bounds = self.journal.bounds();
709 let track_shadow = batch.bounds.ancestors.iter().any(|a| a.end > db_size);
710 let seen_cap = if track_shadow {
711 batch.diff.len()
712 + batch
713 .bounds
714 .ancestors
715 .iter()
716 .zip(&batch.ancestor_diffs)
717 .filter(|(a, _)| a.end > db_size)
718 .map(|(_, d)| d.len())
719 .sum::<usize>()
720 } else {
721 0
722 };
723 let mut seen: AHashSet<&K> = AHashSet::with_capacity(seen_cap);
724 for (key, entry) in batch.diff.iter() {
725 if track_shadow {
726 seen.insert(key);
727 }
728 self.snapshot
729 .insert_and_retain(key, entry.loc, |v| *v >= bounds.start);
730 }
731 for (i, ancestor_diff) in batch.ancestor_diffs.iter().enumerate() {
732 if batch.bounds.ancestors[i].end <= db_size {
733 continue;
734 }
735 for (key, entry) in ancestor_diff.iter() {
736 if seen.insert(key) {
737 self.snapshot
738 .insert_and_retain(key, entry.loc, |v| *v >= bounds.start);
739 }
740 }
741 }
742
743 self.last_commit_loc = Location::new(batch.bounds.total_size - 1);
745 self.inactivity_floor_loc = batch.bounds.inactivity_floor;
746 self.root = batch.root;
747 let range = start_loc..Location::new(batch.bounds.total_size);
748 self.update_metrics();
749 self.metrics
750 .operations_applied
751 .inc_by(*range.end - *range.start);
752 Ok(range)
753 }
754}
755
756#[cfg(test)]
757pub(super) mod test {
758 use super::*;
759 use crate::{
760 merkle::{Family, Location},
761 qmdb::verify_proof,
762 translator::TwoCap,
763 };
764 use commonware_codec::EncodeShared;
765 use commonware_cryptography::{sha256, sha256::Digest, Sha256};
766 use commonware_runtime::{deterministic, Supervisor as _};
767 use commonware_utils::NZU64;
768 use core::{future::Future, pin::Pin};
769 use std::ops::Range;
770
771 const ITEMS_PER_SECTION: u64 = 5;
772
773 type TestDb<F, V, C> = Immutable<
774 F,
775 deterministic::Context,
776 Digest,
777 V,
778 C,
779 Sha256,
780 TwoCap,
781 commonware_parallel::Sequential,
782 >;
783
784 #[boxed]
785 pub(crate) async fn test_immutable_empty<F: Family, V, C>(
786 context: deterministic::Context,
787 open_db: impl Fn(
788 deterministic::Context,
789 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
790 ) where
791 V: ValueEncoding<Value = Digest>,
792 C: Mutable<Item = Operation<F, Digest, V>>,
793 C::Item: EncodeShared,
794 {
795 let db = open_db(context.child("first")).await;
796 let bounds = db.bounds();
797 assert_eq!(bounds.end, 1);
798 assert_eq!(bounds.start, Location::new(0));
799 assert_eq!(db.inactivity_floor_loc(), Location::new(0));
800 assert!(db.get_metadata().await.unwrap().is_none());
801
802 let k1 = Sha256::fill(1u8);
804 let v1 = Sha256::fill(2u8);
805 let root = db.root();
806 {
807 let _batch = db.new_batch().set(k1, v1);
808 }
810 drop(db);
811 let mut db = open_db(context.child("second")).await;
812 assert_eq!(db.root(), root);
813 assert_eq!(db.bounds().end, 1);
814
815 db.apply_batch(db.new_batch().merkleize(&db, None, Location::new(0)).await)
817 .await
818 .unwrap();
819 db.commit().await.unwrap();
820 assert_eq!(db.bounds().end, 2); let root = db.root();
822 drop(db);
823
824 let db = open_db(context.child("third")).await;
825 assert_eq!(db.root(), root);
826
827 db.destroy().await.unwrap();
828 }
829
830 #[boxed]
831 pub(crate) async fn test_immutable_commit_after_sync_recovery<F: Family, V, C>(
832 context: deterministic::Context,
833 open_db: impl Fn(
834 deterministic::Context,
835 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
836 ) where
837 V: ValueEncoding<Value = Digest>,
838 C: Mutable<Item = Operation<F, Digest, V>>,
839 C::Item: EncodeShared,
840 {
841 let mut db = open_db(context.child("first")).await;
842 let k1 = Sha256::fill(1u8);
843 let k2 = Sha256::fill(2u8);
844 let v1 = Sha256::fill(3u8);
845 let v2 = Sha256::fill(4u8);
846
847 commit_sets(&mut db, [(k1, v1)], None).await;
849 db.sync().await.unwrap();
850
851 commit_sets(&mut db, [(k2, v2)], None).await;
853 let committed_bounds = db.bounds();
854 let committed_root = db.root();
855 drop(db);
856
857 let db = open_db(context.child("second")).await;
858 assert_eq!(db.bounds(), committed_bounds);
859 assert_eq!(db.root(), committed_root);
860 assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
861 assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
862
863 db.destroy().await.unwrap();
864 }
865
866 #[boxed]
867 pub(crate) async fn test_immutable_build_basic<F: Family, V, C>(
868 context: deterministic::Context,
869 open_db: impl Fn(
870 deterministic::Context,
871 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
872 ) where
873 V: ValueEncoding<Value = Digest>,
874 C: Mutable<Item = Operation<F, Digest, V>>,
875 C::Item: EncodeShared,
876 {
877 let mut db = open_db(context.child("first")).await;
879
880 let k1 = Sha256::fill(1u8);
881 let k2 = Sha256::fill(2u8);
882 let v1 = Sha256::fill(3u8);
883 let v2 = Sha256::fill(4u8);
884
885 assert!(db.get(&k1).await.unwrap().is_none());
886 assert!(db.get(&k2).await.unwrap().is_none());
887
888 let metadata = Some(Sha256::fill(99u8));
890 db.apply_batch(
891 db.new_batch()
892 .set(k1, v1)
893 .merkleize(&db, metadata, Location::new(0))
894 .await,
895 )
896 .await
897 .unwrap();
898 db.commit().await.unwrap();
899 assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
900 assert!(db.get(&k2).await.unwrap().is_none());
901 assert_eq!(db.bounds().end, 3);
902 assert_eq!(db.get_metadata().await.unwrap(), Some(Sha256::fill(99u8)));
903
904 db.apply_batch(
906 db.new_batch()
907 .set(k2, v2)
908 .merkleize(&db, None, Location::new(0))
909 .await,
910 )
911 .await
912 .unwrap();
913 db.commit().await.unwrap();
914 assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
915 assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);
916 assert_eq!(db.bounds().end, 5);
917 assert_eq!(db.get_metadata().await.unwrap(), None);
918
919 let root = db.root();
921
922 let k3 = Sha256::fill(5u8);
924 let v3 = Sha256::fill(6u8);
925 {
926 let _batch = db.new_batch().set(k3, v3);
927 }
929
930 drop(db); let db = open_db(context.child("second")).await;
933 assert!(db.get(&k3).await.unwrap().is_none());
934 assert_eq!(db.root(), root);
935 assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
936 assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);
937 assert_eq!(db.bounds().end, 5);
938 assert_eq!(db.get_metadata().await.unwrap(), None);
939
940 db.destroy().await.unwrap();
942 }
943
944 #[boxed]
945 pub(crate) async fn test_immutable_proof_verify<F: Family, V, C>(
946 context: deterministic::Context,
947 open_db: impl Fn(
948 deterministic::Context,
949 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
950 ) where
951 V: ValueEncoding<Value = Digest>,
952 C: Mutable<Item = Operation<F, Digest, V>>,
953 C::Item: EncodeShared,
954 {
955 let mut db = open_db(context.child("first")).await;
956
957 let k1 = Sha256::fill(1u8);
958 let v1 = Sha256::fill(10u8);
959 db.apply_batch(
960 db.new_batch()
961 .set(k1, v1)
962 .merkleize(&db, None, Location::new(0))
963 .await,
964 )
965 .await
966 .unwrap();
967 db.commit().await.unwrap();
968
969 let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap();
970 let root = db.root();
971 assert!(verify_proof::<Sha256, _, _>(
972 &proof,
973 Location::new(0),
974 &ops,
975 &root
976 ));
977
978 db.destroy().await.unwrap();
979 }
980
981 #[boxed]
982 pub(crate) async fn test_immutable_prune<F: Family, V, C>(
983 context: deterministic::Context,
984 open_db: impl Fn(
985 deterministic::Context,
986 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
987 ) where
988 V: ValueEncoding<Value = Digest>,
989 C: Mutable<Item = Operation<F, Digest, V>>,
990 C::Item: EncodeShared,
991 {
992 let mut db = open_db(context.child("first")).await;
993
994 for i in 0..20u8 {
995 let key = Sha256::fill(i);
996 let value = Sha256::fill(i.wrapping_add(100));
997 let floor = db.bounds().end;
998 db.apply_batch(
999 db.new_batch()
1000 .set(key, value)
1001 .merkleize(&db, None, floor)
1002 .await,
1003 )
1004 .await
1005 .unwrap();
1006 db.commit().await.unwrap();
1007 }
1008
1009 let root_before = db.root();
1010 let bounds_before = db.bounds();
1011
1012 let prune_loc = Location::new(*bounds_before.end - 5);
1013 db.prune(prune_loc).await.unwrap();
1014
1015 assert_eq!(db.root(), root_before);
1016
1017 let key_0 = Sha256::fill(0u8);
1018 assert!(db.get(&key_0).await.unwrap().is_none());
1019
1020 let key_19 = Sha256::fill(19u8);
1021 assert_eq!(
1022 db.get(&key_19).await.unwrap(),
1023 Some(Sha256::fill(19u8.wrapping_add(100)))
1024 );
1025
1026 db.destroy().await.unwrap();
1027 }
1028
1029 #[boxed]
1033 pub(crate) async fn test_immutable_prune_after_uncommitted_apply_batch_recovery<
1034 F: Family,
1035 V,
1036 C,
1037 >(
1038 context: deterministic::Context,
1039 open_db: impl Fn(
1040 deterministic::Context,
1041 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1042 ) where
1043 V: ValueEncoding<Value = Digest>,
1044 C: Mutable<Item = Operation<F, Digest, V>>,
1045 C::Item: EncodeShared,
1046 {
1047 let mut db = open_db(context.child("first")).await;
1048
1049 let mut batch = db.new_batch();
1052 for i in 0..6u8 {
1053 batch = batch.set(Sha256::fill(i), Sha256::fill(i.wrapping_add(10)));
1054 }
1055 db.apply_batch(batch.merkleize(&db, None, Location::new(0)).await)
1056 .await
1057 .unwrap();
1058 db.sync().await.unwrap();
1059 let durable_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
1060
1061 let buffered_floor = db.bounds().end;
1064 let key = Sha256::fill(100);
1065 let value = Sha256::fill(101);
1066 db.apply_batch(
1067 db.new_batch()
1068 .set(key, value)
1069 .merkleize(&db, None, buffered_floor)
1070 .await,
1071 )
1072 .await
1073 .unwrap();
1074 let buffered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
1075 assert_ne!(buffered_state, durable_state);
1076
1077 db.prune(buffered_floor).await.unwrap();
1078 assert!(db.bounds().start > Location::new(0));
1079 drop(db);
1080
1081 let db = open_db(context.child("second")).await;
1084 assert!(db.bounds().start <= db.inactivity_floor_loc());
1085 let recovered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
1086 assert!(
1087 recovered_state == durable_state || recovered_state == buffered_state,
1088 "recovered state is neither the durable baseline nor the buffered state"
1089 );
1090
1091 db.destroy().await.unwrap();
1092 }
1093
1094 #[boxed]
1095 pub(crate) async fn test_immutable_batch_chain<F: Family, V, C>(
1096 context: deterministic::Context,
1097 open_db: impl Fn(
1098 deterministic::Context,
1099 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1100 ) where
1101 V: ValueEncoding<Value = Digest>,
1102 C: Mutable<Item = Operation<F, Digest, V>>,
1103 C::Item: EncodeShared,
1104 {
1105 let mut db = open_db(context.child("first")).await;
1106
1107 let k1 = Sha256::fill(1u8);
1108 let k2 = Sha256::fill(2u8);
1109 let k3 = Sha256::fill(3u8);
1110 let v1 = Sha256::fill(11u8);
1111 let v2 = Sha256::fill(12u8);
1112 let v3 = Sha256::fill(13u8);
1113
1114 let parent = db
1115 .new_batch()
1116 .set(k1, v1)
1117 .merkleize(&db, None, Location::new(0))
1118 .await;
1119 let child = parent
1120 .new_batch::<Sha256>()
1121 .set(k2, v2)
1122 .merkleize(&db, None, Location::new(0))
1123 .await;
1124
1125 assert_eq!(child.get(&k1, &db).await.unwrap(), Some(v1));
1126 assert_eq!(child.get(&k2, &db).await.unwrap(), Some(v2));
1127 assert!(child.get(&k3, &db).await.unwrap().is_none());
1128
1129 db.apply_batch(child).await.unwrap();
1130 db.commit().await.unwrap();
1131
1132 assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
1133 assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
1134
1135 db.apply_batch(
1136 db.new_batch()
1137 .set(k3, v3)
1138 .merkleize(&db, None, Location::new(0))
1139 .await,
1140 )
1141 .await
1142 .unwrap();
1143 db.commit().await.unwrap();
1144 assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
1145
1146 db.destroy().await.unwrap();
1147 }
1148
1149 #[boxed]
1150 pub(crate) async fn test_immutable_build_and_authenticate<F: Family, V, C>(
1151 context: deterministic::Context,
1152 open_db: impl Fn(
1153 deterministic::Context,
1154 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1155 ) where
1156 V: ValueEncoding<Value = Digest>,
1157 C: Mutable<Item = Operation<F, Digest, V>>,
1158 C::Item: EncodeShared,
1159 {
1160 let mut db = open_db(context.child("first")).await;
1162
1163 let mut batch = db.new_batch();
1164 for i in 0u64..2_000 {
1165 let k = Sha256::hash(&i.to_be_bytes());
1166 let v = Sha256::fill(i as u8);
1167 batch = batch.set(k, v);
1168 }
1169 let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1170 db.apply_batch(merkleized).await.unwrap();
1171 db.commit().await.unwrap();
1172 assert_eq!(db.bounds().end, 2_000 + 2);
1173
1174 let root = db.root();
1176 drop(db);
1177
1178 let db = open_db(context.child("second")).await;
1179 assert_eq!(root, db.root());
1180 assert_eq!(db.bounds().end, 2_000 + 2);
1181 for i in 0u64..2_000 {
1182 let k = Sha256::hash(&i.to_be_bytes());
1183 let v = Sha256::fill(i as u8);
1184 assert_eq!(db.get(&k).await.unwrap().unwrap(), v);
1185 }
1186
1187 let max_ops = NZU64!(5);
1190 for i in 0..*db.bounds().end {
1191 let (proof, log) = db.proof(Location::new(i), max_ops).await.unwrap();
1192 assert!(verify_proof::<Sha256, _, _>(
1193 &proof,
1194 Location::new(i),
1195 &log,
1196 &root
1197 ));
1198 }
1199
1200 db.destroy().await.unwrap();
1201 }
1202
1203 #[boxed]
1204 pub(crate) async fn test_immutable_recovery_from_failed_merkle_sync<F: Family, V, C>(
1205 context: deterministic::Context,
1206 open_db: impl Fn(
1207 deterministic::Context,
1208 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1209 ) where
1210 V: ValueEncoding<Value = Digest>,
1211 C: Mutable<Item = Operation<F, Digest, V>>,
1212 C::Item: EncodeShared,
1213 {
1214 const ELEMENTS: u64 = 1000;
1216 let mut db = open_db(context.child("first")).await;
1217
1218 let mut batch = db.new_batch();
1219 for i in 0u64..ELEMENTS {
1220 let k = Sha256::hash(&i.to_be_bytes());
1221 let v = Sha256::fill(i as u8);
1222 batch = batch.set(k, v);
1223 }
1224 let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1225 db.apply_batch(merkleized).await.unwrap();
1226 db.commit().await.unwrap();
1227 assert_eq!(db.bounds().end, ELEMENTS + 2);
1228 db.sync().await.unwrap();
1229 let halfway_root = db.root();
1230
1231 let mut batch = db.new_batch();
1233 for i in ELEMENTS..ELEMENTS * 2 {
1234 let k = Sha256::hash(&i.to_be_bytes());
1235 let v = Sha256::fill(i as u8);
1236 batch = batch.set(k, v);
1237 }
1238 let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1239 db.apply_batch(merkleized).await.unwrap();
1240 db.commit().await.unwrap();
1241 drop(db); let db = open_db(context.child("second")).await;
1246 assert_eq!(db.bounds().end, 2003);
1247 let root = db.root();
1248 assert_ne!(root, halfway_root);
1249
1250 drop(db);
1252 let db = open_db(context.child("third")).await;
1253 assert_eq!(db.bounds().end, 2003);
1254 assert_eq!(db.root(), root);
1255
1256 db.destroy().await.unwrap();
1257 }
1258
1259 #[boxed]
1260 pub(crate) async fn test_immutable_recovery_from_failed_log_sync<F: Family, V, C>(
1261 context: deterministic::Context,
1262 open_db: impl Fn(
1263 deterministic::Context,
1264 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1265 ) where
1266 V: ValueEncoding<Value = Digest>,
1267 C: Mutable<Item = Operation<F, Digest, V>>,
1268 C::Item: EncodeShared,
1269 {
1270 let mut db = open_db(context.child("first")).await;
1271
1272 let k1 = Sha256::fill(1u8);
1274 let v1 = Sha256::fill(3u8);
1275 db.apply_batch(
1276 db.new_batch()
1277 .set(k1, v1)
1278 .merkleize(&db, None, Location::new(0))
1279 .await,
1280 )
1281 .await
1282 .unwrap();
1283 db.commit().await.unwrap();
1284 let first_commit_root = db.root();
1285
1286 drop(db);
1289
1290 let db = open_db(context.child("second")).await;
1292 assert_eq!(db.bounds().end, 3);
1293 let root = db.root();
1294 assert_eq!(root, first_commit_root);
1295
1296 db.destroy().await.unwrap();
1297 }
1298
1299 #[boxed]
1300 pub(crate) async fn test_immutable_pruning<F: Family, V, C>(
1301 context: deterministic::Context,
1302 open_db: impl Fn(
1303 deterministic::Context,
1304 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1305 ) where
1306 V: ValueEncoding<Value = Digest>,
1307 C: Mutable<Item = Operation<F, Digest, V>>,
1308 C::Item: EncodeShared,
1309 {
1310 const ELEMENTS: u64 = 2_000;
1312 let mut db = open_db(context.child("first")).await;
1313
1314 let mut sorted_keys: Vec<sha256::Digest> = (1u64..ELEMENTS + 1)
1317 .map(|i| Sha256::hash(&i.to_be_bytes()))
1318 .collect();
1319 sorted_keys.sort();
1320 let mut batch = db.new_batch();
1325 for i in 1u64..ELEMENTS + 1 {
1326 let k = Sha256::hash(&i.to_be_bytes());
1327 let v = Sha256::fill(i as u8);
1328 batch = batch.set(k, v);
1329 }
1330 let inactivity_floor = Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1);
1333 let merkleized = batch.merkleize(&db, None, inactivity_floor).await;
1334 db.apply_batch(merkleized).await.unwrap();
1335 assert_eq!(db.bounds().end, ELEMENTS + 2);
1336
1337 db.prune(Location::new((ELEMENTS + 2) / 2)).await.unwrap();
1339 let bounds = db.bounds();
1340 assert_eq!(bounds.end, ELEMENTS + 2);
1341
1342 let oldest_retained_loc = bounds.start;
1345 assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2));
1346
1347 let pruned_key = sorted_keys[*oldest_retained_loc as usize - 2];
1349 assert!(db.get(&pruned_key).await.unwrap().is_none());
1350
1351 let unpruned_key = sorted_keys[*oldest_retained_loc as usize - 1];
1353 assert!(db.get(&unpruned_key).await.unwrap().is_some());
1354
1355 let root = db.root();
1357 db.sync().await.unwrap();
1358 drop(db);
1359
1360 let mut db = open_db(context.child("second")).await;
1361 assert_eq!(root, db.root());
1362 let bounds = db.bounds();
1363 assert_eq!(bounds.end, ELEMENTS + 2);
1364 let oldest_retained_loc = bounds.start;
1365 assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2));
1366
1367 let loc = Location::new(ELEMENTS / 2 + (ITEMS_PER_SECTION * 2 - 1));
1369 db.prune(loc).await.unwrap();
1370 let oldest_retained_loc = db.bounds().start;
1372 assert_eq!(
1373 oldest_retained_loc,
1374 Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION)
1375 );
1376
1377 db.sync().await.unwrap();
1379 drop(db);
1380 let db = open_db(context.child("third")).await;
1381 let oldest_retained_loc = db.bounds().start;
1382 assert_eq!(
1383 oldest_retained_loc,
1384 Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION)
1385 );
1386
1387 let floor_val = ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1;
1389 let inactive_key = sorted_keys[floor_val as usize - 2];
1390 assert!(db.get(&inactive_key).await.unwrap().is_none());
1391
1392 let active_key = sorted_keys[floor_val as usize - 1];
1394 assert!(db.get(&active_key).await.unwrap().is_some());
1395
1396 let pruned_pos = ELEMENTS / 2;
1398 let proof_result = db
1399 .proof(Location::new(pruned_pos), NZU64!(pruned_pos + 100))
1400 .await;
1401 assert!(
1402 matches!(proof_result, Err(Error::Journal(crate::journal::Error::ItemPruned(pos))) if pos == pruned_pos)
1403 );
1404
1405 db.destroy().await.unwrap();
1406 }
1407
1408 #[boxed]
1409 pub(crate) async fn test_immutable_prune_beyond_floor<F: Family, V, C>(
1410 context: deterministic::Context,
1411 open_db: impl Fn(
1412 deterministic::Context,
1413 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1414 ) where
1415 V: ValueEncoding<Value = Digest>,
1416 C: Mutable<Item = Operation<F, Digest, V>>,
1417 C::Item: EncodeShared,
1418 {
1419 let mut db = open_db(context.child("test")).await;
1420
1421 let result = db.prune(Location::new(1)).await;
1423 assert!(
1424 matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor))
1425 if prune_loc == Location::new(1) && floor == Location::new(0))
1426 );
1427
1428 let k1 = Digest::from(*b"12345678901234567890123456789012");
1430 let k2 = Digest::from(*b"abcdefghijklmnopqrstuvwxyz123456");
1431 let k3 = Digest::from(*b"99999999999999999999999999999999");
1432 let v1 = Sha256::fill(1u8);
1433 let v2 = Sha256::fill(2u8);
1434 let v3 = Sha256::fill(3u8);
1435
1436 db.apply_batch(
1438 db.new_batch()
1439 .set(k1, v1)
1440 .set(k2, v2)
1441 .merkleize(&db, None, Location::new(3))
1442 .await,
1443 )
1444 .await
1445 .unwrap();
1446
1447 assert_eq!(*db.last_commit_loc, 3);
1449
1450 db.apply_batch(
1452 db.new_batch()
1453 .set(k3, v3)
1454 .merkleize(&db, None, Location::new(5))
1455 .await,
1456 )
1457 .await
1458 .unwrap();
1459
1460 assert!(db.prune(Location::new(3)).await.is_ok());
1462
1463 let floor = db.inactivity_floor_loc();
1465 let beyond = floor + 1;
1466 let result = db.prune(beyond).await;
1467 assert!(
1468 matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, f))
1469 if prune_loc == beyond && f == floor)
1470 );
1471
1472 db.destroy().await.unwrap();
1473 }
1474
1475 async fn commit_sets<F: Family, V, C>(
1476 db: &mut TestDb<F, V, C>,
1477 sets: impl IntoIterator<Item = (Digest, V::Value)>,
1478 metadata: Option<V::Value>,
1479 ) -> Range<Location<F>>
1480 where
1481 V: ValueEncoding<Value = Digest>,
1482 C: Mutable<Item = Operation<F, Digest, V>>,
1483 C::Item: EncodeShared,
1484 {
1485 commit_sets_with_floor(db, sets, metadata, Location::new(0)).await
1486 }
1487
1488 async fn commit_sets_with_floor<F: Family, V, C>(
1489 db: &mut TestDb<F, V, C>,
1490 sets: impl IntoIterator<Item = (Digest, V::Value)>,
1491 metadata: Option<V::Value>,
1492 floor: Location<F>,
1493 ) -> Range<Location<F>>
1494 where
1495 V: ValueEncoding<Value = Digest>,
1496 C: Mutable<Item = Operation<F, Digest, V>>,
1497 C::Item: EncodeShared,
1498 {
1499 let mut batch = db.new_batch();
1500 for (key, value) in sets {
1501 batch = batch.set(key, value);
1502 }
1503 let range = db
1504 .apply_batch(batch.merkleize(db, metadata, floor).await)
1505 .await
1506 .unwrap();
1507 db.commit().await.unwrap();
1508 range
1509 }
1510
1511 #[boxed]
1512 pub(crate) async fn test_immutable_rewind_recovery<F: Family, V, C>(
1513 context: deterministic::Context,
1514 open_db: impl Fn(
1515 deterministic::Context,
1516 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1517 ) where
1518 V: ValueEncoding<Value = Digest>,
1519 C: Mutable<Item = Operation<F, Digest, V>>,
1520 C::Item: EncodeShared,
1521 {
1522 let mut db = open_db(context.child("db")).await;
1523
1524 let key1 = Sha256::hash(&1u64.to_be_bytes());
1525 let key2 = Sha256::hash(&2u64.to_be_bytes());
1526 let key3 = Sha256::hash(&3u64.to_be_bytes());
1527 let key4 = Sha256::hash(&4u64.to_be_bytes());
1528
1529 let value1 = Sha256::fill(11u8);
1530 let value2 = Sha256::fill(22u8);
1531 let value3 = Sha256::fill(33u8);
1532 let value4 = Sha256::fill(66u8);
1533
1534 let metadata_a = Sha256::fill(44u8);
1535 let first_range =
1536 commit_sets(&mut db, [(key1, value1), (key2, value2)], Some(metadata_a)).await;
1537 let size_before = db.bounds().end;
1538 let root_before = db.root();
1539 let last_commit_before = db.last_commit_loc;
1540 assert_eq!(size_before, first_range.end);
1541
1542 let metadata_b = Sha256::fill(55u8);
1543 let second_range =
1544 commit_sets(&mut db, [(key3, value3), (key4, value4)], Some(metadata_b)).await;
1545 assert_eq!(second_range.start, size_before);
1546 assert_ne!(db.root(), root_before);
1547 assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_b));
1548 assert_eq!(db.get(&key3).await.unwrap(), Some(value3));
1549 assert_eq!(db.get(&key4).await.unwrap(), Some(value4));
1550
1551 db.rewind(size_before).await.unwrap();
1552 assert_eq!(db.root(), root_before);
1553 assert_eq!(db.bounds().end, size_before);
1554 assert_eq!(db.last_commit_loc, last_commit_before);
1555 assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
1556 assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1557 assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
1558 assert_eq!(db.get(&key3).await.unwrap(), None);
1559 assert_eq!(db.get(&key4).await.unwrap(), None);
1560
1561 db.commit().await.unwrap();
1562 drop(db);
1563 let db = open_db(context.child("reopen")).await;
1564 assert_eq!(db.root(), root_before);
1565 assert_eq!(db.bounds().end, size_before);
1566 assert_eq!(db.last_commit_loc, last_commit_before);
1567 assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
1568 assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1569 assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
1570 assert_eq!(db.get(&key3).await.unwrap(), None);
1571 assert_eq!(db.get(&key4).await.unwrap(), None);
1572
1573 db.destroy().await.unwrap();
1574 }
1575
1576 #[boxed]
1580 pub(crate) async fn test_immutable_rewind_preserves_collision_bucket<F: Family, V, C>(
1581 context: deterministic::Context,
1582 open_db: impl Fn(
1583 deterministic::Context,
1584 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1585 ) where
1586 V: ValueEncoding<Value = Digest>,
1587 C: Mutable<Item = Operation<F, Digest, V>>,
1588 C::Item: EncodeShared,
1589 {
1590 let mut db = open_db(context.child("db")).await;
1591
1592 let mut k1_bytes = [0u8; 32];
1594 let mut k2_bytes = [0u8; 32];
1595 k1_bytes[0] = 0xAA;
1596 k1_bytes[1] = 0xBB;
1597 k2_bytes[0] = 0xAA;
1598 k2_bytes[1] = 0xBB;
1599 k1_bytes[31] = 0x01;
1600 k2_bytes[31] = 0x02;
1601 let key1 = Digest::from(k1_bytes);
1602 let key2 = Digest::from(k2_bytes);
1603 let value1 = Sha256::fill(11u8);
1604 let value2 = Sha256::fill(22u8);
1605
1606 commit_sets(&mut db, [(key1, value1)], None).await;
1607 let size_after_first = db.bounds().end;
1608 commit_sets(&mut db, [(key2, value2)], None).await;
1609 assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1610 assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
1611
1612 db.rewind(size_after_first).await.unwrap();
1613
1614 assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
1617 assert_eq!(db.get(&key2).await.unwrap(), None);
1618
1619 db.destroy().await.unwrap();
1620 }
1621
1622 #[boxed]
1623 pub(crate) async fn test_immutable_rewind_pruned_target_errors<F: Family, V, C>(
1624 context: deterministic::Context,
1625 open_small_sections_db: impl Fn(
1626 deterministic::Context,
1627 )
1628 -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1629 ) where
1630 V: ValueEncoding<Value = Digest>,
1631 C: Mutable<Item = Operation<F, Digest, V>>,
1632 C::Item: EncodeShared,
1633 {
1634 let mut db = open_small_sections_db(context.child("db")).await;
1635
1636 let first_range = commit_sets(
1637 &mut db,
1638 (0u64..16).map(|i| (Sha256::hash(&i.to_be_bytes()), Sha256::fill(i as u8))),
1639 None,
1640 )
1641 .await;
1642
1643 let mut round = 0u64;
1644 loop {
1645 round += 1;
1646 assert!(
1647 round <= 64,
1648 "failed to prune enough history for rewind test"
1649 );
1650
1651 let floor = Location::new(*db.bounds().end + 16);
1654 commit_sets_with_floor(
1655 &mut db,
1656 (0u64..16).map(|i| {
1657 let seed = round * 100 + i;
1658 (Sha256::hash(&seed.to_be_bytes()), Sha256::fill(seed as u8))
1659 }),
1660 None,
1661 floor,
1662 )
1663 .await;
1664 db.prune(db.last_commit_loc).await.unwrap();
1665
1666 if db.bounds().start > first_range.start {
1667 break;
1668 }
1669 }
1670
1671 let oldest_retained = db.bounds().start;
1672 let boundary_err = db.rewind(oldest_retained).await.unwrap_err();
1673 assert!(
1674 matches!(
1675 boundary_err,
1676 Error::Journal(crate::journal::Error::ItemPruned(_))
1677 ),
1678 "unexpected rewind error at retained boundary: {boundary_err:?}"
1679 );
1680
1681 let err = db.rewind(first_range.start).await.unwrap_err();
1682 assert!(
1683 matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
1684 "unexpected rewind error: {err:?}"
1685 );
1686
1687 db.destroy().await.unwrap();
1688 }
1689
1690 #[boxed]
1692 pub(crate) async fn test_immutable_batch_get_read_through<F: Family, V, C>(
1693 context: deterministic::Context,
1694 open_db: impl Fn(
1695 deterministic::Context,
1696 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1697 ) where
1698 V: ValueEncoding<Value = Digest>,
1699 C: Mutable<Item = Operation<F, Digest, V>>,
1700 C::Item: EncodeShared,
1701 {
1702 let mut db = open_db(context.child("db")).await;
1703
1704 let key_a = Sha256::hash(&0u64.to_be_bytes());
1706 let val_a = Sha256::fill(1u8);
1707 db.apply_batch(
1708 db.new_batch()
1709 .set(key_a, val_a)
1710 .merkleize(&db, None, Location::new(0))
1711 .await,
1712 )
1713 .await
1714 .unwrap();
1715
1716 let mut batch = db.new_batch();
1718 assert_eq!(batch.get(&key_a, &db).await.unwrap(), Some(val_a));
1719
1720 let key_b = Sha256::hash(&1u64.to_be_bytes());
1722 let val_b = Sha256::fill(2u8);
1723 batch = batch.set(key_b, val_b);
1724 assert_eq!(batch.get(&key_b, &db).await.unwrap(), Some(val_b));
1725
1726 let key_c = Sha256::hash(&2u64.to_be_bytes());
1728 assert_eq!(batch.get(&key_c, &db).await.unwrap(), None);
1729
1730 db.destroy().await.unwrap();
1731 }
1732
1733 #[boxed]
1735 pub(crate) async fn test_immutable_batch_stacked_get<F: Family, V, C>(
1736 context: deterministic::Context,
1737 open_db: impl Fn(
1738 deterministic::Context,
1739 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1740 ) where
1741 V: ValueEncoding<Value = Digest>,
1742 C: Mutable<Item = Operation<F, Digest, V>>,
1743 C::Item: EncodeShared,
1744 {
1745 let db = open_db(context.child("db")).await;
1746
1747 let key_a = Sha256::hash(&0u64.to_be_bytes());
1749 let val_a = Sha256::fill(10u8);
1750 let parent = db.new_batch().set(key_a, val_a);
1751 let parent_m = parent.merkleize(&db, None, Location::new(0)).await;
1752
1753 let mut child = parent_m.new_batch::<Sha256>();
1755 assert_eq!(child.get(&key_a, &db).await.unwrap(), Some(val_a));
1756
1757 let key_b = Sha256::hash(&1u64.to_be_bytes());
1759 let val_b = Sha256::fill(20u8);
1760 child = child.set(key_b, val_b);
1761 assert_eq!(child.get(&key_b, &db).await.unwrap(), Some(val_b));
1762
1763 let key_c = Sha256::hash(&2u64.to_be_bytes());
1765 assert_eq!(child.get(&key_c, &db).await.unwrap(), None);
1766
1767 db.destroy().await.unwrap();
1768 }
1769
1770 #[boxed]
1772 pub(crate) async fn test_immutable_batch_stacked_apply<F: Family, V, C>(
1773 context: deterministic::Context,
1774 open_db: impl Fn(
1775 deterministic::Context,
1776 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1777 ) where
1778 V: ValueEncoding<Value = Digest>,
1779 C: Mutable<Item = Operation<F, Digest, V>>,
1780 C::Item: EncodeShared,
1781 {
1782 let mut db = open_db(context.child("db")).await;
1783
1784 let mut kvs_first: Vec<(Digest, Digest)> = (0u64..5)
1786 .map(|i| (Sha256::hash(&i.to_be_bytes()), Sha256::fill(i as u8)))
1787 .collect();
1788 kvs_first.sort_by_key(|a| a.0);
1789
1790 let mut kvs_second: Vec<(Digest, Digest)> = (5u64..10)
1791 .map(|i| (Sha256::hash(&i.to_be_bytes()), Sha256::fill(i as u8)))
1792 .collect();
1793 kvs_second.sort_by_key(|a| a.0);
1794
1795 let mut parent = db.new_batch();
1797 for (k, v) in &kvs_first {
1798 parent = parent.set(*k, *v);
1799 }
1800 let parent_m = parent.merkleize(&db, None, Location::new(0)).await;
1801
1802 let mut child = parent_m.new_batch::<Sha256>();
1804 for (k, v) in &kvs_second {
1805 child = child.set(*k, *v);
1806 }
1807 let child_m = child.merkleize(&db, None, Location::new(0)).await;
1808 let expected_root = child_m.root();
1809 db.apply_batch(child_m).await.unwrap();
1810
1811 assert_eq!(db.root(), expected_root);
1812
1813 for (k, v) in kvs_first.iter().chain(kvs_second.iter()) {
1815 assert_eq!(db.get(k).await.unwrap(), Some(*v));
1816 }
1817
1818 db.destroy().await.unwrap();
1819 }
1820
1821 #[boxed]
1823 pub(crate) async fn test_immutable_batch_speculative_root<F: Family, V, C>(
1824 context: deterministic::Context,
1825 open_db: impl Fn(
1826 deterministic::Context,
1827 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1828 ) where
1829 V: ValueEncoding<Value = Digest>,
1830 C: Mutable<Item = Operation<F, Digest, V>>,
1831 C::Item: EncodeShared,
1832 {
1833 let mut db = open_db(context.child("db")).await;
1834
1835 let mut batch = db.new_batch();
1836 for i in 0u8..10 {
1837 let k = Sha256::hash(&[i]);
1838 batch = batch.set(k, Sha256::fill(i));
1839 }
1840 let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1841
1842 let speculative = merkleized.root();
1843 db.apply_batch(merkleized).await.unwrap();
1844 assert_eq!(db.root(), speculative);
1845
1846 let metadata = Some(Sha256::fill(55u8));
1848 let mut batch = db.new_batch();
1849 let k = Sha256::hash(&[0xAA]);
1850 batch = batch.set(k, Sha256::fill(0xAA));
1851 let merkleized = batch.merkleize(&db, metadata, Location::new(0)).await;
1852 let speculative = merkleized.root();
1853 db.apply_batch(merkleized).await.unwrap();
1854 assert_eq!(db.root(), speculative);
1855
1856 db.destroy().await.unwrap();
1857 }
1858
1859 #[boxed]
1861 pub(crate) async fn test_immutable_merkleized_batch_get<F: Family, V, C>(
1862 context: deterministic::Context,
1863 open_db: impl Fn(
1864 deterministic::Context,
1865 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1866 ) where
1867 V: ValueEncoding<Value = Digest>,
1868 C: Mutable<Item = Operation<F, Digest, V>>,
1869 C::Item: EncodeShared,
1870 {
1871 let mut db = open_db(context.child("db")).await;
1872
1873 let key_a = Sha256::hash(&0u64.to_be_bytes());
1875 let val_a = Sha256::fill(10u8);
1876 db.apply_batch(
1877 db.new_batch()
1878 .set(key_a, val_a)
1879 .merkleize(&db, None, Location::new(0))
1880 .await,
1881 )
1882 .await
1883 .unwrap();
1884
1885 let key_b = Sha256::hash(&1u64.to_be_bytes());
1887 let val_b = Sha256::fill(20u8);
1888 let merkleized = db
1889 .new_batch()
1890 .set(key_b, val_b)
1891 .merkleize(&db, None, Location::new(0))
1892 .await;
1893
1894 assert_eq!(merkleized.get(&key_a, &db).await.unwrap(), Some(val_a));
1896
1897 assert_eq!(merkleized.get(&key_b, &db).await.unwrap(), Some(val_b));
1899
1900 let key_c = Sha256::hash(&2u64.to_be_bytes());
1902 assert_eq!(merkleized.get(&key_c, &db).await.unwrap(), None);
1903
1904 db.destroy().await.unwrap();
1905 }
1906
1907 #[boxed]
1909 pub(crate) async fn test_immutable_batch_sequential_apply<F: Family, V, C>(
1910 context: deterministic::Context,
1911 open_db: impl Fn(
1912 deterministic::Context,
1913 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1914 ) where
1915 V: ValueEncoding<Value = Digest>,
1916 C: Mutable<Item = Operation<F, Digest, V>>,
1917 C::Item: EncodeShared,
1918 {
1919 let mut db = open_db(context.child("db")).await;
1920
1921 let key_a = Sha256::hash(&0u64.to_be_bytes());
1922 let val_a = Sha256::fill(1u8);
1923
1924 let m = db
1926 .new_batch()
1927 .set(key_a, val_a)
1928 .merkleize(&db, None, Location::new(0))
1929 .await;
1930 let root1 = m.root();
1931 db.apply_batch(m).await.unwrap();
1932 assert_eq!(db.root(), root1);
1933 assert_eq!(db.get(&key_a).await.unwrap(), Some(val_a));
1934
1935 let key_b = Sha256::hash(&1u64.to_be_bytes());
1937 let val_b = Sha256::fill(2u8);
1938 let m = db
1939 .new_batch()
1940 .set(key_b, val_b)
1941 .merkleize(&db, None, Location::new(0))
1942 .await;
1943 let root2 = m.root();
1944 db.apply_batch(m).await.unwrap();
1945 assert_eq!(db.root(), root2);
1946 assert_eq!(db.get(&key_b).await.unwrap(), Some(val_b));
1947
1948 db.destroy().await.unwrap();
1949 }
1950
1951 #[boxed]
1953 pub(crate) async fn test_immutable_batch_many_sequential<F: Family, V, C>(
1954 context: deterministic::Context,
1955 open_db: impl Fn(
1956 deterministic::Context,
1957 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
1958 ) where
1959 V: ValueEncoding<Value = Digest>,
1960 C: Mutable<Item = Operation<F, Digest, V>>,
1961 C::Item: EncodeShared,
1962 {
1963 let mut db = open_db(context.child("db")).await;
1964
1965 const BATCHES: u64 = 20;
1966 const KEYS_PER_BATCH: u64 = 5;
1967
1968 let mut all_kvs: Vec<(Digest, Digest)> = Vec::new();
1969
1970 for batch_idx in 0..BATCHES {
1971 let mut batch = db.new_batch();
1972 for j in 0..KEYS_PER_BATCH {
1973 let seed = batch_idx * 100 + j;
1974 let k = Sha256::hash(&seed.to_be_bytes());
1975 let v = Sha256::fill(seed as u8);
1976 batch = batch.set(k, v);
1977 all_kvs.push((k, v));
1978 }
1979 let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
1980 db.apply_batch(merkleized).await.unwrap();
1981 }
1982
1983 for (k, v) in &all_kvs {
1985 assert_eq!(db.get(k).await.unwrap(), Some(*v));
1986 }
1987
1988 let root = db.root();
1990 let (proof, ops) = db.proof(Location::new(0), NZU64!(10000)).await.unwrap();
1991 assert!(verify_proof::<Sha256, _, _>(
1992 &proof,
1993 Location::new(0),
1994 &ops,
1995 &root
1996 ));
1997
1998 let expected = 1 + BATCHES * (KEYS_PER_BATCH + 1);
2000 assert_eq!(db.bounds().end, expected);
2001
2002 db.destroy().await.unwrap();
2003 }
2004
2005 #[boxed]
2007 pub(crate) async fn test_immutable_batch_empty_batch<F: Family, V, C>(
2008 context: deterministic::Context,
2009 open_db: impl Fn(
2010 deterministic::Context,
2011 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2012 ) where
2013 V: ValueEncoding<Value = Digest>,
2014 C: Mutable<Item = Operation<F, Digest, V>>,
2015 C::Item: EncodeShared,
2016 {
2017 let mut db = open_db(context.child("db")).await;
2018
2019 let k = Sha256::hash(&[1u8]);
2021 db.apply_batch(
2022 db.new_batch()
2023 .set(k, Sha256::fill(1u8))
2024 .merkleize(&db, None, Location::new(0))
2025 .await,
2026 )
2027 .await
2028 .unwrap();
2029 let root_before = db.root();
2030 let size_before = db.bounds().end;
2031
2032 let merkleized = db.new_batch().merkleize(&db, None, Location::new(0)).await;
2034 let speculative = merkleized.root();
2035 db.apply_batch(merkleized).await.unwrap();
2036
2037 assert_ne!(db.root(), root_before);
2039 assert_eq!(db.root(), speculative);
2040 assert_eq!(db.bounds().end, size_before + 1);
2042
2043 db.destroy().await.unwrap();
2044 }
2045
2046 #[boxed]
2048 pub(crate) async fn test_immutable_batch_chained_merkleized_get<F: Family, V, C>(
2049 context: deterministic::Context,
2050 open_db: impl Fn(
2051 deterministic::Context,
2052 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2053 ) where
2054 V: ValueEncoding<Value = Digest>,
2055 C: Mutable<Item = Operation<F, Digest, V>>,
2056 C::Item: EncodeShared,
2057 {
2058 let mut db = open_db(context.child("db")).await;
2059
2060 let key_a = Sha256::hash(&0u64.to_be_bytes());
2062 let val_a = Sha256::fill(10u8);
2063 db.apply_batch(
2064 db.new_batch()
2065 .set(key_a, val_a)
2066 .merkleize(&db, None, Location::new(0))
2067 .await,
2068 )
2069 .await
2070 .unwrap();
2071
2072 let key_b = Sha256::hash(&1u64.to_be_bytes());
2074 let val_b = Sha256::fill(1u8);
2075 let parent_m = db
2076 .new_batch()
2077 .set(key_b, val_b)
2078 .merkleize(&db, None, Location::new(0))
2079 .await;
2080
2081 let key_c = Sha256::hash(&2u64.to_be_bytes());
2083 let val_c = Sha256::fill(2u8);
2084 let child_m = parent_m
2085 .new_batch::<Sha256>()
2086 .set(key_c, val_c)
2087 .merkleize(&db, None, Location::new(0))
2088 .await;
2089
2090 assert_eq!(child_m.get(&key_a, &db).await.unwrap(), Some(val_a));
2093 assert_eq!(child_m.get(&key_b, &db).await.unwrap(), Some(val_b));
2095 assert_eq!(child_m.get(&key_c, &db).await.unwrap(), Some(val_c));
2097 let key_d = Sha256::hash(&3u64.to_be_bytes());
2099 assert_eq!(child_m.get(&key_d, &db).await.unwrap(), None);
2100
2101 db.destroy().await.unwrap();
2102 }
2103
2104 #[boxed]
2106 pub(crate) async fn test_immutable_batch_large<F: Family, V, C>(
2107 context: deterministic::Context,
2108 open_db: impl Fn(
2109 deterministic::Context,
2110 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2111 ) where
2112 V: ValueEncoding<Value = Digest>,
2113 C: Mutable<Item = Operation<F, Digest, V>>,
2114 C::Item: EncodeShared,
2115 {
2116 let mut db = open_db(context.child("db")).await;
2117
2118 const N: u64 = 500;
2119 let mut kvs: Vec<(Digest, Digest)> = Vec::new();
2120
2121 let mut batch = db.new_batch();
2122 for i in 0..N {
2123 let k = Sha256::hash(&i.to_be_bytes());
2124 let v = Sha256::fill((i % 256) as u8);
2125 batch = batch.set(k, v);
2126 kvs.push((k, v));
2127 }
2128 let merkleized = batch.merkleize(&db, None, Location::new(0)).await;
2129 db.apply_batch(merkleized).await.unwrap();
2130
2131 for (k, v) in &kvs {
2133 assert_eq!(db.get(k).await.unwrap(), Some(*v));
2134 }
2135
2136 let root = db.root();
2138 let (proof, ops) = db.proof(Location::new(0), NZU64!(1000)).await.unwrap();
2139 assert!(verify_proof::<Sha256, _, _>(
2140 &proof,
2141 Location::new(0),
2142 &ops,
2143 &root
2144 ));
2145
2146 assert_eq!(db.bounds().end, 1 + N + 1);
2148
2149 db.destroy().await.unwrap();
2150 }
2151
2152 #[boxed]
2154 pub(crate) async fn test_immutable_batch_chained_key_override<F: Family, V, C>(
2155 context: deterministic::Context,
2156 open_db: impl Fn(
2157 deterministic::Context,
2158 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2159 ) where
2160 V: ValueEncoding<Value = Digest>,
2161 C: Mutable<Item = Operation<F, Digest, V>>,
2162 C::Item: EncodeShared,
2163 {
2164 let mut db = open_db(context.child("db")).await;
2165
2166 let key = Sha256::hash(&0u64.to_be_bytes());
2167 let val_parent = Sha256::fill(1u8);
2168 let val_child = Sha256::fill(2u8);
2169
2170 let parent_m = db
2172 .new_batch()
2173 .set(key, val_parent)
2174 .merkleize(&db, None, Location::new(0))
2175 .await;
2176
2177 let mut child = parent_m.new_batch::<Sha256>();
2179 child = child.set(key, val_child);
2180
2181 assert_eq!(child.get(&key, &db).await.unwrap(), Some(val_child));
2183
2184 let child_m = child.merkleize(&db, None, Location::new(0)).await;
2185
2186 assert_eq!(child_m.get(&key, &db).await.unwrap(), Some(val_child));
2188
2189 db.apply_batch(child_m).await.unwrap();
2191 assert_eq!(db.get(&key).await.unwrap(), Some(val_child));
2192
2193 db.destroy().await.unwrap();
2194 }
2195
2196 #[boxed]
2205 pub(crate) async fn test_immutable_batch_sequential_key_override<F: Family, V, C>(
2206 context: deterministic::Context,
2207 open_db_small_sections: impl Fn(
2208 deterministic::Context,
2209 )
2210 -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2211 ) where
2212 V: ValueEncoding<Value = Digest>,
2213 C: Mutable<Item = Operation<F, Digest, V>>,
2214 C::Item: EncodeShared,
2215 {
2216 let mut db = open_db_small_sections(context.child("db")).await;
2217
2218 let key = Sha256::hash(&0u64.to_be_bytes());
2219 let v1 = Sha256::fill(1u8);
2220 let v2 = Sha256::fill(2u8);
2221
2222 db.apply_batch(
2225 db.new_batch()
2226 .set(key, v1)
2227 .merkleize(&db, None, Location::new(0))
2228 .await,
2229 )
2230 .await
2231 .unwrap();
2232 assert_eq!(db.get(&key).await.unwrap(), Some(v1));
2233
2234 db.apply_batch(
2237 db.new_batch()
2238 .set(key, v2)
2239 .merkleize(&db, None, Location::new(0))
2240 .await,
2241 )
2242 .await
2243 .unwrap();
2244
2245 let live = db.get(&key).await.unwrap().unwrap();
2247 assert!(live == v1 || live == v2);
2248
2249 db.commit().await.unwrap();
2251 drop(db);
2252 let db = open_db_small_sections(context.child("reopen")).await;
2253 let reopened = db.get(&key).await.unwrap().unwrap();
2254 assert!(reopened == v1 || reopened == v2);
2255 db.destroy().await.unwrap();
2256
2257 let mut db = open_db_small_sections(context.child("prune")).await;
2262 db.apply_batch(
2263 db.new_batch()
2264 .set(key, v1)
2265 .merkleize(&db, None, Location::new(0))
2266 .await,
2267 )
2268 .await
2269 .unwrap();
2270 db.apply_batch(
2271 db.new_batch()
2272 .set(key, v2)
2273 .merkleize(&db, None, Location::new(4))
2274 .await,
2275 )
2276 .await
2277 .unwrap();
2278
2279 db.prune(Location::new(2)).await.unwrap();
2283 assert_eq!(db.get(&key).await.unwrap(), Some(v2));
2284
2285 db.destroy().await.unwrap();
2286 }
2287
2288 #[boxed]
2290 pub(crate) async fn test_immutable_batch_metadata<F: Family, V, C>(
2291 context: deterministic::Context,
2292 open_db: impl Fn(
2293 deterministic::Context,
2294 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2295 ) where
2296 V: ValueEncoding<Value = Digest>,
2297 C: Mutable<Item = Operation<F, Digest, V>>,
2298 C::Item: EncodeShared,
2299 {
2300 let mut db = open_db(context.child("db")).await;
2301
2302 let metadata = Sha256::fill(42u8);
2304 let k = Sha256::hash(&[1u8]);
2305 db.apply_batch(
2306 db.new_batch()
2307 .set(k, Sha256::fill(1u8))
2308 .merkleize(&db, Some(metadata), Location::new(0))
2309 .await,
2310 )
2311 .await
2312 .unwrap();
2313 assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
2314
2315 db.apply_batch(db.new_batch().merkleize(&db, None, Location::new(0)).await)
2317 .await
2318 .unwrap();
2319 assert_eq!(db.get_metadata().await.unwrap(), None);
2320
2321 db.destroy().await.unwrap();
2322 }
2323
2324 #[boxed]
2325 pub(crate) async fn test_immutable_stale_batch_rejected<F: Family, V, C>(
2326 context: deterministic::Context,
2327 open_db: impl Fn(
2328 deterministic::Context,
2329 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2330 ) where
2331 V: ValueEncoding<Value = Digest>,
2332 C: Mutable<Item = Operation<F, Digest, V>>,
2333 C::Item: EncodeShared,
2334 {
2335 let mut db = open_db(context.child("db")).await;
2336
2337 let key1 = Sha256::hash(&[1]);
2338 let key2 = Sha256::hash(&[2]);
2339 let v1 = Sha256::fill(10u8);
2340 let v2 = Sha256::fill(20u8);
2341
2342 let batch_a = db
2344 .new_batch()
2345 .set(key1, v1)
2346 .merkleize(&db, None, Location::new(0))
2347 .await;
2348 let batch_b = db
2349 .new_batch()
2350 .set(key2, v2)
2351 .merkleize(&db, None, Location::new(0))
2352 .await;
2353
2354 db.apply_batch(batch_a).await.unwrap();
2356 let expected_root = db.root();
2357 let expected_bounds = db.bounds();
2358 assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2359 assert_eq!(db.get(&key2).await.unwrap(), None);
2360 assert_eq!(db.get_metadata().await.unwrap(), None);
2361
2362 let result = db.apply_batch(batch_b).await;
2364 assert!(
2365 matches!(result, Err(Error::StaleBatch { .. })),
2366 "expected StaleBatch error, got {result:?}"
2367 );
2368 assert_eq!(db.root(), expected_root);
2369 assert_eq!(db.bounds(), expected_bounds);
2370 assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2371 assert_eq!(db.get(&key2).await.unwrap(), None);
2372 assert_eq!(db.get_metadata().await.unwrap(), None);
2373
2374 db.destroy().await.unwrap();
2375 }
2376
2377 #[boxed]
2378 pub(crate) async fn test_immutable_stale_batch_chained<F: Family, V, C>(
2379 context: deterministic::Context,
2380 open_db: impl Fn(
2381 deterministic::Context,
2382 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2383 ) where
2384 V: ValueEncoding<Value = Digest>,
2385 C: Mutable<Item = Operation<F, Digest, V>>,
2386 C::Item: EncodeShared,
2387 {
2388 let mut db = open_db(context.child("db")).await;
2389
2390 let key1 = Sha256::hash(&[1]);
2391 let key2 = Sha256::hash(&[2]);
2392 let key3 = Sha256::hash(&[3]);
2393
2394 let parent_m = db
2396 .new_batch()
2397 .set(key1, Sha256::fill(1u8))
2398 .merkleize(&db, None, Location::new(0))
2399 .await;
2400
2401 let child_a = parent_m
2403 .new_batch::<Sha256>()
2404 .set(key2, Sha256::fill(2u8))
2405 .merkleize(&db, None, Location::new(0))
2406 .await;
2407 let child_b = parent_m
2408 .new_batch::<Sha256>()
2409 .set(key3, Sha256::fill(3u8))
2410 .merkleize(&db, None, Location::new(0))
2411 .await;
2412
2413 db.apply_batch(child_a).await.unwrap();
2415
2416 let result = db.apply_batch(child_b).await;
2418 assert!(
2419 matches!(result, Err(Error::StaleBatch { .. })),
2420 "expected StaleBatch error, got {result:?}"
2421 );
2422
2423 db.destroy().await.unwrap();
2424 }
2425
2426 #[boxed]
2427 pub(crate) async fn test_immutable_partial_ancestor_commit<F: Family, V, C>(
2428 context: deterministic::Context,
2429 open_db: impl Fn(
2430 deterministic::Context,
2431 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2432 ) where
2433 V: ValueEncoding<Value = Digest>,
2434 C: Mutable<Item = Operation<F, Digest, V>>,
2435 C::Item: EncodeShared,
2436 {
2437 let mut db = open_db(context.child("db")).await;
2438
2439 let key1 = Sha256::hash(&[1]);
2440 let key2 = Sha256::hash(&[2]);
2441 let key3 = Sha256::hash(&[3]);
2442 let v1 = Sha256::fill(1u8);
2443 let v2 = Sha256::fill(2u8);
2444 let v3 = Sha256::fill(3u8);
2445
2446 let a = db
2448 .new_batch()
2449 .set(key1, v1)
2450 .merkleize(&db, None, Location::new(0))
2451 .await;
2452 let b = a
2453 .new_batch::<Sha256>()
2454 .set(key2, v2)
2455 .merkleize(&db, None, Location::new(0))
2456 .await;
2457 let c = b
2458 .new_batch::<Sha256>()
2459 .set(key3, v3)
2460 .merkleize(&db, None, Location::new(0))
2461 .await;
2462
2463 let expected_root = c.root();
2464
2465 db.apply_batch(a).await.unwrap();
2467 db.apply_batch(c).await.unwrap();
2468
2469 assert_eq!(db.root(), expected_root);
2470 assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2471 assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2472 assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
2473
2474 db.destroy().await.unwrap();
2475 }
2476
2477 #[boxed]
2478 pub(crate) async fn test_immutable_sequential_commit_parent_then_child<F: Family, V, C>(
2479 context: deterministic::Context,
2480 open_db: impl Fn(
2481 deterministic::Context,
2482 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2483 ) where
2484 V: ValueEncoding<Value = Digest>,
2485 C: Mutable<Item = Operation<F, Digest, V>>,
2486 C::Item: EncodeShared,
2487 {
2488 let mut db = open_db(context.child("db")).await;
2489
2490 let key1 = Sha256::hash(&[1]);
2491 let key2 = Sha256::hash(&[2]);
2492 let v1 = Sha256::fill(1u8);
2493 let v2 = Sha256::fill(2u8);
2494
2495 let parent_m = db
2497 .new_batch()
2498 .set(key1, v1)
2499 .merkleize(&db, None, Location::new(0))
2500 .await;
2501
2502 let child_m = parent_m
2504 .new_batch::<Sha256>()
2505 .set(key2, v2)
2506 .merkleize(&db, None, Location::new(0))
2507 .await;
2508
2509 db.apply_batch(parent_m).await.unwrap();
2511 db.apply_batch(child_m).await.unwrap();
2512
2513 assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2515 assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2516
2517 db.destroy().await.unwrap();
2518 }
2519
2520 #[boxed]
2521 pub(crate) async fn test_immutable_child_root_matches_pending_and_committed<F: Family, V, C>(
2522 context: deterministic::Context,
2523 open_db: impl Fn(
2524 deterministic::Context,
2525 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2526 ) where
2527 V: ValueEncoding<Value = Digest>,
2528 C: Mutable<Item = Operation<F, Digest, V>>,
2529 C::Item: EncodeShared,
2530 {
2531 let mut db = open_db(context.child("db")).await;
2532
2533 let key1 = Sha256::hash(&[1]);
2534 let key2 = Sha256::hash(&[2]);
2535
2536 let parent = db
2538 .new_batch()
2539 .set(key1, Sha256::fill(1u8))
2540 .merkleize(&db, None, Location::new(0))
2541 .await;
2542 let pending_child = parent
2543 .new_batch::<Sha256>()
2544 .set(key2, Sha256::fill(2u8))
2545 .merkleize(&db, None, Location::new(0))
2546 .await;
2547
2548 db.apply_batch(parent).await.unwrap();
2551 db.commit().await.unwrap();
2552
2553 let committed_child = db
2554 .new_batch()
2555 .set(key2, Sha256::fill(2u8))
2556 .merkleize(&db, None, Location::new(0))
2557 .await;
2558
2559 assert_eq!(pending_child.root(), committed_child.root());
2560
2561 db.destroy().await.unwrap();
2562 }
2563
2564 #[boxed]
2565 pub(crate) async fn test_immutable_stale_batch_child_applied_before_parent<F: Family, V, C>(
2566 context: deterministic::Context,
2567 open_db: impl Fn(
2568 deterministic::Context,
2569 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2570 ) where
2571 V: ValueEncoding<Value = Digest>,
2572 C: Mutable<Item = Operation<F, Digest, V>>,
2573 C::Item: EncodeShared,
2574 {
2575 let mut db = open_db(context.child("db")).await;
2576
2577 let key1 = Sha256::hash(&[1]);
2578 let key2 = Sha256::hash(&[2]);
2579
2580 let parent_m = db
2582 .new_batch()
2583 .set(key1, Sha256::fill(1u8))
2584 .merkleize(&db, None, Location::new(0))
2585 .await;
2586
2587 let child_m = parent_m
2589 .new_batch::<Sha256>()
2590 .set(key2, Sha256::fill(2u8))
2591 .merkleize(&db, None, Location::new(0))
2592 .await;
2593
2594 db.apply_batch(child_m).await.unwrap();
2596
2597 let result = db.apply_batch(parent_m).await;
2599 assert!(
2600 matches!(result, Err(Error::StaleBatch { .. })),
2601 "expected StaleBatch error, got {result:?}"
2602 );
2603
2604 db.destroy().await.unwrap();
2605 }
2606
2607 #[boxed]
2610 pub(crate) async fn test_immutable_to_batch<F: Family, V, C>(
2611 context: deterministic::Context,
2612 open_db: impl Fn(
2613 deterministic::Context,
2614 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2615 ) where
2616 V: ValueEncoding<Value = Digest>,
2617 C: Mutable<Item = Operation<F, Digest, V>>,
2618 C::Item: EncodeShared,
2619 {
2620 let mut db = open_db(context.child("db")).await;
2621
2622 let key1 = Sha256::hash(&[1]);
2624 let v1 = Sha256::fill(10u8);
2625 db.apply_batch(
2626 db.new_batch()
2627 .set(key1, v1)
2628 .merkleize(&db, None, Location::new(0))
2629 .await,
2630 )
2631 .await
2632 .unwrap();
2633
2634 let snapshot = db.to_batch();
2636 assert_eq!(snapshot.root(), db.root());
2637
2638 let key2 = Sha256::hash(&[2]);
2640 let v2 = Sha256::fill(20u8);
2641 let child = snapshot
2642 .new_batch::<Sha256>()
2643 .set(key2, v2)
2644 .merkleize(&db, None, Location::new(0))
2645 .await;
2646 db.apply_batch(child).await.unwrap();
2647
2648 assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2649 assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2650
2651 db.destroy().await.unwrap();
2652 }
2653
2654 #[boxed]
2657 pub(crate) async fn test_immutable_apply_after_ancestor_dropped<F: Family, V, C>(
2658 context: deterministic::Context,
2659 open_db: impl Fn(
2660 deterministic::Context,
2661 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2662 ) where
2663 V: ValueEncoding<Value = Digest>,
2664 C: Mutable<Item = Operation<F, Digest, V>>,
2665 C::Item: EncodeShared,
2666 {
2667 let mut db = open_db(context.child("db")).await;
2668
2669 let key1 = Sha256::hash(&[1]);
2670 let key2 = Sha256::hash(&[2]);
2671 let key3 = Sha256::hash(&[3]);
2672 let v1 = Sha256::fill(1u8);
2673 let v2 = Sha256::fill(2u8);
2674 let v3 = Sha256::fill(3u8);
2675
2676 let a = db
2678 .new_batch()
2679 .set(key1, v1)
2680 .merkleize(&db, None, Location::new(0))
2681 .await;
2682 let b = a
2683 .new_batch::<Sha256>()
2684 .set(key2, v2)
2685 .merkleize(&db, None, Location::new(0))
2686 .await;
2687 let c = b
2688 .new_batch::<Sha256>()
2689 .set(key3, v3)
2690 .merkleize(&db, None, Location::new(0))
2691 .await;
2692
2693 drop(a);
2695 drop(b);
2696
2697 db.apply_batch(c).await.unwrap();
2699
2700 assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
2702 assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
2703 assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
2704
2705 db.destroy().await.unwrap();
2706 }
2707
2708 #[boxed]
2711 pub(crate) async fn test_immutable_inactivity_floor_tracking<F: Family, V, C>(
2712 context: deterministic::Context,
2713 open_db: impl Fn(
2714 deterministic::Context,
2715 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2716 ) where
2717 V: ValueEncoding<Value = Digest>,
2718 C: Mutable<Item = Operation<F, Digest, V>>,
2719 C::Item: EncodeShared,
2720 {
2721 let mut db = open_db(context.child("test")).await;
2722
2723 assert_eq!(db.inactivity_floor_loc(), Location::new(0));
2725
2726 let k1 = Sha256::fill(1u8);
2728 let v1 = Sha256::fill(2u8);
2729 db.apply_batch(
2730 db.new_batch()
2731 .set(k1, v1)
2732 .merkleize(&db, None, Location::new(0))
2733 .await,
2734 )
2735 .await
2736 .unwrap();
2737 assert_eq!(db.inactivity_floor_loc(), Location::new(0));
2738
2739 let k2 = Sha256::fill(3u8);
2741 let v2 = Sha256::fill(4u8);
2742 db.apply_batch(
2743 db.new_batch()
2744 .set(k2, v2)
2745 .merkleize(&db, None, Location::new(3))
2746 .await,
2747 )
2748 .await
2749 .unwrap();
2750 assert_eq!(db.inactivity_floor_loc(), Location::new(3));
2751
2752 db.commit().await.unwrap();
2754 db.sync().await.unwrap();
2755 drop(db);
2756 let db = open_db(context.child("reopen")).await;
2757 assert_eq!(db.inactivity_floor_loc(), Location::new(3));
2758
2759 db.destroy().await.unwrap();
2760 }
2761
2762 #[boxed]
2765 pub(crate) async fn test_immutable_floor_monotonicity<F: Family, V, C>(
2766 context: deterministic::Context,
2767 open_db: impl Fn(
2768 deterministic::Context,
2769 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2770 ) where
2771 V: ValueEncoding<Value = Digest>,
2772 C: Mutable<Item = Operation<F, Digest, V>>,
2773 C::Item: EncodeShared,
2774 {
2775 let mut db = open_db(context.child("test")).await;
2776
2777 let k1 = Sha256::fill(1u8);
2780 let v1 = Sha256::fill(2u8);
2781 db.apply_batch(
2782 db.new_batch()
2783 .set(k1, v1)
2784 .merkleize(&db, None, Location::new(2))
2785 .await,
2786 )
2787 .await
2788 .unwrap();
2789 assert_eq!(db.inactivity_floor_loc(), Location::new(2));
2790
2791 let k2 = Sha256::fill(3u8);
2793 let v2 = Sha256::fill(4u8);
2794 db.apply_batch(
2795 db.new_batch()
2796 .set(k2, v2)
2797 .merkleize(&db, None, Location::new(2))
2798 .await,
2799 )
2800 .await
2801 .unwrap();
2802 assert_eq!(db.inactivity_floor_loc(), Location::new(2));
2803
2804 let k3 = Sha256::fill(5u8);
2806 let v3 = Sha256::fill(6u8);
2807 db.apply_batch(
2808 db.new_batch()
2809 .set(k3, v3)
2810 .merkleize(&db, None, Location::new(5))
2811 .await,
2812 )
2813 .await
2814 .unwrap();
2815 assert_eq!(db.inactivity_floor_loc(), Location::new(5));
2816
2817 db.destroy().await.unwrap();
2818 }
2819
2820 #[boxed]
2822 pub(crate) async fn test_immutable_rewind_restores_floor<F: Family, V, C>(
2823 context: deterministic::Context,
2824 open_db: impl Fn(
2825 deterministic::Context,
2826 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2827 ) where
2828 V: ValueEncoding<Value = Digest>,
2829 C: Mutable<Item = Operation<F, Digest, V>>,
2830 C::Item: EncodeShared,
2831 {
2832 let mut db = open_db(context.child("test")).await;
2833
2834 let k1 = Sha256::fill(1u8);
2836 let v1 = Sha256::fill(2u8);
2837 db.apply_batch(
2838 db.new_batch()
2839 .set(k1, v1)
2840 .merkleize(&db, None, Location::new(2))
2841 .await,
2842 )
2843 .await
2844 .unwrap();
2845 db.commit().await.unwrap();
2846 let first_size = db.bounds().end;
2847 assert_eq!(db.inactivity_floor_loc(), Location::new(2));
2848
2849 let k2 = Sha256::fill(3u8);
2851 let v2 = Sha256::fill(4u8);
2852 db.apply_batch(
2853 db.new_batch()
2854 .set(k2, v2)
2855 .merkleize(&db, None, Location::new(4))
2856 .await,
2857 )
2858 .await
2859 .unwrap();
2860 db.commit().await.unwrap();
2861 assert_eq!(db.inactivity_floor_loc(), Location::new(4));
2862
2863 db.rewind(first_size).await.unwrap();
2865 assert_eq!(db.inactivity_floor_loc(), Location::new(2));
2866
2867 db.destroy().await.unwrap();
2868 }
2869
2870 #[boxed]
2873 pub(crate) async fn test_immutable_floor_monotonicity_violation<F: Family, V, C>(
2874 context: deterministic::Context,
2875 open_db: impl Fn(
2876 deterministic::Context,
2877 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2878 ) where
2879 V: ValueEncoding<Value = Digest>,
2880 C: Mutable<Item = Operation<F, Digest, V>>,
2881 C::Item: EncodeShared,
2882 {
2883 let mut db = open_db(context.child("test")).await;
2884
2885 let k1 = Sha256::fill(1u8);
2887 let v1 = Sha256::fill(2u8);
2888 db.apply_batch(
2889 db.new_batch()
2890 .set(k1, v1)
2891 .merkleize(&db, None, Location::new(2))
2892 .await,
2893 )
2894 .await
2895 .unwrap();
2896
2897 let k2 = Sha256::fill(3u8);
2899 let v2 = Sha256::fill(4u8);
2900 let result = db
2901 .apply_batch(
2902 db.new_batch()
2903 .set(k2, v2)
2904 .merkleize(&db, None, Location::new(1))
2905 .await,
2906 )
2907 .await;
2908 assert!(matches!(result, Err(Error::FloorRegressed(new, current))
2909 if new == Location::new(1) && current == Location::new(2)));
2910
2911 db.destroy().await.unwrap();
2912 }
2913
2914 #[boxed]
2917 pub(crate) async fn test_immutable_floor_beyond_size<F: Family, V, C>(
2918 context: deterministic::Context,
2919 open_db: impl Fn(
2920 deterministic::Context,
2921 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2922 ) where
2923 V: ValueEncoding<Value = Digest>,
2924 C: Mutable<Item = Operation<F, Digest, V>>,
2925 C::Item: EncodeShared,
2926 {
2927 let mut db = open_db(context.child("test")).await;
2928
2929 let k1 = Sha256::fill(1u8);
2932 let v1 = Sha256::fill(2u8);
2933 let result = db
2934 .apply_batch(
2935 db.new_batch()
2936 .set(k1, v1)
2937 .merkleize(&db, None, Location::new(100))
2938 .await,
2939 )
2940 .await;
2941 assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit))
2942 if floor == Location::new(100) && commit == Location::new(2)));
2943
2944 let k2 = Sha256::fill(3u8);
2948 let v2 = Sha256::fill(4u8);
2949 let result = db
2950 .apply_batch(
2951 db.new_batch()
2952 .set(k2, v2)
2953 .merkleize(&db, None, Location::new(3))
2954 .await,
2955 )
2956 .await;
2957 assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit))
2958 if floor == Location::new(3) && commit == Location::new(2)));
2959
2960 db.apply_batch(
2962 db.new_batch()
2963 .set(k2, v2)
2964 .merkleize(&db, None, Location::new(2))
2965 .await,
2966 )
2967 .await
2968 .unwrap();
2969
2970 db.destroy().await.unwrap();
2971 }
2972
2973 #[boxed]
2979 pub(crate) async fn test_immutable_chained_ancestor_floor_regression<F: Family, V, C>(
2980 context: deterministic::Context,
2981 open_db: impl Fn(
2982 deterministic::Context,
2983 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
2984 ) where
2985 V: ValueEncoding<Value = Digest>,
2986 C: Mutable<Item = Operation<F, Digest, V>>,
2987 C::Item: EncodeShared,
2988 {
2989 let mut db = open_db(context.child("test")).await;
2990
2991 let a = db
2998 .new_batch()
2999 .set(Sha256::fill(1u8), Sha256::fill(2u8))
3000 .merkleize(&db, None, Location::new(2))
3001 .await;
3002 let b = a
3003 .new_batch::<Sha256>()
3004 .set(Sha256::fill(3u8), Sha256::fill(4u8))
3005 .merkleize(&db, None, Location::new(1))
3006 .await;
3007 let c = b
3008 .new_batch::<Sha256>()
3009 .set(Sha256::fill(5u8), Sha256::fill(6u8))
3010 .merkleize(&db, None, Location::new(2))
3011 .await;
3012
3013 let root_before = db.root();
3014 let last_commit_before = db.last_commit_loc;
3015 let floor_before = db.inactivity_floor_loc();
3016
3017 let err = db.apply_batch(c).await.unwrap_err();
3018 assert!(
3019 matches!(err, Error::FloorRegressed(new, prev)
3020 if new == Location::new(1) && prev == Location::new(2)),
3021 "unexpected error: {err:?}"
3022 );
3023
3024 assert_eq!(db.root(), root_before);
3026 assert_eq!(db.last_commit_loc, last_commit_before);
3027 assert_eq!(db.inactivity_floor_loc(), floor_before);
3028
3029 db.destroy().await.unwrap();
3030 }
3031
3032 #[boxed]
3037 pub(crate) async fn test_immutable_chained_ancestor_floor_beyond_size<F: Family, V, C>(
3038 context: deterministic::Context,
3039 open_db: impl Fn(
3040 deterministic::Context,
3041 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3042 ) where
3043 V: ValueEncoding<Value = Digest>,
3044 C: Mutable<Item = Operation<F, Digest, V>>,
3045 C::Item: EncodeShared,
3046 {
3047 let mut db = open_db(context.child("test")).await;
3048
3049 let a = db
3052 .new_batch()
3053 .set(Sha256::fill(1u8), Sha256::fill(2u8))
3054 .merkleize(&db, None, Location::new(3))
3055 .await;
3056 let b = a
3057 .new_batch::<Sha256>()
3058 .set(Sha256::fill(3u8), Sha256::fill(4u8))
3059 .merkleize(&db, None, Location::new(0))
3060 .await;
3061
3062 let root_before = db.root();
3063 let last_commit_before = db.last_commit_loc;
3064 let floor_before = db.inactivity_floor_loc();
3065
3066 let err = db.apply_batch(b).await.unwrap_err();
3067 assert!(
3069 matches!(err, Error::FloorBeyondSize(floor, commit)
3070 if floor == Location::new(3) && commit == Location::new(2)),
3071 "unexpected error: {err:?}"
3072 );
3073
3074 assert_eq!(db.root(), root_before);
3076 assert_eq!(db.last_commit_loc, last_commit_before);
3077 assert_eq!(db.inactivity_floor_loc(), floor_before);
3078
3079 db.destroy().await.unwrap();
3080 }
3081
3082 #[boxed]
3089 pub(crate) async fn test_immutable_rewind_after_reopen_with_floor_change<F: Family, V, C>(
3090 context: deterministic::Context,
3091 open_db: impl Fn(
3092 deterministic::Context,
3093 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3094 ) where
3095 V: ValueEncoding<Value = Digest>,
3096 C: Mutable<Item = Operation<F, Digest, V>>,
3097 C::Item: EncodeShared,
3098 {
3099 let mut db = open_db(context.child("first")).await;
3100
3101 let k1 = Sha256::fill(1u8);
3102 let k2 = Sha256::fill(2u8);
3103 let k3 = Sha256::fill(3u8);
3104 let v1 = Sha256::fill(11u8);
3105 let v2 = Sha256::fill(12u8);
3106 let v3 = Sha256::fill(13u8);
3107
3108 commit_sets(&mut db, [(k1, v1), (k2, v2), (k3, v3)], None).await;
3110 let first_size = db.bounds().end;
3111 let first_root = db.root();
3112
3113 let k4 = Sha256::fill(4u8);
3115 let k5 = Sha256::fill(5u8);
3116 let k6 = Sha256::fill(6u8);
3117 let v4 = Sha256::fill(14u8);
3118 let v5 = Sha256::fill(15u8);
3119 let v6 = Sha256::fill(16u8);
3120 commit_sets_with_floor(&mut db, [(k4, v4), (k5, v5), (k6, v6)], None, first_size).await;
3121 db.sync().await.unwrap();
3122
3123 drop(db);
3125 let mut db = open_db(context.child("second")).await;
3126
3127 assert!(db.get(&k1).await.unwrap().is_none());
3129
3130 db.rewind(first_size).await.unwrap();
3132
3133 assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
3135 assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
3136 assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3137 assert_eq!(db.root(), first_root);
3138 assert_eq!(db.inactivity_floor_loc(), Location::new(0));
3139
3140 assert!(db.get(&k4).await.unwrap().is_none());
3142
3143 db.destroy().await.unwrap();
3144 }
3145
3146 #[boxed]
3150 pub(crate) async fn test_immutable_rewind_after_reopen_partial_floor_gap<F: Family, V, C>(
3151 context: deterministic::Context,
3152 open_db: impl Fn(
3153 deterministic::Context,
3154 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3155 ) where
3156 V: ValueEncoding<Value = Digest>,
3157 C: Mutable<Item = Operation<F, Digest, V>>,
3158 C::Item: EncodeShared,
3159 {
3160 let mut db = open_db(context.child("first")).await;
3161
3162 let k1 = Sha256::fill(1u8);
3163 let v1 = Sha256::fill(11u8);
3164
3165 commit_sets(&mut db, [(k1, v1)], None).await;
3167 let first_size = db.bounds().end;
3168 let first_root = db.root();
3169
3170 let k2 = Sha256::fill(2u8);
3172 let v2 = Sha256::fill(12u8);
3173 commit_sets_with_floor(&mut db, [(k2, v2)], None, first_size).await;
3174 let second_size = db.bounds().end;
3175
3176 let k3 = Sha256::fill(3u8);
3179 let v3 = Sha256::fill(13u8);
3180 commit_sets_with_floor(&mut db, [(k3, v3)], None, second_size).await;
3181 db.sync().await.unwrap();
3182
3183 drop(db);
3185 let mut db = open_db(context.child("second")).await;
3186 assert!(db.get(&k1).await.unwrap().is_none());
3187 assert!(db.get(&k2).await.unwrap().is_none());
3188 assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3189
3190 db.rewind(second_size).await.unwrap();
3194 assert!(db.get(&k1).await.unwrap().is_none()); assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
3196 assert!(db.get(&k3).await.unwrap().is_none()); db.rewind(first_size).await.unwrap();
3200 assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
3201 assert!(db.get(&k2).await.unwrap().is_none()); assert_eq!(db.root(), first_root);
3203 assert_eq!(db.inactivity_floor_loc(), Location::new(0));
3204
3205 db.destroy().await.unwrap();
3206 }
3207
3208 #[boxed]
3211 pub(crate) async fn test_immutable_rewind_after_reopen_repeated_key_gap<F: Family, V, C>(
3212 context: deterministic::Context,
3213 open_db: impl Fn(
3214 deterministic::Context,
3215 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3216 ) where
3217 V: ValueEncoding<Value = Digest>,
3218 C: Mutable<Item = Operation<F, Digest, V>>,
3219 C::Item: EncodeShared,
3220 {
3221 let mut db = open_db(context.child("first")).await;
3222
3223 let key = Sha256::fill(7u8);
3224 let v1 = Sha256::fill(17u8);
3225 let v2 = Sha256::fill(18u8);
3226 let k3 = Sha256::fill(8u8);
3227 let v3 = Sha256::fill(19u8);
3228
3229 commit_sets(&mut db, [(key, v1)], None).await;
3231 let first_size = db.bounds().end;
3232
3233 commit_sets(&mut db, [(key, v2)], None).await;
3235 let second_size = db.bounds().end;
3236 let live = db.get(&key).await.unwrap().unwrap();
3237 assert!(live == v1 || live == v2);
3238
3239 commit_sets_with_floor(&mut db, [(k3, v3)], None, second_size).await;
3241 db.sync().await.unwrap();
3242
3243 drop(db);
3245 let mut db = open_db(context.child("second")).await;
3246 assert!(db.get(&key).await.unwrap().is_none());
3247 assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3248
3249 db.rewind(second_size).await.unwrap();
3251 let rewound = db.get(&key).await.unwrap().unwrap();
3252 assert!(rewound == v1 || rewound == v2);
3253
3254 db.rewind(first_size).await.unwrap();
3257 assert_eq!(db.get(&key).await.unwrap(), Some(v1));
3258
3259 db.destroy().await.unwrap();
3260 }
3261
3262 #[boxed]
3266 pub(crate) async fn test_immutable_rewind_after_reopen_mixed_gap_retained<F: Family, V, C>(
3267 context: deterministic::Context,
3268 open_db: impl Fn(
3269 deterministic::Context,
3270 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3271 ) where
3272 V: ValueEncoding<Value = Digest>,
3273 C: Mutable<Item = Operation<F, Digest, V>>,
3274 C::Item: EncodeShared,
3275 {
3276 let mut db = open_db(context.child("first")).await;
3277
3278 let key = Sha256::fill(7u8);
3279 let v1 = Sha256::fill(17u8);
3280 let v2 = Sha256::fill(18u8);
3281 let k3 = Sha256::fill(8u8);
3282 let v3 = Sha256::fill(19u8);
3283
3284 commit_sets(&mut db, [(key, v1)], None).await;
3286 let first_size = db.bounds().end;
3287
3288 commit_sets(&mut db, [(key, v2)], None).await;
3290 let second_size = db.bounds().end;
3291 let live = db.get(&key).await.unwrap().unwrap();
3292 assert!(live == v1 || live == v2);
3293
3294 commit_sets_with_floor(&mut db, [(k3, v3)], None, first_size).await;
3297 db.sync().await.unwrap();
3298
3299 drop(db);
3302 let mut db = open_db(context.child("second")).await;
3303 assert_eq!(db.get(&key).await.unwrap(), Some(v2));
3304
3305 db.rewind(second_size).await.unwrap();
3308 let rewound = db.get(&key).await.unwrap().unwrap();
3309 assert!(rewound == v1 || rewound == v2);
3310
3311 db.rewind(first_size).await.unwrap();
3314 assert_eq!(db.get(&key).await.unwrap(), Some(v1));
3315
3316 db.destroy().await.unwrap();
3317 }
3318
3319 #[boxed]
3329 pub(crate) async fn test_immutable_single_commit_live_set<F: Family, V, C>(
3330 context: deterministic::Context,
3331 open_db: impl Fn(
3332 deterministic::Context,
3333 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3334 ) where
3335 V: ValueEncoding<Value = Digest>,
3336 C: Mutable<Item = Operation<F, Digest, V>>,
3337 C::Item: EncodeShared,
3338 {
3339 let mut db = open_db(context.child("test")).await;
3340
3341 let metadata = Sha256::fill(42u8);
3344 let commit_loc = Location::<F>::new(4);
3345 let k1 = Sha256::fill(1u8);
3346 let k2 = Sha256::fill(2u8);
3347 let k3 = Sha256::fill(3u8);
3348 let v1 = Sha256::fill(11u8);
3349 let v2 = Sha256::fill(12u8);
3350 let v3 = Sha256::fill(13u8);
3351 db.apply_batch(
3352 db.new_batch()
3353 .set(k1, v1)
3354 .set(k2, v2)
3355 .set(k3, v3)
3356 .merkleize(&db, Some(metadata), commit_loc)
3357 .await,
3358 )
3359 .await
3360 .unwrap();
3361 db.commit().await.unwrap();
3362 assert_eq!(db.last_commit_loc, commit_loc);
3363 assert_eq!(db.inactivity_floor_loc(), commit_loc);
3364 let root_after_commit = db.root();
3365
3366 assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
3368 assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
3369 assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
3370
3371 db.prune(commit_loc).await.unwrap();
3376 let bounds = db.bounds();
3377 assert!(
3378 bounds.start <= commit_loc,
3379 "prune must not advance bounds.start past the floor"
3380 );
3381 assert_eq!(bounds.end, Location::new(*commit_loc + 1));
3382
3383 let err = db.prune(Location::new(*commit_loc + 1)).await.unwrap_err();
3385 assert!(matches!(err, Error::PruneBeyondMinRequired(p, f)
3386 if *p == *commit_loc + 1 && *f == *commit_loc));
3387
3388 assert_eq!(db.last_commit_loc, commit_loc);
3390 assert_eq!(db.inactivity_floor_loc(), commit_loc);
3391 assert_eq!(db.root(), root_after_commit);
3392 assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
3393
3394 db.sync().await.unwrap();
3398 drop(db);
3399 let mut db = open_db(context.child("reopened")).await;
3400 assert_eq!(db.last_commit_loc, commit_loc);
3401 assert_eq!(db.inactivity_floor_loc(), commit_loc);
3402 assert_eq!(db.root(), root_after_commit);
3403 assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
3406
3407 assert!(db.get(&k1).await.unwrap().is_none());
3409 assert!(db.get(&k2).await.unwrap().is_none());
3410 assert!(db.get(&k3).await.unwrap().is_none());
3411
3412 let k4 = Sha256::fill(4u8);
3416 let v4 = Sha256::fill(14u8);
3417 let next_commit_loc = Location::<F>::new(6);
3418 db.apply_batch(
3419 db.new_batch()
3420 .set(k4, v4)
3421 .merkleize(&db, None, next_commit_loc)
3422 .await,
3423 )
3424 .await
3425 .unwrap();
3426 db.commit().await.unwrap();
3427 assert_eq!(db.last_commit_loc, next_commit_loc);
3428 assert_eq!(db.inactivity_floor_loc(), next_commit_loc);
3429
3430 assert_eq!(db.get(&k4).await.unwrap(), Some(v4));
3432 assert!(db.get(&k1).await.unwrap().is_none());
3433 assert_eq!(db.get_metadata().await.unwrap(), None);
3436
3437 db.destroy().await.unwrap();
3438 }
3439
3440 #[boxed]
3443 pub(crate) async fn test_immutable_get_many<F: Family, V, C>(
3444 context: deterministic::Context,
3445 open_db: impl Fn(
3446 deterministic::Context,
3447 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3448 ) where
3449 V: ValueEncoding<Value = Digest>,
3450 C: Mutable<Item = Operation<F, Digest, V>>,
3451 C::Item: EncodeShared,
3452 {
3453 let mut db = open_db(context.child("db")).await;
3454
3455 let k1 = Sha256::fill(1u8);
3456 let k2 = Sha256::fill(2u8);
3457 let k3 = Sha256::fill(3u8);
3458 let k_missing = Sha256::fill(99u8);
3459
3460 let v1 = Sha256::fill(11u8);
3461 let v2 = Sha256::fill(12u8);
3462 let v3 = Sha256::fill(13u8);
3463
3464 db.apply_batch(
3466 db.new_batch()
3467 .set(k1, v1)
3468 .set(k2, v2)
3469 .merkleize(&db, None, db.inactivity_floor_loc())
3470 .await,
3471 )
3472 .await
3473 .unwrap();
3474 db.commit().await.unwrap();
3475
3476 let results = db.get_many(&[&k1, &k2, &k_missing]).await.unwrap();
3478 assert_eq!(results, vec![Some(v1), Some(v2), None]);
3479
3480 let results = db.get_many(&([] as [&Digest; 0])).await.unwrap();
3482 assert!(results.is_empty());
3483
3484 let batch = db.new_batch().set(k3, v3);
3486 let results = batch.get_many(&[&k3, &k1, &k_missing], &db).await.unwrap();
3487 assert_eq!(results, vec![Some(v3), Some(v1), None]);
3488
3489 let parent = db
3491 .new_batch()
3492 .set(k3, v3)
3493 .merkleize(&db, None, db.inactivity_floor_loc())
3494 .await;
3495 let results = parent.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap();
3496 assert_eq!(results, vec![Some(v1), Some(v3), None]);
3497
3498 let v3_new = Sha256::fill(30u8);
3500 let child = parent.new_batch::<Sha256>().set(k3, v3_new);
3501 let results = child.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap();
3502 assert_eq!(results, vec![Some(v1), Some(v3_new), None]);
3503
3504 db.destroy().await.unwrap();
3505 }
3506
3507 #[boxed]
3509 pub(crate) async fn test_immutable_get_many_unexpected_data<F: Family, V, C>(
3510 context: deterministic::Context,
3511 open_db: impl Fn(
3512 deterministic::Context,
3513 ) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
3514 ) where
3515 V: ValueEncoding<Value = Digest>,
3516 C: Mutable<Item = Operation<F, Digest, V>>,
3517 C::Item: EncodeShared,
3518 {
3519 let mut db = open_db(context.child("db")).await;
3520
3521 let key = Sha256::fill(1u8);
3522 let value = Sha256::fill(11u8);
3523 db.apply_batch(
3524 db.new_batch()
3525 .set(key, value)
3526 .merkleize(&db, None, db.inactivity_floor_loc())
3527 .await,
3528 )
3529 .await
3530 .unwrap();
3531 db.commit().await.unwrap();
3532
3533 let bad_key = Sha256::fill(99u8);
3534 let bad_loc = db.last_commit_loc;
3535 db.snapshot.insert(&bad_key, bad_loc);
3536
3537 let err = db.get(&bad_key).await.unwrap_err();
3538 assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc));
3539
3540 let err = db.get_many(&[&bad_key]).await.unwrap_err();
3541 assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc));
3542
3543 db.destroy().await.unwrap();
3544 }
3545}