use crate::{
Context,
index::Factory as IndexFactory,
journal::{
authenticated,
contiguous::{fixed::Config as FConfig, variable::Config as VConfig},
},
merkle::{self, Location, full::Config as MerkleConfig},
qmdb::{
any::{
self, Config as AnyConfig,
operation::{Operation, Update},
},
bitmap::Shared,
},
translator::Translator,
};
use commonware_codec::{Codec, FixedSize};
use commonware_cryptography::Hasher;
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use commonware_runtime::Spawner;
use commonware_utils::bitmap::Prunable as BitMap;
use core::num::NonZeroUsize;
use std::sync::Arc;
pub mod batch;
pub mod db;
pub mod grafting;
pub mod ordered;
pub mod proof;
pub(crate) mod sync;
pub mod unordered;
use self::db::Metrics;
#[derive(Clone)]
pub struct Config<T: Translator, J, S: Strategy, B = ()> {
pub merkle_config: MerkleConfig<S>,
pub journal_config: J,
pub grafted_metadata_partition: String,
pub translator: T,
pub init_cache: Option<NonZeroUsize>,
pub init_buffer: NonZeroUsize,
pub init_concurrency: B,
}
impl<T: Translator, J, S: Strategy, B> From<Config<T, J, S, B>> for AnyConfig<T, J, S, B> {
fn from(cfg: Config<T, J, S, B>) -> Self {
Self {
merkle_config: cfg.merkle_config,
journal_config: cfg.journal_config,
translator: cfg.translator,
init_cache: cfg.init_cache,
init_buffer: cfg.init_buffer,
init_concurrency: cfg.init_concurrency,
}
}
}
pub type FixedConfig<T, S, B = ()> = Config<T, FConfig, S, B>;
pub type VariableConfig<T, C, S, B = ()> = Config<T, VConfig<C>, S, B>;
#[boxed]
pub(super) async fn init<F, E, U, H, I, J, const N: usize, S>(
context: E,
config: Config<I::Translator, J::Config, S, <I as crate::qmdb::SnapshotBuild<F>>::Concurrency>,
max_size: Option<Location<F>>,
) -> Result<db::Db<F, E, J, I, H, U, N, S>, crate::qmdb::Error<F>>
where
F: merkle::Graftable,
E: Context + Spawner,
U: Update,
H: Hasher,
I: IndexFactory<Value = Location<F>> + crate::qmdb::SnapshotBuild<F>,
J: authenticated::Backing<E, Item = Operation<F, U>> + 'static,
S: Strategy,
Operation<F, U>: Codec,
{
const {
assert!(
N.is_multiple_of(H::Digest::SIZE),
"chunk size must be some multiple of the digest size",
);
assert!(N.is_power_of_two(), "chunk size must be a power of 2");
}
let strategy = config.merkle_config.strategy.clone();
let metadata_partition = config.grafted_metadata_partition.clone();
let (metadata, pruned_chunks, pinned_nodes) =
db::init_metadata(context.child("metadata"), &metadata_partition).await?;
let bitmap = BitMap::<N>::new_with_pruned_chunks(pruned_chunks)
.map_err(|_| crate::qmdb::Error::<F>::DataCorrupted("pruned chunks overflow"))?;
let bitmap = Arc::new(Shared::<N>::new(bitmap));
let any = any::init_with_bitmap(
context.child("any"),
config.into(),
Some(bitmap),
max_size,
db::pair_absorption_threshold::<F, N>(pruned_chunks as u64),
)
.await?;
let (grafted_tree, graft_roots, root) = db::rebuild_grafted_tree::<F, H, S, N>(
any.bitmap.as_ref(),
&pinned_nodes,
&any.log.merkle,
any.inactivity_floor_loc,
any.root(),
&strategy,
)
.await?;
let metrics = Metrics::new(context);
let db = db::Db {
any,
grafted_tree: Arc::new(grafted_tree),
graft_roots,
metadata,
strategy,
root,
metrics,
#[cfg(test)]
halt_before_prune_log: false,
};
db.update_metrics();
Ok(db)
}
#[cfg(any(test, feature = "test-traits"))]
pub trait BitmapPrunedBits {
fn pruned_bits(&self) -> u64;
fn get_bit(&self, index: u64) -> bool;
fn oldest_retained(&self) -> u64;
}
#[cfg(test)]
pub mod tests {
pub use super::BitmapPrunedBits;
use super::{
Codec, FConfig, FixedConfig, MerkleConfig, Operation, Strategy, Update, VConfig,
VariableConfig, batch, db, grafting, ordered, unordered,
};
use crate::{
index::{Ordered as OrderedIndex, Unordered as UnorderedIndex},
journal::contiguous::Mutable,
merkle::{self, mmb, mmr, storage::Storage as _},
qmdb::{
any::{
ValueEncoding,
batch::Slot,
operation::update,
test::{
Changes, Choice, Inspect, Links, Neighbors, Script, assert_bits, assert_exact,
build, colliding_digest, counter, hold, hold_batch, live, replay,
test_any_activity_depths, test_any_ordered_policy_eviction_matrix,
test_any_ordered_policy_repair_across_ancestors,
test_any_policy_ancestor_twins, test_any_policy_decisions_match_writes,
test_any_policy_evicts_parent_created_key, test_any_policy_hold,
test_any_policy_keep_evict_and_recover,
test_any_policy_limits_after_colliding_writes,
test_any_policy_limits_with_writes, test_any_policy_matches_proportional,
test_any_policy_own_writes, test_any_proportional_bound,
test_any_proportional_one_batch,
},
traits::{DbAny, MerkleizedBatch as _, UnmerkleizedBatch as _},
},
chain::Bounds,
floor::{Bounded, Hold, Proportional},
store::tests::{TestKey, TestValue},
verify_proof,
},
translator::Translator,
};
use commonware_parallel::Sequential;
use commonware_runtime::{
Runner as _, Supervisor as _,
buffer::paged::CacheRef,
deterministic::{self, Context},
};
use commonware_utils::{NZU16, NZU64, NZUsize, TestRng, bitmap::Readable};
use core::future::Future;
use ordered::tests::test_build_small_close_reopen as test_ordered_build_small_close_reopen;
use rand::Rng;
use std::{
collections::BTreeMap,
num::{NonZeroU16, NonZeroUsize},
ops::Range,
pin::Pin,
sync::Arc,
};
use tracing::warn;
use unordered::tests::test_build_small_close_reopen as test_unordered_build_small_close_reopen;
type Error<F> = crate::qmdb::Error<F>;
type Location<F> = merkle::Location<F>;
type WriteVec<F, C> = Vec<(<C as DbAny<F>>::Key, Option<<C as DbAny<F>>::Value>)>;
const PAGE_SIZE: NonZeroU16 = NZU16!(88);
const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(8);
macro_rules! staged_merkleize_parity_test {
($name:ident, $open_db:path) => {
#[test_traced("WARN")]
pub fn $name() {
fn key(i: u64) -> Digest {
Sha256::hash(&[&i.to_be_bytes()])
}
fn val(i: u64) -> Digest {
Sha256::hash(&[&(i + 10000).to_be_bytes()])
}
deterministic::Runner::default().start(|ctx| async move {
let db = $open_db(ctx.child("current"), "staged-parity".to_string()).await;
let mut seed = db.new_batch();
for i in 0..2000u64 {
seed = seed.write(key(i), Some(val(i)));
}
let seed = seed.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(seed).await.unwrap().commit().await.unwrap();
for depth in [0u8, 1u8, 2u8] {
let mut stack = Vec::new();
match depth {
0 => {}
1 => {
let mut p = db.new_batch();
for i in 0..50u64 {
p = p.write(key(i), Some(val(i + 1_000)));
}
for i in 100..110u64 {
p = p.write(key(i), None);
}
stack.push(
p.merkleize(&db, None, &mut Proportional)
.await
.unwrap(),
);
}
2 => {
let mut grandparent = db.new_batch();
for i in 0..10u64 {
grandparent = grandparent.write(key(i), Some(val(i + 1_000)));
}
for i in 100..110u64 {
grandparent = grandparent.write(key(i), None);
}
let grandparent = grandparent
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let mut p = grandparent.new_batch::<Sha256>();
for i in 20..30u64 {
p = p.write(key(i), Some(val(i + 2_000)));
}
let p = p.merkleize(&db, None, &mut Proportional).await.unwrap();
stack.push(grandparent);
stack.push(p);
}
_ => unreachable!("covered depths"),
};
let new_batch = || {
stack
.last()
.map_or_else(|| db.new_batch(), |p| p.new_batch::<Sha256>())
};
let read_keys = [
key(5),
key(6),
key(9000),
key(5),
key(0),
key(20),
key(60),
key(105),
];
let keys: Vec<&Digest> = read_keys.iter().collect();
let slots = vec![
Slot::Put(val(5_000)),
Slot::Keep,
Slot::Put(val(5_001)),
Slot::Put(val(5_002)),
Slot::Put(val(5_003)),
Slot::Delete,
Slot::Put(val(5_004)),
Slot::Put(val(5_005)),
];
let unstaged = vec![
(key(7000), Some(val(6_000))),
(key(30), Some(val(6_001))),
(key(5), Some(val(6_002))),
(key(31), None),
];
let mut explicit = new_batch();
let explicit_values = explicit.get_many(&keys, &db).await.unwrap();
for (key, slot) in read_keys.iter().zip(&slots) {
match slot {
Slot::Keep => {}
Slot::Put(value) => explicit = explicit.write(*key, Some(*value)),
Slot::Delete => explicit = explicit.write(*key, None),
}
}
for (k, v) in &unstaged {
explicit = explicit.write(*k, *v);
}
let explicit_root = explicit
.merkleize(&db, None, &mut Proportional)
.await
.unwrap()
.root();
let (staged_values, staged) = new_batch().stage(&keys, &db).await.unwrap();
let staged_root = staged
.merkleize(
slots.clone(),
unstaged.clone(),
None,
&db,
&mut Proportional,
)
.await
.unwrap()
.root();
assert_eq!(
explicit_values, staged_values,
"value mismatch at depth={depth}"
);
assert_eq!(explicit_root, staged_root, "root mismatch at depth={depth}");
let split = 3;
let (mut expanded_values, staged) =
new_batch().stage(&keys[..split], &db).await.unwrap();
let (range, suffix_values, staged) =
staged.expand(&keys[split..], &db).await.unwrap();
assert_eq!(range, split..keys.len());
assert_eq!(staged.len(), keys.len());
assert!(!staged.is_empty());
expanded_values.extend(suffix_values);
let expanded_root = staged
.merkleize(
slots.clone(),
unstaged.clone(),
None,
&db,
&mut Proportional,
)
.await
.unwrap()
.root();
assert_eq!(
explicit_values, expanded_values,
"expanded value mismatch at depth={depth}"
);
assert_eq!(
explicit_root, expanded_root,
"expanded root mismatch at depth={depth}"
);
let planned = val(7_000);
let duplicate_update = val(7_001);
let (first_values, staged) =
new_batch().stage(&keys[..1], &db).await.unwrap();
let (duplicate_range, duplicate_values, staged) =
staged.expand(&keys[..1], &db).await.unwrap();
assert_eq!(duplicate_range, 1..2);
assert_eq!(
first_values[0], duplicate_values[0],
"duplicate expansion must assign a new slot without changing the base read"
);
assert_ne!(
duplicate_values[0],
Some(planned),
"expand must not observe values computed for earlier staged slots"
);
let duplicate_root = staged
.merkleize(
vec![Slot::Put(planned), Slot::Put(duplicate_update)],
Vec::new(),
None,
&db,
&mut Proportional,
)
.await
.unwrap()
.root();
let expected_duplicate_root = new_batch()
.write(read_keys[0], Some(planned))
.write(read_keys[0], Some(duplicate_update))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap()
.root();
assert_eq!(
expected_duplicate_root, duplicate_root,
"duplicate expanded slots should use normal update-order semantics"
);
}
});
}
};
}
pub(crate) use staged_merkleize_parity_test;
pub(crate) fn fixed_config<T: Translator + Default>(
partition_prefix: &str,
) -> FixedConfig<T, Sequential> {
fixed_config_full(partition_prefix, ())
}
pub(crate) fn fixed_config_full<T: Translator + Default, B>(
partition_prefix: &str,
init_concurrency: B,
) -> FixedConfig<T, Sequential, B> {
let page_cache = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE);
FixedConfig {
merkle_config: MerkleConfig {
journal_partition: format!("{partition_prefix}-journal-partition"),
metadata_partition: format!("{partition_prefix}-metadata-partition"),
items_per_blob: NZU64!(11),
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
strategy: Sequential,
page_cache: page_cache.clone(),
},
journal_config: FConfig {
partition: format!("{partition_prefix}-partition-prefix"),
items_per_blob: NZU64!(7),
page_cache,
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
},
grafted_metadata_partition: format!("{partition_prefix}-grafted-metadata-partition"),
translator: T::default(),
init_cache: Some(NZUsize!(1024)),
init_buffer: NZUsize!(1 << 21),
init_concurrency,
}
}
pub(crate) fn variable_config<T: Translator + Default>(
partition_prefix: &str,
) -> VariableConfig<T, ((), ()), Sequential> {
variable_config_full(partition_prefix, ())
}
pub(crate) fn variable_config_full<T: Translator + Default, B>(
partition_prefix: &str,
init_concurrency: B,
) -> VariableConfig<T, ((), ()), Sequential, B> {
let page_cache = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE);
VariableConfig {
merkle_config: MerkleConfig {
journal_partition: format!("{partition_prefix}-journal-partition"),
metadata_partition: format!("{partition_prefix}-metadata-partition"),
items_per_blob: NZU64!(11),
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
strategy: Sequential,
page_cache: page_cache.clone(),
},
journal_config: VConfig {
partition: format!("{partition_prefix}-partition-prefix"),
items_per_section: NZU64!(7),
compression: None,
codec_config: ((), ()),
page_cache,
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
},
grafted_metadata_partition: format!("{partition_prefix}-grafted-metadata-partition"),
translator: T::default(),
init_cache: Some(NZUsize!(1024)),
init_buffer: NZUsize!(1 << 21),
init_concurrency,
}
}
pub(crate) fn fixed_config_partitioned<T: Translator + Default>(
partition_prefix: &str,
) -> FixedConfig<T, Sequential, NonZeroUsize> {
fixed_config_full(partition_prefix, NZUsize!(1))
}
pub(crate) fn variable_config_partitioned<T: Translator + Default>(
partition_prefix: &str,
) -> VariableConfig<T, ((), ()), Sequential, NonZeroUsize> {
variable_config_full(partition_prefix, NZUsize!(1))
}
fn commit_writes<'a, F: merkle::Graftable, C: DbAny<F> + 'a>(
db: C,
writes: impl IntoIterator<Item = (C::Key, Option<<C as DbAny<F>>::Value>)> + 'a,
) -> std::pin::Pin<Box<dyn Future<Output = Result<C, Error<F>>> + 'a>> {
Box::pin(async move {
let mut batch = db.new_batch();
for (k, v) in writes {
batch = batch.write(k, v);
}
let merkleized = batch.merkleize(&db, None, &mut Proportional).await?;
let db = db.apply_batch(merkleized).await?;
db.commit().await
})
}
#[boxed]
pub async fn apply_random_ops<F, C>(
num_elements: u64,
commit_changes: bool,
rng_seed: u64,
mut db: C,
) -> Result<C, Error<F>>
where
F: merkle::Graftable,
C: DbAny<F>,
C::Key: TestKey,
<C as DbAny<F>>::Value: TestValue,
{
warn!("rng_seed={}", rng_seed);
let mut rng = TestRng::new(rng_seed);
let writes: Vec<_> = (0u64..num_elements)
.map(|i| {
let k = TestKey::from_seed(i);
let v = TestValue::from_seed(rng.next_u64());
(k, Some(v))
})
.collect();
if commit_changes {
db = commit_writes(db, writes).await?;
}
let mut pending: WriteVec<F, C> = Vec::new();
for _ in 0u64..num_elements * 10 {
let rand_key = TestKey::from_seed(rng.next_u64() % num_elements);
if rng.next_u32().is_multiple_of(7) {
pending.push((rand_key, None));
continue;
}
let v = TestValue::from_seed(rng.next_u64());
pending.push((rand_key, Some(v)));
if commit_changes && rng.next_u32().is_multiple_of(20) {
db = commit_writes(db, pending.drain(..)).await?;
}
}
if commit_changes {
db = commit_writes(db, pending).await?;
}
Ok(db)
}
#[boxed]
async fn build_random_close_reopen_round<M, C, F, Fut>(
mut context: Context,
mut open_db: F,
) -> String
where
M: merkle::Graftable + 'static,
C: DbAny<M> + 'static,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut,
Fut: Future<Output = C>,
{
const ELEMENTS: u64 = 1000;
let partition = "build-random".to_string();
let rng_seed = context.next_u64();
let db: C = open_db(context.child("first"), partition.clone()).await;
let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
.await
.unwrap();
let merkleized = db
.new_batch()
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.sync().await.unwrap();
let root = db.root();
drop(db);
let db: C = open_db(context.child("second"), partition).await;
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
context.auditor().state()
}
pub async fn test_build_random_close_reopen<M, C, F, Fut>(context: Context, open_db: F)
where
M: merkle::Graftable,
C: DbAny<M> + 'static,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
let state1 =
build_random_close_reopen_round::<M, C, F, Fut>(context, open_db.clone()).await;
let executor = deterministic::Runner::default();
let state2 = executor
.start(|context| build_random_close_reopen_round::<M, C, F, Fut>(context, open_db));
assert_eq!(state1, state2);
}
pub async fn test_commit_after_sync_recovery<M, C, F, Fut>(context: Context, mut open_db: F)
where
M: merkle::Graftable,
C: DbAny<M> + 'static,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
let mut open_db_clone = open_db.clone();
let partition = "commit-after-sync".to_string();
let db: C = Box::pin(open_db_clone(context.child("first"), partition.clone())).await;
let key0 = <<C as DbAny<M>>::Key as TestKey>::from_seed(0);
let key1 = <<C as DbAny<M>>::Key as TestKey>::from_seed(1);
let value0 = <C as DbAny<M>>::Value::from_seed(100);
let value1 = <C as DbAny<M>>::Value::from_seed(200);
let db = commit_writes::<M, C>(db, [(key0, Some(value0.clone()))])
.await
.unwrap();
let db = db.sync().await.unwrap();
let db = commit_writes::<M, C>(db, [(key1, Some(value1.clone()))])
.await
.unwrap();
let committed_root = db.root();
let committed_size = db.size();
drop(db);
let db: C = Box::pin(open_db(context.child("second"), partition)).await;
assert_eq!(db.root(), committed_root);
assert_eq!(db.size(), committed_size);
assert_eq!(db.get(&key0).await.unwrap(), Some(value0));
assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
db.destroy().await.unwrap();
}
pub async fn test_simulate_write_failures<M, C, F, Fut>(mut context: Context, mut open_db: F)
where
M: merkle::Graftable,
C: DbAny<M> + 'static,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
const ELEMENTS: u64 = 1000;
let partition = "build-random-fail-commit".to_string();
let rng_seed = context.next_u64();
let db: C = Box::pin(open_db(context.child("first"), partition.clone())).await;
let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
.await
.unwrap();
let db = commit_writes(db, []).await.unwrap();
let committed_root = db.root();
let committed_op_count = db.bounds().end;
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
let db = apply_random_ops::<M, C>(ELEMENTS, false, rng_seed + 1, db)
.await
.unwrap();
drop(db);
let db: C = Box::pin(open_db(
context.child("scenario").with_attribute("index", 1),
partition.clone(),
))
.await;
assert_eq!(db.root(), committed_root);
assert_eq!(db.bounds().end, committed_op_count);
let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed + 1, db)
.await
.unwrap();
let committed_op_count = db.bounds().end;
drop(db);
let db: C = Box::pin(open_db(
context.child("scenario").with_attribute("index", 2),
partition.clone(),
))
.await;
let scenario_2_root = db.root();
let fresh_partition = "build-random-fail-commit-fresh".to_string();
let db: C = Box::pin(open_db(context.child("fresh"), fresh_partition.clone())).await;
let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
.await
.unwrap();
let db = commit_writes(db, []).await.unwrap();
let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed + 1, db)
.await
.unwrap();
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
assert_eq!(db.bounds().end, committed_op_count);
assert_eq!(db.root(), scenario_2_root);
db.destroy().await.unwrap();
}
pub async fn test_different_pruning_delays_same_root<M, C, F, Fut>(
context: Context,
mut open_db: F,
) where
M: merkle::Graftable,
C: DbAny<M>,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
const NUM_OPERATIONS: u64 = 1000;
let mut open_db_clone = open_db.clone();
let mut db_no_pruning: C = Box::pin(open_db_clone(
context.child("no_pruning"),
"no-pruning-test".into(),
))
.await;
let mut db_pruning: C =
Box::pin(open_db(context.child("pruning"), "pruning-test".into())).await;
let mut pending_no_pruning: WriteVec<M, C> = Vec::new();
let mut pending_pruning: WriteVec<M, C> = Vec::new();
for i in 0..NUM_OPERATIONS {
let key: C::Key = TestKey::from_seed(i);
let value: <C as DbAny<M>>::Value = TestValue::from_seed(i * 1000);
pending_no_pruning.push((key, Some(value.clone())));
pending_pruning.push((key, Some(value)));
if i % 50 == 49 {
db_no_pruning = commit_writes(db_no_pruning, pending_no_pruning.drain(..))
.await
.unwrap();
db_pruning = commit_writes(db_pruning, pending_pruning.drain(..))
.await
.unwrap();
db_pruning = db_pruning
.prune(db_no_pruning.sync_boundary())
.await
.unwrap();
}
}
let db_no_pruning = commit_writes(db_no_pruning, pending_no_pruning)
.await
.unwrap();
let db_pruning = commit_writes(db_pruning, pending_pruning).await.unwrap();
let root_no_pruning = db_no_pruning.root();
let root_pruning = db_pruning.root();
assert_eq!(root_no_pruning, root_pruning);
assert_eq!(
db_no_pruning.inactivity_floor_loc(),
db_pruning.inactivity_floor_loc()
);
db_no_pruning.destroy().await.unwrap();
db_pruning.destroy().await.unwrap();
}
pub async fn test_sync_persists_bitmap_pruning_boundary<M, C, F, Fut>(
mut context: Context,
mut open_db: F,
) where
M: merkle::Graftable,
C: DbAny<M> + BitmapPrunedBits + 'static,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
const ELEMENTS: u64 = 500;
let mut open_db_clone = open_db.clone();
let partition = "sync-bitmap-pruning".to_string();
let rng_seed = context.next_u64();
let db: C = Box::pin(open_db_clone(context.child("first"), partition.clone())).await;
let db = apply_random_ops::<M, C>(ELEMENTS, true, rng_seed, db)
.await
.unwrap();
let merkleized = db
.new_batch()
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
let pruned_bits_before = db.pruned_bits();
warn!(
"pruned_bits_before={}, inactivity_floor={}, op_count={}",
pruned_bits_before,
*db.inactivity_floor_loc(),
*db.bounds().end
);
assert!(
pruned_bits_before > 0,
"Expected bitmap to have pruned bits after prune()"
);
let db = db.sync().await.unwrap();
let root_before = db.root();
drop(db);
let db: C = Box::pin(open_db(context.child("second"), partition)).await;
let pruned_bits_after = db.pruned_bits();
warn!("pruned_bits_after={}", pruned_bits_after);
assert_eq!(
pruned_bits_after, pruned_bits_before,
"Bitmap pruned bits mismatch after reopen - sync() may not have called write_pruned()"
);
assert_eq!(db.root(), root_before);
db.destroy().await.unwrap();
}
pub async fn test_current_db_build_big<M, C, F, Fut>(context: Context, mut open_db: F)
where
M: merkle::Graftable,
C: DbAny<M>,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
const ELEMENTS: u64 = 1000;
let mut open_db_clone = open_db.clone();
let db: C = Box::pin(open_db_clone(context.child("first"), "build-big".into())).await;
let mut map = std::collections::HashMap::<C::Key, <C as DbAny<M>>::Value>::default();
let mut batch = db.new_batch();
for i in 0u64..ELEMENTS {
let k: C::Key = TestKey::from_seed(i);
let v: <C as DbAny<M>>::Value = TestValue::from_seed(i * 1000);
batch = batch.write(k, Some(v.clone()));
map.insert(k, v);
}
for i in 0u64..ELEMENTS {
if i % 3 != 0 {
continue;
}
let k: C::Key = TestKey::from_seed(i);
let v: <C as DbAny<M>>::Value = TestValue::from_seed((i + 1) * 10000);
batch = batch.write(k, Some(v.clone()));
map.insert(k, v);
}
for i in 0u64..ELEMENTS {
if i % 7 != 1 {
continue;
}
let k: C::Key = TestKey::from_seed(i);
batch = batch.write(k, None);
map.remove(&k);
}
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.sync().await.unwrap();
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
let root = db.root();
db.sync().await.unwrap();
let db: C = Box::pin(open_db(context.child("second"), "build-big".into())).await;
assert_eq!(root, db.root());
for i in 0u64..ELEMENTS {
let k: C::Key = TestKey::from_seed(i);
if let Some(map_value) = map.get(&k) {
let Some(db_value) = db.get(&k).await.unwrap() else {
panic!("key not found in db: {k}");
};
assert_eq!(*map_value, db_value);
} else {
assert!(db.get(&k).await.unwrap().is_none());
}
}
}
pub async fn test_stale_batch_side_effect_free<M, C, F, Fut>(context: Context, mut open_db: F)
where
M: merkle::Graftable,
C: DbAny<M>,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut,
Fut: Future<Output = C>,
{
let db: C = Box::pin(open_db(
context.child("db"),
"stale-side-effect-free".into(),
))
.await;
let key1 = <C::Key as TestKey>::from_seed(1);
let key2 = <C::Key as TestKey>::from_seed(2);
let value1 = <<C as DbAny<M>>::Value as TestValue>::from_seed(10);
let value2 = <<C as DbAny<M>>::Value as TestValue>::from_seed(20);
let mut batch = db.new_batch();
batch = batch.write(key1, Some(value1.clone()));
let batch_a = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let mut batch = db.new_batch();
batch = batch.write(key2, Some(value2));
let batch_b = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(batch_a).await.unwrap();
let db = db.commit().await.unwrap();
let expected_root = db.root();
let expected_bounds = db.bounds();
let expected_metadata = db.get_metadata().await.unwrap();
assert_eq!(db.get(&key1).await.unwrap(), Some(value1.clone()));
assert_eq!(db.get(&key2).await.unwrap(), None);
let Err(err) = db.apply_batch(batch_b).await else {
panic!("expected StaleBatch error");
};
assert!(
matches!(err, Error::StaleBatch),
"expected StaleBatch error, got {err:?}"
);
let db: C = Box::pin(open_db(
context.child("reopen"),
"stale-side-effect-free".into(),
))
.await;
assert_eq!(db.root(), expected_root);
assert_eq!(db.bounds(), expected_bounds);
assert_eq!(db.get_metadata().await.unwrap(), expected_metadata);
assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
assert_eq!(db.get(&key2).await.unwrap(), None);
db.destroy().await.unwrap();
}
struct Observed<M: merkle::Family> {
root: Digest,
bounds: Range<Location<M>>,
floor: Location<M>,
pruned: u64,
bits: Vec<bool>,
}
impl<M: merkle::Family> Observed<M> {
fn capture<C>(db: &C) -> Self
where
C: DbAny<M, Digest = Digest> + BitmapPrunedBits,
{
let bounds = db.bounds();
let pruned = db.pruned_bits();
Self {
root: db.root(),
floor: db.inactivity_floor_loc(),
bits: (pruned..*bounds.end).map(|i| db.get_bit(i)).collect(),
pruned,
bounds,
}
}
fn assert_bits(&self, expected: &Self, label: &str) {
assert_eq!(
self.bounds.end, expected.bounds.end,
"{label}: size diverged",
);
let from = self.pruned.max(expected.pruned);
assert_eq!(
self.bits[(from - self.pruned) as usize..],
expected.bits[(from - expected.pruned) as usize..],
"{label}: activity bits diverged",
);
}
}
fn indexed(i: u64) -> Digest {
colliding_digest((i / 4) as u8, i)
}
pub async fn test_chained_schedules_match_sequential<M, U, const N: usize, S, C, F, Fut>(
context: Context,
mut open_db: F,
) where
M: merkle::Graftable,
U: Update,
S: Strategy,
Operation<M, U>: Codec,
C: DbAny<
M,
Key = Digest,
Value = Digest,
Digest = Digest,
Merkleized = Arc<batch::MerkleizedBatch<M, Digest, U, N, S>>,
Batch = batch::UnmerkleizedBatch<M, Sha256, U, N, S>,
> + BitmapPrunedBits,
F: FnMut(Context, String) -> Fut,
Fut: Future<Output = C>,
{
const SEED: u64 = 2 * CHUNK_BITS + 50;
let k0 = indexed(0);
let k1 = indexed(2 * CHUNK_BITS + 10);
let kx = indexed(SEED);
let c1 = indexed(CHUNK_BITS + 50);
let fresh = SEED + 1;
let keys: Vec<_> = (0..=fresh + CHUNK_BITS).map(indexed).collect();
let seed: WriteVec<M, C> = (0..SEED).map(|i| (indexed(i), Some(val(i)))).collect();
let mut a: WriteVec<M, C> = (0..CHUNK_BITS)
.map(|i| (indexed(i), Some(val(SEED + i))))
.collect();
a.extend([(k1, None), (kx, Some(val(5000)))]);
let mut b: WriteVec<M, C> = vec![(k0, Some(val(5001))), (k1, Some(val(5002))), (kx, None)];
b.extend((fresh..fresh + CHUNK_BITS).map(|i| (indexed(i), Some(val(i)))));
let e: WriteVec<M, C> = Vec::new();
let c: WriteVec<M, C> = vec![(k0, Some(val(5003))), (k1, None), (c1, Some(val(5004)))];
let d: WriteVec<M, C> = vec![(k0, Some(val(5005))), (kx, Some(val(5006)))];
let f: WriteVec<M, C> = vec![
(k0, Some(val(5007))),
(c1, Some(val(5008))),
(indexed(fresh + CHUNK_BITS), Some(val(5009))),
];
let reference: C =
Box::pin(open_db(context.child("reference"), "chained-ref".into())).await;
let reference = commit_writes(reference, seed.clone()).await.unwrap();
let reference = commit_writes(reference, a.clone()).await.unwrap();
let size_a = *reference.size();
let reference = commit_writes(reference, b.clone()).await.unwrap();
let size_b = *reference.size();
let reference = commit_writes(reference, e.clone()).await.unwrap();
let reference = commit_writes(reference, c.clone()).await.unwrap();
let size_c = *reference.size();
let reference = commit_writes(reference, d.clone()).await.unwrap();
assert!(
size_b / CHUNK_BITS > size_a / CHUNK_BITS,
"B must complete a chunk",
);
assert_eq!(
size_c / CHUNK_BITS,
size_b / CHUNK_BITS,
"E and C must not complete a chunk",
);
let expected = Observed::capture(&reference);
let f_batch = build(&reference, reference.new_batch(), &f).await;
let f_root = f_batch.root();
let reference = reference.apply_batch(f_batch).await.unwrap();
let expected_f = Observed::capture(&reference);
let mut values = Vec::with_capacity(keys.len());
for key in &keys {
values.push(reference.get(key).await.unwrap());
}
reference.destroy().await.unwrap();
for schedule in 1..=5u64 {
let partition = format!("chained-{schedule}");
let db: C = Box::pin(open_db(
context.child("schedule").with_attribute("index", schedule),
partition.clone(),
))
.await;
let db = commit_writes(db, seed.clone()).await.unwrap();
let a_batch = build(&db, db.new_batch(), &a).await;
let b_batch = build(&db, a_batch.new_batch::<Sha256>(), &b).await;
let e_batch = build(&db, b_batch.new_batch::<Sha256>(), &e).await;
let c_batch = build(&db, e_batch.new_batch::<Sha256>(), &c).await;
let (db, pruned) = match schedule {
1 => {
let d_batch = build(&db, c_batch.new_batch::<Sha256>(), &d).await;
(db.apply_batch(d_batch).await.unwrap(), false)
}
2 => {
let d_batch = build(&db, c_batch.new_batch::<Sha256>(), &d).await;
let db = db.apply_batch(a_batch).await.unwrap();
(db.apply_batch(d_batch).await.unwrap(), false)
}
3 => {
let db = db.apply_batch(a_batch).await.unwrap();
let d_batch = build(&db, c_batch.new_batch::<Sha256>(), &d).await;
(db.apply_batch(d_batch).await.unwrap(), false)
}
4 => {
let db = db.apply_batch(a_batch).await.unwrap();
let d_batch = build(&db, c_batch.new_batch::<Sha256>(), &d).await;
let db = db.apply_batch(b_batch).await.unwrap();
(db.apply_batch(d_batch).await.unwrap(), false)
}
5 => {
let d_batch = build(&db, c_batch.new_batch::<Sha256>(), &d).await;
let db = db.apply_batch(a_batch).await.unwrap();
let db = db.commit().await.unwrap();
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
assert!(db.pruned_bits() > 0, "S5 must prune bitmap bits");
(db.apply_batch(d_batch).await.unwrap(), true)
}
_ => unreachable!("five schedules"),
};
let label = format!("S{schedule}");
let observed = Observed::capture(&db);
assert_eq!(observed.root, expected.root, "{label}: root diverged");
if !pruned {
assert_eq!(observed.bounds, expected.bounds, "{label}: bounds diverged");
}
assert_eq!(observed.floor, expected.floor, "{label}: floor diverged");
observed.assert_bits(&expected, &label);
let f_batch = build(&db, db.new_batch(), &f).await;
assert_eq!(f_batch.root(), f_root, "{label}: follow-on root diverged");
let db = db.apply_batch(f_batch).await.unwrap();
let observed_f = Observed::capture(&db);
observed_f.assert_bits(&expected_f, &format!("{label} after F"));
db.sync().await.unwrap();
let db: C = Box::pin(open_db(
context.child("reopen").with_attribute("index", schedule),
partition,
))
.await;
let reopened = Observed::capture(&db);
assert_eq!(
reopened.root, observed_f.root,
"{label}: root diverged on reopen",
);
reopened.assert_bits(&observed_f, &format!("{label} on reopen"));
for (key, value) in keys.iter().zip(&values) {
assert_eq!(
&db.get(key).await.unwrap(),
value,
"{label}: value of {key} diverged on reopen",
);
}
db.destroy().await.unwrap();
}
}
use crate::translator::OneCap;
use commonware_cryptography::{Hasher as _, Sha256, sha256::Digest};
use commonware_macros::{boxed, test_group, test_traced};
type OrderedFixedDb =
ordered::fixed::Db<mmr::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
type OrderedVariableDb =
ordered::variable::Db<mmr::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
type UnorderedFixedDb =
unordered::fixed::Db<mmr::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
type UnorderedVariableDb = unordered::variable::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
32,
Sequential,
>;
type OrderedFixedP1Db = ordered::fixed::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type OrderedVariableP1Db = ordered::variable::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type UnorderedFixedP1Db = unordered::fixed::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type UnorderedVariableP1Db = unordered::variable::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type OrderedFixedP2Db = ordered::fixed::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
type OrderedVariableP2Db = ordered::variable::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
type UnorderedFixedP2Db = unordered::fixed::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
type UnorderedVariableP2Db = unordered::variable::partitioned::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
type OrderedFixedMmbDb =
ordered::fixed::Db<mmb::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
type OrderedVariableMmbDb =
ordered::variable::Db<mmb::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
type UnorderedFixedMmbDb =
unordered::fixed::Db<mmb::Family, Context, Digest, Digest, Sha256, OneCap, 32, Sequential>;
type UnorderedVariableMmbDb = unordered::variable::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
32,
Sequential,
>;
type OrderedFixedMmbP1Db = ordered::fixed::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type OrderedVariableMmbP1Db = ordered::variable::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type UnorderedFixedMmbP1Db = unordered::fixed::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type UnorderedVariableMmbP1Db = unordered::variable::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
1,
32,
Sequential,
>;
type OrderedFixedMmbP2Db = ordered::fixed::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
type OrderedVariableMmbP2Db = ordered::variable::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
type UnorderedFixedMmbP2Db = unordered::fixed::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
type UnorderedVariableMmbP2Db = unordered::variable::partitioned::Db<
mmb::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
2,
32,
Sequential,
>;
#[test_traced]
fn test_reconstruction_views_expose_all_chunk_states() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let db = UnorderedFixedMmbDb::init(
context.child("db"),
fixed_config::<OneCap>("reconstruction-views"),
None,
)
.await
.unwrap();
let mut batch = db.new_batch();
for i in 0u64..512 {
let key = Sha256::hash(&[&i.to_be_bytes()]);
let value = Sha256::hash(&[&(i + 1_000).to_be_bytes()]);
batch = batch.write(key, Some(value));
}
let batch = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(batch).await.unwrap();
let end = db.bounds().end;
let bitmap = db.bitmap();
assert_eq!(bitmap.len(), *end);
assert_eq!(bitmap.pruned_chunks(), 0);
let first = bitmap.get_chunk(0);
let chunk_bits = first.len() as u64 * 8;
let grafting_height = chunk_bits.trailing_zeros();
let complete = bitmap.complete_chunks() as u64;
let graftable =
grafting::graftable_chunks::<mmb::Family>(*end, grafting_height).min(complete);
assert_eq!(complete - graftable, 1, "expected one pending chunk");
assert!(first.iter().any(|byte| *byte != 0));
let pending = bitmap.get_chunk(graftable as usize);
assert!(pending.iter().any(|byte| *byte != 0));
let (partial, partial_bits) = bitmap.last_chunk();
assert!(partial_bits > 0 && partial_bits < chunk_bits);
assert!(partial.iter().any(|byte| *byte != 0));
let witness = db.ops_root_witness().await.unwrap();
assert_eq!(
witness.pending_chunk_digest,
Some(Sha256::hash(&[pending.as_slice()]))
);
assert_eq!(
witness.partial_chunk,
Some((partial_bits, Sha256::hash(&[partial.as_slice()])))
);
let storage = db.grafted_storage();
assert_eq!(
storage.size(),
<mmb::Family as merkle::Family>::location_to_position(end)
);
let ops_pos = <mmb::Family as merkle::Graftable>::subtree_root_position(
Location::new(0),
grafting_height,
);
assert!(storage.get_node(ops_pos).await.unwrap().is_some());
let pinned_positions =
<mmb::Family as merkle::Family>::nodes_to_pin(end).collect::<Vec<_>>();
let raw_pinned = db.pinned_nodes_at(end).await.unwrap();
assert_eq!(pinned_positions.len(), raw_pinned.len());
let mut checked_grafted_node = false;
for (pos, raw_digest) in pinned_positions.into_iter().zip(raw_pinned) {
let digest = storage.get_node(pos).await.unwrap().unwrap();
if pos == ops_pos {
assert_ne!(digest, raw_digest);
checked_grafted_node = true;
}
}
assert!(checked_grafted_node);
drop(storage);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_partitioned_ordered_short_key_exclusion_not_forgeable() {
type ForgedExclusionDb = ordered::variable::partitioned::Db<
mmr::Family,
Context,
Vec<u8>,
Vec<u8>,
Sha256,
OneCap,
2,
32,
Sequential,
>;
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let page_cache = CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE);
let cfg = VariableConfig {
merkle_config: MerkleConfig {
journal_partition: "forged-exclusion-journal".to_string(),
metadata_partition: "forged-exclusion-metadata".to_string(),
items_per_blob: NZU64!(11),
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
strategy: Sequential,
page_cache: page_cache.clone(),
},
journal_config: VConfig {
partition: "forged-exclusion-log".to_string(),
items_per_section: NZU64!(7),
compression: None,
codec_config: (((0..=8).into(), ()), ((0..=8).into(), ())),
page_cache,
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
},
grafted_metadata_partition: "forged-exclusion-grafted".to_string(),
translator: OneCap,
init_cache: Some(NZUsize!(1024)),
init_buffer: NZUsize!(1 << 21),
init_concurrency: NZUsize!(1),
};
let db = ForgedExclusionDb::init(context.child("db"), cfg, None)
.await
.unwrap();
let a = vec![0x00u8, 0x80];
let b = vec![0x01u8];
let c = vec![0x01u8, 0x00];
let (va, vb, vc) = (vec![0x0au8], vec![0x0bu8], vec![0x0cu8]);
let merkleized = db
.new_batch()
.write(a.clone(), Some(va.clone()))
.write(b.clone(), Some(vb.clone()))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let merkleized = db
.new_batch()
.write(c.clone(), Some(vc.clone()))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let root = db.root();
assert_eq!(db.get(&a).await.unwrap(), Some(va));
assert_eq!(db.get(&b).await.unwrap(), Some(vb));
assert_eq!(db.get(&c).await.unwrap(), Some(vc));
let (_, span_b) = db.get_span(&b).await.unwrap().unwrap();
assert_eq!(
span_b.next_key, c,
"b.next_key must be c, not the wrapped-around a"
);
assert!(matches!(
db.exclusion_proof(&c).await,
Err(Error::KeyExists)
));
let kvp = db.key_value_proof(b.clone()).await.unwrap();
let forged = ordered::proof::constant::ExclusionProof::KeyValue(kvp.proof, span_b);
assert!(!forged.verify::<Sha256>(&c, &root));
db.destroy().await.unwrap();
});
}
macro_rules! open_db_fn {
($db:ty, $cfg:ident) => {
|ctx: Context, partition: String| async move {
<$db>::init(ctx.child("storage"), $cfg::<OneCap>(&partition), None)
.await
.unwrap()
}
};
}
macro_rules! with_all_variants {
($cb:ident!($($args:tt)*)) => {
$cb!($($args)*, of, OrderedFixedDb, fixed_config);
$cb!($($args)*, ov, OrderedVariableDb, variable_config);
$cb!($($args)*, uf, UnorderedFixedDb, fixed_config);
$cb!($($args)*, uv, UnorderedVariableDb, variable_config);
$cb!($($args)*, ofp1, OrderedFixedP1Db, fixed_config_partitioned);
$cb!($($args)*, ovp1, OrderedVariableP1Db, variable_config_partitioned);
$cb!($($args)*, ufp1, UnorderedFixedP1Db, fixed_config_partitioned);
$cb!($($args)*, uvp1, UnorderedVariableP1Db, variable_config_partitioned);
$cb!($($args)*, ofp2, OrderedFixedP2Db, fixed_config_partitioned);
$cb!($($args)*, ovp2, OrderedVariableP2Db, variable_config_partitioned);
$cb!($($args)*, ufp2, UnorderedFixedP2Db, fixed_config_partitioned);
$cb!($($args)*, uvp2, UnorderedVariableP2Db, variable_config_partitioned);
$cb!($($args)*, of_mmb, OrderedFixedMmbDb, fixed_config);
$cb!($($args)*, ov_mmb, OrderedVariableMmbDb, variable_config);
$cb!($($args)*, uf_mmb, UnorderedFixedMmbDb, fixed_config);
$cb!($($args)*, uv_mmb, UnorderedVariableMmbDb, variable_config);
$cb!($($args)*, ofp1_mmb, OrderedFixedMmbP1Db, fixed_config_partitioned);
$cb!($($args)*, ovp1_mmb, OrderedVariableMmbP1Db, variable_config_partitioned);
$cb!($($args)*, ufp1_mmb, UnorderedFixedMmbP1Db, fixed_config_partitioned);
$cb!($($args)*, uvp1_mmb, UnorderedVariableMmbP1Db, variable_config_partitioned);
$cb!($($args)*, ofp2_mmb, OrderedFixedMmbP2Db, fixed_config_partitioned);
$cb!($($args)*, ovp2_mmb, OrderedVariableMmbP2Db, variable_config_partitioned);
$cb!($($args)*, ufp2_mmb, UnorderedFixedMmbP2Db, fixed_config_partitioned);
$cb!($($args)*, uvp2_mmb, UnorderedVariableMmbP2Db, variable_config_partitioned);
};
}
macro_rules! with_ordered_variants {
($cb:ident!($($args:tt)*)) => {
$cb!($($args)*, of, OrderedFixedDb, fixed_config);
$cb!($($args)*, ov, OrderedVariableDb, variable_config);
$cb!($($args)*, ofp1, OrderedFixedP1Db, fixed_config_partitioned);
$cb!($($args)*, ovp1, OrderedVariableP1Db, variable_config_partitioned);
$cb!($($args)*, ofp2, OrderedFixedP2Db, fixed_config_partitioned);
$cb!($($args)*, ovp2, OrderedVariableP2Db, variable_config_partitioned);
$cb!($($args)*, of_mmb, OrderedFixedMmbDb, fixed_config);
$cb!($($args)*, ov_mmb, OrderedVariableMmbDb, variable_config);
$cb!($($args)*, ofp1_mmb, OrderedFixedMmbP1Db, fixed_config_partitioned);
$cb!($($args)*, ovp1_mmb, OrderedVariableMmbP1Db, variable_config_partitioned);
$cb!($($args)*, ofp2_mmb, OrderedFixedMmbP2Db, fixed_config_partitioned);
$cb!($($args)*, ovp2_mmb, OrderedVariableMmbP2Db, variable_config_partitioned);
};
}
macro_rules! with_unordered_variants {
($cb:ident!($($args:tt)*)) => {
$cb!($($args)*, uf, UnorderedFixedDb, fixed_config);
$cb!($($args)*, uv, UnorderedVariableDb, variable_config);
$cb!($($args)*, ufp1, UnorderedFixedP1Db, fixed_config_partitioned);
$cb!($($args)*, uvp1, UnorderedVariableP1Db, variable_config_partitioned);
$cb!($($args)*, ufp2, UnorderedFixedP2Db, fixed_config_partitioned);
$cb!($($args)*, uvp2, UnorderedVariableP2Db, variable_config_partitioned);
$cb!($($args)*, uf_mmb, UnorderedFixedMmbDb, fixed_config);
$cb!($($args)*, uv_mmb, UnorderedVariableMmbDb, variable_config);
$cb!($($args)*, ufp1_mmb, UnorderedFixedMmbP1Db, fixed_config_partitioned);
$cb!($($args)*, uvp1_mmb, UnorderedVariableMmbP1Db, variable_config_partitioned);
$cb!($($args)*, ufp2_mmb, UnorderedFixedMmbP2Db, fixed_config_partitioned);
$cb!($($args)*, uvp2_mmb, UnorderedVariableMmbP2Db, variable_config_partitioned);
};
}
macro_rules! test_for_variant {
($f:ident, $traced:literal, $label:ident, $db:ty, $cfg:ident) => {
paste::paste! {
#[test_group("slow")]
#[test_traced($traced)]
fn [<$f _ $label>]() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
Box::pin($f(context, open_db_fn!($db, $cfg))).await
});
}
}
};
}
macro_rules! test_for_all_variants {
($f:ident, $traced:literal) => {
with_all_variants!(test_for_variant!($f, $traced));
};
}
macro_rules! test_for_ordered_variants {
($f:ident, $traced:literal) => {
with_ordered_variants!(test_for_variant!($f, $traced));
};
}
macro_rules! test_for_unordered_variants {
($f:ident, $traced:literal) => {
with_unordered_variants!(test_for_variant!($f, $traced));
};
}
async fn test_ordered_build_big<M, C, F, Fut>(context: Context, open_db: F)
where
M: merkle::Graftable,
C: DbAny<M>,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
test_current_db_build_big::<M, C, F, Fut>(context, open_db).await;
}
async fn test_unordered_build_big<M, C, F, Fut>(context: Context, open_db: F)
where
M: merkle::Graftable,
C: DbAny<M>,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
test_current_db_build_big::<M, C, F, Fut>(context, open_db).await;
}
test_for_all_variants!(test_build_random_close_reopen, "WARN");
test_for_all_variants!(test_simulate_write_failures, "WARN");
test_for_all_variants!(test_different_pruning_delays_same_root, "WARN");
test_for_all_variants!(test_sync_persists_bitmap_pruning_boundary, "WARN");
test_for_all_variants!(test_commit_after_sync_recovery, "WARN");
test_for_all_variants!(test_stale_batch_side_effect_free, "WARN");
test_for_all_variants!(test_chained_schedules_match_sequential, "WARN");
test_for_ordered_variants!(test_ordered_build_big, "WARN");
test_for_ordered_variants!(test_ordered_build_small_close_reopen, "DEBUG");
test_for_unordered_variants!(test_unordered_build_big, "WARN");
test_for_unordered_variants!(test_unordered_build_small_close_reopen, "DEBUG");
#[test_traced]
fn test_current_foreign_db_merkleize_rejected() {
deterministic::Runner::default().start(|context| async move {
let db_a = UnorderedFixedDb::init(
context.child("a"),
fixed_config::<OneCap>("foreign-a"),
None,
)
.await
.unwrap();
let db_b = UnorderedFixedDb::init(
context.child("b"),
fixed_config::<OneCap>("foreign-b"),
None,
)
.await
.unwrap();
let seed_a = db_a
.new_batch()
.write(key(1), Some(val(1)))
.merkleize(&db_a, None, &mut Proportional)
.await
.unwrap();
let db_a = db_a.apply_batch(seed_a).await.unwrap();
let seed_b = db_b
.new_batch()
.write(key(2), Some(val(2)))
.merkleize(&db_b, None, &mut Proportional)
.await
.unwrap();
let db_b = db_b.apply_batch(seed_b).await.unwrap();
assert_eq!(db_a.bounds().end, db_b.bounds().end);
assert_ne!(db_a.ops_root(), db_b.ops_root());
assert_ne!(db_a.root(), db_b.root());
let staged_keys = [key(1)];
let staged_refs: Vec<_> = staged_keys.iter().collect();
assert!(matches!(
db_a.new_batch().stage(&staged_refs, &db_b).await,
Err(Error::StaleRead)
));
let batch = db_a.new_batch().write(key(1), Some(val(3)));
assert!(matches!(
batch.merkleize(&db_b, None, &mut Proportional).await,
Err(Error::StaleBatch)
));
});
}
#[test_traced]
fn test_current_unordered_merkleize_rejects_foreign_instance() {
deterministic::Runner::default().start(|context| async move {
let db_a = UnorderedFixedDb::init(
context.child("a"),
fixed_config::<OneCap>("foreign-instance-unordered-a"),
None,
)
.await
.unwrap();
let db_b = UnorderedFixedDb::init(
context.child("b"),
fixed_config::<OneCap>("foreign-instance-unordered-b"),
None,
)
.await
.unwrap();
assert_eq!(db_a.root(), db_b.root());
let pending = db_a.new_batch().write(key(2), Some(val(2)));
let staged_keys = [key(2)];
let staged_refs: Vec<_> = staged_keys.iter().collect();
let (_, staged) = db_a.new_batch().stage(&staged_refs, &db_a).await.unwrap();
let sibling = db_a
.new_batch()
.write(key(1), Some(val(1)))
.merkleize(&db_a, None, &mut Proportional)
.await
.unwrap();
let _db_a = db_a.apply_batch(sibling).await.unwrap();
assert!(matches!(
pending.merkleize(&db_b, None, &mut Proportional).await,
Err(Error::StaleBatch)
));
assert!(matches!(
staged
.merkleize(
vec![Slot::Put(val(3))],
Vec::new(),
None,
&db_b,
&mut Proportional
)
.await,
Err(Error::StaleBatch)
));
});
}
#[test_traced]
fn test_current_ordered_merkleize_rejects_foreign_instance() {
deterministic::Runner::default().start(|context| async move {
let db_a = OrderedFixedDb::init(
context.child("a"),
fixed_config::<OneCap>("foreign-instance-ordered-a"),
None,
)
.await
.unwrap();
let db_b = OrderedFixedDb::init(
context.child("b"),
fixed_config::<OneCap>("foreign-instance-ordered-b"),
None,
)
.await
.unwrap();
assert_eq!(db_a.root(), db_b.root());
let pending = db_a.new_batch().write(key(2), Some(val(2)));
let staged_keys = [key(2)];
let staged_refs: Vec<_> = staged_keys.iter().collect();
let (_, staged) = db_a.new_batch().stage(&staged_refs, &db_a).await.unwrap();
let sibling = db_a
.new_batch()
.write(key(1), Some(val(1)))
.merkleize(&db_a, None, &mut Proportional)
.await
.unwrap();
let _db_a = db_a.apply_batch(sibling).await.unwrap();
assert!(matches!(
pending.merkleize(&db_b, None, &mut Proportional).await,
Err(Error::StaleBatch)
));
assert!(matches!(
staged
.merkleize(
vec![Slot::Put(val(3))],
Vec::new(),
None,
&db_b,
&mut Proportional
)
.await,
Err(Error::StaleBatch)
));
});
}
#[test_traced]
fn test_current_merkleize_rejects_stale_sibling() {
deterministic::Runner::default().start(|context| async move {
let db = OrderedFixedDb::init(
context.child("db"),
fixed_config::<OneCap>("stale-unmerkleized"),
None,
)
.await
.unwrap();
let seed = db
.new_batch()
.write(key(1), Some(val(1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap();
let stale = db.new_batch().write(key(1), Some(val(2)));
let sibling = db
.new_batch()
.write(key(1), Some(val(3)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(sibling).await.unwrap();
assert!(matches!(
stale.merkleize(&db, None, &mut Proportional).await,
Err(Error::StaleBatch)
));
});
}
#[test_traced]
fn test_current_staged_merkleize_rejects_stale_sibling() {
deterministic::Runner::default().start(|context| async move {
let db = UnorderedFixedDb::init(
context.child("db"),
fixed_config::<OneCap>("stale-staged"),
None,
)
.await
.unwrap();
let target = key(1);
let seed = db
.new_batch()
.write(target, Some(val(1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap();
let staged_keys = [&target];
let (_, staged) = db.new_batch().stage(&staged_keys, &db).await.unwrap();
let sibling = db
.new_batch()
.write(target, Some(val(2)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(sibling).await.unwrap();
assert!(matches!(
staged
.merkleize(
vec![Slot::Put(val(3))],
Vec::new(),
None,
&db,
&mut Proportional
)
.await,
Err(Error::StaleBatch)
));
});
}
const CHUNK_BITS: u64 = commonware_utils::bitmap::BitMap::<32>::CHUNK_SIZE_BITS;
fn key(i: u64) -> Digest {
Sha256::hash(&[&i.to_be_bytes()])
}
fn val(i: u64) -> Digest {
Sha256::hash(&[&(i + 10000).to_be_bytes()])
}
#[boxed]
async fn mmb_commit(
db: UnorderedVariableMmbDb,
writes: impl IntoIterator<Item = (Digest, Option<Digest>)>,
) -> UnorderedVariableMmbDb {
let mut batch = db.new_batch();
for (k, v) in writes {
batch = batch.write(k, v);
}
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
db.commit().await.unwrap()
}
#[boxed]
async fn commit_writes_with_metadata(
db: UnorderedVariableDb,
writes: impl IntoIterator<Item = (Digest, Option<Digest>)>,
metadata: Option<Digest>,
) -> UnorderedVariableDb {
let mut batch = db.new_batch();
for (k, v) in writes {
batch = batch.write(k, v);
}
let merkleized = batch
.merkleize(&db, metadata, &mut Proportional)
.await
.unwrap();
let expected_end = merkleized.bounds().tip.size;
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.bounds().end, expected_end);
db.commit().await.unwrap()
}
struct Generation {
size: Location<mmr::Family>,
floor: Location<mmr::Family>,
sync_boundary: Location<mmr::Family>,
root: Digest,
}
async fn build_generations(
ctx: &Context,
partition: &str,
generations: u64,
) -> (UnorderedVariableDb, Vec<Generation>) {
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let mut history = Vec::new();
for generation in 0..generations {
db = commit_writes_with_metadata(
db,
(0..384).map(|i| (key(i), Some(val(generation * 1_000 + i)))),
None,
)
.await;
history.push(Generation {
size: db.bounds().end,
floor: db.inactivity_floor_loc(),
sync_boundary: db.sync_boundary(),
root: db.root(),
});
}
(db, history)
}
#[test_traced("INFO")]
fn test_current_bounded_initialization_recovery() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let partition = "current-rewind-recovery";
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let initial_size = db.bounds().end;
let initial_root = db.root();
let initial_ops_root = db.ops_root();
let initial_floor = db.inactivity_floor_loc();
let metadata_a = val(900);
let db = commit_writes_with_metadata(
db,
[(key(0), Some(val(0))), (key(1), Some(val(1)))],
Some(metadata_a),
)
.await;
let size_before = db.bounds().end;
let root_before = db.root();
let ops_root_before = db.ops_root();
let floor_before = db.inactivity_floor_loc();
let metadata_b = val(901);
let db = commit_writes_with_metadata(
db,
[
(key(0), Some(val(100))),
(key(1), None),
(key(2), Some(val(2))),
],
Some(metadata_b),
)
.await;
assert_ne!(db.root(), root_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(val(901)));
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(100)));
assert_eq!(db.get(&key(1)).await.unwrap(), None);
assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2)));
let db = {
_ = db.sync().await.unwrap();
UnorderedVariableDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(size_before),
)
.await
}
.unwrap();
assert_eq!(db.bounds().end, size_before);
assert_eq!(db.root(), root_before);
assert_eq!(db.ops_root(), ops_root_before);
assert_eq!(db.inactivity_floor_loc(), floor_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
assert_eq!(db.get(&key(2)).await.unwrap(), None);
let db = db.commit().await.unwrap();
drop(db);
let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
context.child("reopen"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened.bounds().end, size_before);
assert_eq!(reopened.root(), root_before);
assert_eq!(reopened.ops_root(), ops_root_before);
assert_eq!(reopened.inactivity_floor_loc(), floor_before);
assert_eq!(reopened.get_metadata().await.unwrap(), Some(val(900)));
assert_eq!(reopened.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(reopened.get(&key(1)).await.unwrap(), Some(val(1)));
assert_eq!(reopened.get(&key(2)).await.unwrap(), None);
let reopened = {
_ = reopened.sync().await.unwrap();
UnorderedVariableDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(initial_size),
)
.await
}
.unwrap();
assert_eq!(reopened.bounds().end, initial_size);
assert_eq!(reopened.root(), initial_root);
assert_eq!(reopened.ops_root(), initial_ops_root);
assert_eq!(reopened.inactivity_floor_loc(), initial_floor);
assert_eq!(reopened.get_metadata().await.unwrap(), None);
assert_eq!(reopened.get(&key(0)).await.unwrap(), None);
assert_eq!(reopened.get(&key(1)).await.unwrap(), None);
assert_eq!(reopened.get(&key(2)).await.unwrap(), None);
let reopened = reopened.commit().await.unwrap();
drop(reopened);
let reopened_initial: UnorderedVariableDb = UnorderedVariableDb::init(
context.child("reopen_initial"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened_initial.bounds().end, initial_size);
assert_eq!(reopened_initial.root(), initial_root);
assert_eq!(reopened_initial.ops_root(), initial_ops_root);
assert_eq!(reopened_initial.inactivity_floor_loc(), initial_floor);
assert_eq!(reopened_initial.get_metadata().await.unwrap(), None);
assert_eq!(reopened_initial.get(&key(0)).await.unwrap(), None);
assert_eq!(reopened_initial.get(&key(1)).await.unwrap(), None);
assert_eq!(reopened_initial.get(&key(2)).await.unwrap(), None);
reopened_initial.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_bounded_initialization_recovery_pruned_repeated_updates() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 200;
let partition = "current-rewind-pruned-recovery";
let ctx = context.child("db");
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let key0 = key(0);
let mut history = Vec::new();
for round in 0..COMMITS {
db = commit_writes_with_metadata(db, [(key0, Some(val(20_000 + round)))], None)
.await;
history.push((
db.bounds().end,
db.inactivity_floor_loc(),
db.root(),
db.ops_root(),
val(20_000 + round),
));
}
let (_, older_floor, _, _, _) = history[history.len() - 4];
let prune_loc = Location::new(*older_floor / CHUNK_BITS * CHUNK_BITS);
let db = db.prune(prune_loc).await.unwrap();
let pruned_bits = db.pruned_bits();
assert!(
pruned_bits > 0,
"expected bitmap pruning: prune_loc={prune_loc} bounds={:?}",
db.bounds()
);
assert_eq!(pruned_bits, *prune_loc);
let bounds = db.bounds();
let (target_size, target_root, target_ops_root, target_value) = history
.iter()
.enumerate()
.find_map(|(idx, (size, floor, root, ops_root, value))| {
let removed_commits = history.len() - idx - 1;
if removed_commits >= 3 && *size > bounds.start && *floor >= pruned_bits {
Some((*size, *root, *ops_root, *value))
} else {
None
}
})
.unwrap_or_else(|| {
panic!(
"expected recoverable pruned target with repeated updates. \
bounds={bounds:?}, pruned_bits={pruned_bits}, latest_floor={:?}, \
history={history:?}",
db.inactivity_floor_loc()
)
});
_ = db.sync().await.unwrap();
let db = UnorderedVariableDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(target_size),
)
.await
.unwrap();
assert_eq!(db.root(), target_root);
assert_eq!(db.ops_root(), target_ops_root);
assert_eq!(db.bounds().end, target_size);
assert_eq!(db.get(&key0).await.unwrap(), Some(target_value));
let db = db.commit().await.unwrap();
drop(db);
let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
context.child("reopen_pruned_recovery"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened.root(), target_root);
assert_eq!(reopened.ops_root(), target_ops_root);
assert_eq!(reopened.bounds().end, target_size);
assert_eq!(reopened.get(&key0).await.unwrap(), Some(target_value));
let metadata_after_reopen = val(30_000);
let new_key = key(1);
let new_value = val(30_001);
let reopened = commit_writes_with_metadata(
reopened,
[(new_key, Some(new_value))],
Some(metadata_after_reopen),
)
.await;
let size_after_new_write = reopened.bounds().end;
let root_after_new_write = reopened.root();
let ops_root_after_new_write = reopened.ops_root();
assert_eq!(
reopened.get_metadata().await.unwrap(),
Some(metadata_after_reopen)
);
assert_eq!(reopened.get(&key0).await.unwrap(), Some(target_value));
assert_eq!(reopened.get(&new_key).await.unwrap(), Some(new_value));
drop(reopened);
let reopened_after_new_write: UnorderedVariableDb = UnorderedVariableDb::init(
context.child("reopen_pruned_after_new_write"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened_after_new_write.root(), root_after_new_write);
assert_eq!(
reopened_after_new_write.ops_root(),
ops_root_after_new_write
);
assert_eq!(reopened_after_new_write.bounds().end, size_after_new_write);
assert_eq!(
reopened_after_new_write.get_metadata().await.unwrap(),
Some(metadata_after_reopen)
);
assert_eq!(
reopened_after_new_write.get(&key0).await.unwrap(),
Some(target_value)
);
assert_eq!(
reopened_after_new_write.get(&new_key).await.unwrap(),
Some(new_value)
);
reopened_after_new_write.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_mmb_settlement_guard_defers_pruning() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 100;
let partition = "current-mmb-reopen-prove-after-prune";
let ctx = context.child("db");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let k = key(0);
let mut expected = None;
for round in 0..COMMITS {
expected = Some(val(50_000 + round));
let mut batch = db.new_batch();
batch = batch.write(k, expected);
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
db = db.apply_batch(merkleized).await.unwrap();
db = db.commit().await.unwrap();
}
let root_before = db.root();
assert!(
*db.inactivity_floor_loc() >= 256,
"expected inactivity floor past chunk 0"
);
assert_eq!(
*db.sync_boundary(),
0,
"settlement guard should hold boundary at 0 during unsettled window"
);
let Err(err) = db.prune(Location::<mmb::Family>::new(1)).await else {
panic!("expected PruneBeyondMinRequired");
};
assert!(
matches!(err, Error::PruneBeyondMinRequired(_, _)),
"expected PruneBeyondMinRequired, got {err:?}"
);
let reopened: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
context.child("reopen"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened.pruned_bits(), 0);
assert_eq!(reopened.root(), root_before);
assert_eq!(reopened.get(&k).await.unwrap(), expected);
let _proof = reopened.key_value_proof(k).await.unwrap();
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_mmb_bounded_initialization_rejects_unsettled_pruned_window() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 320;
const N: usize = 32;
let partition = "current-mmb-rewind-unsettled-window";
let ctx = context.child("db");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let key0 = key(0);
let mut history = Vec::new();
for round in 0..COMMITS {
let mut batch = db.new_batch();
batch = batch.write(key0, Some(val(60_000 + round)));
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
db = db.apply_batch(merkleized).await.unwrap();
db = db.commit().await.unwrap();
history.push((db.bounds().end, db.inactivity_floor_loc()));
}
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
let pruned_bits = db.pruned_bits();
assert!(pruned_bits > 0, "expected MMB bitmap pruning to be active");
let db = db.sync().await.unwrap();
let chunk_bits = commonware_utils::bitmap::BitMap::<N>::CHUNK_SIZE_BITS;
let pruned_chunks = (pruned_bits / chunk_bits) as u64;
let gh = super::grafting::height::<N>();
let youngest = pruned_chunks - 1;
let pair_chunk = youngest & !1;
let pair_start = pair_chunk << gh;
let pair_pos = <mmb::Family as merkle::Graftable>::subtree_root_position(
merkle::Location::<mmb::Family>::new(pair_start),
gh + 1,
);
let absorbed_after =
<mmb::Family as merkle::Graftable>::peak_birth_size(pair_pos, gh + 1);
let unsafe_target = history
.iter()
.filter_map(|(size, floor)| {
let s = **size;
if s >= pruned_bits && s < absorbed_after && **floor >= pruned_bits {
Some(s)
} else {
None
}
})
.max()
.unwrap_or_else(|| {
panic!(
"expected initialization target in unsettled window. \
pruned_bits={pruned_bits}, absorbed_after={absorbed_after}, \
history={history:?}"
)
});
let original_root = db.root();
_ = db.sync().await.unwrap();
let Err(err) = UnorderedVariableMmbDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(merkle::Location::<mmb::Family>::new(unsafe_target)),
)
.await
else {
panic!("expected initialization rejection in unsettled delayed-merge window");
};
assert!(
matches!(err, Error::HistoricalFloorPruned(_)),
"unexpected bounded initialization error for unsettled delayed-merge window. \
{err:?}"
);
let db = UnorderedVariableMmbDb::init(
ctx.child("unchanged"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(db.root(), original_root);
});
}
#[test_traced]
fn test_current_mmb_prune_respects_sync_boundary() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 320;
let ctx = context.child("db");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
ctx.child("storage"),
variable_config::<OneCap>("prune-clip-mmb"),
None,
)
.await
.unwrap();
let k = key(0);
for round in 0..COMMITS {
db = mmb_commit(db, [(k, Some(val(70_000 + round)))]).await;
}
let prune_boundary = db.sync_boundary();
let db = db.prune(prune_boundary).await.unwrap();
let boundary = db.sync_boundary();
let floor = db.inactivity_floor_loc();
assert!(
boundary < floor,
"delayed-merge lag must be strictly active: boundary={boundary}, floor={floor}"
);
assert!(
db.bounds().start <= boundary,
"ops journal was pruned past the settled bitmap boundary: \
bounds.start={}, boundary={boundary}",
db.bounds().start
);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_mmr_prune_boundary_lag_is_only_chunk_alignment() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 320;
const N: usize = 32;
let ctx = context.child("db");
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("prune-clip-mmr"), None,
)
.await
.unwrap();
for round in 0..COMMITS {
db = commit_writes_with_metadata(
db,
[(key(0), Some(val(80_000 + round)))],
None,
)
.await;
}
let prune_boundary = db.sync_boundary();
let db = db.prune(prune_boundary).await.unwrap();
let boundary = db.sync_boundary();
let floor = db.inactivity_floor_loc();
let chunk_bits = commonware_utils::bitmap::BitMap::<N>::CHUNK_SIZE_BITS;
assert!(
boundary <= floor && *floor - *boundary < chunk_bits,
"MMR lag should be only chunk alignment: boundary={boundary}, floor={floor}, chunk_bits={chunk_bits}"
);
assert!(
db.bounds().start <= boundary,
"ops journal bounds must be <= sync_boundary: bounds.start={}, boundary={boundary}",
db.bounds().start
);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_prune_below_settled_boundary_is_honored() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 100;
let ctx = context.child("db");
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("prune-below-boundary"), None,
)
.await
.unwrap();
for round in 0..COMMITS {
db = commit_writes_with_metadata(db, [(key(0), Some(val(90_000 + round)))], None)
.await;
}
assert!(*db.inactivity_floor_loc() > 1);
let small = Location::new(1);
let db = db.prune(small).await.unwrap();
assert!(
db.bounds().start <= small,
"journal pruning exceeded the caller-supplied target: bounds.start={}, requested={small}",
db.bounds().start
);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_mmb_reopen_and_prove_after_prune_delayed_merge() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let db_ctx = context.child("db_init");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
db_ctx.child("db"),
variable_config::<OneCap>("test_prune_delayed_merge"),
None,
)
.await
.unwrap();
let k = key(0);
for round in 0..200u64 {
db = mmb_commit(db, [(k, Some(val(60_000 + round)))]).await;
}
let boundary = db.sync_boundary();
db = db.prune(boundary).await.unwrap();
db = db.sync().await.unwrap();
for round in 200..300u64 {
db = mmb_commit(db, [(key(1), Some(val(round)))]).await;
}
let proof = db.key_value_proof(k).await.unwrap();
assert!(UnorderedVariableMmbDb::verify_key_value_proof(
k,
val(60_000 + 199),
&proof,
&db.root()
));
let target_root = db.root();
drop(db);
let reopen_ctx = context.child("db_reopen");
let reopened: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
reopen_ctx.child("db"),
variable_config::<OneCap>("test_prune_delayed_merge"),
None,
)
.await
.unwrap();
assert_eq!(reopened.root(), target_root);
let proof = reopened.key_value_proof(k).await.unwrap();
assert!(UnorderedVariableMmbDb::verify_key_value_proof(
k,
val(60_000 + 199),
&proof,
&reopened.root()
));
reopened.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_mmb_reopen_after_prune_two_chunks() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let db_ctx = context.child("db");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
db_ctx.child("db"),
variable_config::<OneCap>("test_prune_two"),
None,
)
.await
.unwrap();
let k = key(0);
let mut expected;
let mut round = 0u64;
loop {
expected = Some(val(60_000 + round));
db = mmb_commit(db, [(k, expected)]).await;
round += 1;
let boundary = db.sync_boundary();
db = db.prune(boundary).await.unwrap();
if db.pruned_bits() >= 512 {
break;
}
assert!(
round < 500,
"failed to reach 2 pruned chunks after {round} commits"
);
}
let db = db.sync().await.unwrap();
let target_root = db.root();
drop(db);
let reopen_ctx = context.child("db_reopen");
let reopened: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
reopen_ctx.child("db"),
variable_config::<OneCap>("test_prune_two"),
None,
)
.await
.unwrap();
assert_eq!(reopened.root(), target_root);
assert_eq!(reopened.get(&k).await.unwrap(), expected);
reopened.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_mmb_repeated_prune() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let mut db_ctx = context.child("db_init");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
db_ctx.child("db"),
variable_config::<OneCap>("test_repeated_prune"),
None,
)
.await
.unwrap();
for round in 0..3u64 {
let k = key(round * 1000);
let mut expected = None;
for i in 0..90 {
expected = Some(val(round * 1000 + i));
db = mmb_commit(db, [(k, expected)]).await;
}
let boundary = db.sync_boundary();
db = db.prune(boundary).await.unwrap();
db = db.sync().await.unwrap();
let root_before = db.root();
db_ctx = context.child("db").with_attribute("round", round);
drop(db);
db = UnorderedVariableMmbDb::init(
db_ctx.child("db"),
variable_config::<OneCap>("test_repeated_prune"),
None,
)
.await
.unwrap();
assert_eq!(db.root(), root_before);
assert_eq!(db.get(&k).await.unwrap(), expected);
}
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_mmb_stepwise_growth_matches_unpruned_reference() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let db_ctx = context.child("db_stepwise");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
db_ctx.child("db"),
variable_config::<OneCap>("test_stepwise"),
None,
)
.await
.unwrap();
let ref_ctx = context.child("ref_stepwise");
let mut ref_db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
ref_ctx.child("db"),
variable_config::<OneCap>("test_stepwise_ref"),
None,
)
.await
.unwrap();
let k = key(0);
let mut commit_idx = 0u64;
while *db.inactivity_floor_loc() < 1024 {
let value = Some(val(80_000 + commit_idx));
db = mmb_commit(db, [(k, value)]).await;
ref_db = mmb_commit(ref_db, [(k, value)]).await;
commit_idx += 1;
}
let boundary = db.sync_boundary();
db = db.prune(boundary).await.unwrap();
db = db.sync().await.unwrap();
assert_eq!(
db.root(),
ref_db.root(),
"root mismatch immediately after prune"
);
loop {
let db_leaves =
*Location::<mmb::Family>::try_from(db.any.log.merkle.size()).unwrap();
if db_leaves >= 1560 {
break;
}
let value = Some(val(80_000 + commit_idx));
db = mmb_commit(db, [(k, value)]).await;
ref_db = mmb_commit(ref_db, [(k, value)]).await;
commit_idx += 1;
let db_leaves =
*Location::<mmb::Family>::try_from(db.any.log.merkle.size()).unwrap();
assert_eq!(
db.root(),
ref_db.root(),
"stepwise root mismatch: leaves={db_leaves}, commit_idx={commit_idx}"
);
}
db.destroy().await.unwrap();
ref_db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_mmb_large_repeated_prune_matches_unpruned_reference() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const ROUNDS: u64 = 8;
const COMMITS_PER_ROUND: u64 = 120;
let mut db_ctx = context.child("db_init");
let mut db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
db_ctx.child("db"),
variable_config::<OneCap>("test_large_prune"),
None,
)
.await
.unwrap();
let ref_ctx = context.child("ref");
let mut ref_db: UnorderedVariableMmbDb = UnorderedVariableMmbDb::init(
ref_ctx.child("db"),
variable_config::<OneCap>("test_large_prune_ref"),
None,
)
.await
.unwrap();
let k = key(0);
let mut expected = None;
for round in 0..ROUNDS {
for i in 0..COMMITS_PER_ROUND {
let value = Some(val(round * 10_000 + i));
expected = value;
db = mmb_commit(db, [(k, value)]).await;
ref_db = mmb_commit(ref_db, [(k, value)]).await;
}
assert_eq!(
db.root(),
ref_db.root(),
"root mismatch before prune at round {round}"
);
let boundary = db.sync_boundary();
db = db.prune(boundary).await.unwrap();
db = db.sync().await.unwrap();
assert_eq!(
db.root(),
ref_db.root(),
"root mismatch after prune at round {round}"
);
let proof = db.key_value_proof(k).await.unwrap();
assert!(
UnorderedVariableMmbDb::verify_key_value_proof(
k,
expected.expect("value should exist"),
&proof,
&db.root()
),
"proof verification failed at round {round}"
);
db_ctx = context.child("db_reopen").with_attribute("round", round);
drop(db);
db = UnorderedVariableMmbDb::init(
db_ctx.child("db"),
variable_config::<OneCap>("test_large_prune"),
None,
)
.await
.unwrap();
assert_eq!(
db.root(),
ref_db.root(),
"root mismatch after reopen at round {round}"
);
assert_eq!(
db.get(&k).await.unwrap(),
expected,
"value mismatch after reopen at round {round}"
);
let proof = db.key_value_proof(k).await.unwrap();
assert!(
UnorderedVariableMmbDb::verify_key_value_proof(
k,
expected.expect("value should exist"),
&proof,
&db.root()
),
"proof verification failed after reopen at round {round}"
);
}
db.destroy().await.unwrap();
ref_db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_prune_rejects_beyond_sync_boundary_without_mutation() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 160;
let partition = "current-prune-beyond-boundary";
let ctx = context.child("db");
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let key0 = key(0);
for round in 0..COMMITS {
db = commit_writes_with_metadata(db, [(key0, Some(val(40_000 + round)))], None)
.await;
}
let expected_root = db.root();
let expected_ops_root = db.ops_root();
let expected_boundary = db.sync_boundary();
let expected_pruned_bits = db.pruned_bits();
let expected_value = db.get(&key0).await.unwrap();
let invalid_prune_loc = expected_boundary + 256;
let Err(err) = db.prune(invalid_prune_loc).await else {
panic!("expected prune rejection above sync boundary");
};
assert!(
matches!(err, Error::PruneBeyondMinRequired(loc, boundary)
if loc == invalid_prune_loc && boundary == expected_boundary),
"expected prune rejection above sync boundary, got {err:?}"
);
let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
context.child("reopen"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened.root(), expected_root);
assert_eq!(reopened.ops_root(), expected_ops_root);
assert_eq!(reopened.pruned_bits(), expected_pruned_bits);
assert_eq!(reopened.get(&key0).await.unwrap(), expected_value);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_bounded_initialization_small_delta_large_history() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 200;
let partition = "current-rewind-small-delta";
let ctx = context.child("db");
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let key0 = key(0);
let key1 = key(1);
let mut history = Vec::new();
for round in 0..COMMITS {
let key0_value = val(40_000 + round);
let key1_value = if round % 3 == 1 {
None
} else {
Some(val(50_000 + round))
};
db = commit_writes_with_metadata(
db,
[(key0, Some(key0_value)), (key1, key1_value)],
None,
)
.await;
history.push((
db.bounds().end,
db.root(),
db.ops_root(),
key0_value,
key1_value,
));
}
let target = *history
.get(history.len() - 3)
.expect("history should contain at least three commits");
let (target_size, target_root, target_ops_root, target_key0, target_key1) = target;
let db = {
_ = db.sync().await.unwrap();
UnorderedVariableDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(target_size),
)
.await
}
.unwrap();
assert_eq!(db.bounds().end, target_size);
assert_eq!(db.root(), target_root);
assert_eq!(db.ops_root(), target_ops_root);
assert_eq!(db.get(&key0).await.unwrap(), Some(target_key0));
assert_eq!(db.get(&key1).await.unwrap(), target_key1);
let db = db.commit().await.unwrap();
drop(db);
let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
context.child("reopen_small_delta"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened.bounds().end, target_size);
assert_eq!(reopened.root(), target_root);
assert_eq!(reopened.ops_root(), target_ops_root);
assert_eq!(reopened.get(&key0).await.unwrap(), Some(target_key0));
assert_eq!(reopened.get(&key1).await.unwrap(), target_key1);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_bounded_initialization_pruned_target_errors() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const KEYS: u64 = 384;
let partition = "current-rewind-pruned";
let ctx = context.child("db");
let db: UnorderedVariableDb =
UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition), None,
)
.await
.unwrap();
let first_start = db.bounds().end;
let db = commit_writes_with_metadata(
db,
(0..KEYS).map(|i| (key(i), Some(val(i)))),
None,
)
.await;
let db = commit_writes_with_metadata(
db,
(0..KEYS).map(|i| (key(i), Some(val(1000 + i)))),
None,
)
.await;
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
let pruned_bits = db.pruned_bits();
assert!(
pruned_bits > *first_start,
"expected bitmap pruning boundary above initialization bound: pruned_bits={pruned_bits}, target={:?}",
first_start
);
let oldest_retained = db.bounds().start;
_ = db.sync().await.unwrap();
let Err(boundary_err) = UnorderedVariableDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(oldest_retained),
)
.await else {
panic!("expected initialization rejection at retained boundary");
};
assert!(
matches!(
boundary_err,
Error::Journal(crate::journal::Error::ItemPruned(_))
),
"unexpected bounded initialization error at retained boundary: {boundary_err:?}"
);
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("reopen"),
variable_config::<OneCap>(partition), None,
)
.await
.unwrap();
let expected_pruned_loc = *first_start;
_ = db.sync().await.unwrap();
let Err(err) = UnorderedVariableDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(first_start),
)
.await else {
panic!("expected initialization rejection at pruned target");
};
assert!(
matches!(
err,
Error::Journal(crate::journal::Error::ItemPruned(loc))
if loc == expected_pruned_loc
),
"unexpected bounded initialization error: {err:?}"
);
});
}
#[test_traced("INFO")]
fn test_current_bounded_initialization_rejects_target_below_bitmap_floor() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
const COMMITS: u64 = 120;
let partition = "current-rewind-bitmap-floor";
let ctx = context.child("db");
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let mut history = Vec::new();
for round in 0..COMMITS {
db = commit_writes_with_metadata(db, [(key(0), Some(val(10_000 + round)))], None)
.await;
history.push((db.bounds().end, db.inactivity_floor_loc()));
}
let prune_loc = Location::new(CHUNK_BITS);
assert!(prune_loc <= db.sync_boundary());
let db = db.prune(prune_loc).await.unwrap();
let pruned_bits = db.pruned_bits();
assert_eq!(pruned_bits, CHUNK_BITS);
let retained_start = db.bounds().start;
assert!(retained_start < prune_loc);
let (target_size, target_floor) = history
.iter()
.copied()
.find(|(size, floor)| {
**size >= pruned_bits && *floor >= retained_start && **floor < pruned_bits
})
.unwrap_or_else(|| {
panic!(
"expected initialization target below bitmap boundary. \
retained_start={retained_start:?}, pruned_bits={pruned_bits}, \
history={history:?}"
)
});
assert!(retained_start <= target_floor);
assert!(*target_floor < pruned_bits);
assert!(pruned_bits <= *target_size);
let original_root = db.root();
_ = db.sync().await.unwrap();
let Err(err) = UnorderedVariableDb::init(
ctx.child("cap"),
variable_config::<OneCap>(partition),
Some(target_size),
)
.await
else {
panic!("expected initialization rejection below bitmap floor");
};
assert!(
matches!(err, Error::HistoricalFloorPruned(loc) if loc == target_size),
"unexpected bounded initialization error: {err:?}"
);
let db = UnorderedVariableDb::init(
ctx.child("unchanged"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(db.root(), original_root);
});
}
#[test_traced("INFO")]
fn test_current_prune_keeps_retained_checkpoints_initializable() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let partition = "current-prune-retained-checkpoints";
let ctx = context.child("db");
let (db, history) = build_generations(&ctx, partition, 4).await;
assert!(history[2].floor < db.sync_boundary());
let prune_loc = history[1].sync_boundary;
assert!(prune_loc >= Location::new(CHUNK_BITS));
let db = db.prune(prune_loc).await.unwrap();
assert_eq!(db.pruned_bits(), *prune_loc);
let retained_start = db.bounds().start;
assert!(retained_start > Location::new(0));
assert!(history[0].size <= retained_start);
let db = db.sync().await.unwrap();
drop(db);
for (label, generation) in [
("gen3", &history[3]),
("gen2", &history[2]),
("gen1", &history[1]),
] {
let db = UnorderedVariableDb::init(
ctx.child(label),
variable_config::<OneCap>(partition),
Some(generation.size),
)
.await
.unwrap();
assert_eq!(db.bounds().end, generation.size);
assert_eq!(db.root(), generation.root);
assert_eq!(db.sync_boundary(), generation.sync_boundary);
}
let Err(err) = UnorderedVariableDb::init(
ctx.child("gen0"),
variable_config::<OneCap>(partition),
Some(history[0].size),
)
.await
else {
panic!("expected initialization rejection below the retained start");
};
assert!(
matches!(
err,
Error::Journal(crate::journal::Error::ItemPruned(loc))
if loc == *history[0].size
),
"unexpected bounded initialization error: {err:?}"
);
});
}
pub async fn test_speculative_root_matches_committed<M, C, F, Fut>(
context: Context,
mut open_db: F,
) where
M: merkle::Graftable,
C: DbAny<M> + 'static,
C::Key: TestKey,
<C as DbAny<M>>::Value: TestValue,
F: FnMut(Context, String) -> Fut + Clone,
Fut: Future<Output = C>,
{
let mut open_db_clone = open_db.clone();
let partition = "speculative-root".to_string();
let db: C = Box::pin(open_db_clone(context.child("init"), partition.clone())).await;
let mut batch = db.new_batch();
for i in 0..260 {
batch = batch.write(TestKey::from_seed(i), Some(TestValue::from_seed(i + 1000)));
}
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let speculative_root = db.root();
db.sync().await.unwrap();
let db: C = Box::pin(open_db(context.child("reopen"), partition)).await;
assert_eq!(db.root(), speculative_root);
db.destroy().await.unwrap();
}
test_for_all_variants!(test_speculative_root_matches_committed, "INFO");
#[test_traced("INFO")]
fn test_current_batch_merkleized_get() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let mut db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("mg"),
None,
)
.await
.unwrap();
let ka = key(0);
let kb = key(1);
let kc = key(2);
{
let mut batch = db.new_batch();
batch = batch.write(ka, Some(val(0)));
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
db = db.apply_batch(merkleized).await.unwrap();
}
let va2 = val(100);
let vb = val(1);
let mut batch = db.new_batch();
batch = batch.write(ka, Some(va2));
batch = batch.write(kb, Some(vb));
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
assert_eq!(merkleized.get(&ka, &db).await.unwrap(), Some(va2));
assert_eq!(merkleized.get(&kb, &db).await.unwrap(), Some(vb));
assert_eq!(merkleized.get(&kc, &db).await.unwrap(), None);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_batch_chaining() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("ch"),
None,
)
.await
.unwrap();
let mut parent = db.new_batch();
for i in 0..5 {
parent = parent.write(key(i), Some(val(i)));
}
let parent_m = parent
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let mut child = parent_m.new_batch::<Sha256>();
for i in 5..10 {
child = child.write(key(i), Some(val(i)));
}
child = child.write(key(0), Some(val(999)));
let child_m = child.merkleize(&db, None, &mut Proportional).await.unwrap();
let child_root = child_m.root();
assert_eq!(child_m.get(&key(0), &db).await.unwrap(), Some(val(999)));
assert_eq!(child_m.get(&key(3), &db).await.unwrap(), Some(val(3)));
assert_eq!(child_m.get(&key(7), &db).await.unwrap(), Some(val(7)));
let db = db.apply_batch(child_m).await.unwrap();
assert_eq!(db.root(), child_root);
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(999)));
for i in 1..10 {
assert_eq!(db.get(&key(i)).await.unwrap(), Some(val(i)));
}
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_unordered_root_matches_between_pending_and_committed_paths() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedFixedDb =
UnorderedFixedDb::init(ctx.child("storage"), fixed_config::<OneCap>("ucr"), None)
.await
.unwrap();
let key_a = colliding_digest(0xAA, 1);
let key_b = colliding_digest(0xAA, 0);
let mut initial = db.new_batch();
for i in 0..4 {
initial = initial.write(colliding_digest(0xAA, i), Some(colliding_digest(0xBB, i)));
}
let merkleized = initial
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let parent = db
.new_batch()
.write(key_a, Some(colliding_digest(0xCC, 1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let pending_child = parent
.new_batch::<Sha256>()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let pending_root = pending_child.root();
let pending_ops_root = pending_child.ops_root();
let db = db.apply_batch(parent).await.unwrap();
let db = db.commit().await.unwrap();
let committed_child = db
.new_batch()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
assert_eq!(pending_root, committed_child.root());
assert_eq!(pending_ops_root, committed_child.ops_root());
let db = db.apply_batch(pending_child).await.unwrap();
assert_eq!(db.root(), committed_child.root());
assert_eq!(db.ops_root(), committed_child.ops_root());
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_ordered_root_matches_between_pending_and_committed_paths() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: OrderedFixedDb =
OrderedFixedDb::init(ctx.child("storage"), fixed_config::<OneCap>("ocr"), None)
.await
.unwrap();
let key_a = colliding_digest(0xAA, 1);
let key_b = colliding_digest(0xAA, 0);
let mut initial = db.new_batch();
for i in 0..4 {
initial = initial.write(colliding_digest(0xAA, i), Some(colliding_digest(0xBB, i)));
}
let merkleized = initial
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let parent = db
.new_batch()
.write(key_a, Some(colliding_digest(0xCC, 1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let pending_child = parent
.new_batch::<Sha256>()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let pending_root = pending_child.root();
let pending_ops_root = pending_child.ops_root();
let db = db.apply_batch(parent).await.unwrap();
let db = db.commit().await.unwrap();
let committed_child = db
.new_batch()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
assert_eq!(pending_root, committed_child.root());
assert_eq!(pending_ops_root, committed_child.ops_root());
let db = db.apply_batch(pending_child).await.unwrap();
assert_eq!(db.root(), committed_child.root());
assert_eq!(db.ops_root(), committed_child.ops_root());
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_batch_apply_requires_commit_for_recovery() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let partition = "apply_requires_commit";
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let committed_root = db.root();
let merkleized = db
.new_batch()
.write(key(0), Some(val(0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
drop(db);
let reopened: UnorderedVariableDb = UnorderedVariableDb::init(
context.child("reopen"),
variable_config::<OneCap>(partition),
None,
)
.await
.unwrap();
assert_eq!(reopened.root(), committed_root);
assert_eq!(reopened.get(&key(0)).await.unwrap(), None);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_batch_single_stage_pipeline() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("pipe"),
None,
)
.await
.unwrap();
let mut batch = db.new_batch();
batch = batch.write(key(0), Some(val(0)));
let parent_merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(parent_merkleized).await.unwrap();
let child_merkleized = {
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
let mut child = db.new_batch();
child = child.write(key(1), Some(val(1)));
child.merkleize(&db, None, &mut Proportional).await.unwrap()
};
let db = db.commit().await.unwrap();
let db = db.apply_batch(child_merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_sequential_commit() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("ff"),
None,
)
.await
.unwrap();
let parent_m = db
.new_batch()
.write(key(0), Some(val(0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let child_m = parent_m
.new_batch::<Sha256>()
.write(key(1), Some(val(1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(parent_m).await.unwrap();
let db = db.apply_batch(child_m).await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
let ctx2 = context.child("db").with_attribute("index", 2);
let db2: UnorderedVariableDb =
UnorderedVariableDb::init(ctx2.child("db"), variable_config::<OneCap>("ff2"), None)
.await
.unwrap();
let m1 = db2
.new_batch()
.write(key(0), Some(val(0)))
.merkleize(&db2, None, &mut Proportional)
.await
.unwrap();
let db2 = db2.apply_batch(m1).await.unwrap();
let m2 = db2
.new_batch()
.write(key(1), Some(val(1)))
.merkleize(&db2, None, &mut Proportional)
.await
.unwrap();
let db2 = db2.apply_batch(m2).await.unwrap();
assert_eq!(db.root(), db2.root());
db.destroy().await.unwrap();
db2.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_to_batch_then_chain() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("tb"),
None,
)
.await
.unwrap();
let m = db
.new_batch()
.write(key(0), Some(val(0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(m).await.unwrap();
let snapshot = db.to_batch();
assert_eq!(snapshot.root(), db.root());
let child = snapshot
.new_batch::<Sha256>()
.write(key(1), Some(val(1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
assert_ne!(child.root(), snapshot.root());
let db = db.apply_batch(child).await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_live_batch_safe_across_prune() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("prune-live"),
None,
)
.await
.unwrap();
let mut seed = db.new_batch();
for i in 0u64..300 {
seed = seed.write(key(i), Some(val(i)));
}
let seed_m = seed.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(seed_m).await.unwrap();
let db = db.commit().await.unwrap();
let mut p = db.new_batch();
for i in 0u64..250 {
p = p.write(key(i), Some(val(i + 10_000)));
}
let p_m = p.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(Arc::clone(&p_m)).await.unwrap();
let db = db.commit().await.unwrap();
let c = p_m
.new_batch::<Sha256>()
.write(key(250), Some(val(99_999)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
assert_eq!(c.get(&key(250), &db).await.unwrap(), Some(val(99_999)));
let db = db.apply_batch(c).await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(10_000)));
assert_eq!(db.get(&key(250)).await.unwrap(), Some(val(99_999)));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_extend_applied_batch() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("xtend"),
None,
)
.await
.unwrap();
let a = db
.new_batch()
.write(key(0), Some(val(0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(Arc::clone(&a)).await.unwrap();
let b = a
.new_batch::<Sha256>()
.write(key(1), Some(val(1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(b).await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
let c = db
.new_batch()
.write(key(2), Some(val(2)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(c).await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2)));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_live_batch_child_after_prune() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("child-after-prune"),
None,
)
.await
.unwrap();
let mut seed = db.new_batch();
for i in 0u64..300 {
seed = seed.write(key(i), Some(val(i)));
}
let seed_m = seed.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(seed_m).await.unwrap();
let db = db.commit().await.unwrap();
let mut a_batch = db.new_batch();
for i in 0u64..250 {
a_batch = a_batch.write(key(i), Some(val(i + 10_000)));
}
let a = a_batch
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(Arc::clone(&a)).await.unwrap();
let db = db.commit().await.unwrap();
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
let b = a
.new_batch::<Sha256>()
.write(key(300), Some(val(300)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(b).await.unwrap();
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(10_000)));
assert_eq!(db.get(&key(249)).await.unwrap(), Some(val(10_249)));
assert_eq!(db.get(&key(300)).await.unwrap(), Some(val(300)));
db.destroy().await.unwrap();
});
}
#[test_traced("WARN")]
fn test_current_apply_after_ancestor_dropped() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("adrop"),
None,
)
.await
.unwrap();
let mut a = db.new_batch();
for i in 0..3 {
a = a.write(key(i), Some(val(i)));
}
let a_m = a.merkleize(&db, None, &mut Proportional).await.unwrap();
let mut b = a_m.new_batch::<Sha256>();
for i in 3..6 {
b = b.write(key(i), Some(val(i)));
}
let b_m = b.merkleize(&db, None, &mut Proportional).await.unwrap();
let mut c = b_m.new_batch::<Sha256>();
for i in 6..9 {
c = c.write(key(i), Some(val(i)));
}
let c_m = c.merkleize(&db, None, &mut Proportional).await.unwrap();
drop(a_m);
drop(b_m);
let db = db.apply_batch(c_m).await.unwrap().commit().await.unwrap();
for i in 0..9 {
assert_eq!(
db.get(&key(i)).await.unwrap(),
Some(val(i)),
"key({i}) missing after apply_batch with dropped ancestors"
);
}
db.destroy().await.unwrap();
});
}
#[test_traced("WARN")]
fn test_current_chain_bitmap_order_matches_sequential() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx1 = context.child("db").with_attribute("index", 1);
let db1: UnorderedVariableDb = UnorderedVariableDb::init(
ctx1.child("db"),
variable_config::<OneCap>("ord1"),
None,
)
.await
.unwrap();
let db1 = commit_writes_with_metadata(
db1,
[(key(10), Some(val(10))), (key(11), Some(val(11)))],
None,
)
.await;
let a = db1
.new_batch()
.write(key(10), Some(val(100)))
.write(key(11), None) .merkleize(&db1, None, &mut Proportional)
.await
.unwrap();
let b = a
.new_batch::<Sha256>()
.write(key(12), Some(val(120)))
.write(key(13), Some(val(130)))
.merkleize(&db1, None, &mut Proportional)
.await
.unwrap();
let c = b
.new_batch::<Sha256>()
.write(key(14), Some(val(140)))
.merkleize(&db1, None, &mut Proportional)
.await
.unwrap();
let db1 = db1.apply_batch(c).await.unwrap().commit().await.unwrap();
let d1 = db1
.new_batch()
.write(key(20), Some(val(200)))
.merkleize(&db1, None, &mut Proportional)
.await
.unwrap();
let chain_then_d_root = d1.root();
let ctx2 = context.child("db").with_attribute("index", 2);
let db2: UnorderedVariableDb = UnorderedVariableDb::init(
ctx2.child("db"),
variable_config::<OneCap>("ord2"),
None,
)
.await
.unwrap();
let db2 = commit_writes_with_metadata(
db2,
[(key(10), Some(val(10))), (key(11), Some(val(11)))],
None,
)
.await;
let a2 = db2
.new_batch()
.write(key(10), Some(val(100)))
.write(key(11), None)
.merkleize(&db2, None, &mut Proportional)
.await
.unwrap();
let db2 = db2.apply_batch(a2).await.unwrap().commit().await.unwrap();
let b2 = db2
.new_batch()
.write(key(12), Some(val(120)))
.write(key(13), Some(val(130)))
.merkleize(&db2, None, &mut Proportional)
.await
.unwrap();
let db2 = db2.apply_batch(b2).await.unwrap().commit().await.unwrap();
let c2 = db2
.new_batch()
.write(key(14), Some(val(140)))
.merkleize(&db2, None, &mut Proportional)
.await
.unwrap();
let db2 = db2.apply_batch(c2).await.unwrap().commit().await.unwrap();
let d2 = db2
.new_batch()
.write(key(20), Some(val(200)))
.merkleize(&db2, None, &mut Proportional)
.await
.unwrap();
let sequential_then_d_root = d2.root();
assert_eq!(
chain_then_d_root, sequential_then_d_root,
"batch D's root on top of chain-applied state must match sequential state"
);
db1.destroy().await.unwrap();
db2.destroy().await.unwrap();
});
}
#[test_traced("WARN")]
fn test_current_stale_bitmap_clears_after_prune() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("stale-clears"),
None,
)
.await
.unwrap();
let mut seed = db.new_batch();
for i in 0u64..255 {
seed = seed.write(key(i), Some(val(i)));
}
let seed_m = seed.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(seed_m).await.unwrap();
let db = db.commit().await.unwrap();
let mut p = db.new_batch();
for i in 1u64..255 {
p = p.write(key(i), Some(val(i + 10000)));
}
let p_m = p.merkleize(&db, None, &mut Proportional).await.unwrap();
let c_m = p_m
.new_batch::<Sha256>()
.write(key(0), Some(val(9999)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(p_m).await.unwrap().commit().await.unwrap();
let floor = *db.inactivity_floor_loc();
assert!(floor >= 256, "floor must be past chunk 0: floor={floor}",);
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
let db = db.apply_batch(c_m).await.unwrap();
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_partial_ancestor_commit() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("pac"),
None,
)
.await
.unwrap();
let a = db
.new_batch()
.write(key(0), Some(val(0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let b = a
.new_batch::<Sha256>()
.write(key(1), Some(val(1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let c = b
.new_batch::<Sha256>()
.write(key(2), Some(val(2)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let expected_root = c.root();
let db = db.apply_batch(a).await.unwrap();
let db = db.apply_batch(c).await.unwrap();
assert_eq!(db.root(), expected_root);
assert_eq!(db.get(&key(0)).await.unwrap(), Some(val(0)));
assert_eq!(db.get(&key(1)).await.unwrap(), Some(val(1)));
assert_eq!(db.get(&key(2)).await.unwrap(), Some(val(2)));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_partial_ancestor_bitmap_ordering() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("bmo"),
None,
)
.await
.unwrap();
let a = db
.new_batch()
.write(key(0), Some(val(0)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let b = a
.new_batch::<Sha256>()
.write(key(1), Some(val(1)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let c = b
.new_batch::<Sha256>()
.write(key(2), Some(val(2)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let d = c
.new_batch::<Sha256>()
.write(key(3), Some(val(3)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(a).await.unwrap();
let db = db.apply_batch(d.clone()).await.unwrap();
let e = db
.new_batch()
.write(key(4), Some(val(4)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(e).await.unwrap();
let ref_ctx = context.child("ref");
let mut ref_db: UnorderedVariableDb = UnorderedVariableDb::init(
ref_ctx.child("db"),
variable_config::<OneCap>("bmo_ref"),
None,
)
.await
.unwrap();
for i in 0..5 {
let batch = ref_db
.new_batch()
.write(key(i), Some(val(i)))
.merkleize(&ref_db, None, &mut Proportional)
.await
.unwrap();
ref_db = ref_db.apply_batch(batch).await.unwrap();
}
assert_eq!(
db.root(),
ref_db.root(),
"root mismatch: bitmap ordering bug"
);
db.destroy().await.unwrap();
ref_db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_apply_chunks_match_speculative_chunks() {
const N: usize = 32;
const CHUNK_SIZE_BITS: u64 = commonware_utils::bitmap::Prunable::<N>::CHUNK_SIZE_BITS;
const SEED_KEYS: u64 = CHUNK_SIZE_BITS + 50;
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedVariableDb = UnorderedVariableDb::init(
ctx.child("storage"),
variable_config::<OneCap>("spec_eq"),
None,
)
.await
.unwrap();
let seed = (0..SEED_KEYS).fold(db.new_batch(), |b, i| b.write(key(i), Some(val(i))));
let seed = seed.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(seed).await.unwrap().commit().await.unwrap();
assert!(
Readable::<N>::len(db.any.bitmap.as_ref()) > CHUNK_SIZE_BITS,
"setup must cross a chunk boundary",
);
let parent = db
.new_batch()
.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, &mut Proportional)
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.write(key(10), Some(val(210)))
.write(key(SEED_KEYS), None)
.write(key(75), None)
.write(key(CHUNK_SIZE_BITS + 30), Some(val(220)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let speculative_chunks: Vec<[u8; N]> = {
let len = Readable::<N>::len(&child.bitmap);
let chunk_count = len.div_ceil(CHUNK_SIZE_BITS) as usize;
(0..chunk_count)
.map(|idx| Readable::<N>::get_chunk(&child.bitmap, idx))
.collect()
};
assert!(speculative_chunks.len() >= 2);
let db = db.apply_batch(child).await.unwrap();
let committed_chunks: Vec<[u8; N]> = {
let len = Readable::<N>::len(db.any.bitmap.as_ref());
let chunk_count = len.div_ceil(CHUNK_SIZE_BITS) as usize;
(0..chunk_count)
.map(|idx| Readable::<N>::get_chunk(db.any.bitmap.as_ref(), idx))
.collect()
};
assert_eq!(
speculative_chunks, committed_chunks,
"speculative chunks must equal post-apply committed chunks across all chunks",
);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_mmb_ops_historical_proof_verifies_with_backward_bagging() {
let executor = deterministic::Runner::default();
executor.start(|context| async move {
let ctx = context.child("db");
let db: UnorderedFixedMmbDb = UnorderedFixedMmbDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("mmb-ops-proof"),
None,
)
.await
.unwrap();
let writes: Vec<(Digest, Option<Digest>)> =
(0u64..16).map(|i| (key(i), Some(val(i)))).collect();
let db = commit_writes(db, writes).await.unwrap();
let ops_root = db.ops_root();
let historical_size = db.bounds().end;
let (proof, ops) = db
.ops_historical_proof(historical_size, Location::new(0), NZU64!(32))
.await
.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(0),
&ops,
&ops_root
));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_ordered_policy_replace_and_ancestor_proofs() {
deterministic::Runner::default().start(|context| async move {
let ctx = context.child("db");
let partition = "current-ordered-policy-ancestor";
let db: OrderedFixedDb = OrderedFixedDb::init(
ctx.child("storage"),
fixed_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let mut keys = [key(1), key(2), key(3)];
keys.sort();
let seed = keys
.into_iter()
.enumerate()
.fold(db.new_batch(), |batch, (i, key)| {
batch.write(key, Some(val(i as u64)))
});
let seed = seed.merkleize(&db, None, &mut Hold).await.unwrap();
let db = db.apply_batch(seed).await.unwrap().commit().await.unwrap();
let parent = db
.new_batch()
.write(keys[1], Some(val(11)))
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
let oldest = keys[0];
let mut policy = Script::new(usize::MAX, 2, move |key: &Digest| {
if *key == oldest {
Choice::Replace(val(10))
} else {
Choice::Evict
}
});
let child = parent
.new_batch::<Sha256>()
.merkleize(&db, None, &mut policy)
.await
.unwrap();
let first = policy.visited[0].0;
assert_eq!(
policy.visited,
[
(first, keys[0], val(0)),
(Location::new(*first + 2), keys[2], val(2)),
]
);
assert_eq!(child.bounds().inactivity_floor, Location::new(*first + 3));
let speculative_root = child.root();
let db = db.apply_batch(parent).await.unwrap();
let db = db.apply_batch(child).await.unwrap();
let root = db.root();
assert_eq!(root, speculative_root);
let proof = db.key_value_proof(keys[0]).await.unwrap();
assert_eq!(proof.next_key, keys[1]);
assert!(
proof.verify::<Sha256, crate::qmdb::any::value::FixedEncoding<Digest>>(
keys[0],
val(10),
&root
)
);
let proof = db.key_value_proof(keys[1]).await.unwrap();
assert_eq!(proof.next_key, keys[0]);
assert!(
proof.verify::<Sha256, crate::qmdb::any::value::FixedEncoding<Digest>>(
keys[1],
val(11),
&root
)
);
let exclusion = db.exclusion_proof(&keys[2]).await.unwrap();
assert!(exclusion.verify::<Sha256>(&keys[2], &root));
let db = db.sync().await.unwrap();
assert_eq!(db.root(), root);
drop(db);
let reopened: OrderedFixedDb =
OrderedFixedDb::init(ctx.child("reopen"), fixed_config::<OneCap>(partition), None)
.await
.unwrap();
assert_eq!(reopened.root(), root);
assert!(
reopened
.exclusion_proof(&keys[2])
.await
.unwrap()
.verify::<Sha256>(&keys[2], &root)
);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_ordered_policy_to_empty_proves_exclusion() {
deterministic::Runner::default().start(|context| async move {
let ctx = context.child("db");
let db: OrderedFixedDb = OrderedFixedDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("current-ordered-policy-empty"),
None,
)
.await
.unwrap();
let k = key(7);
let seed = db
.new_batch()
.write(k, Some(val(7)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap();
assert!(!db.is_empty());
let mut policy = Script::new(usize::MAX, u64::MAX, |_: &Digest| Choice::Evict);
let batch = db
.new_batch()
.merkleize(&db, None, &mut policy)
.await
.unwrap();
assert_eq!(policy.visited.len(), 1);
assert_eq!((policy.visited[0].1, policy.visited[0].2), (k, val(7)));
let db = db.apply_batch(batch).await.unwrap();
assert_eq!(db.get(&k).await.unwrap(), None);
assert!(db.is_empty());
let proof = db.exclusion_proof(&k).await.unwrap();
assert!(matches!(
proof,
ordered::proof::constant::ExclusionProof::Commit(..)
));
assert!(proof.verify::<Sha256>(&k, &db.root()));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_policy_skips_updates_superseded_by_pending_parent() {
deterministic::Runner::default().start(|context| async move {
let ctx = context.child("db");
let db: UnorderedFixedDb = UnorderedFixedDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("current-policy-speculative-skip"),
None,
)
.await
.unwrap();
let mut keys: Vec<_> = (40..50).map(key).collect();
keys.sort();
let seed = keys
.iter()
.enumerate()
.fold(db.new_batch(), |batch, (i, key)| {
batch.write(*key, Some(val(i as u64)))
})
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap().commit().await.unwrap();
let parent = keys[..9]
.iter()
.fold(db.new_batch(), |batch, key| {
batch.write(*key, Some(val(100)))
})
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
let items_read = || counter(&context, "log_journal_items_read_total");
let before = items_read();
let mut reads = Vec::new();
let mut policy = Script::new(1, u64::MAX, |_: &Digest| {
reads.push(items_read());
Choice::Keep
});
let child = parent
.new_batch::<Sha256>()
.merkleize(&db, None, &mut policy)
.await
.unwrap();
assert_eq!(policy.visited.len(), 1);
assert_eq!(policy.visited[0].1, keys[9]);
assert_eq!(reads, [before + 1]);
drop((child, parent));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_current_staged_walk_continues_past_applied_candidates() {
deterministic::Runner::default().start(|context| async move {
let ctx = context.child("db");
let mut db: UnorderedFixedDb = UnorderedFixedDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("current-staged-walk"),
None,
)
.await
.unwrap();
let rounds = CHUNK_BITS / 2;
for i in 0..rounds {
let batch = db
.new_batch()
.write(key(0), Some(val(i)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
db = db.apply_batch(batch).await.unwrap();
}
let db = db.commit().await.unwrap();
let floor = db.inactivity_floor_loc();
let db = db
.prune(Location::new(*floor / CHUNK_BITS * CHUNK_BITS))
.await
.unwrap();
assert!(db.pruned_bits() > 0);
let mut keys: Vec<_> = (1..5).map(key).collect();
keys.sort();
let seed = keys
.iter()
.fold(db.new_batch(), |batch, k| batch.write(*k, Some(val(1))))
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap();
let parent = keys
.iter()
.fold(db.new_batch(), |batch, k| batch.write(*k, Some(val(2))))
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
let (_, staged) = parent
.new_batch::<Sha256>()
.stage(&[&keys[0]], &db)
.await
.unwrap();
let staged = staged
.merkleize(
vec![Slot::Put(val(3))],
Vec::new(),
None,
&db,
&mut Proportional,
)
.await
.unwrap();
let direct = parent
.new_batch::<Sha256>()
.write(keys[0], Some(val(3)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
assert_eq!(staged.root(), direct.root());
let (_, operations) = staged.operations();
assert_eq!(
operations[1..3],
[
Operation::Update(update::Unordered(key(0), val(rounds - 1))),
Operation::Update(update::Unordered(keys[1], val(2))),
]
);
let db = db.apply_batch(parent).await.unwrap();
let twin = db
.new_batch()
.write(keys[0], Some(val(3)))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
assert_eq!(twin.root(), staged.root());
drop((staged, direct, twin));
db.destroy().await.unwrap();
});
}
trait Proves {
async fn assert_proves(&self, live: &BTreeMap<Digest, Digest>, evicted: &[Digest]);
}
macro_rules! unordered_proves {
($($db:ty),+) => {$(
impl Proves for $db {
async fn assert_proves(&self, live: &BTreeMap<Digest, Digest>, evicted: &[Digest]) {
let root = self.root();
for (key, value) in live {
let proof = self.key_value_proof(*key).await.unwrap();
assert!(
Self::verify_key_value_proof(*key, *value, &proof, &root),
"{key} fails to prove its value",
);
}
for key in evicted {
assert_eq!(self.get(key).await.unwrap(), None, "{key} is live");
assert!(matches!(
self.key_value_proof(*key).await,
Err(Error::KeyNotFound)
));
}
}
}
)+};
}
macro_rules! ordered_proves {
($($db:ty),+) => {$(
impl Proves for $db {
async fn assert_proves(&self, live: &BTreeMap<Digest, Digest>, evicted: &[Digest]) {
let keys: Vec<_> = live.keys().copied().collect();
for (i, (key, value)) in live.iter().enumerate() {
self.assert_link(*key, *value, keys[(i + 1) % keys.len()])
.await;
}
for key in evicted {
self.assert_absent(*key).await;
}
}
}
)+};
}
unordered_proves!(UnorderedFixedDb, UnorderedFixedMmbDb);
ordered_proves!(OrderedFixedDb, OrderedFixedMmbDb);
macro_rules! policy_consistency_test {
($name:ident, $db:ty) => {
#[test_traced("WARN")]
fn $name() {
deterministic::Runner::default().start(|context| async move {
let ctx = context.child("db");
let partition = stringify!($name);
let mut db: $db = <$db>::init(
ctx.child("storage"),
fixed_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let keys: Vec<Digest> = (0..7).map(key).collect();
let created = [key(7), key(8)];
let fillers: Vec<Digest> = (100..110).map(key).collect();
let mut values = BTreeMap::new();
let mut round = 0;
while *db.size() + 11 <= CHUNK_BITS - 6 {
let writes: Vec<Digest> = if round == 0 {
keys.iter().chain(&fillers).copied().collect()
} else {
fillers.clone()
};
let batch = writes
.iter()
.fold(db.new_batch(), |batch, key| {
batch.write(*key, Some(val(round)))
})
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
db = db.apply_batch(batch).await.unwrap();
for key in writes {
values.insert(key, val(round));
}
round += 1;
}
assert_eq!(*db.size(), CHUNK_BITS - 6);
let parent_writes = [(keys[2], val(1002)), (keys[3], val(1003))];
let parent = parent_writes
.iter()
.fold(db.new_batch(), |batch, (key, value)| {
batch.write(*key, Some(*value))
})
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
values.extend(parent_writes);
let child_writes = [
(keys[4], val(2004)),
(keys[5], val(2005)),
(keys[6], val(2006)),
(created[0], val(2007)),
(created[1], val(2008)),
];
let write = |batch: <$db as crate::qmdb::any::traits::BatchableDb>::Batch| {
child_writes
.iter()
.fold(batch, |batch, (key, value)| batch.write(*key, Some(*value)))
};
values.extend(child_writes);
let evict = [keys[0], keys[2], keys[6], created[0]];
let replace = [(keys[3], val(3003)), (keys[5], val(3005))];
let choose = move |key: &Digest| {
if evict.contains(key) {
Choice::Evict
} else if let Some((_, value)) = replace.iter().find(|(k, _)| k == key) {
Choice::Replace(*value)
} else {
Choice::Keep
}
};
let mut live = db.live().await;
let (start, ops) = parent.operations();
replay(&mut live, start, &ops);
let held = write(parent.new_batch::<Sha256>())
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
let (start, ops) = held.operations();
let tip = start + (ops.len() as u64 - 1);
assert!(*parent.bounds().tip.size < CHUNK_BITS && CHUNK_BITS < *tip);
let mut written = live.clone();
replay(&mut written, start, &ops);
let mut expected: Vec<_> = written
.iter()
.map(|(key, loc)| (*loc, *key, values[key]))
.collect();
expected.sort();
drop(held);
let mut policy = Script::new(usize::MAX, u64::MAX, choose);
let child = write(parent.new_batch::<Sha256>())
.merkleize(&db, None, &mut policy)
.await
.unwrap();
assert_eq!(policy.visited, expected);
assert_eq!(child.bounds().inactivity_floor, tip);
let mut model = values.clone();
for key in &evict {
model.remove(key);
}
model.extend(replace);
let (start, ops) = child.operations();
replay(&mut live, start, &ops);
assert!(live.keys().eq(model.keys()));
assert_eq!(Readable::<32>::len(&child.bitmap), *child.bounds().tip.size);
assert_bits(&child.bitmap, &live);
for key in values.keys() {
assert_eq!(db.read(&child, key).await, model.get(key).copied());
}
let root = child.root();
let db = db.apply_batch(parent).await.unwrap();
let mut twin_policy = Script::new(usize::MAX, u64::MAX, choose);
let twin = write(db.new_batch())
.merkleize(&db, None, &mut twin_policy)
.await
.unwrap();
assert_eq!(twin_policy.visited, policy.visited);
assert_eq!(twin.operations(), child.operations());
assert_eq!(twin.bounds().inactivity_floor, tip);
assert_eq!(twin.root(), root);
assert_bits(&twin.bitmap, &live);
drop(twin);
let db = db.apply_batch(child).await.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.inactivity_floor_loc(), tip);
assert_eq!(db.live().await, live);
assert_bits(db.bitmap(), &live);
db.assert_exact().await;
db.assert_proves(&model, &evict).await;
let db = db.sync().await.unwrap();
drop(db);
let db: $db =
<$db>::init(ctx.child("reopen"), fixed_config::<OneCap>(partition), None)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_bits(db.bitmap(), &live);
db.assert_exact().await;
db.assert_proves(&model, &evict).await;
db.destroy().await.unwrap();
});
}
};
}
policy_consistency_test!(
test_current_unordered_policy_bits_and_proofs,
UnorderedFixedDb
);
policy_consistency_test!(test_current_ordered_policy_bits_and_proofs, OrderedFixedDb);
macro_rules! policy_prune_test {
($name:ident, $db:ty, $family:ty, $pending:literal, $pruned:literal) => {
#[test_traced("WARN")]
fn $name() {
deterministic::Runner::default().start(|context| async move {
let ctx = context.child("db");
let partition = stringify!($name);
let mut db: $db = <$db>::init(
ctx.child("storage"),
fixed_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let keys: Vec<Digest> = (0..8).map(key).collect();
let created = [key(8), key(9)];
let fillers: Vec<Digest> = (100..220).map(key).collect();
let start = 3 * CHUNK_BITS - 8;
let mut values = BTreeMap::new();
let mut round = 0;
while *db.size() < start {
let writes: Vec<Digest> = if round == 0 {
keys.iter().chain(&fillers).copied().collect()
} else {
fillers[..(start - *db.size() - 1).min(10) as usize].to_vec()
};
let batch = writes
.iter()
.fold(db.new_batch(), |batch, key| {
batch.write(*key, Some(val(round)))
})
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
db = db.apply_batch(batch).await.unwrap();
for key in writes {
values.insert(key, val(round));
}
round += 1;
}
assert_eq!(*db.size(), start);
let grand_writes: Vec<_> = [keys[1], keys[2]]
.iter()
.chain(&fillers[10..16])
.zip(1000..)
.map(|(key, i)| (*key, val(i)))
.collect();
let parent_writes = [(keys[3], val(2003)), (keys[7], val(2007))];
let grandparent = grand_writes
.iter()
.fold(db.new_batch(), |batch, (key, value)| {
batch.write(*key, Some(*value))
})
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
let parent = parent_writes
.iter()
.fold(grandparent.new_batch::<Sha256>(), |batch, (key, value)| {
batch.write(*key, Some(*value))
})
.merkleize(&db, None, &mut Hold)
.await
.unwrap();
values.extend(grand_writes.iter().copied());
values.extend(parent_writes);
let child_writes = [
(keys[4], val(3004)),
(keys[5], val(3005)),
(keys[6], val(3006)),
(created[0], val(3008)),
(created[1], val(3009)),
];
values.extend(child_writes);
let [held, pending, child] = [(); 3].map(|_| {
child_writes
.iter()
.fold(parent.new_batch::<Sha256>(), |batch, (key, value)| {
batch.write(*key, Some(*value))
})
});
let mut live = db.live().await;
for batch in [&grandparent, &parent] {
let (start, ops) = batch.operations();
replay(&mut live, start, &ops);
}
let held = held.merkleize(&db, None, &mut Hold).await.unwrap();
let (child_start, ops) = held.operations();
let mut written = live.clone();
replay(&mut written, child_start, &ops);
drop(held);
let mut own: Vec<_> = child_writes
.iter()
.map(|(key, _)| (written[key], *key))
.collect();
own.sort();
let (stopped, stop) = own[4];
let choices: BTreeMap<Digest, Choice> = [
(keys[0], Choice::Evict),
(keys[2], Choice::Evict),
(keys[3], Choice::Evict),
(keys[1], Choice::Replace(val(4001))),
(keys[7], Choice::Replace(val(4007))),
(own[0].1, Choice::Evict),
(own[1].1, Choice::Replace(val(4100))),
(own[2].1, Choice::Evict),
(stop, Choice::Stop),
]
.into_iter()
.collect();
for key in [keys[0], keys[1], keys[2], keys[3], keys[7]] {
assert!(written[&key] < child_start, "the child rewrites {key}");
}
let choose = |key: &Digest| choices.get(key).copied().unwrap_or(Choice::Keep);
let mut expected: Vec<_> = written
.iter()
.filter(|(_, loc)| **loc <= stopped)
.map(|(key, loc)| (*loc, *key, values[key]))
.collect();
expected.sort();
let mut policy = Script::new(usize::MAX, u64::MAX, choose);
let pending = pending.merkleize(&db, None, &mut policy).await.unwrap();
assert_eq!(policy.visited, expected);
assert_eq!(pending.bounds().inactivity_floor, stopped);
let ungrafted = |db: &$db| {
let complete = db.bitmap().complete_chunks() as u64;
let graftable = grafting::graftable_chunks::<$family>(
*db.size(),
grafting::height::<32>(),
);
complete - graftable.min(complete)
};
let freed = Arc::downgrade(&grandparent.inner);
let db = db.apply_batch(grandparent).await.unwrap();
assert!(freed.upgrade().is_none());
assert_eq!(*db.size(), 3 * CHUNK_BITS + 1);
assert_eq!(ungrafted(&db), $pending);
let mut policy = Script::new(usize::MAX, u64::MAX, choose);
let child = child.merkleize(&db, None, &mut policy).await.unwrap();
assert_eq!(policy.visited, expected);
assert_eq!(child.operations(), pending.operations());
assert_eq!(child.bounds().inactivity_floor, stopped);
let root = child.root();
assert_eq!(pending.root(), root);
drop(pending);
let mut model = values.clone();
let mut evicted = Vec::new();
for (key, choice) in &choices {
match choice {
Choice::Evict => {
model.remove(key);
evicted.push(*key);
}
Choice::Replace(value) => {
model.insert(*key, *value);
}
Choice::Keep | Choice::Stop => {}
}
}
let (start, ops) = child.operations();
replay(&mut live, start, &ops);
assert!(live.keys().eq(model.keys()));
let db = db.apply_batch(child).await.unwrap();
drop(parent);
assert_eq!(db.root(), root);
assert_eq!(db.inactivity_floor_loc(), stopped);
assert_eq!(db.bitmap().complete_chunks(), 3);
assert_eq!(ungrafted(&db), 0);
assert_eq!(db.live().await, live);
assert_bits(db.bitmap(), &live);
db.assert_exact().await;
db.assert_proves(&model, &evicted).await;
let db = db.sync().await.unwrap();
let boundary = db.sync_boundary();
assert_eq!(*boundary, $pruned * CHUNK_BITS);
let db = db.prune(boundary).await.unwrap();
assert_eq!(db.pruned_bits(), *boundary);
assert!(*db.bounds().start > 0 && db.bounds().start <= boundary);
assert_eq!(db.root(), root);
assert_eq!(db.live().await, live);
assert_bits(db.bitmap(), &live);
db.assert_exact().await;
db.assert_proves(&model, &evicted).await;
let bounds = db.bounds();
drop(db);
let db: $db =
<$db>::init(ctx.child("reopen"), fixed_config::<OneCap>(partition), None)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.bounds(), bounds);
assert_eq!(db.inactivity_floor_loc(), stopped);
assert_eq!(db.pruned_bits(), *boundary);
assert_eq!(db.live().await, live);
assert_bits(db.bitmap(), &live);
db.assert_exact().await;
db.assert_proves(&model, &evicted).await;
db.destroy().await.unwrap();
});
}
};
}
policy_prune_test!(
test_current_unordered_policy_prune,
UnorderedFixedDb,
mmr::Family,
0,
3
);
policy_prune_test!(
test_current_ordered_policy_prune,
OrderedFixedDb,
mmr::Family,
0,
3
);
policy_prune_test!(
test_current_unordered_mmb_policy_prune,
UnorderedFixedMmbDb,
mmb::Family,
1,
2
);
policy_prune_test!(
test_current_ordered_mmb_policy_prune,
OrderedFixedMmbDb,
mmb::Family,
1,
2
);
async fn assert_policy_matches_any<M, A, C>(mut any: A, mut current: C)
where
M: merkle::Graftable,
A: Inspect<M>,
C: Inspect<M, Update = A::Update>,
Operation<M, A::Update>: PartialEq + core::fmt::Debug,
{
let seed: Vec<_> = (0..24).map(|i| (key(i), Some(val(i)))).collect();
let churn: Vec<_> = (0..24)
.step_by(4)
.map(|i| (key(i), Some(val(i + 100))))
.collect();
for writes in [&seed, &churn] {
any = hold(any, writes).await;
current = hold(current, writes).await;
}
let parent: Vec<_> = (1..24)
.step_by(3)
.map(|i| (key(i), Some(val(i + 200))))
.collect();
let any_parent = hold_batch(&any, any.new_batch(), &parent).await;
let current_parent = hold_batch(¤t, current.new_batch(), &parent).await;
let stop = key(5);
let choose = |action: usize| {
move |k: &Digest| match action {
0 => Choice::Keep,
1 => Choice::Evict,
2 => Choice::Replace(val(1000)),
3 if *k == stop => Choice::Stop,
3 => Choice::Keep,
_ => match k.as_ref()[0] % 3 {
0 => Choice::Keep,
1 => Choice::Evict,
_ => Choice::Replace(val(1001)),
},
}
};
for (entries, skips) in [
(usize::MAX, u64::MAX),
(5, u64::MAX),
(usize::MAX, 6),
(3, 2),
(0, 4),
] {
for action in 0..5 {
let label = format!("action={action} entries={entries} skips={skips}");
let mut any_policy = Script::new(entries, skips, choose(action));
let any_batch = A::child(&any_parent)
.merkleize(&any, None, &mut any_policy)
.await
.unwrap();
let mut current_policy = Script::new(entries, skips, choose(action));
let current_batch = C::child(¤t_parent)
.merkleize(¤t, None, &mut current_policy)
.await
.unwrap();
assert_eq!(any_policy.visited, current_policy.visited, "{label}");
assert_eq!(
A::span(&any_batch).inactivity_floor,
C::span(¤t_batch).inactivity_floor,
"{label}"
);
assert_eq!(A::ops(&any_batch), C::ops(¤t_batch), "{label}");
if (entries, skips) == (usize::MAX, u64::MAX) {
let keys: Vec<_> = any_policy.visited.iter().map(|(_, k, _)| *k).collect();
match action {
3 => assert_eq!(keys.last(), Some(&stop)),
4 => assert!((0..3).all(|r| keys.iter().any(|k| k.as_ref()[0] % 3 == r))),
_ => assert_eq!(keys.len(), 24),
}
}
}
}
drop((any_parent, current_parent));
any.destroy().await.unwrap();
current.destroy().await.unwrap();
}
#[test_traced("INFO")]
fn test_current_policy_matches_any() {
type UnorderedAny = crate::qmdb::any::unordered::fixed::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
Sequential,
>;
type OrderedAny = crate::qmdb::any::ordered::fixed::Db<
mmr::Family,
Context,
Digest,
Digest,
Sha256,
OneCap,
Sequential,
>;
deterministic::Runner::default().start(|context| async move {
let any_config = |name: &str| crate::qmdb::any::test::fixed_db_config::<OneCap>(name);
let any = UnorderedAny::init(
context.child("unordered_any"),
any_config("unordered-any"),
None,
)
.await
.unwrap();
let current = UnorderedFixedDb::init(
context.child("unordered_current"),
fixed_config::<OneCap>("unordered-current"),
None,
)
.await
.unwrap();
assert_policy_matches_any(any, current).await;
let any = OrderedAny::init(
context.child("ordered_any"),
any_config("ordered-any"),
None,
)
.await
.unwrap();
let current = OrderedFixedDb::init(
context.child("ordered_current"),
fixed_config::<OneCap>("ordered-current"),
None,
)
.await
.unwrap();
assert_policy_matches_any(any, current).await;
});
}
fn is_send<T: Send>(_: T) {}
#[allow(dead_code)]
fn assert_policy_futures_are_send(
unordered: &UnorderedFixedDb,
ordered: &OrderedFixedDb,
key: Digest,
) {
let mut policy = Bounded {
entries: 1,
skips: 1,
};
is_send(
unordered
.new_batch()
.merkleize(unordered, None, &mut policy),
);
is_send(ordered.new_batch().merkleize(ordered, None, &mut policy));
is_send(async move {
let (_, staged) = unordered.new_batch().stage(&[&key], unordered).await?;
staged
.merkleize(Vec::new(), Vec::new(), None, unordered, &mut policy)
.await
});
is_send(async move {
let (_, staged) = ordered.new_batch().stage(&[&key], ordered).await?;
staged
.merkleize(Vec::new(), Vec::new(), None, ordered, &mut policy)
.await
});
}
impl<F, C, I, U, const N: usize, S> Inspect<F> for db::Db<F, Context, C, I, Sha256, U, N, S>
where
F: merkle::Graftable,
C: Mutable<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
U: Update<Key = Digest, Value = Digest>,
S: Strategy,
Operation<F, U>: Codec,
Self: DbAny<
F,
Key = Digest,
Value = Digest,
Digest = Digest,
Merkleized = Arc<batch::MerkleizedBatch<F, Digest, U, N, S>>,
Batch = batch::UnmerkleizedBatch<F, Sha256, U, N, S>,
>,
{
type Update = U;
async fn assert_exact(&self) {
assert_exact(&self.any).await;
}
async fn live(&self) -> BTreeMap<Digest, Location<F>> {
live(&self.any).await
}
fn child(batch: &Self::Merkleized) -> Self::Batch {
batch.new_batch::<Sha256>()
}
fn span(batch: &Self::Merkleized) -> &Bounds<F, Digest> {
batch.bounds()
}
fn ops(batch: &Self::Merkleized) -> (Location<F>, Arc<Vec<Operation<F, U>>>) {
batch.operations()
}
async fn read(&self, batch: &Self::Merkleized, key: &Digest) -> Option<Digest> {
batch.get(key, self).await.unwrap()
}
}
impl<F, C, I, V, const N: usize, S> Links<F>
for db::Db<F, Context, C, I, Sha256, update::Ordered<Digest, V>, N, S>
where
F: merkle::Graftable,
C: Mutable<Item = Operation<F, update::Ordered<Digest, V>>>,
I: OrderedIndex<Value = Location<F>> + 'static,
V: ValueEncoding<Value = Digest>,
S: Strategy,
Operation<F, update::Ordered<Digest, V>>: Codec,
Self: Inspect<F>,
{
async fn assert_link(&self, key: Digest, value: Digest, next: Digest) {
assert_eq!(self.get(&key).await.unwrap(), Some(value), "{key} diverged");
let proof = self.key_value_proof(key).await.unwrap();
assert_eq!(proof.next_key, next, "{key} links to the wrong key");
assert!(
proof.verify::<Sha256, V>(key, value, &self.root()),
"{key} fails to prove its link",
);
}
async fn assert_absent(&self, key: Digest) {
assert_eq!(self.get(&key).await.unwrap(), None, "{key} is live");
let proof = self.exclusion_proof(&key).await.unwrap();
assert!(
proof.verify::<Sha256>(&key, &self.root()),
"{key} fails to prove its exclusion",
);
}
}
impl<F, C, I, V, const N: usize, S> Neighbors<F>
for db::Db<F, Context, C, I, Sha256, update::Ordered<Digest, V>, N, S>
where
F: merkle::Graftable,
C: Mutable<Item = Operation<F, update::Ordered<Digest, V>>>,
I: OrderedIndex<Value = Location<F>> + 'static,
V: ValueEncoding<Value = Digest>,
S: Strategy,
Operation<F, update::Ordered<Digest, V>>: Codec,
Self: Links<F>
+ DbAny<
F,
Merkleized = Arc<
batch::MerkleizedBatch<F, Digest, update::Ordered<Digest, V>, N, S>,
>,
>,
{
async fn neighbors(
&self,
batch: Option<&Self::Merkleized>,
key: &Digest,
) -> (Option<Digest>, Option<Digest>) {
match batch {
Some(batch) => (
batch.get_next_key(key, self).await.unwrap(),
batch.get_prev_key(key, self).await.unwrap(),
),
None => (
self.get_next_key(key).await.unwrap(),
self.get_prev_key(key).await.unwrap(),
),
}
}
}
async fn current_db<C, F, Fut>(
context: Context,
open_db: F,
partition: &'static str,
) -> (
C,
impl Fn(Context) -> Pin<Box<dyn Future<Output = C> + Send>> + Clone,
)
where
F: Fn(Context, String) -> Fut + Clone + Send + 'static,
Fut: Future<Output = C> + Send + 'static,
{
let db = open_db(context, partition.into()).await;
let reopen = move |ctx: Context| -> Pin<Box<dyn Future<Output = C> + Send>> {
Box::pin(open_db(ctx, partition.into()))
};
(db, reopen)
}
macro_rules! current_test {
($variants:ident, $name:ident, $any:ident, $partition:literal, $bound:ident) => {
current_test!(@define $variants, $name, $partition, $bound, |context, db, _| {
$any(context, db, val)
});
};
($variants:ident, $name:ident, $any:ident, $partition:literal, $bound:ident, reopen) => {
current_test!(@define $variants, $name, $partition, $bound, |context, db, reopen| {
$any(context, db, reopen, val)
});
};
(@define $variants:ident, $name:ident, $partition:literal, $bound:ident, $run:expr) => {
async fn $name<M, C, F, Fut>(context: Context, open_db: F)
where
M: merkle::Graftable,
C: $bound<M>,
C::Merkleized: Clone,
Operation<M, C::Update>: Codec,
F: Fn(Context, String) -> Fut + Clone + Send + 'static,
Fut: Future<Output = C> + Send + 'static,
{
let (db, reopen) = current_db(context.child("db"), open_db, $partition).await;
($run)(context, db, reopen).await;
}
$variants!($name, "WARN");
};
}
current_test!(
test_for_all_variants,
test_current_activity_depths,
test_any_activity_depths,
"activity",
Inspect,
reopen
);
current_test!(
test_for_all_variants,
test_current_proportional_bound,
test_any_proportional_bound,
"bound",
Inspect
);
current_test!(
test_for_all_variants,
test_current_policy_matches_proportional,
test_any_policy_matches_proportional,
"proportional",
Inspect
);
current_test!(
test_for_all_variants,
test_current_policy_decisions_match_writes,
test_any_policy_decisions_match_writes,
"decisions",
Inspect
);
current_test!(
test_for_all_variants,
test_current_policy_limits_after_colliding_writes,
test_any_policy_limits_after_colliding_writes,
"colliding",
Inspect
);
current_test!(
test_for_all_variants,
test_current_policy_hold,
test_any_policy_hold,
"hold",
Inspect
);
current_test!(
test_for_all_variants,
test_current_policy_limits_with_writes,
test_any_policy_limits_with_writes,
"writes",
Inspect
);
current_test!(
test_for_all_variants,
test_current_proportional_one_batch,
test_any_proportional_one_batch,
"one-batch",
Inspect
);
current_test!(
test_for_all_variants,
test_current_policy_own_writes,
test_any_policy_own_writes,
"own",
Inspect,
reopen
);
current_test!(
test_for_all_variants,
test_current_policy_evicts_parent_created_key,
test_any_policy_evicts_parent_created_key,
"parent-created",
Changes,
reopen
);
current_test!(
test_for_all_variants,
test_current_policy_ancestor_twins,
test_any_policy_ancestor_twins,
"twins",
Inspect
);
current_test!(
test_for_all_variants,
test_current_policy_keep_evict_and_recover,
test_any_policy_keep_evict_and_recover,
"recover",
Inspect,
reopen
);
current_test!(
test_for_ordered_variants,
test_current_ordered_policy_eviction_matrix,
test_any_ordered_policy_eviction_matrix,
"matrix",
Neighbors,
reopen
);
current_test!(
test_for_ordered_variants,
test_current_ordered_policy_repair_across_ancestors,
test_any_ordered_policy_repair_across_ancestors,
"repair",
Neighbors,
reopen
);
}