use crate::{
Context,
journal::{
authenticated,
contiguous::fixed::{self, Config as JournalConfig},
},
merkle::Family,
qmdb::{
Error, ROOT_BAGGING,
any::value::{FixedEncoding, FixedValue},
keyless::operation::Operation as BaseOperation,
operation::Committable,
},
};
use commonware_cryptography::Hasher;
use commonware_parallel::Strategy;
pub type Operation<F, V> = BaseOperation<F, FixedEncoding<V>>;
pub type Db<F, E, V, H, S> =
super::Keyless<F, E, FixedEncoding<V>, fixed::Journal<E, Operation<F, V>>, H, S>;
pub type CompactDb<F, E, V, H, S> = super::CompactDb<F, E, FixedEncoding<V>, H, (), S>;
type Journal<F, E, V, H, S> =
authenticated::Journal<F, E, fixed::Journal<E, Operation<F, V>>, H, S>;
pub type Config<S> = super::Config<JournalConfig, S>;
pub type CompactConfig<S> = super::CompactConfig<(), S>;
impl<F: Family, E: Context, V: FixedValue, H: Hasher, S: Strategy> Db<F, E, V, H, S> {
pub async fn init(context: E, cfg: Config<S>) -> Result<Self, Error<F>> {
let journal: Journal<F, E, V, H, S> = Journal::new(
context.child("journal"),
cfg.merkle,
cfg.log,
Operation::<F, V>::is_commit,
ROOT_BAGGING,
)
.await?;
Self::init_from_journal(journal, context).await
}
}
impl<F: Family, E: Context, V: FixedValue, H: Hasher, S: Strategy> CompactDb<F, E, V, H, S> {
pub async fn init(context: E, cfg: CompactConfig<S>) -> Result<Self, Error<F>> {
let merkle = crate::merkle::compact::Merkle::new(cfg.strategy);
Self::init_from_merkle(merkle, context.child("witness"), cfg.witness, ()).await
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
merkle::{Location, mmb, mmr},
qmdb::keyless::tests::{self, keyless_tests},
};
use commonware_cryptography::Sha256;
use commonware_macros::{boxed, test_traced};
use commonware_parallel::{Rayon, Sequential, Strategy};
use commonware_runtime::{
BufferPooler, Metrics as _, Runner as _, Spawner as _, Strategizer as _, Supervisor as _,
buffer::paged::CacheRef,
deterministic,
mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs},
reschedule,
};
use commonware_utils::{NZU16, NZU64, NZUsize, sequence::U64};
use core::future::Future;
use futures::FutureExt as _;
use std::num::{NonZeroU16, NonZeroUsize};
const PAGE_SIZE: NonZeroU16 = NZU16!(101);
const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(11);
fn db_config<S: Strategy>(suffix: &str, pooler: &impl BufferPooler, strategy: S) -> Config<S> {
let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
Config {
merkle: crate::merkle::full::Config {
journal_partition: format!("fixed-journal-{suffix}"),
metadata_partition: format!("fixed-metadata-{suffix}"),
items_per_blob: NZU64!(11),
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
strategy,
page_cache: page_cache.clone(),
},
log: JournalConfig {
partition: format!("fixed-log-journal-{suffix}"),
items_per_blob: NZU64!(7),
page_cache,
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
},
}
}
type TestDb<F> =
Db<F, deterministic::Context, commonware_utils::sequence::U64, Sha256, Sequential>;
type TestRayonDb<F> =
Db<F, deterministic::Context, commonware_utils::sequence::U64, Sha256, Rayon>;
type TestCompactDb<F> =
CompactDb<F, deterministic::Context, commonware_utils::sequence::U64, Sha256, Sequential>;
async fn open_db<F: Family>(context: deterministic::Context) -> TestDb<F> {
open_db_with_suffix("partition", context).await
}
async fn open_db_with_suffix<F: Family>(
suffix: &str,
context: deterministic::Context,
) -> TestDb<F> {
let cfg = db_config(suffix, &context, Sequential);
TestDb::init(context, cfg).await.unwrap()
}
async fn open_rayon_db<F: Family>(context: deterministic::Context) -> TestRayonDb<F> {
let strategy = context.strategy(NZUsize!(2));
let cfg = db_config("rayon", &context, strategy);
TestRayonDb::init(context, cfg).await.unwrap()
}
async fn open_compact<F: crate::merkle::Family>(
context: deterministic::Context,
) -> TestCompactDb<F> {
let cfg = CompactConfig {
strategy: Sequential,
witness: crate::journal::contiguous::variable::Config {
partition: "compact-keyless-fixed-witness".into(),
items_per_section: NZU64!(64),
compression: None,
codec_config: (),
page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
},
commit_codec_config: (),
};
TestCompactDb::init(context, cfg).await.unwrap()
}
fn reopen<F: Family>() -> tests::Reopen<TestDb<F>> {
Box::new(|ctx| Box::pin(open_db(ctx)))
}
type DelayedDb =
Db<mmr::Family, DelayedSyncContext<deterministic::Context>, U64, Sha256, Sequential>;
fn open_delayed_db(
context: &deterministic::Context,
label: &'static str,
suffix: &str,
pending: &PendingSyncs,
) -> impl Future<Output = Result<DelayedDb, Error<mmr::Family>>> {
let mut cfg = db_config(suffix, context, Sequential);
let page_cache = CacheRef::from_pooler(context, NZU16!(1024), NZUsize!(8));
cfg.log.items_per_blob = NZU64!(1000);
cfg.log.page_cache = page_cache.clone();
cfg.merkle.items_per_blob = NZU64!(1000);
cfg.merkle.page_cache = page_cache;
DelayedDb::init(
DelayedSyncContext {
inner: context.child(label),
pending: pending.clone(),
},
cfg,
)
}
async fn apply_append(
db: DelayedDb,
value: U64,
floor: Location<mmr::Family>,
) -> (DelayedDb, Location<mmr::Family>) {
let batch = db
.new_batch()
.append(value)
.merkleize(&db, None, floor)
.await;
let (db, range) = db.apply_batch(batch).await.unwrap();
(db, range.start)
}
#[test_traced]
fn test_keyless_fixed_start_sync_overlaps_work() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
let open = open_delayed_db(&ctx, "delayed", "start-sync-overlap", &pending);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
let value0 = U64::new(1);
let loc0;
let floor = db.inactivity_floor_loc();
(db, loc0) = apply_append(db, value0.clone(), floor).await;
let starts_before = pending.starts();
let entered_before = pending.entered();
let completions_before = pending.completions();
let handle;
(db, handle) = db.start_sync().await.unwrap();
assert!(pending.starts() > starts_before);
assert_eq!(pending.completions(), completions_before);
let waiter = ctx
.child("await_sync")
.spawn(|_| async move { handle.await.unwrap() });
while pending.entered() == entered_before {
reschedule().await;
}
assert_eq!(db.get(loc0).await.unwrap(), Some(value0));
let value1 = U64::new(2);
let loc1;
let floor = db.inactivity_floor_loc();
(db, loc1) = apply_append(db, value1.clone(), floor).await;
assert_eq!(
pending.completions(),
completions_before,
"the database made progress while the sync was still in flight"
);
pending.unblock();
waiter.await.unwrap();
let handle;
(db, handle) = db.start_sync().await.unwrap();
handle.await.unwrap();
let root = db.root();
drop(db);
let db = open_delayed_db(&ctx, "reopen", "start-sync-overlap", &pending)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.get(loc1).await.unwrap(), Some(value1));
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_keyless_fixed_start_sync_failure_propagates() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
pending.unblock();
let mut db = open_delayed_db(&ctx, "delayed", "start-sync-fail", &pending)
.await
.unwrap();
let floor = db.inactivity_floor_loc();
(db, _) = apply_append(db, U64::new(1), floor).await;
pending.arm_fail();
let handle;
(db, handle) = db.start_sync().await.unwrap();
assert!(
handle.await.is_err(),
"the sync handle surfaces the failure"
);
let starts_before = pending.starts();
assert!(
db.commit().await.is_err(),
"the next durability op surfaces the failed in-flight sync"
);
assert_eq!(
pending.starts(),
starts_before,
"the surfaced error is the retained failure, not a fresh sync's"
);
});
}
#[test_traced]
fn test_keyless_fixed_start_sync_recovery() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
pending.unblock();
let mut db = open_delayed_db(&ctx, "delayed", "start-sync-recovery", &pending)
.await
.unwrap();
let value = U64::new(1);
let loc;
let floor = db.inactivity_floor_loc();
(db, loc) = apply_append(db, value.clone(), floor).await;
let handle;
(db, handle) = db.start_sync().await.unwrap();
handle.await.unwrap();
let root = db.root();
drop(db);
let db = open_delayed_db(&ctx, "reopen", "start-sync-recovery", &pending)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.get(loc).await.unwrap(), Some(value));
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_keyless_fixed_start_sync_prune_waits() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
let open = open_delayed_db(&ctx, "delayed", "start-sync-prune", &pending);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
(db, _) = apply_append(db, U64::new(1), Location::new(0)).await;
(db, _) = apply_append(db, U64::new(2), Location::new(2)).await;
let starts_before = pending.starts();
let handle;
(db, handle) = db.start_sync().await.unwrap();
assert!(pending.starts() > starts_before);
let floor = db.inactivity_floor_loc();
assert!(*floor > 0);
let db = {
let mut prune = std::pin::pin!(db.prune(floor));
assert!(
prune.as_mut().now_or_never().is_none(),
"prune proceeded while the started sync was pending"
);
pending.unblock();
prune.await.unwrap()
};
handle.await.unwrap();
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_keyless_fixed_start_sync_rewind_waits() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
let open = open_delayed_db(&ctx, "delayed", "start-sync-rewind", &pending);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
(db, _) = apply_append(db, U64::new(1), Location::new(0)).await;
db = drive_pending_syncs(&pending, db.commit()).await.unwrap();
let committed_root = db.root();
let committed_size = db.bounds().end;
(db, _) = apply_append(db, U64::new(2), Location::new(0)).await;
let handle;
(db, handle) = db.start_sync().await.unwrap();
let db = {
let mut rewind = std::pin::pin!(db.rewind(committed_size));
assert!(
rewind.as_mut().now_or_never().is_none(),
"rewind proceeded while the started sync was pending"
);
pending.unblock();
rewind.await.unwrap()
};
handle.await.unwrap();
assert_eq!(db.root(), committed_root);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_keyless_fixed_metrics() {
deterministic::Runner::default().start(|ctx| async move {
let db = open_db::<mmr::Family>(ctx.child("db")).await;
let value = commonware_utils::sequence::U64::new(7);
let floor = db.inactivity_floor_loc();
let batch = db
.new_batch()
.append(value.clone())
.merkleize(&db, None, floor)
.await;
let (db, range) = db.apply_batch(batch).await.unwrap();
assert_eq!(db.get(range.start).await.unwrap(), Some(value.clone()));
assert_eq!(
db.get_many(&[range.start]).await.unwrap(),
vec![Some(value)]
);
let db = db.commit().await.unwrap();
let db = db.sync().await.unwrap();
let (db, handle) = db.start_sync().await.unwrap();
handle.await.unwrap();
let _db = db.prune(crate::merkle::Location::new(0)).await.unwrap();
let metrics = ctx.encode();
for expected in [
"db_size 3",
"db_pruning_boundary 0",
"db_retained 3",
"db_inactivity_floor 0",
"db_last_commit 2",
"db_get_calls_total 1",
"db_get_many_calls_total 1",
"db_lookups_requested_total 2",
"db_apply_batch_calls_total 1",
"db_operations_applied_total 2",
"db_commit_calls_total 1",
"db_sync_calls_total 1",
"db_start_sync_calls_total 1",
"db_prune_calls_total 1",
"db_get_duration_count 1",
"db_get_many_duration_count 1",
"db_apply_batch_duration_count 1",
"db_commit_duration_count 1",
"db_sync_duration_count 1",
"db_prune_duration_count 1",
] {
assert!(metrics.contains(expected), "missing {expected}\n{metrics}");
}
});
}
keyless_tests! {
test_keyless_fixed_empty => run_empty, reopen_indexed;
test_keyless_fixed_build_basic => run_build_basic, reopen_indexed;
test_keyless_fixed_recovery => run_recovery, reopen_indexed;
test_keyless_fixed_non_empty_recovery => run_non_empty_recovery, reopen_indexed;
test_keyless_fixed_proof => run_proof, db;
test_keyless_fixed_proof_comprehensive => run_proof_comprehensive, db;
test_keyless_fixed_proof_with_pruning => run_proof_with_pruning, reopen_indexed;
test_keyless_fixed_empty_db_recovery => run_empty_db_recovery, reopen_indexed;
test_keyless_fixed_replay_with_trailing_appends => run_replay_with_trailing_appends, reopen_indexed;
test_keyless_fixed_get_out_of_bounds => run_get_out_of_bounds, db;
test_keyless_fixed_metadata => run_metadata, db;
test_keyless_fixed_pruning => run_pruning, reopen;
test_keyless_fixed_batch_get => run_batch_get, db;
test_keyless_fixed_batch_stacked_get => run_batch_stacked_get, db;
test_keyless_fixed_batch_speculative_root => run_batch_speculative_root, db;
test_keyless_fixed_merkleized_batch_get => run_merkleized_batch_get, db;
test_keyless_fixed_batch_chained => run_batch_chained, db;
test_keyless_fixed_operations_match_applied_log => run_operations_match_applied_log, db;
test_keyless_fixed_batch_chained_apply_sequential => run_batch_chained_apply_sequential, db;
test_keyless_fixed_batch_many_sequential => run_batch_many_sequential, db;
test_keyless_fixed_batch_empty => run_batch_empty, db;
test_keyless_fixed_batch_chained_merkleized_get => run_batch_chained_merkleized_get, db;
test_keyless_fixed_batch_large => run_batch_large, db;
test_keyless_fixed_stale_batch => run_stale_batch, reopen;
test_keyless_fixed_stale_batch_chained => run_stale_batch_chained, db;
test_keyless_fixed_sequential_commit_parent_then_child => run_sequential_commit_parent_then_child, db;
test_keyless_fixed_stale_batch_child_before_parent => run_stale_batch_child_before_parent, db;
test_keyless_fixed_to_batch => run_to_batch, db;
test_keyless_fixed_child_root_matches_pending_and_committed => run_child_root_matches_pending_and_committed, db;
test_keyless_fixed_rewind_recovery => run_rewind_recovery, reopen;
test_keyless_fixed_rewind_pruned_target_errors => run_rewind_pruned_target_errors, reopen;
test_keyless_fixed_floor_tracking => run_floor_tracking, reopen_indexed;
test_keyless_fixed_floor_regression_rejected => run_floor_regression_rejected, reopen;
test_keyless_fixed_floor_beyond_commit_loc_rejected => run_floor_beyond_commit_loc_rejected, reopen;
test_keyless_fixed_rewind_restores_floor => run_rewind_restores_floor, db;
test_keyless_fixed_floor_at_commit_loc_accepted => run_floor_at_commit_loc_accepted, db;
test_keyless_fixed_rewind_after_reopen_with_floor => run_rewind_after_reopen_with_floor, reopen_indexed;
test_keyless_fixed_ancestor_floor_regression_rejected => run_ancestor_floor_regression_rejected, reopen;
test_keyless_fixed_ancestor_floor_beyond_commit_loc_rejected => run_ancestor_floor_beyond_commit_loc_rejected, db;
test_keyless_fixed_chained_apply_with_valid_floors_succeeds => run_chained_apply_with_valid_floors_succeeds, db;
test_keyless_fixed_single_commit_live_set => run_single_commit_live_set, reopen_indexed;
test_keyless_fixed_commit_after_sync_recovery => run_commit_after_sync_recovery, reopen_indexed;
test_keyless_fixed_get_many => run_get_many, db;
}
#[test_traced("INFO")]
fn test_keyless_fixed_shared_helper_accepts_rayon_strategy() {
deterministic::Runner::default().start(|ctx| async move {
let db = open_rayon_db::<mmr::Family>(ctx.child("db").with_attribute("index", 1)).await;
tests::run_metadata(db).await;
});
}
#[boxed]
async fn assert_compact_root_compatibility<F: crate::merkle::Family>(
ctx: deterministic::Context,
) {
let db = open_db::<F>(ctx.child("db")).await;
let compact = open_compact::<F>(ctx.child("compact")).await;
assert_eq!(db.root(), compact.root());
let v1 = commonware_utils::sequence::U64::new(1);
let v2 = commonware_utils::sequence::U64::new(2);
let metadata = commonware_utils::sequence::U64::new(99);
let floor = db.inactivity_floor_loc();
let retained = db
.new_batch()
.append(v1.clone())
.append(v2.clone())
.merkleize(&db, Some(metadata.clone()), floor)
.await;
let compact_batch = compact
.new_batch()
.append(v1)
.append(v2)
.merkleize(&compact, Some(metadata.clone()), floor)
.await;
assert_eq!(retained.root(), compact_batch.root());
let (db, _) = db.apply_batch(retained).await.unwrap();
let (compact, _) = compact.apply_batch(compact_batch).await.unwrap();
let db = db.commit().await.unwrap();
let compact = compact.sync().await.unwrap();
assert_eq!(db.root(), compact.root());
assert_eq!(compact.get_metadata(), Some(metadata.clone()));
drop(compact);
let reopened = open_compact::<F>(ctx.child("reopen")).await;
assert_eq!(db.root(), reopened.root());
assert_eq!(reopened.get_metadata(), Some(metadata));
reopened.destroy().await.unwrap();
db.destroy().await.unwrap();
}
#[test_traced("INFO")]
fn test_keyless_fixed_compact_root_compatibility() {
deterministic::Runner::default().start(|ctx| async move {
assert_compact_root_compatibility::<mmr::Family>(ctx).await;
});
}
#[test_traced("INFO")]
fn test_keyless_fixed_compact_root_compatibility_mmb() {
deterministic::Runner::default().start(|ctx| async move {
assert_compact_root_compatibility::<mmb::Family>(ctx).await;
});
}
#[test_traced("INFO")]
fn test_keyless_fixed_floor_changes_root() {
deterministic::Runner::default().start(|ctx| async move {
let db_a = open_db_with_suffix::<mmr::Family>("root-a", ctx.child("a")).await;
let db_b = open_db_with_suffix::<mmr::Family>("root-b", ctx.child("b")).await;
tests::run_floor_changes_root(db_a, db_b).await;
});
}
#[test_traced("INFO")]
fn test_keyless_fixed_floor_changes_root_mmb() {
deterministic::Runner::default().start(|ctx| async move {
let db_a = open_db_with_suffix::<mmb::Family>("root-a", ctx.child("a")).await;
let db_b = open_db_with_suffix::<mmb::Family>("root-b", ctx.child("b")).await;
tests::run_floor_changes_root(db_a, db_b).await;
});
}
#[test_traced("WARN")]
fn test_keyless_fixed_sync() {
use crate::{
merkle::Location,
qmdb::sync::{self, Target, engine::Config},
};
use commonware_utils::{non_empty_range, sequence::U64};
use std::sync::Arc;
deterministic::Runner::default().start(|ctx| async move {
let target_config = db_config("sync-target", &ctx, Sequential);
let target_db: TestDb<mmr::Family> = TestDb::init(ctx.child("target"), target_config)
.await
.unwrap();
let mut batch = target_db.new_batch();
for i in 0..20u64 {
batch = batch.append(U64::new(i * 10 + 1));
}
let floor = target_db.inactivity_floor_loc();
let merkleized = batch.merkleize(&target_db, None, floor).await;
let (target_db, _) = target_db.apply_batch(merkleized).await.unwrap();
let target_root = target_db.root();
let bounds = target_db.bounds();
let lower_bound = bounds.start;
let upper_bound = bounds.end;
let client_config = db_config("sync-client", &ctx, Sequential);
let target_db = Arc::new(target_db);
let config = Config {
db_config: client_config,
fetch_batch_size: NZU64!(5),
target: Target {
root: target_root,
range: non_empty_range!(lower_bound, upper_bound),
},
context: ctx.child("client"),
source: target_db.clone(),
apply_batch_size: NZU64!(1024),
max_outstanding_requests: 1,
update_rx: None,
finish_rx: None,
reached_target_tx: None,
max_retained_roots: 8,
};
let synced_db: TestDb<mmr::Family> = sync::sync(config).await.unwrap();
assert_eq!(synced_db.root(), target_root);
let bounds = synced_db.bounds();
assert_eq!(bounds.end, upper_bound);
assert_eq!(bounds.start, lower_bound);
for i in 0..20u64 {
let got = synced_db.get(Location::new(i + 1)).await.unwrap();
assert_eq!(got, Some(U64::new(i * 10 + 1)));
}
synced_db.destroy().await.unwrap();
let target_db =
Arc::try_unwrap(target_db).unwrap_or_else(|_| panic!("failed to unwrap Arc"));
target_db.destroy().await.unwrap();
});
}
}