use super::operation::Operation;
pub use crate::qmdb::compact::Config;
use crate::{
Context,
merkle::{Family, Location, Proof, batch, compact as compact_merkle},
qmdb::{
self, Error,
any::value::ValueEncoding,
chain::{self, Bounds, Commitment},
compact::{batch as compact_batch, witness},
sync::{CompactTarget, Request, Source, source},
},
};
use commonware_codec::{Encode, EncodeShared, Read};
use commonware_cryptography::{Digest, Hasher};
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use commonware_runtime::Handle;
use std::sync::{Arc, Weak};
pub struct Db<F, E, V, H, S: Strategy>
where
F: Family,
E: Context,
V: ValueEncoding,
H: Hasher,
Operation<F, V>: EncodeShared,
Operation<F, V>: Read,
{
merkle: compact_merkle::Merkle<F, H::Digest, S>,
last_commit_metadata: Option<V::Value>,
inactivity_floor_loc: Location<F>,
commit_codec_config: <Operation<F, V> as Read>::Cfg,
witness: witness::Store<E, F, H::Digest>,
}
impl<F, E, V, H, S: Strategy> std::fmt::Debug for Db<F, E, V, H, S>
where
F: Family,
E: Context,
V: ValueEncoding,
H: Hasher,
Operation<F, V>: EncodeShared,
Operation<F, V>: Read,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Db")
.field("size", &self.size())
.field("inactivity_floor_loc", &self.inactivity_floor_loc())
.finish_non_exhaustive()
}
}
type MerkleizeResult<F, D, V, S> = Result<Arc<MerkleizedBatch<F, D, V, S>>, Error<F>>;
#[allow(clippy::type_complexity)]
pub struct UnmerkleizedBatch<F, H, V, S: Strategy>
where
F: Family,
V: ValueEncoding,
H: Hasher,
Operation<F, V>: EncodeShared,
{
merkle_batch: compact_merkle::UnmerkleizedBatch<F, H::Digest, S>,
appends: Vec<V::Value>,
parent: Option<Arc<MerkleizedBatch<F, H::Digest, V, S>>>,
base: chain::Commitment<F, H::Digest>,
}
#[derive(Clone)]
pub struct MerkleizedBatch<F: Family, D: Digest, V: ValueEncoding, S: Strategy>
where
Operation<F, V>: EncodeShared,
{
pub(super) merkle_batch: Arc<batch::MerkleizedBatch<F, D, S>>,
operations: Arc<Vec<Operation<F, V>>>,
pub(super) commit_metadata: Option<V::Value>,
pub(super) parent: Option<Weak<Self>>,
pub(super) bounds: chain::Bounds<F, D>,
}
impl<F: Family, D: Digest, V: ValueEncoding, S: Strategy> MerkleizedBatch<F, D, V, S>
where
Operation<F, V>: EncodeShared,
{
pub(super) fn ancestors(&self) -> impl Iterator<Item = Arc<Self>> + use<F, D, V, S> {
chain::ancestors(self.parent.clone(), |batch| batch.parent.as_ref())
}
pub(super) const fn commitment(&self) -> Commitment<F, D> {
self.bounds.tip
}
pub const fn root(&self) -> D {
self.bounds.tip.root
}
pub const fn bounds(&self) -> &Bounds<F, D> {
&self.bounds
}
pub fn operations(&self) -> (Location<F>, Arc<Vec<Operation<F, V>>>) {
(self.bounds.base.size, Arc::clone(&self.operations))
}
pub fn proof<E, H>(&self, db: &Db<F, E, V, H, S>) -> Result<Proof<F, D>, Error<F>>
where
E: Context,
H: Hasher<Digest = D>,
Operation<F, V>: Read,
{
let db = self.bounds.compatible(db, db.commitment())?;
let inactive_peaks = F::inactive_peaks(self.bounds.tip.size, self.bounds.inactivity_floor);
let hasher = qmdb::hasher::<H>();
self.merkle_batch
.range_proof(
db.merkle.mem(),
&hasher,
self.bounds.base.size..self.bounds.tip.size,
inactive_peaks,
)
.map_err(Into::into)
}
pub fn pinned_nodes<E, H>(&self, db: &Db<F, E, V, H, S>) -> Result<Vec<D>, Error<F>>
where
E: Context,
H: Hasher<Digest = D>,
Operation<F, V>: Read,
{
let db = self.bounds.compatible(db, db.commitment())?;
let base = db.merkle.mem();
F::nodes_to_pin(self.bounds.base.size)
.map(|pos| {
self.merkle_batch
.get_node(pos)
.or_else(|| base.get_node(pos))
.ok_or(crate::merkle::Error::ElementPruned(pos))
})
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub fn new_batch<H>(self: &Arc<Self>) -> UnmerkleizedBatch<F, H, V, S>
where
H: Hasher<Digest = D>,
{
UnmerkleizedBatch {
merkle_batch: compact_merkle::UnmerkleizedBatch::wrap(self.merkle_batch.new_batch()),
appends: Vec::new(),
parent: Some(Arc::clone(self)),
base: self.commitment(),
}
}
}
impl<F, H, V, S> UnmerkleizedBatch<F, H, V, S>
where
F: Family,
V: ValueEncoding,
H: Hasher,
S: Strategy,
Operation<F, V>: EncodeShared,
{
pub(super) fn new<E>(db: &Db<F, E, V, H, S>, base: chain::Commitment<F, H::Digest>) -> Self
where
E: Context,
Operation<F, V>: Read,
{
Self {
merkle_batch: db.merkle.new_batch(),
appends: Vec::new(),
parent: None,
base,
}
}
fn db(&self) -> Commitment<F, H::Digest> {
self.parent
.as_ref()
.map_or(self.base, |parent| parent.bounds.db)
}
pub fn append(mut self, value: V::Value) -> Self {
self.appends.push(value);
self
}
#[tracing::instrument(
name = "qmdb.keyless.compact.batch.merkleize",
level = "info",
skip_all
)]
pub async fn merkleize<E>(
self,
db: &Db<F, E, V, H, S>,
metadata: Option<V::Value>,
inactivity_floor: Location<F>,
) -> MerkleizeResult<F, H::Digest, V, S>
where
E: Context,
Operation<F, V>: Read,
{
let live_ancestors: Vec<_> =
chain::parent_and_ancestors(self.parent.as_ref(), |parent| parent.ancestors())
.collect();
let boundary = chain::effective_boundary(
self.db(),
live_ancestors.last().map(|oldest| oldest.bounds.base),
);
let ancestors: Vec<_> = live_ancestors
.iter()
.map(|batch| batch.commitment())
.collect();
let db = chain::merkleizable(db, db.commitment(), boundary, ancestors.iter().copied())?;
let mut ops: Vec<Operation<F, V>> = Vec::with_capacity(self.appends.len() + 1);
ops.extend(self.appends.into_iter().map(Operation::Append));
ops.push(Operation::Commit(metadata.clone(), inactivity_floor));
let operations = Arc::new(ops);
let total_size = self.base.size + operations.len() as u64;
chain::validate_merkleize_floor::<F>(
self.parent.as_ref().map_or_else(
|| db.inactivity_floor_loc(),
|parent| parent.bounds.inactivity_floor,
),
inactivity_floor,
total_size - 1,
)?;
let inactive_peaks = F::inactive_peaks(total_size, inactivity_floor);
let (merkle, root) = compact_batch::merkleize_ops::<F, H, S, _>(
&db.merkle,
self.merkle_batch,
Arc::clone(&operations),
inactive_peaks,
)
.await?;
drop(live_ancestors);
Ok(Arc::new(MerkleizedBatch {
merkle_batch: merkle,
operations,
commit_metadata: metadata,
parent: self.parent.as_ref().map(Arc::downgrade),
bounds: chain::Bounds {
base: self.base,
db: boundary,
tip: Commitment::new(total_size, root),
ancestors,
inactivity_floor,
},
}))
}
}
impl<F, E, V, H, S> Db<F, E, V, H, S>
where
F: Family,
E: Context,
V: ValueEncoding,
H: Hasher,
S: Strategy,
Operation<F, V>: EncodeShared,
Operation<F, V>: Read,
{
#[boxed]
pub async fn init(
context: E,
cfg: Config<<Operation<F, V> as Read>::Cfg, S>,
max_size: Option<Location<F>>,
) -> Result<Self, Error<F>> {
let Config {
strategy,
witness: witness_config,
commit_codec_config,
} = cfg;
let mut merkle = compact_merkle::Merkle::new(strategy);
let (witness, last_commit_op) = witness::init::<E, F, H, S, Operation<F, V>>(
context.child("witness"),
witness_config,
max_size,
&mut merkle,
&commit_codec_config,
Operation::<F, V>::Commit(None, Location::new(0))
.encode()
.to_vec(),
)
.await?;
let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
return Err(Error::DataCorrupted("last operation was not a commit"));
};
Ok(Self {
merkle,
last_commit_metadata,
inactivity_floor_loc,
commit_codec_config,
witness,
})
}
fn encode_commit_op(metadata: Option<V::Value>, inactivity_floor_loc: Location<F>) -> Vec<u8> {
Operation::<F, V>::Commit(metadata, inactivity_floor_loc)
.encode()
.to_vec()
}
pub(crate) fn init_from_sync(
strategy: S,
journal: witness::Journal<E, F, H::Digest>,
commit_codec_config: <Operation<F, V> as Read>::Cfg,
last_commit_loc: Location<F>,
pinned_nodes: Vec<H::Digest>,
last_commit_op: Operation<F, V>,
) -> Result<Self, Error<F>> {
let Operation::Commit(last_commit_metadata, inactivity_floor_loc) = last_commit_op else {
return Err(Error::UnexpectedData(last_commit_loc));
};
witness::validate_inactivity_floor(inactivity_floor_loc, last_commit_loc)?;
let op_bytes = Self::encode_commit_op(last_commit_metadata.clone(), inactivity_floor_loc);
let mut merkle =
compact_merkle::Merkle::from_compact_state(strategy, last_commit_loc, pinned_nodes)?;
let hasher = qmdb::hasher::<H>();
merkle.append_leaf(&hasher, &op_bytes)?;
let imported = witness::build_witness::<F, H, S>(&merkle, inactivity_floor_loc, op_bytes)?;
merkle.prune_to_frontier();
let witness = witness::Store::from_import(journal, imported);
Ok(Self {
merkle,
last_commit_metadata,
inactivity_floor_loc,
commit_codec_config,
witness,
})
}
pub const fn root(&self) -> H::Digest {
self.witness.tip().root
}
pub const fn inactivity_floor_loc(&self) -> Location<F> {
self.inactivity_floor_loc
}
pub const fn size(&self) -> Location<F> {
self.witness.tip().size()
}
pub fn get_metadata(&self) -> Option<V::Value> {
self.last_commit_metadata.clone()
}
pub const fn target(&self) -> CompactTarget<F, H::Digest> {
self.witness.tip().target()
}
pub(crate) const fn commitment(&self) -> chain::Commitment<F, H::Digest> {
chain::Commitment::new(self.size(), self.root())
}
pub fn new_batch(&self) -> UnmerkleizedBatch<F, H, V, S> {
UnmerkleizedBatch::new(self, self.commitment())
}
pub fn to_batch(&self) -> Arc<MerkleizedBatch<F, H::Digest, V, S>> {
Arc::new(MerkleizedBatch {
merkle_batch: self.merkle.to_batch(),
operations: Arc::new(Vec::new()),
commit_metadata: self.last_commit_metadata.clone(),
parent: None,
bounds: chain::Bounds::from_db(self.commitment(), self.inactivity_floor_loc),
})
}
pub fn validate_batch(
&self,
batch: &MerkleizedBatch<F, H::Digest, V, S>,
) -> Result<(), Error<F>> {
batch.bounds.validate_apply_to(self.commitment())
}
#[tracing::instrument(name = "qmdb.keyless.compact.db.apply_batch", level = "info", skip_all)]
pub async fn apply_batch(
mut self,
batch: Arc<MerkleizedBatch<F, H::Digest, V, S>>,
) -> Result<Self, Error<F>> {
self.validate_batch(&batch)?;
self.merkle.apply_batch(&batch.merkle_batch)?;
self.last_commit_metadata = batch.commit_metadata.clone();
self.inactivity_floor_loc = batch.bounds.inactivity_floor;
let last_commit_metadata = self.last_commit_metadata.clone();
let inactivity_floor_loc = self.inactivity_floor_loc;
self.witness = self
.witness
.apply::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
})
.await?;
assert_eq!(self.commitment(), batch.bounds.tip);
Ok(self)
}
#[tracing::instrument(name = "qmdb.keyless.compact.db.start_sync", level = "info", skip_all)]
pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
let last_commit_metadata = self.last_commit_metadata.clone();
let inactivity_floor_loc = self.inactivity_floor_loc;
let handle;
(self.witness, handle) = self
.witness
.start_sync::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
})
.await?;
Ok((self, handle))
}
#[tracing::instrument(name = "qmdb.keyless.compact.db.commit", level = "info", skip_all)]
pub async fn commit(mut self) -> Result<Self, Error<F>> {
let last_commit_metadata = self.last_commit_metadata.clone();
let inactivity_floor_loc = self.inactivity_floor_loc;
self.witness = self
.witness
.commit::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
})
.await?;
Ok(self)
}
#[tracing::instrument(name = "qmdb.keyless.compact.db.sync", level = "info", skip_all)]
pub async fn sync(mut self) -> Result<Self, Error<F>> {
let last_commit_metadata = self.last_commit_metadata.clone();
let inactivity_floor_loc = self.inactivity_floor_loc;
self.witness = self
.witness
.sync::<H, S>(&mut self.merkle, inactivity_floor_loc, || {
Self::encode_commit_op(last_commit_metadata, inactivity_floor_loc)
})
.await?;
Ok(self)
}
#[tracing::instrument(name = "qmdb.keyless.compact.db.prune", level = "info", skip_all)]
pub async fn prune(mut self, pruning_boundary: Location<F>) -> Result<Self, Error<F>> {
self.witness = self.witness.prune(pruning_boundary).await?;
Ok(self)
}
#[boxed]
pub async fn destroy(self) -> Result<(), Error<F>> {
self.witness.destroy().await?;
Ok(())
}
}
impl<F, E, V, H, S> Source for Db<F, E, V, H, S>
where
F: Family,
E: Context,
V: ValueEncoding,
H: Hasher,
Operation<F, V>: EncodeShared + Read,
S: Strategy,
{
type Family = F;
type Digest = H::Digest;
type Op = Operation<F, V>;
type Error = qmdb::Error<F>;
async fn serve(&self, request: Request<F>) -> source::Result<Self> {
let response = self
.witness
.compact_state::<H, S, Self::Op>(
self.merkle.strategy(),
&self.commit_codec_config,
request,
)
.await?;
Ok((response, None))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
journal::contiguous::variable::Config as JournalConfig,
merkle::{mmb, mmr},
metadata::{Config as MetadataConfig, Metadata},
qmdb::{
any::value::FixedEncoding, compact::witness, sync::Response, verify_proof,
verify_proof_and_pinned_nodes,
},
};
use commonware_cryptography::{Sha256, sha256::Digest};
use commonware_macros::test_traced;
use commonware_parallel::Sequential;
use commonware_runtime::{
Runner as _, Spawner as _, Supervisor as _,
buffer::paged::CacheRef,
deterministic,
mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs, fail_pending_syncs},
reschedule,
};
use commonware_utils::{
NZU16, NZU64, NZUsize,
sequence::{U64, VecU64},
};
use core::future::Future;
use futures::FutureExt as _;
use std::num::{NonZeroU16, NonZeroUsize};
type TestDb<F> = Db<F, deterministic::Context, FixedEncoding<U64>, Sha256, Sequential>;
const WITNESS_PAGE_SIZE: NonZeroU16 = NZU16!(77);
const WITNESS_PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9);
fn witness_config(partition: &str) -> JournalConfig<()> {
JournalConfig {
partition: format!("{partition}-witness"),
items_per_section: NZU64!(64),
compression: None,
codec_config: (),
page_cache: CacheRef::new(WITNESS_PAGE_SIZE, WITNESS_PAGE_CACHE_SIZE),
write_buffer: NZUsize!(1024),
replay_buffer: NZUsize!(1024),
}
}
async fn open_bounded<F: Family>(
context: deterministic::Context,
witness_cfg: JournalConfig<()>,
cap: Location<F>,
) -> Result<TestDb<F>, Error<F>> {
let cfg = Config {
strategy: Sequential,
witness: witness_cfg,
commit_codec_config: (),
};
Db::init(context, cfg, Some(cap)).await
}
async fn open_db<F: Family>(context: deterministic::Context, partition: &str) -> TestDb<F> {
let witness_cfg = witness_config(partition);
let cfg = Config {
strategy: Sequential,
witness: witness_cfg,
commit_codec_config: (),
};
Db::init(context, cfg, None).await.unwrap()
}
async fn compact_merkleize_foreign_db_inner<F: Family>(context: deterministic::Context) {
let db = open_db::<F>(context.child("db"), "keyless-merkleize-foreign-db").await;
let foreign = open_db::<F>(
context.child("foreign"),
"keyless-merkleize-foreign-db-foreign",
)
.await;
let batch = db
.new_batch()
.append(U64::new(11))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap();
let batch = foreign
.new_batch()
.append(U64::new(99))
.merkleize(&foreign, None, Location::new(0))
.await
.unwrap();
let foreign = foreign.apply_batch(batch).await.unwrap();
assert_eq!(db.size(), foreign.size());
assert_ne!(db.root(), foreign.root());
let batch = db.new_batch().append(U64::new(22));
assert!(matches!(
batch.merkleize(&foreign, None, Location::new(0)).await,
Err(Error::StaleBatch)
));
let parent = db
.new_batch()
.append(U64::new(33))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child = parent.new_batch::<Sha256>().append(U64::new(44));
assert!(matches!(
child.merkleize(&foreign, None, Location::new(0)).await,
Err(Error::StaleBatch)
));
db.destroy().await.unwrap();
foreign.destroy().await.unwrap();
}
async fn compact_merkleize_stale_sibling_inner<F: Family>(context: deterministic::Context) {
let db = open_db::<F>(context.child("db"), "keyless-merkleize-stale-sibling").await;
let direct = db.new_batch().append(U64::new(11));
let sibling = db
.new_batch()
.append(U64::new(22))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(sibling).await.unwrap();
assert!(matches!(
direct.merkleize(&db, None, Location::new(0)).await,
Err(Error::StaleBatch)
));
let parent = db
.new_batch()
.append(U64::new(33))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child = parent.new_batch::<Sha256>().append(U64::new(44));
let sibling = db
.new_batch()
.append(U64::new(55))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(sibling).await.unwrap();
assert!(matches!(
child.merkleize(&db, None, Location::new(0)).await,
Err(Error::StaleBatch)
));
db.destroy().await.unwrap();
}
async fn compact_merkleize_ancestor_states_inner<F: Family>(context: deterministic::Context) {
let db = open_db::<F>(context.child("db"), "keyless-merkleize-ancestor-states").await;
let grandparent = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let parent = grandparent
.new_batch::<Sha256>()
.append(U64::new(2))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let pending = parent
.new_batch::<Sha256>()
.append(U64::new(3))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(Arc::clone(&grandparent)).await.unwrap();
let applied = parent
.new_batch::<Sha256>()
.append(U64::new(3))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert_eq!(pending.root(), applied.root());
drop(grandparent);
let retired = parent
.new_batch::<Sha256>()
.append(U64::new(3))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert_eq!(retired.bounds().db, db.commitment());
assert_eq!(pending.root(), retired.root());
let child = parent.new_batch::<Sha256>().append(U64::new(3));
let db = db.apply_batch(parent).await.unwrap();
let child = child.merkleize(&db, None, Location::new(0)).await.unwrap();
assert_eq!(pending.root(), child.root());
let expected_root = child.root();
let db = db.apply_batch(child).await.unwrap();
assert_eq!(db.root(), expected_root);
db.destroy().await.unwrap();
}
#[test_traced]
fn test_compact_merkleize_foreign_db_mmr() {
deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::<mmr::Family>);
}
#[test_traced]
fn test_compact_merkleize_foreign_db_mmb() {
deterministic::Runner::default().start(compact_merkleize_foreign_db_inner::<mmb::Family>);
}
#[test_traced]
fn test_compact_merkleize_stale_sibling_mmr() {
deterministic::Runner::default()
.start(compact_merkleize_stale_sibling_inner::<mmr::Family>);
}
#[test_traced]
fn test_compact_merkleize_stale_sibling_mmb() {
deterministic::Runner::default()
.start(compact_merkleize_stale_sibling_inner::<mmb::Family>);
}
#[test_traced]
fn test_compact_merkleize_ancestor_states_mmr() {
deterministic::Runner::default()
.start(compact_merkleize_ancestor_states_inner::<mmr::Family>);
}
#[test_traced]
fn test_compact_merkleize_ancestor_states_mmb() {
deterministic::Runner::default()
.start(compact_merkleize_ancestor_states_inner::<mmb::Family>);
}
async fn proof_refused_after_off_chain_reopen<F: Family>(context: deterministic::Context) {
let db = open_db::<F>(context.child("db"), "off-chain-proof").await;
let mut seed = db.new_batch();
for value in 1..=6 {
seed = seed.append(U64::new(value));
}
let seed = seed.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(seed).await.unwrap();
let db = db.sync().await.unwrap();
let batch = db
.new_batch()
.append(U64::new(100))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let (start, ops) = batch.operations();
let root = batch.root();
let original_proof = batch.proof(&db).unwrap();
let original_pins = batch.pinned_nodes(&db).unwrap();
assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
&original_proof,
start,
&ops,
&original_pins,
&root
));
drop(db);
let db = open_bounded::<F>(
context.child("reopen"),
witness_config("off-chain-proof"),
Location::new(1),
)
.await
.unwrap();
let mut other = db.new_batch();
for value in 11..=16 {
other = other.append(U64::new(value));
}
let other = other.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(other).await.unwrap();
assert_eq!(db.size(), start);
assert!(matches!(
batch
.new_batch::<Sha256>()
.merkleize(&db, None, Location::new(0))
.await,
Err(Error::StaleBatch)
));
let proof = batch.proof(&db);
let pins = batch.pinned_nodes(&db);
assert!(
matches!(proof, Err(Error::StaleRead)),
"proof must return StaleRead at a state outside the batch's chain"
);
assert!(
matches!(pins, Err(Error::StaleRead)),
"pinned nodes must return StaleRead at a state outside the batch's chain"
);
}
#[test]
fn compact_mmr_proof_refused_after_off_chain_reopen() {
deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::<mmr::Family>);
}
#[test]
fn compact_mmb_proof_refused_after_off_chain_reopen() {
deterministic::Runner::default().start(proof_refused_after_off_chain_reopen::<mmb::Family>);
}
async fn compact_operations_and_proof_inner<F: Family>(context: deterministic::Context) {
let db = open_db::<F>(context.child("db"), "keyless-operations-and-proof").await;
let mut seed = db.new_batch();
for value in 1u64..=6 {
seed = seed.append(U64::new(value));
}
let seed = seed
.merkleize(&db, Some(U64::new(7)), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap().sync().await.unwrap();
let floor = db.size();
assert_eq!(floor, Location::new(8));
assert!(matches!(
db.to_batch().proof(&db),
Err(Error::Merkle(crate::merkle::Error::Empty))
));
let mut parent = db.new_batch();
for value in 8u64..=12 {
parent = parent.append(U64::new(value));
}
let parent = parent
.merkleize(&db, Some(U64::new(13)), floor)
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.append(U64::new(14))
.merkleize(&db, Some(U64::new(15)), floor)
.await
.unwrap();
let (child_start, child_ops) = child.operations();
let (_, child_ops_again) = child.operations();
let child_end = child.bounds().tip.size;
assert!(Arc::ptr_eq(&child_ops, &child_ops_again));
assert_eq!(child_start, parent.bounds().tip.size);
assert_eq!(*child_start + child_ops.len() as u64, *child_end);
assert_eq!(child_end, Location::new(16));
assert!(matches!(
child_ops.as_slice(),
[Operation::Append(value), Operation::Commit(Some(metadata), operation_floor)]
if value == &U64::new(14)
&& metadata == &U64::new(15)
&& operation_floor == &floor
));
let child_root = child.root();
let child_proof = child.proof(&db).unwrap();
let child_pins = child.pinned_nodes(&db).unwrap();
assert_eq!(child_proof.leaves, child_end);
assert_eq!(
child_proof.inactive_peaks,
F::inactive_peaks(child_end, floor),
);
assert!(verify_proof::<Sha256, _, _>(
&child_proof,
child_start,
&child_ops,
&child_root
));
assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
&child_proof,
child_start,
&child_ops,
&child_pins,
&child_root
));
assert!(child_pins.len() > 1);
let mut reordered_child_pins = child_pins.clone();
reordered_child_pins.swap(0, 1);
assert!(!verify_proof_and_pinned_nodes::<Sha256, _, _>(
&child_proof,
child_start,
&child_ops,
&reordered_child_pins,
&child_root
));
let db = db.apply_batch(parent).await.unwrap();
let child_proof_after = child.proof(&db).unwrap();
assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins);
assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
&child_proof_after,
child_start,
&child_ops,
&child_pins,
&child_root
));
let applied_start = db.size();
let db = db.apply_batch(child).await.unwrap();
let child_range = applied_start..db.size();
assert_eq!(child_range, child_start..child_end);
let commit_floor = db.size();
let commit_only = db
.new_batch()
.merkleize(&db, Some(U64::new(16)), commit_floor)
.await
.unwrap();
let (commit_start, commit_ops) = commit_only.operations();
let commit_end = commit_only.bounds().tip.size;
let commit_root = commit_only.root();
let commit_proof = commit_only.proof(&db).unwrap();
let commit_pins = commit_only.pinned_nodes(&db).unwrap();
assert_eq!(commit_start, commit_floor);
assert!(matches!(
commit_ops.as_slice(),
[Operation::Commit(Some(metadata), operation_floor)]
if metadata == &U64::new(16) && operation_floor == &commit_floor
));
assert_eq!(*commit_start + commit_ops.len() as u64, *commit_end);
assert_eq!(commit_proof.leaves, commit_end);
assert_eq!(
commit_proof.inactive_peaks,
F::inactive_peaks(commit_end, commit_floor)
);
assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
&commit_proof,
commit_start,
&commit_ops,
&commit_pins,
&commit_root
));
let applied_start = db.size();
let db = db.apply_batch(commit_only).await.unwrap();
let commit_range = applied_start..db.size();
assert_eq!(commit_range, commit_start..commit_end);
let db = db.sync().await.unwrap();
let late_parent = db
.new_batch()
.append(U64::new(17))
.merkleize(&db, Some(U64::new(18)), db.size())
.await
.unwrap();
let late = late_parent
.new_batch::<Sha256>()
.append(U64::new(19))
.merkleize(&db, Some(U64::new(20)), db.size())
.await
.unwrap();
let db = db.apply_batch(Arc::clone(&late_parent)).await.unwrap();
assert!(matches!(
late_parent.proof(&db),
Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
));
assert!(matches!(
late_parent.pinned_nodes(&db),
Err(Error::Merkle(crate::merkle::Error::ElementPruned(_)))
));
drop(late_parent);
let (late_start, late_ops) = late.operations();
let late_root = late.root();
let late_proof = late.proof(&db).unwrap();
let late_pins = late.pinned_nodes(&db).unwrap();
assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
&late_proof,
late_start,
&late_ops,
&late_pins,
&late_root
));
let db = db.apply_batch(late).await.unwrap();
db.destroy().await.unwrap();
}
#[test_traced]
fn test_compact_operations_and_proof_mmr() {
deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmr::Family>);
}
#[test_traced]
fn test_compact_operations_and_proof_mmb() {
deterministic::Runner::default().start(compact_operations_and_proof_inner::<mmb::Family>);
}
async fn open_witness_journal(
context: deterministic::Context,
partition: &str,
) -> witness::Journal<deterministic::Context, mmr::Family, Digest> {
let cfg = witness_config(partition);
witness::Journal::init(context, cfg).await.unwrap()
}
#[test_traced("INFO")]
fn test_serve_refuses_requests_outside_witness() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-serve-refusal").await;
let floor = db.inactivity_floor_loc();
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), floor)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let n = db.target().size;
let boundary = |size: Location<mmr::Family>, start: Location<mmr::Family>| {
Request::Boundary { size, start }
};
let operations =
|size: Location<mmr::Family>, start: Location<mmr::Family>| Request::Operations {
size,
start,
max_ops: NZU64!(1),
};
let beyond = n + 1;
assert!(matches!(
db.serve(boundary(beyond, n)).await,
Err(Error::Merkle(crate::merkle::Error::RangeOutOfBounds(_)))
));
assert!(matches!(
db.serve(operations(Location::new(0), Location::new(0)))
.await,
Err(Error::Merkle(crate::merkle::Error::RangeOutOfBounds(_)))
));
assert!(matches!(
db.serve(operations(n - 1, n - 2)).await,
Err(Error::Journal(crate::journal::Error::ItemPruned(_)))
));
assert!(matches!(
db.serve(boundary(n, n)).await,
Err(Error::Merkle(crate::merkle::Error::RangeOutOfBounds(_)))
));
assert!(matches!(
db.serve(boundary(n, n - 2)).await,
Err(Error::Journal(crate::journal::Error::ItemPruned(_)))
));
let response = db
.serve(Request::Operations {
size: n,
start: n - 1,
max_ops: NZU64!(5),
})
.await
.unwrap()
.0;
let Response::Operations { operations, .. } = response else {
panic!("operations request should get an operations response");
};
assert_eq!(operations.len(), 1);
let response = db.serve(boundary(n, n - 1)).await.unwrap().0;
assert!(matches!(response, Response::Boundary { .. }));
});
}
#[test_traced("INFO")]
fn test_serve_retained_witnesses() {
deterministic::Runner::default().start(|context| async move {
let mut db =
open_db::<mmr::Family>(context.child("db"), "keyless-serve-retained").await;
let mut targets = Vec::new();
for seed in [1, 2] {
let floor = db.inactivity_floor_loc();
let batch = db
.new_batch()
.append(U64::new(seed))
.merkleize(&db, Some(U64::new(seed + 10)), floor)
.await
.unwrap();
db = db.apply_batch(batch).await.unwrap();
targets.push(db.target());
}
let db = db.sync().await.unwrap();
let first = &targets[0];
let start = first.size - 1;
let request = Request::Boundary {
size: first.size,
start,
};
let Response::Boundary {
proof,
op,
pinned_nodes,
} = db.serve(request).await.unwrap().0
else {
panic!("boundary request should get a boundary response");
};
assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
&proof,
start,
std::slice::from_ref(&op),
&pinned_nodes,
&first.root
));
let Response::Operations { proof, operations } = db
.serve(Request::Operations {
size: first.size,
start,
max_ops: NZU64!(1),
})
.await
.unwrap()
.0
else {
panic!("operations request should get an operations response");
};
assert_eq!(operations, vec![op]);
assert!(verify_proof::<Sha256, _, _>(
&proof,
start,
&operations,
&first.root
));
assert!(matches!(
db.serve(Request::Boundary {
size: first.size + 1,
start: first.size,
})
.await,
Err(Error::Journal(crate::journal::Error::ItemPruned(_)))
));
});
}
type DelayedDb = Db<
mmr::Family,
DelayedSyncContext<deterministic::Context>,
FixedEncoding<U64>,
Sha256,
Sequential,
>;
fn open_delayed_db(
context: deterministic::Context,
partition: &str,
pending: &PendingSyncs,
) -> impl Future<Output = Result<DelayedDb, Error<mmr::Family>>> {
let witness_cfg = witness_config(partition);
let cfg = Config {
strategy: Sequential,
witness: witness_cfg,
commit_codec_config: (),
};
let context = DelayedSyncContext {
inner: context,
pending: pending.clone(),
};
DelayedDb::init(context, cfg, None)
}
async fn apply_append(db: DelayedDb, seed: u64) -> DelayedDb {
let floor = db.inactivity_floor_loc();
let batch = db
.new_batch()
.append(U64::new(seed))
.merkleize(&db, Some(U64::new(seed)), floor)
.await
.unwrap();
db.apply_batch(batch).await.unwrap()
}
async fn fail_dropped_watermark_sync(mut db: DelayedDb, pending: &PendingSyncs) -> DelayedDb {
let first;
(db, first) = db.start_sync().await.unwrap();
drive_pending_syncs(pending, first).await.unwrap();
let dropped;
(db, dropped) = db.start_sync().await.unwrap();
assert_eq!(
pending.lock().len(),
1,
"expected only the recovery-watermark sync"
);
fail_pending_syncs(pending);
drop(dropped);
db
}
#[test_traced]
fn test_compact_apply_overlaps_start_sync() {
deterministic::Runner::default().start(|ctx| async move {
let partition = "keyless-start-sync-overlap";
let pending = PendingSyncs::default();
let open = open_delayed_db(ctx.child("delayed"), partition, &pending);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
db = apply_append(db, 1).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 first_target = db.target();
let waiter = ctx
.child("await_sync")
.spawn(|_| async move { handle.await.unwrap() });
while pending.entered() == entered_before {
reschedule().await;
}
db = apply_append(db, 2).await;
assert_ne!(db.root(), first_target.root);
let second_target = db.target();
assert_ne!(second_target, first_target);
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();
drop(db);
let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
.await
.unwrap();
assert_eq!(db.target(), second_target);
assert_eq!(db.get_metadata(), Some(U64::new(2)));
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_compact_start_sync_recovery() {
deterministic::Runner::default().start(|ctx| async move {
let partition = "keyless-start-sync-recovery";
let pending = PendingSyncs::default();
pending.unblock();
let mut db = open_delayed_db(ctx.child("delayed"), partition, &pending)
.await
.unwrap();
db = apply_append(db, 1).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.child("reopen"), partition, &pending)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.get_metadata(), Some(U64::new(1)));
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_compact_start_sync_failure_propagates() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
pending.unblock();
let mut db = open_delayed_db(ctx.child("delayed"), "keyless-start-sync-fail", &pending)
.await
.unwrap();
db = apply_append(db, 1).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_compact_start_sync_then_noop_sync_drains() {
deterministic::Runner::default().start(|ctx| async move {
let partition = "keyless-start-sync-noop-drain";
let pending = PendingSyncs::default();
let open = open_delayed_db(ctx.child("delayed"), partition, &pending);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
db = apply_append(db, 1).await;
let starts_before = pending.starts();
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 root = db.root();
let db = {
let mut sync = std::pin::pin!(db.sync());
assert!(
sync.as_mut().now_or_never().is_none(),
"sync proceeded while the started sync was pending"
);
pending.unblock();
sync.await.unwrap()
};
handle.await.unwrap();
assert!(pending.completions() > completions_before);
drop(db);
let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
.await
.unwrap();
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_compact_start_sync_then_noop_sync_fails_without_new_work() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
let open = open_delayed_db(
ctx.child("delayed"),
"keyless-start-sync-noop-sync-fail",
&pending,
);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
db = apply_append(db, 1).await;
db = fail_dropped_watermark_sync(db, &pending).await;
let starts_before = pending.starts();
assert!(
drive_pending_syncs(&pending, db.sync()).await.is_err(),
"sync absorbed the retained metadata failure"
);
assert_eq!(
pending.starts(),
starts_before,
"sync started new journal work before returning the retained failure"
);
});
}
#[test_traced]
fn test_compact_start_sync_then_noop_commit_waits() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
let open = open_delayed_db(
ctx.child("delayed"),
"keyless-start-sync-noop-commit",
&pending,
);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
db = apply_append(db, 1).await;
let handle;
(db, handle) = db.start_sync().await.unwrap();
let starts_before = pending.starts();
let db = {
let mut commit = std::pin::pin!(db.commit());
assert!(
commit.as_mut().now_or_never().is_none(),
"commit proceeded while the started sync was pending"
);
pending.unblock();
commit.await.unwrap()
};
handle.await.unwrap();
assert_eq!(
pending.starts(),
starts_before,
"a successful pipelined sync still triggered journal work"
);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_compact_start_sync_noop_second_call() {
deterministic::Runner::default().start(|ctx| async move {
let partition = "keyless-start-sync-noop-second";
let pending = PendingSyncs::default();
let open = open_delayed_db(ctx.child("delayed"), partition, &pending);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
db = apply_append(db, 1).await;
let h1;
(db, h1) = db.start_sync().await.unwrap();
pending.unblock();
h1.await.unwrap();
let h2;
(db, h2) = db.start_sync().await.unwrap();
h2.await.unwrap();
let root = db.root();
drop(db);
let journal = open_witness_journal(ctx.child("probe"), partition).await;
assert_eq!(journal.size(), 2);
drop(journal);
let db = open_delayed_db(ctx.child("reopen"), partition, &pending)
.await
.unwrap();
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_compact_start_sync_metadata_failure_resurfaces_on_commit() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
let open = open_delayed_db(
ctx.child("delayed"),
"keyless-start-sync-meta-fail",
&pending,
);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
db = apply_append(db, 1).await;
let handle;
(db, handle) = db.start_sync().await.unwrap();
{
let mut parked = pending.lock();
assert_eq!(
parked.len(),
2,
"expected the data and offsets syncs parked"
);
let offsets = parked.pop().unwrap();
let data = parked.pop().unwrap();
data.release.send(Ok(())).unwrap();
offsets
.release
.send(Err(commonware_runtime::Error::Io(
std::io::Error::other("injected sync failure").into(),
)))
.unwrap();
}
assert!(
handle.await.is_err(),
"the sync handle surfaces the failure"
);
pending.unblock();
let db = apply_append(db, 2).await;
assert!(
db.commit().await.is_err(),
"commit absorbed the retained metadata failure"
);
});
}
#[test_traced]
fn test_compact_start_sync_retains_dropped_metadata_failure() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
let open = open_delayed_db(
ctx.child("delayed"),
"keyless-start-sync-dropped-meta-fail",
&pending,
);
let mut db = drive_pending_syncs(&pending, open).await.unwrap();
db = apply_append(db, 1).await;
db = fail_dropped_watermark_sync(db, &pending).await;
let next;
(db, next) = db.start_sync().await.unwrap();
assert!(
drive_pending_syncs(&pending, next).await.is_err(),
"a later start_sync masked the retained metadata failure"
);
drop(db);
});
}
#[test_traced]
fn test_compact_start_sync_proven_skips_journal() {
deterministic::Runner::default().start(|ctx| async move {
let pending = PendingSyncs::default();
pending.unblock();
let mut db =
open_delayed_db(ctx.child("delayed"), "keyless-start-sync-proven", &pending)
.await
.unwrap();
db = apply_append(db, 1).await;
let handle;
(db, handle) = db.start_sync().await.unwrap();
handle.await.unwrap();
let starts_before = pending.starts();
let db = db.commit().await.unwrap();
assert_eq!(
pending.starts(),
starts_before,
"a proven pipelined sync still triggered journal work"
);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_start_sync_persists_import() {
deterministic::Runner::default().start(|context| async move {
let dst = "keyless-import-start-sync-dst";
let src = "keyless-import-start-sync-src";
let meta_a = U64::new(11);
let meta_b = U64::new(22);
let target_b = {
let source = open_db::<mmr::Family>(context.child("src"), src).await;
let batch = source
.new_batch()
.append(U64::new(2))
.merkleize(&source, Some(meta_b.clone()), Location::new(0))
.await
.unwrap();
let source = source.apply_batch(batch).await.unwrap();
let source = source.sync().await.unwrap();
source.target()
};
let (_, size_b, pinned_b) = {
let journal = open_witness_journal(context.child("tip"), src).await;
witness::tests::tip(&journal).await
};
assert_eq!(size_b, target_b.size);
{
let seeded = open_db::<mmr::Family>(context.child("seed"), dst).await;
let batch = seeded
.new_batch()
.append(U64::new(1))
.merkleize(&seeded, Some(meta_a), Location::new(0))
.await
.unwrap();
let seeded = seeded.apply_batch(batch).await.unwrap();
let seeded = seeded.sync().await.unwrap();
assert_ne!(seeded.target(), target_b);
}
{
let journal = open_witness_journal(context.child("import"), dst).await;
let imported = TestDb::<mmr::Family>::init_from_sync(
Sequential,
journal,
(),
size_b - 1,
pinned_b,
Operation::Commit(Some(meta_b.clone()), Location::new(0)),
)
.unwrap();
assert_eq!(imported.target(), target_b);
let (_imported, handle) = imported.start_sync().await.unwrap();
handle.await.unwrap();
}
let db = open_db::<mmr::Family>(context.child("reopen"), dst).await;
assert_eq!(db.target(), target_b);
assert_eq!(db.root(), target_b.root);
assert_eq!(db.get_metadata(), Some(meta_b));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_stale_batch_rejected() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-stale").await;
let floor = db.inactivity_floor_loc();
let batch_a = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), floor)
.await
.unwrap();
let batch_b = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, Some(U64::new(22)), floor)
.await
.unwrap();
let stale = db.new_batch().append(U64::new(3));
let expected_root = batch_a.root();
let db = db.apply_batch(batch_a).await.unwrap();
assert_eq!(db.root(), expected_root);
assert!(matches!(
stale.merkleize(&db, Some(U64::new(33)), floor).await,
Err(Error::StaleBatch)
));
assert!(matches!(
db.apply_batch(batch_b).await,
Err(Error::StaleBatch)
));
});
}
#[test_traced("INFO")]
fn test_compact_delayed_merkleize_after_ancestor_apply() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-delayed-child").await;
let floor = db.inactivity_floor_loc();
let a = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, None, floor)
.await
.unwrap();
let b = a
.new_batch::<Sha256>()
.append(U64::new(2))
.merkleize(&db, None, floor)
.await
.unwrap();
let c = b.new_batch::<Sha256>().append(U64::new(3));
let db = db.apply_batch(a).await.unwrap();
let c = c.merkleize(&db, None, floor).await.unwrap();
let expected_root = c.root();
let db = db.apply_batch(c).await.unwrap();
assert_eq!(db.root(), expected_root);
});
}
#[test_traced("INFO")]
fn test_compact_to_batch_reflects_live_state() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-to-batch-live").await;
let floor = db.inactivity_floor_loc();
let pre_apply_root = db.root();
let pre_snapshot = db.to_batch();
assert_eq!(
pre_snapshot.root(),
pre_apply_root,
"snapshot before any mutation should match the live root"
);
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), floor)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap();
let live_root = db.root();
assert_ne!(
live_root, pre_apply_root,
"applying a non-empty batch must change the live root"
);
let snapshot = db.to_batch();
assert_eq!(
snapshot.root(),
live_root,
"to_batch().root() must match the live db.root() even before sync"
);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_stale_batch_chained() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-chained-stale").await;
let floor = db.inactivity_floor_loc();
let common_parent = db
.new_batch()
.append(U64::new(10))
.merkleize(&db, Some(U64::new(110)), floor)
.await
.unwrap();
let sibling_a = common_parent
.new_batch::<Sha256>()
.append(U64::new(11))
.merkleize(&db, Some(U64::new(111)), floor)
.await
.unwrap();
let sibling_b = common_parent
.new_batch::<Sha256>()
.append(U64::new(12))
.merkleize(&db, Some(U64::new(112)), floor)
.await
.unwrap();
let db = db.apply_batch(sibling_a).await.unwrap();
assert!(matches!(
db.validate_batch(&sibling_b),
Err(Error::StaleBatch)
));
let parent_a = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), floor)
.await
.unwrap();
let parent_b = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, Some(U64::new(22)), floor)
.await
.unwrap();
let child_b = parent_b
.new_batch::<Sha256>()
.append(U64::new(3))
.merkleize(&db, Some(U64::new(33)), floor)
.await
.unwrap();
let db = db.apply_batch(parent_a).await.unwrap();
assert!(matches!(
db.validate_batch(&child_b),
Err(Error::StaleBatch)
));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_stale_parent_after_child_applied() {
deterministic::Runner::default().start(|context| async move {
let db =
open_db::<mmr::Family>(context.child("db"), "keyless-child-before-parent").await;
let floor = db.inactivity_floor_loc();
let parent = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), floor)
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.append(U64::new(2))
.merkleize(&db, Some(U64::new(22)), floor)
.await
.unwrap();
let db = db.apply_batch(child).await.unwrap();
assert!(matches!(
db.apply_batch(parent).await,
Err(Error::StaleBatch)
));
});
}
#[test_traced("INFO")]
fn test_compact_sequential_commit_parent_then_child() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-parent-child").await;
let floor = db.inactivity_floor_loc();
let parent = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), floor)
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.append(U64::new(2))
.merkleize(&db, Some(U64::new(22)), floor)
.await
.unwrap();
let expected_root = child.root();
let db = db.apply_batch(parent).await.unwrap();
let db = db.apply_batch(child).await.unwrap().sync().await.unwrap();
assert_eq!(db.root(), expected_root);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_floor_regressed() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-floor-regressed").await;
let advance_floor = db.new_batch().append(U64::new(1));
let advance_floor = advance_floor
.merkleize(&db, None, Location::new(1))
.await
.unwrap();
let db = db.apply_batch(advance_floor).await.unwrap();
let db = db.sync().await.unwrap();
let regressed = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, None, Location::new(0))
.await;
assert!(matches!(
regressed,
Err(Error::FloorRegressed(new, current))
if new == Location::new(0) && current == Location::new(1)
));
});
}
#[test_traced("INFO")]
fn test_compact_chained_floor_regression() {
deterministic::Runner::default().start(|context| async move {
let db =
open_db::<mmr::Family>(context.child("db"), "keyless-ancestor-floor-regressed")
.await;
let parent = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.append(U64::new(2))
.merkleize(&db, None, Location::new(1))
.await;
assert!(matches!(
child,
Err(Error::FloorRegressed(new, prev))
if new == Location::new(1) && prev == Location::new(2)
));
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_restores_commit_metadata_and_floor() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-rewind-meta").await;
let v1 = U64::new(1);
let meta1 = U64::new(11);
let floor1 = Location::new(0);
let batch = db
.new_batch()
.append(v1)
.merkleize(&db, Some(meta1.clone()), floor1)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root_after_first = db.root();
let size_after_first = db.size();
let v2 = U64::new(2);
let meta2 = U64::new(22);
let floor2 = Location::new(1);
let batch = db
.new_batch()
.append(v2)
.merkleize(&db, Some(meta2.clone()), floor2)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
assert_eq!(db.get_metadata(), Some(meta2));
assert_eq!(db.inactivity_floor_loc(), floor2);
let db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(
context.child("cap"),
witness_config("keyless-rewind-meta"),
size_after_first,
)
.await
}
.unwrap();
assert_eq!(db.root(), root_after_first);
assert_eq!(db.get_metadata(), Some(meta1));
assert_eq!(db.inactivity_floor_loc(), floor1);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_persists_across_reopen() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-rewind-reopen";
let meta1 = U64::new(11);
let floor1 = Location::new(0);
let meta2 = U64::new(22);
let floor2 = Location::new(1);
let root_after_first = {
let db = open_db::<mmr::Family>(context.child("first"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(meta1.clone()), floor1)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root = db.root();
let size_after_first = db.size();
let batch = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, Some(meta2), floor2)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let _db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(
context.child("cap"),
witness_config(partition),
size_after_first,
)
.await
}
.unwrap();
root
};
let db = open_db::<mmr::Family>(context.child("second"), partition).await;
assert_eq!(db.root(), root_after_first);
assert_eq!(db.get_metadata(), Some(meta1));
assert_eq!(db.inactivity_floor_loc(), floor1);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_commit_persists_across_reopen() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-commit-reopen";
let meta1 = U64::new(11);
let meta2 = U64::new(22);
let root_after_second = {
let db = open_db::<mmr::Family>(context.child("first"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(meta1), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
let batch = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, Some(meta2.clone()), Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
db.root()
};
let db = open_db::<mmr::Family>(context.child("second"), partition).await;
assert_eq!(db.root(), root_after_second);
assert_eq!(db.get_metadata(), Some(meta2));
assert_eq!(db.inactivity_floor_loc(), Location::new(1));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_to_committed_entry_after_reopen() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-commit-rewind-reopen";
let meta1 = U64::new(11);
let meta2 = U64::new(22);
let (root_a, size_a) = {
let db = open_db::<mmr::Family>(context.child("first"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(meta1.clone()), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
let root_a = db.root();
let size_a = db.size();
let batch = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, Some(meta2), Location::new(1))
.await
.unwrap();
let _db = db.apply_batch(batch).await.unwrap().commit().await.unwrap();
(root_a, size_a)
};
let db = open_db::<mmr::Family>(context.child("second"), partition).await;
let db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(context.child("cap"), witness_config(partition), size_a)
.await
}
.unwrap();
assert_eq!(db.root(), root_a);
assert_eq!(db.get_metadata(), Some(meta1));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_sync_after_commit() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-sync-after-commit";
let meta = U64::new(11);
let root = {
let db = open_db::<mmr::Family>(context.child("first"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(meta.clone()), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap();
let db = db.commit().await.unwrap();
let db = db.sync().await.unwrap();
db.root()
};
let metadata = Metadata::<_, u64, VecU64>::init(
context.child("checkpoint"),
MetadataConfig {
partition: format!("{partition}-witness_offsets-metadata"),
codec_config: (),
},
)
.await
.unwrap();
assert_eq!(metadata.get(&3).copied().map(u64::from), Some(2));
drop(metadata);
let db = open_db::<mmr::Family>(context.child("second"), partition).await;
assert_eq!(db.root(), root);
assert_eq!(db.get_metadata(), Some(meta));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_import_persists_with_commit() {
deterministic::Runner::default().start(|context| async move {
let dst = "keyless-import-commit-dst";
let src = "keyless-import-commit-src";
let meta_a = U64::new(11);
let meta_b = U64::new(22);
let (target_b, pinned_b) = {
let source = open_db::<mmr::Family>(context.child("src"), src).await;
let batch = source
.new_batch()
.append(U64::new(2))
.merkleize(&source, Some(meta_b.clone()), Location::new(0))
.await
.unwrap();
let source = source.apply_batch(batch).await.unwrap();
let source = source.sync().await.unwrap();
let target = source.target();
let response = source
.serve(Request::Boundary {
size: target.size,
start: target.size - 1,
})
.await
.unwrap()
.0;
let Response::Boundary { pinned_nodes, .. } = response else {
panic!("boundary request should get a boundary response");
};
(target, pinned_nodes)
};
{
let seeded = open_db::<mmr::Family>(context.child("seed"), dst).await;
let batch = seeded
.new_batch()
.append(U64::new(1))
.merkleize(&seeded, Some(meta_a), Location::new(0))
.await
.unwrap();
let seeded = seeded.apply_batch(batch).await.unwrap();
let seeded = seeded.sync().await.unwrap();
assert_ne!(seeded.target(), target_b);
}
{
let journal = open_witness_journal(context.child("import"), dst).await;
let imported = TestDb::<mmr::Family>::init_from_sync(
Sequential,
journal,
(),
target_b.size - 1,
pinned_b,
Operation::Commit(Some(meta_b.clone()), Location::new(0)),
)
.unwrap();
assert_eq!(imported.target(), target_b);
let _imported = imported.commit().await.unwrap();
}
let db = open_db::<mmr::Family>(context.child("reopen"), dst).await;
assert_eq!(db.target(), target_b);
assert_eq!(db.root(), target_b.root);
assert_eq!(db.get_metadata(), Some(meta_b));
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_reopen_rejects_tampered_witness() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-witness-tamper";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(7))
.merkleize(&db, Some(U64::new(11)), Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
drop(db);
let journal = open_witness_journal(context.child("tamper"), partition).await;
let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
pinned_nodes.push(Sha256::fill(0xff));
witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;
let cfg = Config {
strategy: Sequential,
witness: witness_config(partition),
commit_codec_config: (),
};
let reopened =
TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
assert!(matches!(reopened, Err(Error::DataCorrupted(_))));
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_rejects_corrupt_target_entry() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-corrupt-rewind-target";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, None, Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let initialization_bound = db.target().size;
let batch = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, None, Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let tip_target = db.target();
drop(db);
let mut journal = open_witness_journal(context.child("corrupt"), partition).await;
journal = witness::tests::corrupt_entry(journal, 1, |entry| {
entry.pinned_nodes.push(Sha256::fill(0xff));
})
.await;
drop(journal);
let cfg = Config {
strategy: Sequential,
witness: witness_config(partition),
commit_codec_config: (),
};
let reopened = TestDb::<mmr::Family>::init(context.child("reopen"), cfg, None)
.await
.unwrap();
assert_eq!(reopened.target(), tip_target);
assert!(matches!(
{
drop(reopened);
open_bounded::<mmr::Family>(
context.child("cap"),
witness_config(partition),
initialization_bound,
)
.await
},
Err(Error::DataCorrupted(_))
));
let cfg = Config {
strategy: Sequential,
witness: witness_config(partition),
commit_codec_config: (),
};
let reopened = TestDb::<mmr::Family>::init(context.child("reopen2"), cfg, None)
.await
.unwrap();
assert_eq!(reopened.target(), tip_target);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_reopen_rejects_interrupted_import() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-interrupted-import";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(7))
.merkleize(&db, Some(U64::new(11)), Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
drop(db);
let journal = open_witness_journal(context.child("clear"), partition).await;
let size = journal.size();
let journal = journal.clear_to_size(size.max(1)).await.unwrap();
drop(journal);
let cfg = Config {
strategy: Sequential,
witness: witness_config(partition),
commit_codec_config: (),
};
let reopened =
TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
assert!(matches!(
reopened,
Err(Error::DataCorrupted("witness journal has no tip"))
));
});
}
#[test_traced("INFO")]
fn test_compact_reopen_rejects_commit_floor_beyond_tip() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-invalid-persisted-floor";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(7))
.merkleize(&db, Some(U64::new(11)), Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
drop(db);
let oversized_floor = Location::new(10);
let journal = open_witness_journal(context.child("tamper"), partition).await;
let (_, size, pinned_nodes) = witness::tests::tip(&journal).await;
let bad_op = Operation::<mmr::Family, FixedEncoding<U64>>::Commit(
Some(U64::new(11)),
oversized_floor,
)
.encode()
.to_vec();
witness::tests::overwrite_tip(journal, bad_op, size, pinned_nodes).await;
let cfg = Config {
strategy: Sequential,
witness: witness_config(partition),
commit_codec_config: (),
};
let reopened =
TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None).await;
assert!(matches!(
reopened,
Err(Error::DataCorrupted("invalid compact witness"))
));
});
}
#[test_traced("INFO")]
fn test_compact_reopen_rejects_tampered_pinned_nodes() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-pinned-nodes-tamper";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(7))
.merkleize(&db, Some(U64::new(11)), Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let tampered_target = db.target();
drop(db);
let journal = open_witness_journal(context.child("tamper"), partition).await;
let (op_bytes, size, mut pinned_nodes) = witness::tests::tip(&journal).await;
pinned_nodes[0] = Sha256::fill(0xff);
witness::tests::overwrite_tip(journal, op_bytes, size, pinned_nodes).await;
let cfg = Config {
strategy: Sequential,
witness: witness_config(partition),
commit_codec_config: (),
};
let reopened = TestDb::<mmr::Family>::init(context.child("reopen_witness"), cfg, None)
.await
.unwrap();
assert_ne!(reopened.target(), tampered_target);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_preserves_current_state() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-rewind-noop").await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root = db.root();
let size = db.size();
let db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(
context.child("cap"),
witness_config("keyless-rewind-noop"),
size,
)
.await
}
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.size(), size);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_prune_past_tip_keeps_tip() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-prune-past-tip";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let target = db.target();
let boundary = db.size() + 100;
let db = db.prune(boundary).await.unwrap();
assert_eq!(db.target(), target);
drop(db);
let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
assert_eq!(reopened.target(), target);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_initialization_zero_and_above_end() {
deterministic::Runner::default().start(|context| async move {
let cfg = witness_config("keyless-caps");
assert!(matches!(
open_bounded::<mmr::Family>(context.child("zero"), cfg.clone(), Location::new(0))
.await,
Err(Error::InvalidInitializationBound)
));
let db = open_bounded::<mmr::Family>(
context.child("fresh"),
cfg.clone(),
Location::new(100),
)
.await
.unwrap();
let root = db.root();
assert_eq!(db.size(), Location::new(1));
drop(db);
let db = open_bounded::<mmr::Family>(context.child("above"), cfg, Location::new(100))
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.size(), Location::new(1));
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_between_commits() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-rewind-between").await;
let floor = db.inactivity_floor_loc();
let initial_root = db.root();
let batch = db
.new_batch()
.append(U64::new(1))
.append(U64::new(2))
.merkleize(&db, Some(U64::new(11)), floor)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root_a = db.root();
let size_a = db.size();
assert_eq!(size_a, Location::new(4));
let batch = db
.new_batch()
.append(U64::new(3))
.merkleize(&db, Some(U64::new(22)), floor)
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let cfg = witness_config("keyless-rewind-between");
drop(db);
for (target, size, root) in [(5, 4, root_a), (3, 1, initial_root), (2, 1, initial_root)]
{
let db = open_bounded::<mmr::Family>(
context.child("cap"),
cfg.clone(),
Location::new(target),
)
.await
.unwrap();
assert_eq!(db.size(), Location::new(size));
assert_eq!(db.root(), root);
drop(db);
let db = open_db::<mmr::Family>(context.child("restart"), "keyless-rewind-between")
.await;
assert_eq!(db.root(), root);
}
});
}
#[test_traced("INFO")]
fn test_compact_reopen_drops_unsynced_witness() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-witness-unsynced";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let target_a = db.target();
drop(db);
let journal = open_witness_journal(context.child("crash"), partition).await;
let (op_bytes, mut size, pinned_nodes) = witness::tests::tip(&journal).await;
size += 2;
witness::tests::append_unsynced(journal, op_bytes, size, pinned_nodes).await;
let reopened = open_db::<mmr::Family>(context.child("reopen"), partition).await;
assert_eq!(reopened.target(), target_a);
reopened.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_multiple_commits() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-rewind-multi";
let db = open_db::<mmr::Family>(context.child("db"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, Some(U64::new(11)), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root_a = db.root();
let size_a = db.size();
let target_a = db.target();
let mut db = db;
for i in [2u64, 3] {
let batch = db
.new_batch()
.append(U64::new(i))
.merkleize(&db, Some(U64::new(i * 11)), Location::new(0))
.await
.unwrap();
db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
}
assert_ne!(db.root(), root_a);
let db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(context.child("cap"), witness_config(partition), size_a)
.await
}
.unwrap();
assert_eq!(db.root(), root_a);
assert_eq!(db.size(), size_a);
assert_eq!(db.get_metadata(), Some(U64::new(11)));
assert_eq!(db.target(), target_a);
drop(db);
let db = open_db::<mmr::Family>(context.child("reopen"), partition).await;
assert_eq!(db.root(), root_a);
assert_eq!(db.target(), target_a);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_prune_then_bounded_initialization() {
deterministic::Runner::default().start(|context| async move {
let mut witness_cfg = witness_config("keyless-prune-rewind");
witness_cfg.items_per_section = NZU64!(1);
let cfg = Config {
strategy: Sequential,
witness: witness_cfg.clone(),
commit_codec_config: (),
};
let mut db: TestDb<mmr::Family> =
Db::init(context.child("db"), cfg, None).await.unwrap();
let mut sizes = Vec::new();
for i in [1u64, 2, 3] {
let batch = db
.new_batch()
.append(U64::new(i))
.merkleize(&db, Some(U64::new(i * 11)), Location::new(0))
.await
.unwrap();
db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
sizes.push(db.size());
}
let db = db.prune(sizes[1]).await.unwrap().sync().await.unwrap();
drop(db);
let db = open_bounded::<mmr::Family>(
context.child("retained"),
witness_cfg.clone(),
sizes[1],
)
.await
.unwrap();
assert_eq!(db.size(), sizes[1]);
assert_eq!(db.get_metadata(), Some(U64::new(22)));
drop(db);
assert!(matches!(
open_bounded::<mmr::Family>(context.child("pruned"), witness_cfg, sizes[0]).await,
Err(Error::HistoricalFloorPruned(_))
));
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_preserves_pre_advance_batch() {
deterministic::Runner::default().start(|context| async move {
let db =
open_db::<mmr::Family>(context.child("db"), "keyless-rewind-preserves-pre-advance")
.await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let size_after_first = db.size();
let held = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let batch = db
.new_batch()
.append(U64::new(3))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(
context.child("cap"),
witness_config("keyless-rewind-preserves-pre-advance"),
size_after_first,
)
.await
}
.unwrap();
let db = db.apply_batch(held).await.unwrap();
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_noop_commit_after_commit() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-noop-after-commit").await;
let batch = db
.new_batch()
.append(U64::new(1))
.append(U64::new(2))
.merkleize(&db, Some(U64::new(11)), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root_after_first = db.root();
assert_eq!(db.size(), Location::new(4));
let db = db.sync().await.unwrap();
assert_eq!(db.size(), Location::new(4));
assert_eq!(db.root(), root_after_first);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_noop_commit_after_reopen() {
deterministic::Runner::default().start(|context| async move {
let partition = "keyless-noop-after-reopen";
let root_before_drop = {
let db = open_db::<mmr::Family>(context.child("first"), partition).await;
let batch = db
.new_batch()
.append(U64::new(1))
.append(U64::new(2))
.merkleize(&db, Some(U64::new(11)), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root = db.root();
assert_eq!(db.size(), Location::new(4));
root
};
let db = open_db::<mmr::Family>(context.child("second"), partition).await;
assert_eq!(db.root(), root_before_drop);
assert_eq!(db.size(), Location::new(4));
let db = db.sync().await.unwrap();
assert_eq!(db.size(), Location::new(4));
assert_eq!(db.root(), root_before_drop);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_noop_commit_after_bounded_initialization() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-noop-after-rewind").await;
let batch = db
.new_batch()
.append(U64::new(1))
.append(U64::new(2))
.merkleize(&db, Some(U64::new(11)), Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let root_after_first = db.root();
let batch = db
.new_batch()
.append(U64::new(3))
.merkleize(&db, Some(U64::new(22)), Location::new(1))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(
context.child("cap"),
witness_config("keyless-noop-after-rewind"),
Location::new(4),
)
.await
}
.unwrap();
assert_eq!(db.size(), Location::new(4));
assert_eq!(db.root(), root_after_first);
let db = db.sync().await.unwrap();
assert_eq!(db.size(), Location::new(4));
assert_eq!(db.root(), root_after_first);
db.destroy().await.unwrap();
});
}
#[test_traced("INFO")]
fn test_compact_bounded_initialization_makes_post_advance_batch_stale() {
deterministic::Runner::default().start(|context| async move {
let db =
open_db::<mmr::Family>(context.child("db"), "keyless-rewind-makes-stale").await;
let batch = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let size_after_first = db.size();
let batch = db
.new_batch()
.append(U64::new(2))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
let held = db
.new_batch()
.append(U64::new(3))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = {
_ = db.sync().await.unwrap();
open_bounded::<mmr::Family>(
context.child("cap"),
witness_config("keyless-rewind-makes-stale"),
size_after_first,
)
.await
}
.unwrap();
assert!(matches!(db.apply_batch(held).await, Err(Error::StaleBatch)));
});
}
#[test_traced("INFO")]
fn test_compact_floor_beyond_size() {
deterministic::Runner::default().start(|context| async move {
let db = open_db::<mmr::Family>(context.child("db"), "keyless-floor-beyond").await;
let batch = db.new_batch().merkleize(&db, None, Location::new(2)).await;
assert!(matches!(
batch,
Err(Error::FloorBeyondSize(floor, tip))
if floor == Location::new(2) && tip == Location::new(1)
));
});
}
#[test_traced("INFO")]
fn test_compact_chained_floor_beyond_commit() {
deterministic::Runner::default().start(|context| async move {
let db =
open_db::<mmr::Family>(context.child("db"), "keyless-ancestor-floor-beyond").await;
let parent = db
.new_batch()
.append(U64::new(1))
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.append(U64::new(2))
.merkleize(&db, None, Location::new(5))
.await;
assert!(matches!(
child,
Err(Error::FloorBeyondSize(floor, commit))
if floor == Location::new(5) && commit == Location::new(4)
));
});
}
}