use crate::{
Context,
journal::{
authenticated,
contiguous::{Contiguous, Mutable},
},
merkle::{Family, Location, Proof, full::Config as MerkleConfig},
qmdb::{
Error, any::value::ValueEncoding, batch_chain, metrics::Metrics, single_operation_root,
},
};
use commonware_codec::EncodeShared;
use commonware_cryptography::Hasher;
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use commonware_runtime::Handle;
use std::{num::NonZeroU64, sync::Arc};
use tracing::{debug, warn};
pub mod batch;
mod compact;
pub mod fixed;
mod operation;
pub(crate) mod sync;
pub mod variable;
pub use compact::{
Config as CompactConfig, Db as CompactDb, MerkleizedBatch as CompactMerkleizedBatch,
UnmerkleizedBatch as CompactUnmerkleizedBatch,
};
pub use operation::Operation;
pub fn initial_root<F, V, H>() -> H::Digest
where
F: Family,
V: ValueEncoding,
H: Hasher,
Operation<F, V>: EncodeShared,
{
single_operation_root::<F, H>(&Operation::<F, V>::Commit(None, Location::new(0)))
}
#[derive(Clone)]
pub struct Config<J, S: Strategy> {
pub merkle: MerkleConfig<S>,
pub log: J,
}
pub struct Keyless<F, E, V, C, H, S>
where
F: Family,
E: Context,
V: ValueEncoding,
C: Contiguous<Item = Operation<F, V>>,
H: Hasher,
S: Strategy,
Operation<F, V>: EncodeShared,
{
journal: authenticated::Journal<F, E, C, H, S>,
root: H::Digest,
last_commit_loc: Location<F>,
inactivity_floor_loc: Location<F>,
metrics: Metrics<E>,
}
impl<F, E, V, C, H, S> std::fmt::Debug for Keyless<F, E, V, C, H, S>
where
F: Family,
E: Context,
V: ValueEncoding,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
S: Strategy,
Operation<F, V>: EncodeShared,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Keyless")
.field("bounds", &self.bounds())
.field("inactivity_floor_loc", &self.inactivity_floor_loc())
.finish_non_exhaustive()
}
}
impl<F, E, V, C, H, S> Keyless<F, E, V, C, H, S>
where
F: Family,
E: Context,
V: ValueEncoding,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
S: Strategy,
Operation<F, V>: EncodeShared,
{
#[boxed]
pub(crate) async fn init_from_journal(
mut journal: authenticated::Journal<F, E, C, H, S>,
context: E,
) -> Result<Self, Error<F>> {
let metrics = Metrics::new(context);
if journal.size() == 0 {
warn!("no operations found in log, creating initial commit");
(journal, _) = journal
.append(&Operation::Commit(None, Location::new(0)))
.await?;
journal = journal.sync().await?;
}
let (last_commit_loc, inactivity_floor_loc) = {
let bounds = journal.bounds();
let last_commit_loc = Location::new(
bounds
.end
.checked_sub(1)
.expect("at least one commit should exist"),
);
let op = journal.read(*last_commit_loc).await?;
let inactivity_floor_loc = op
.has_floor()
.expect("last operation should be a commit with floor");
(last_commit_loc, inactivity_floor_loc)
};
let inactive_peaks = F::inactive_peaks(last_commit_loc + 1, inactivity_floor_loc);
let root = journal.root(inactive_peaks)?;
let db = Self {
journal,
root,
last_commit_loc,
inactivity_floor_loc,
metrics,
};
db.update_metrics();
Ok(db)
}
pub async fn get(&self, loc: Location<F>) -> Result<Option<V::Value>, Error<F>> {
let _timer = self.metrics.get_timer();
self.metrics.get_calls.inc();
self.metrics.lookups_requested.inc();
let op_count = self.journal.bounds().end;
if loc >= op_count {
return Err(Error::LocationOutOfBounds(loc, Location::new(op_count)));
}
let op = self.journal.read(*loc).await?;
let result = op.into_value();
Ok(result)
}
pub async fn get_many(&self, locs: &[Location<F>]) -> Result<Vec<Option<V::Value>>, Error<F>> {
if locs.is_empty() {
return Ok(Vec::new());
}
let _timer = self.metrics.get_many_timer();
self.metrics.get_many_calls.inc();
self.metrics.lookups_requested.inc_by(locs.len() as u64);
assert!(
locs.is_sorted_by(|a, b| a < b),
"locations must be strictly increasing"
);
let op_count = self.journal.bounds().end;
for &loc in locs {
if loc >= op_count {
return Err(Error::LocationOutOfBounds(loc, Location::new(op_count)));
}
}
let positions: Vec<u64> = locs.iter().map(|loc| **loc).collect();
let ops = self.journal.read_many(&positions).await?;
let result = ops.into_iter().map(|op| op.into_value()).collect();
Ok(result)
}
pub const fn last_commit_loc(&self) -> Location<F> {
self.last_commit_loc
}
pub const fn inactivity_floor_loc(&self) -> Location<F> {
self.inactivity_floor_loc
}
pub fn bounds(&self) -> std::ops::Range<Location<F>> {
let bounds = self.journal.bounds();
Location::new(bounds.start)..Location::new(bounds.end)
}
fn update_metrics(&self) {
let bounds = self.journal.bounds();
self.metrics.update(
bounds.end,
bounds.start,
*self.inactivity_floor_loc,
*self.last_commit_loc,
);
}
pub const fn sync_boundary(&self) -> Location<F> {
self.inactivity_floor_loc
}
pub async fn get_metadata(&self) -> Result<Option<V::Value>, Error<F>> {
let op = self.journal.read(*self.last_commit_loc).await?;
let Operation::Commit(metadata, _floor) = op else {
return Ok(None);
};
Ok(metadata)
}
pub const fn root(&self) -> H::Digest {
self.root
}
pub const fn strategy(&self) -> &S {
self.journal.strategy()
}
pub async fn proof(
&self,
start_loc: Location<F>,
max_ops: NonZeroU64,
) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, V>>), Error<F>> {
self.historical_proof(self.bounds().end, start_loc, max_ops)
.await
}
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.keyless.db.historical_proof",
level = "info",
skip_all,
fields(
op_count = *op_count,
start_loc = *start_loc,
max_ops = max_ops.get(),
),
)]
pub async fn historical_proof(
&self,
op_count: Location<F>,
start_loc: Location<F>,
max_ops: NonZeroU64,
) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, V>>), Error<F>> {
if op_count > self.journal.size() {
return Err(crate::merkle::Error::RangeOutOfBounds(op_count).into());
}
let inactive_peaks =
crate::qmdb::inactive_peaks_at::<F, _>(&self.journal, op_count).await?;
Ok(self
.journal
.historical_proof(op_count, start_loc, max_ops, inactive_peaks)
.await?)
}
pub async fn pinned_nodes_at(&self, loc: Location<F>) -> Result<Vec<H::Digest>, Error<F>> {
self.journal
.merkle
.pinned_nodes_at(loc)
.await
.map_err(Into::into)
}
#[tracing::instrument(name = "qmdb.keyless.db.prune", level = "info", skip_all)]
#[boxed]
pub async fn prune(mut self, loc: Location<F>) -> Result<Self, Error<F>> {
let _timer = self.metrics.prune_timer();
self.metrics.prune_calls.inc();
if loc > self.inactivity_floor_loc {
return Err(Error::PruneBeyondMinRequired(
loc,
self.inactivity_floor_loc,
));
}
(self.journal, _) = self.journal.prune(loc).await?;
self.update_metrics();
Ok(self)
}
#[tracing::instrument(name = "qmdb.keyless.db.rewind", level = "info", skip_all)]
#[boxed]
pub async fn rewind(mut self, size: Location<F>) -> Result<Self, Error<F>> {
let rewind_size = *size;
let current_size = *self.last_commit_loc + 1;
if rewind_size == current_size {
return Ok(self);
}
if rewind_size == 0 || rewind_size > current_size {
return Err(Error::Journal(crate::journal::Error::InvalidRewind(
rewind_size,
)));
}
let rewind_last_loc = Location::new(rewind_size - 1);
let rewind_floor = {
let bounds = self.journal.bounds();
if rewind_size <= bounds.start {
return Err(Error::Journal(crate::journal::Error::ItemPruned(
*rewind_last_loc,
)));
}
let rewind_last_op = self.journal.read(*rewind_last_loc).await?;
let Operation::Commit(_, floor) = rewind_last_op else {
return Err(Error::UnexpectedData(rewind_last_loc));
};
floor
};
self.journal = self.journal.rewind(rewind_size).await?;
self.last_commit_loc = rewind_last_loc;
self.inactivity_floor_loc = rewind_floor;
let inactive_peaks = F::inactive_peaks(size, rewind_floor);
self.root = self.journal.root(inactive_peaks)?;
self.update_metrics();
Ok(self)
}
#[tracing::instrument(name = "qmdb.keyless.db.sync", level = "info", skip_all)]
pub async fn sync(mut self) -> Result<Self, Error<F>> {
let _timer = self.metrics.sync_timer();
self.metrics.sync_calls.inc();
self.journal = self.journal.sync().await?;
Ok(self)
}
#[tracing::instrument(name = "qmdb.keyless.db.start_sync", level = "info", skip_all)]
pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
self.metrics.start_sync_calls.inc();
let handle;
(self.journal, handle) = self.journal.start_sync().await?;
Ok((self, handle))
}
#[tracing::instrument(name = "qmdb.keyless.db.commit", level = "info", skip_all)]
pub async fn commit(mut self) -> Result<Self, Error<F>> {
let _timer = self.metrics.commit_timer();
self.metrics.commit_calls.inc();
self.journal = self.journal.commit().await?;
Ok(self)
}
#[boxed]
pub async fn destroy(self) -> Result<(), Error<F>> {
Ok(self.journal.destroy().await?)
}
pub(crate) fn commitment(&self) -> batch_chain::Commitment<F, H::Digest> {
batch_chain::Commitment::new(self.last_commit_loc + 1, self.root)
}
pub fn new_batch(&self) -> batch::UnmerkleizedBatch<F, H, V, S> {
batch::UnmerkleizedBatch::new(self, self.commitment())
}
pub fn to_batch(&self) -> Arc<batch::MerkleizedBatch<F, H::Digest, V, S>> {
Arc::new(batch::MerkleizedBatch {
journal_batch: self.journal.to_merkleized_batch(),
parent: None,
bounds: batch_chain::Bounds::from_db(self.commitment(), self.inactivity_floor_loc),
})
}
pub fn validate_batch(
&self,
batch: &batch::MerkleizedBatch<F, H::Digest, V, S>,
) -> Result<(), Error<F>> {
batch
.bounds
.validate_apply_to(self.commitment(), self.inactivity_floor_loc)
}
#[tracing::instrument(name = "qmdb.keyless.db.apply_batch", level = "info", skip_all)]
pub async fn apply_batch(
mut self,
batch: Arc<batch::MerkleizedBatch<F, H::Digest, V, S>>,
) -> Result<(Self, core::ops::Range<Location<F>>), Error<F>> {
let _timer = self.metrics.apply_batch_timer();
self.metrics.apply_batch_calls.inc();
self.validate_batch(&batch)?;
let start_loc = self.last_commit_loc + 1;
self.journal = self.journal.apply_batch(&batch.journal_batch).await?;
self.last_commit_loc = batch.bounds.tip.size - 1;
self.inactivity_floor_loc = batch.bounds.inactivity_floor;
self.root = batch.root();
let end_loc = batch.bounds.tip.size;
debug!(size = ?end_loc, "applied batch");
let range = start_loc..end_loc;
self.update_metrics();
self.metrics
.operations_applied
.inc_by(*range.end - *range.start);
Ok((self, range))
}
}
impl<F, E, V, C, H, S> crate::qmdb::sync::Source for Keyless<F, E, V, C, H, S>
where
F: Family,
E: Context,
V: ValueEncoding,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
S: Strategy,
Operation<F, V>: EncodeShared,
{
type Family = F;
type Digest = H::Digest;
type Op = Operation<F, V>;
type Error = Error<F>;
async fn serve(
&self,
request: crate::qmdb::sync::Request<F>,
) -> Result<
(
crate::qmdb::sync::Response<F, Self::Op, Self::Digest>,
crate::qmdb::sync::FeedbackTx,
),
Self::Error,
> {
self.journal.serve(request).await
}
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use crate::qmdb::{verify_proof, verify_proof_and_pinned_nodes};
use commonware_cryptography::Sha256;
use commonware_parallel::Strategy;
use commonware_runtime::{Supervisor as _, deterministic};
use commonware_utils::NZU64;
use std::{future::Future, pin::Pin};
pub(crate) type Reopen<D> =
Box<dyn Fn(deterministic::Context) -> Pin<Box<dyn Future<Output = D> + Send>>>;
type TestKeyless<F, V, C, H, S> = Keyless<F, deterministic::Context, V, C, H, S>;
pub(crate) trait TestValue: Clone + PartialEq + std::fmt::Debug + Send + Sync {
fn make(i: u64) -> Self;
}
impl TestValue for Vec<u8> {
fn make(i: u64) -> Self {
vec![(i % 255) as u8; ((i % 13) + 7) as usize]
}
}
impl TestValue for commonware_utils::sequence::U64 {
fn make(i: u64) -> Self {
Self::new(i * 10 + 1)
}
}
macro_rules! keyless_tests {
($($name:ident => $scenario:ident, $fixture:ident;)*) => {
$(
#[test_traced]
fn $name() {
deterministic::Runner::default().start(|ctx| async move {
keyless_tests!(@fixture $fixture, $scenario, mmr, ctx);
});
}
)*
paste::paste! {
$(
#[test_traced]
fn [<$name _mmb>]() {
deterministic::Runner::default().start(|ctx| async move {
keyless_tests!(@fixture $fixture, $scenario, mmb, ctx);
});
}
)*
}
};
(@fixture db, $scenario:ident, $family:ident, $ctx:ident) => {
let db = open_db::<$family::Family>($ctx.child("db")).await;
tests::$scenario(db).await;
};
(@fixture reopen, $scenario:ident, $family:ident, $ctx:ident) => {
let db = open_db::<$family::Family>($ctx.child("db")).await;
tests::$scenario($ctx, db, reopen::<$family::Family>()).await;
};
(@fixture reopen_indexed, $scenario:ident, $family:ident, $ctx:ident) => {
let db =
open_db::<$family::Family>($ctx.child("db").with_attribute("index", 1)).await;
tests::$scenario($ctx, db, reopen::<$family::Family>()).await;
};
}
pub(super) use keyless_tests;
#[boxed]
pub(crate) async fn run_empty<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let bounds = db.bounds();
assert_eq!(bounds.end, 1); assert_eq!(bounds.start, Location::new(0));
assert_eq!(db.get_metadata().await.unwrap(), None);
assert_eq!(db.last_commit_loc(), Location::new(0));
let root = db.root();
{
db.new_batch().append(V::Value::make(1));
}
drop(db);
let db = reopen(context.child("db").with_attribute("index", 2)).await;
assert_eq!(db.root(), root);
assert_eq!(db.bounds().end, 1);
assert_eq!(db.get_metadata().await.unwrap(), None);
let metadata = V::Value::make(99);
let merkleized = db
.new_batch()
.merkleize(&db, Some(metadata.clone()), db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.bounds().end, 2); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));
assert_eq!(
db.get(Location::new(1)).await.unwrap(),
Some(metadata.clone())
); let root = db.root();
let db = reopen(context.child("db").with_attribute("index", 3)).await;
assert_eq!(db.bounds().end, 2); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
assert_eq!(db.root(), root);
assert_eq!(db.last_commit_loc(), Location::new(1));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_operations_match_applied_log<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared + PartialEq + core::fmt::Debug,
{
let seed = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (seed_start, seed_ops) = seed.operations();
let seed_root = seed.root();
let seed_proof = seed.proof(&db).unwrap();
let seed_pins = seed.pinned_nodes(&db).unwrap();
let (db, seed_range) = db.apply_batch(seed).await.unwrap();
assert_eq!(seed_start, seed_range.start);
assert_eq!(*seed_start + seed_ops.len() as u64, *seed_range.end);
let parent = db
.new_batch()
.append(V::Value::make(3))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let child = parent
.new_batch::<H>()
.append(V::Value::make(4))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (parent_start, parent_ops) = parent.operations();
let (child_start, child_ops) = child.operations();
let (parent_root, child_root) = (parent.root(), child.root());
let (parent_pins, child_pins) = (
parent.pinned_nodes(&db).unwrap(),
child.pinned_nodes(&db).unwrap(),
);
let (parent_proof, child_proof) = (parent.proof(&db).unwrap(), child.proof(&db).unwrap());
let (db, parent_range) = db.apply_batch(parent).await.unwrap();
let (db, child_range) = db.apply_batch(child).await.unwrap();
assert_eq!(parent_start, parent_range.start);
assert_eq!(*parent_start + parent_ops.len() as u64, *parent_range.end);
assert_eq!(child_start, child_range.start);
assert_eq!(*child_start + child_ops.len() as u64, *child_range.end);
let empty = db
.new_batch()
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (empty_start, empty_ops) = empty.operations();
let (empty_root, empty_proof) = (empty.root(), empty.proof(&db).unwrap());
let empty_pins = empty.pinned_nodes(&db).unwrap();
let (db, empty_range) = db.apply_batch(empty).await.unwrap();
assert_eq!(empty_start, empty_range.start);
assert_eq!(*empty_start + empty_ops.len() as u64, *empty_range.end);
for (start, ops, proof, pins, root) in [
(seed_start, seed_ops, seed_proof, seed_pins, seed_root),
(
parent_start,
parent_ops,
parent_proof,
parent_pins,
parent_root,
),
(child_start, child_ops, child_proof, child_pins, child_root),
(empty_start, empty_ops, empty_proof, empty_pins, empty_root),
] {
let len = core::num::NonZeroU64::new(ops.len() as u64).unwrap();
let end = Location::new(*start + ops.len() as u64);
let (log_proof, log_ops) = db.historical_proof(end, start, len).await.unwrap();
assert_eq!(log_ops, *ops);
assert_eq!(log_proof, proof);
assert!(verify_proof::<H, _, _>(&proof, start, &ops, &root));
assert!(verify_proof_and_pinned_nodes::<H, _, _>(
&proof, start, &ops, &pins, &root
));
}
let late = db
.new_batch()
.append(V::Value::make(5))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(Arc::clone(&late)).await.unwrap();
let db = db.commit().await.unwrap();
assert!(matches!(
late.proof(&db),
Err(crate::qmdb::Error::Merkle(
crate::merkle::Error::ElementPruned(_)
))
));
assert!(matches!(
late.pinned_nodes(&db),
Err(crate::qmdb::Error::Merkle(
crate::merkle::Error::ElementPruned(_)
))
));
let flushed = db
.new_batch()
.append(V::Value::make(6))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (flushed_start, flushed_ops) = flushed.operations();
let flushed_root = flushed.root();
let flushed_proof = flushed.proof(&db).unwrap();
let flushed_pins = flushed.pinned_nodes(&db).unwrap();
assert!(verify_proof_and_pinned_nodes::<H, _, _>(
&flushed_proof,
flushed_start,
&flushed_ops,
&flushed_pins,
&flushed_root
));
let (db, flushed_range) = db.apply_batch(flushed).await.unwrap();
assert_eq!(flushed_start, flushed_range.start);
assert_eq!(
*flushed_start + flushed_ops.len() as u64,
*flushed_range.end
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_commit_after_sync_recovery<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let value0 = V::Value::make(10);
let value1 = V::Value::make(20);
let first_loc = Location::new(1);
let merkleized = db
.new_batch()
.append(value0.clone())
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let db = db.sync().await.unwrap();
let second_loc = db.bounds().end;
let merkleized = db
.new_batch()
.append(value1.clone())
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let committed_bounds = db.bounds();
let committed_root = db.root();
drop(db);
let db = reopen(context.child("db").with_attribute("index", 2)).await;
assert_eq!(db.bounds(), committed_bounds);
assert_eq!(db.root(), committed_root);
assert_eq!(db.get(first_loc).await.unwrap(), Some(value0));
assert_eq!(db.get(second_loc).await.unwrap(), Some(value1));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_build_basic<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
mut db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let v1 = V::Value::make(1);
let v2 = V::Value::make(2);
{
let batch = db.new_batch();
let loc1 = batch.size();
let batch = batch.append(v1.clone());
let loc2 = batch.size();
let batch = batch.append(v2.clone());
assert_eq!(loc1, Location::new(1));
assert_eq!(loc2, Location::new(2));
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
assert_eq!(db.bounds().end, 4); assert_eq!(db.get_metadata().await.unwrap(), None);
assert_eq!(db.get(Location::new(3)).await.unwrap(), None); let root = db.root();
db.sync().await.unwrap();
let db = reopen(context.child("db").with_attribute("index", 2)).await;
assert_eq!(db.bounds().end, 4);
assert_eq!(db.root(), root);
assert_eq!(db.get(Location::new(1)).await.unwrap().unwrap(), v1);
assert_eq!(db.get(Location::new(2)).await.unwrap().unwrap(), v2);
drop(db);
let db = reopen(context.child("db").with_attribute("index", 3)).await;
assert_eq!(db.bounds().end, 4);
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_recovery<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let root = db.root();
const ELEMENTS: u64 = 100;
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i));
}
}
drop(db);
let mut db = reopen(context.child("db").with_attribute("index", 2)).await;
assert_eq!(root, db.root());
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 100));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let db = db.commit().await.unwrap();
let root = db.root();
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 200));
}
}
drop(db);
let mut db = reopen(context.child("db").with_attribute("index", 3)).await;
assert_eq!(root, db.root());
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 300));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let db = db.commit().await.unwrap();
let root = db.root();
drop(db);
let db = reopen(context.child("db").with_attribute("index", 4)).await;
assert_eq!(db.bounds().end, 2 * ELEMENTS + 3);
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_proof<F: Family, V, C, S: Strategy>(
mut db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared + std::fmt::Debug,
{
const ELEMENTS: u64 = 50;
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let root = db.root();
let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(0),
&ops,
&root,
));
assert_eq!(ops.len() as u64, 1 + ELEMENTS + 1);
let (proof, ops) = db.proof(Location::new(10), NZU64!(5)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(10),
&ops,
&root,
));
assert_eq!(ops.len(), 5);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_metadata<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let metadata = V::Value::make(99);
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, Some(metadata.clone()), db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
let merkleized = db
.new_batch()
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.get_metadata().await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_pruning<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
let result = db.prune(Location::new(1)).await;
assert!(
matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor))
if prune_loc == Location::new(1) && floor == Location::new(0))
);
let db = reopen(context.child("reopen_empty")).await;
let first_commit_loc = Location::<F>::new(3);
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, first_commit_loc)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.last_commit_loc(), first_commit_loc);
assert_eq!(db.inactivity_floor_loc(), first_commit_loc);
let second_commit_loc = Location::<F>::new(5);
let merkleized = db
.new_batch()
.append(V::Value::make(3))
.merkleize(&db, None, second_commit_loc)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let root = db.root();
let db = db.prune(first_commit_loc).await.unwrap();
assert_eq!(db.root(), root);
let new_floor = db.inactivity_floor_loc();
let beyond = new_floor + 1;
let result = db.prune(beyond).await;
assert!(
matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, floor))
if prune_loc == beyond && floor == new_floor)
);
}
#[boxed]
pub(crate) async fn run_empty_db_recovery<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let root = db.root();
const ELEMENTS: u64 = 200;
let db = reopen(context.child("db").with_attribute("index", 2)).await;
assert_eq!(db.bounds().end, 1); assert_eq!(db.root(), root);
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i));
}
}
drop(db);
let db = reopen(context.child("db").with_attribute("index", 3)).await;
assert_eq!(db.bounds().end, 1); assert_eq!(db.root(), root);
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 500));
}
}
drop(db);
let db = reopen(context.child("db").with_attribute("index", 4)).await;
assert_eq!(db.bounds().end, 1); assert_eq!(db.root(), root);
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS * 3 {
batch = batch.append(V::Value::make(i + 1000));
}
}
drop(db);
let mut db = reopen(context.child("db").with_attribute("index", 5)).await;
assert_eq!(db.bounds().end, 1); assert_eq!(db.root(), root);
assert_eq!(db.last_commit_loc(), Location::new(0));
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 2000));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
db.commit().await.unwrap();
let db = reopen(context.child("db").with_attribute("index", 6)).await;
assert!(db.bounds().end > 1);
assert_ne!(db.root(), root);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_replay_with_trailing_appends<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
mut db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
{
let mut batch = db.new_batch();
for i in 0..10u64 {
batch = batch.append(V::Value::make(i));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let db = db.commit().await.unwrap();
let committed_root = db.root();
let committed_size = db.bounds().end;
{
db.new_batch().append(V::Value::make(99));
}
drop(db);
let mut db = reopen(context.child("db").with_attribute("index", 2)).await;
assert_eq!(
db.bounds().end,
committed_size,
"Should rewind to last commit"
);
assert_eq!(db.root(), committed_root, "Root should match last commit");
assert_eq!(
db.last_commit_loc(),
committed_size - 1,
"Last commit location should be correct"
);
let new_value = V::Value::make(77);
{
let batch = db.new_batch();
let loc = batch.size();
let batch = batch.append(new_value.clone());
assert_eq!(
loc, committed_size,
"New append should get the expected location"
);
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let db = db.commit().await.unwrap();
assert_eq!(db.get(committed_size).await.unwrap(), Some(new_value));
let new_committed_root = db.root();
let new_committed_size = db.bounds().end;
{
let mut batch = db.new_batch();
for i in 0..5u64 {
batch = batch.append(V::Value::make(200 + i));
}
}
drop(db);
let db = reopen(context.child("db").with_attribute("index", 3)).await;
assert_eq!(
db.bounds().end,
new_committed_size,
"Should rewind to last commit with multiple trailing appends"
);
assert_eq!(
db.root(),
new_committed_root,
"Root should match last commit after multiple appends"
);
assert_eq!(
db.last_commit_loc(),
new_committed_size - 1,
"Last commit location should be correct after multiple appends"
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_get_many<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let v1 = V::Value::make(1);
let v2 = V::Value::make(2);
let v3 = V::Value::make(3);
let batch = db.new_batch();
let loc1 = batch.size();
let batch = batch.append(v1.clone());
let loc2 = batch.size();
let batch = batch.append(v2.clone());
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let results = db.get_many(&[loc1, loc2]).await.unwrap();
assert_eq!(results, vec![Some(v1.clone()), Some(v2.clone())]);
let results = db.get_many(&[]).await.unwrap();
assert!(results.is_empty());
let batch = db.new_batch();
let loc3 = batch.size();
let batch = batch.append(v3.clone());
let results = batch.get_many(&[loc1, loc3], &db).await.unwrap();
assert_eq!(results, vec![Some(v1.clone()), Some(v3.clone())]);
let parent = db
.new_batch()
.append(v3.clone())
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let child = parent.new_batch::<Sha256>().append(V::Value::make(4));
let results = child.get_many(&[loc1, loc2], &db).await.unwrap();
assert_eq!(results, vec![Some(v1.clone()), Some(v2.clone())]);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_chained<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let v1 = V::Value::make(10);
let v2 = V::Value::make(20);
let v3 = V::Value::make(30);
let parent = db.new_batch();
let loc1 = parent.size();
let parent = parent.append(v1.clone());
let parent_m = parent.merkleize(&db, None, db.inactivity_floor_loc()).await;
let child = parent_m.new_batch::<Sha256>();
let loc2 = child.size();
let child = child.append(v2.clone());
let loc3 = child.size();
let child = child.append(v3.clone());
let child_m = child.merkleize(&db, None, db.inactivity_floor_loc()).await;
let child_root = child_m.root();
let (db, _) = db.apply_batch(child_m).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.root(), child_root);
assert_eq!(db.get(loc1).await.unwrap(), Some(v1));
assert_eq!(db.get(loc2).await.unwrap(), Some(v2));
assert_eq!(db.get(loc3).await.unwrap(), Some(v3));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_stale_batch<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let batch_a = db
.new_batch()
.append(V::Value::make(10))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let batch_b = db
.new_batch()
.append(V::Value::make(20))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(batch_a).await.unwrap();
let db = db.commit().await.unwrap();
let root = db.root();
let last_commit_loc = db.last_commit_loc();
let result = db.apply_batch(batch_b).await;
assert!(matches!(result, Err(Error::StaleBatch)));
let db = reopen(context.child("reopen")).await;
assert_eq!(db.root(), root);
assert_eq!(db.last_commit_loc(), last_commit_loc);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_partial_ancestor_commit<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let a = db
.new_batch()
.append(V::Value::make(10))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let b = a
.new_batch::<H>()
.append(V::Value::make(20))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let c = b
.new_batch::<H>()
.append(V::Value::make(30))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
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);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_delayed_merkleize_after_ancestor_apply<
F: Family,
V,
C,
H,
S: Strategy,
>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let floor = db.inactivity_floor_loc();
let a = db
.new_batch()
.append(V::Value::make(10))
.merkleize(&db, None, floor)
.await;
let b = a
.new_batch::<H>()
.append(V::Value::make(20))
.merkleize(&db, None, floor)
.await;
let c = b.new_batch::<H>().append(V::Value::make(30));
let (db, _) = db.apply_batch(a).await.unwrap();
let c = c.merkleize(&db, None, floor).await;
let expected_root = c.root();
let (db, _) = db.apply_batch(c).await.unwrap();
assert_eq!(db.root(), expected_root);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_to_batch<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let batch = db.new_batch();
let loc1 = batch.size();
let batch = batch.append(V::Value::make(10));
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let snapshot = db.to_batch();
assert_eq!(snapshot.root(), db.root());
let child_batch = snapshot.new_batch::<Sha256>();
let loc2 = child_batch.size();
let child_batch = child_batch.append(V::Value::make(20));
let merkleized = child_batch
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.get(loc1).await.unwrap(), Some(V::Value::make(10)));
assert_eq!(db.get(loc2).await.unwrap(), Some(V::Value::make(20)));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_non_empty_recovery<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
mut db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
const ELEMENTS: u64 = 200;
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i));
}
let new_commit = db.last_commit_loc() + 1 + ELEMENTS;
let merkleized = batch.merkleize(&db, None, new_commit).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let db = db.commit().await.unwrap();
let root = db.root();
let op_count = db.bounds().end;
let db = reopen(context.child("db").with_attribute("index", 2)).await;
assert_eq!(db.bounds().end, op_count);
assert_eq!(db.root(), root);
assert_eq!(db.last_commit_loc(), op_count - 1);
drop(db);
let db = reopen(context.child("recovery_a")).await;
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 1000));
}
}
drop(db);
let db = reopen(context.child("recovery_b")).await;
assert_eq!(db.bounds().end, op_count);
assert_eq!(db.root(), root);
drop(db);
let db = reopen(context.child("db").with_attribute("index", 3)).await;
let last_commit = db.last_commit_loc();
let db = db.prune(last_commit).await.unwrap();
assert_eq!(db.bounds().end, op_count);
assert_eq!(db.root(), root);
db.sync().await.unwrap();
let db = reopen(context.child("recovery_c")).await;
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 2000));
}
}
drop(db);
let db = reopen(context.child("recovery_d")).await;
assert_eq!(db.bounds().end, op_count);
assert_eq!(db.root(), root);
drop(db);
let mut db = reopen(context.child("db").with_attribute("index", 4)).await;
{
let mut batch = db.new_batch();
for i in 0..ELEMENTS {
batch = batch.append(V::Value::make(i + 3000));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
db.commit().await.unwrap();
let db = reopen(context.child("db").with_attribute("index", 5)).await;
let bounds = db.bounds();
assert!(bounds.end > op_count);
assert_ne!(db.root(), root);
assert_eq!(db.last_commit_loc(), bounds.end - 1);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_proof_comprehensive<F: Family, V, C, S: Strategy>(
mut db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared + std::fmt::Debug,
{
const ELEMENTS: u64 = 100;
{
let mut batch = db.new_batch();
for i in 0u64..ELEMENTS {
batch = batch.append(V::Value::make(i));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
assert!(matches!(
db.historical_proof(db.bounds().end + 1, Location::new(5), NZU64!(10))
.await,
Err(Error::<F>::Merkle(crate::merkle::Error::RangeOutOfBounds(
_
)))
));
let root = db.root();
for (start_loc, max_ops) in [
(0, 10),
(10, 5),
(50, 20),
(90, 15),
(0, 1),
(ELEMENTS - 1, 1),
(ELEMENTS, 1),
] {
let (proof, ops) = db
.proof(Location::new(start_loc), NZU64!(max_ops))
.await
.unwrap();
assert!(
verify_proof::<Sha256, _, _>(&proof, Location::new(start_loc), &ops, &root,),
"Failed to verify proof for range starting at {start_loc} with max {max_ops} ops",
);
let expected_ops = std::cmp::min(max_ops, *db.bounds().end - start_loc);
assert_eq!(ops.len() as u64, expected_ops);
let wrong_root = Sha256::hash(&[&[0xFF; 32]]);
assert!(!verify_proof::<Sha256, _, _>(
&proof,
Location::new(start_loc),
&ops,
&wrong_root,
));
if start_loc > 0 {
assert!(!verify_proof::<Sha256, _, _>(
&proof,
Location::new(start_loc - 1),
&ops,
&root,
));
}
}
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_proof_with_pruning<F: Family, V, C, S: Strategy>(
context: deterministic::Context,
mut db: TestKeyless<F, V, C, Sha256, S>,
reopen: Reopen<TestKeyless<F, V, C, Sha256, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared + std::fmt::Debug,
{
const ELEMENTS: u64 = 100;
{
let mut batch = db.new_batch();
for i in 0u64..ELEMENTS {
batch = batch.append(V::Value::make(i));
}
let new_commit = db.last_commit_loc() + 1 + ELEMENTS;
let merkleized = batch.merkleize(&db, None, new_commit).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
{
let mut batch = db.new_batch();
for i in ELEMENTS..ELEMENTS * 2 {
batch = batch.append(V::Value::make(i));
}
let new_commit = db.last_commit_loc() + 1 + ELEMENTS;
let merkleized = batch.merkleize(&db, None, new_commit).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let root = db.root();
const PRUNE_LOC: u64 = 30;
let db = db.prune(Location::new(PRUNE_LOC)).await.unwrap();
let oldest_retained = db.bounds().start;
assert_eq!(db.root(), root);
db.sync().await.unwrap();
let db = reopen(context).await;
assert_eq!(db.root(), root);
for (start_loc, max_ops) in [
(oldest_retained, 10),
(Location::new(50), 20),
(Location::new(150), 10),
(Location::new(190), 15),
] {
if start_loc < oldest_retained {
continue;
}
let (proof, ops) = db.proof(start_loc, NZU64!(max_ops)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(&proof, start_loc, &ops, &root,));
}
let aggressive_prune: Location<F> = Location::new(150);
let db = db.prune(aggressive_prune).await.unwrap();
let new_oldest = db.bounds().start;
let (proof, ops) = db.proof(new_oldest, NZU64!(20)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof, new_oldest, &ops, &root,
));
let almost_all = db.bounds().end - 5;
let db = db.prune(almost_all).await.unwrap();
let final_oldest = db.bounds().start;
if final_oldest < db.bounds().end {
let (final_proof, final_ops) = db.proof(final_oldest, NZU64!(10)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&final_proof,
final_oldest,
&final_ops,
&root,
));
}
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_get_out_of_bounds<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
assert!(db.get(Location::new(0)).await.unwrap().is_none());
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(
db.get(Location::new(1)).await.unwrap(),
Some(V::Value::make(1))
);
assert!(db.get(Location::new(3)).await.unwrap().is_none());
assert!(matches!(
db.get(Location::new(4)).await,
Err(Error::LocationOutOfBounds(loc, size))
if loc == Location::new(4) && size == Location::new(4)
));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_get<F: Family, V, C, H, S: Strategy>(
mut db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let base_vals: Vec<V::Value> = (0..3).map(|i| V::Value::make(10 + i)).collect();
let mut base_locs = Vec::new();
{
let mut batch = db.new_batch();
for v in &base_vals {
let loc = batch.size();
batch = batch.append(v.clone());
base_locs.push(loc);
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
let batch = db.new_batch();
for (i, loc) in base_locs.iter().enumerate() {
assert_eq!(
batch.get(*loc, &db).await.unwrap(),
Some(base_vals[i].clone()),
);
}
let new_val = V::Value::make(99);
let new_loc = batch.size();
let batch = batch.append(new_val.clone());
assert_eq!(batch.get(new_loc, &db).await.unwrap(), Some(new_val));
assert_eq!(batch.get(new_loc + 1, &db).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_stacked_get<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let v1 = V::Value::make(1);
let v2 = V::Value::make(2);
let parent = db.new_batch();
let loc1 = parent.size();
let parent = parent.append(v1.clone());
let parent_m = parent.merkleize(&db, None, db.inactivity_floor_loc()).await;
let child = parent_m.new_batch::<Sha256>();
assert_eq!(child.get(loc1, &db).await.unwrap(), Some(v1));
let loc2 = child.size();
let child = child.append(v2.clone());
assert_eq!(child.get(loc2, &db).await.unwrap(), Some(v2));
assert_eq!(child.get(Location::new(9999), &db).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_speculative_root<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let mut batch = db.new_batch();
for i in 0u64..10 {
batch = batch.append(V::Value::make(i));
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
let speculative = merkleized.root();
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.root(), speculative);
let merkleized = db
.new_batch()
.append(V::Value::make(100))
.merkleize(&db, Some(V::Value::make(55)), db.inactivity_floor_loc())
.await;
let speculative = merkleized.root();
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.root(), speculative);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_merkleized_batch_get<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let base_val = V::Value::make(10);
let merkleized = db
.new_batch()
.append(base_val.clone())
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let new_val = V::Value::make(20);
let merkleized = db
.new_batch()
.append(new_val.clone())
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
assert_eq!(
merkleized.get(Location::new(1), &db).await.unwrap(),
Some(base_val),
);
assert_eq!(
merkleized.get(Location::new(3), &db).await.unwrap(),
Some(new_val),
);
assert_eq!(merkleized.get(Location::new(4), &db).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_chained_apply_sequential<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let v1 = V::Value::make(1);
let v2 = V::Value::make(2);
let parent = db.new_batch();
let loc1 = parent.size();
let parent = parent.append(v1.clone());
let parent_m = parent.merkleize(&db, None, db.inactivity_floor_loc()).await;
let parent_root = parent_m.root();
let (db, _) = db.apply_batch(parent_m).await.unwrap();
assert_eq!(db.root(), parent_root);
assert_eq!(db.get(loc1).await.unwrap(), Some(v1));
let batch2 = db.new_batch();
let loc2 = batch2.size();
let batch2 = batch2.append(v2.clone());
let batch2_m = batch2.merkleize(&db, None, db.inactivity_floor_loc()).await;
let batch2_root = batch2_m.root();
let (db, _) = db.apply_batch(batch2_m).await.unwrap();
assert_eq!(db.root(), batch2_root);
assert_eq!(db.get(loc2).await.unwrap(), Some(v2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_many_sequential<F: Family, V, C, S: Strategy>(
mut db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared + std::fmt::Debug,
{
const BATCHES: u64 = 20;
const APPENDS_PER_BATCH: u64 = 5;
let mut all_values: Vec<V::Value> = Vec::new();
let mut all_locs: Vec<Location<F>> = Vec::new();
for batch_idx in 0..BATCHES {
let mut batch = db.new_batch();
for j in 0..APPENDS_PER_BATCH {
let v = V::Value::make(batch_idx * 10 + j);
let loc = batch.size();
batch = batch.append(v.clone());
all_values.push(v);
all_locs.push(loc);
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
(db, _) = db.apply_batch(merkleized).await.unwrap();
}
for (i, loc) in all_locs.iter().enumerate() {
assert_eq!(db.get(*loc).await.unwrap(), Some(all_values[i].clone()));
}
let root = db.root();
let (proof, ops) = db.proof(Location::new(0), NZU64!(1000)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(0),
&ops,
&root,
));
assert_eq!(db.bounds().end, 1 + BATCHES * (APPENDS_PER_BATCH + 1));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_empty<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let root_before = db.root();
let size_before = db.bounds().end;
let merkleized = db
.new_batch()
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let speculative = merkleized.root();
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_ne!(db.root(), root_before);
assert_eq!(db.root(), speculative);
assert_eq!(db.bounds().end, size_before + 1);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_chained_merkleized_get<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let base_val = V::Value::make(10);
let floor = db.inactivity_floor_loc();
let merkleized = db
.new_batch()
.append(base_val.clone())
.merkleize(&db, None, floor)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let v1 = V::Value::make(1);
let parent = db.new_batch();
let loc1 = parent.size();
let parent_m = parent
.append(v1.clone())
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let v2 = V::Value::make(2);
let child = parent_m.new_batch::<Sha256>();
let loc2 = child.size();
let child_m = child
.append(v2.clone())
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
assert_eq!(
child_m.get(Location::new(1), &db).await.unwrap(),
Some(base_val),
);
assert_eq!(child_m.get(loc1, &db).await.unwrap(), Some(v1));
assert_eq!(child_m.get(loc2, &db).await.unwrap(), Some(v2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_large<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared + std::fmt::Debug,
{
const N: u64 = 500;
let mut values = Vec::new();
let mut locs = Vec::new();
let mut batch = db.new_batch();
for i in 0..N {
let v = V::Value::make(i);
locs.push(batch.size());
batch = batch.append(v.clone());
values.push(v);
}
let merkleized = batch.merkleize(&db, None, db.inactivity_floor_loc()).await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
for (i, loc) in locs.iter().enumerate() {
assert_eq!(db.get(*loc).await.unwrap(), Some(values[i].clone()));
}
let root = db.root();
let (proof, ops) = db.proof(Location::new(0), NZU64!(1000)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(0),
&ops,
&root,
));
assert_eq!(db.bounds().end, 1 + N + 1);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_stale_batch_chained<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let common_parent = db
.new_batch()
.append(V::Value::make(10))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let sibling_a = common_parent
.new_batch::<Sha256>()
.append(V::Value::make(11))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let sibling_b = common_parent
.new_batch::<Sha256>()
.append(V::Value::make(12))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
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(V::Value::make(1))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let parent_b = db
.new_batch()
.append(V::Value::make(2))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let child_b = parent_b
.new_batch::<Sha256>()
.append(V::Value::make(3))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(parent_a).await.unwrap();
assert!(matches!(
db.validate_batch(&child_b),
Err(Error::StaleBatch)
));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_sequential_commit_parent_then_child<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let parent = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let child = parent
.new_batch::<Sha256>()
.append(V::Value::make(2))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(parent).await.unwrap();
let (db, _) = db.apply_batch(child).await.unwrap();
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_stale_batch_child_before_parent<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let parent = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let child = parent
.new_batch::<Sha256>()
.append(V::Value::make(2))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(child).await.unwrap();
assert!(matches!(
db.apply_batch(parent).await,
Err(Error::StaleBatch)
));
}
#[boxed]
pub(crate) async fn run_child_root_matches_pending_and_committed<F: Family, V, C, S: Strategy>(
db: TestKeyless<F, V, C, Sha256, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
Operation<F, V>: EncodeShared,
{
let parent = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let pending_child = parent
.new_batch::<Sha256>()
.append(V::Value::make(2))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
let (db, _) = db.apply_batch(parent).await.unwrap();
let db = db.commit().await.unwrap();
let committed_child = db
.new_batch()
.append(V::Value::make(2))
.merkleize(&db, None, db.inactivity_floor_loc())
.await;
assert_eq!(pending_child.root(), committed_child.root());
db.destroy().await.unwrap();
}
async fn commit_appends<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
values: impl IntoIterator<Item = V::Value>,
metadata: Option<V::Value>,
) -> (TestKeyless<F, V, C, H, S>, core::ops::Range<Location<F>>)
where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let base_size = *db.last_commit_loc() + 1;
let appends_iter: Vec<_> = values.into_iter().collect();
let new_commit_loc = Location::new(base_size + appends_iter.len() as u64);
let mut batch = db.new_batch();
for value in appends_iter {
batch = batch.append(value);
}
let merkleized = batch.merkleize(&db, metadata, new_commit_loc).await;
let (db, range) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
(db, range)
}
#[boxed]
pub(crate) async fn run_rewind_recovery<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let initial_root = db.root();
let initial_size = db.bounds().end;
let value_a = V::Value::make(1);
let value_b = V::Value::make(2);
let metadata_a = V::Value::make(3);
let (db, first_range) = commit_appends(
db,
[value_a.clone(), value_b.clone()],
Some(metadata_a.clone()),
)
.await;
let root_before = db.root();
let size_before = db.bounds().end;
let commit_before = db.last_commit_loc();
assert_eq!(size_before, first_range.end);
let value_c = V::Value::make(4);
let metadata_b = V::Value::make(5);
let (db, second_range) =
commit_appends(db, [value_c.clone()], Some(metadata_b.clone())).await;
assert_eq!(second_range.start, size_before);
assert_ne!(db.root(), root_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_b));
let db = db.rewind(size_before).await.unwrap();
assert_eq!(db.root(), root_before);
assert_eq!(db.bounds().end, size_before);
assert_eq!(db.last_commit_loc(), commit_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a.clone()));
assert_eq!(
db.get(Location::new(1)).await.unwrap(),
Some(value_a.clone())
);
assert_eq!(
db.get(Location::new(2)).await.unwrap(),
Some(value_b.clone())
);
assert!(
matches!(
db.get(Location::new(4)).await,
Err(Error::LocationOutOfBounds(_, size)) if size == size_before
),
"rewound append should be out of bounds",
);
db.commit().await.unwrap();
let db = reopen(context.child("reopen")).await;
assert_eq!(db.root(), root_before);
assert_eq!(db.bounds().end, size_before);
assert_eq!(db.last_commit_loc(), commit_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
assert_eq!(
db.get(Location::new(1)).await.unwrap(),
Some(value_a.clone())
);
assert_eq!(
db.get(Location::new(2)).await.unwrap(),
Some(value_b.clone())
);
assert!(matches!(
db.get(Location::new(4)).await,
Err(Error::LocationOutOfBounds(_, size)) if size == size_before
));
let db = db.rewind(initial_size).await.unwrap();
assert_eq!(db.root(), initial_root);
assert_eq!(db.bounds().end, initial_size);
assert_eq!(db.get_metadata().await.unwrap(), None);
assert!(matches!(
db.get(Location::new(1)).await,
Err(Error::LocationOutOfBounds(_, size)) if size == initial_size
));
db.commit().await.unwrap();
let db = reopen(context.child("reopen_initial_boundary")).await;
assert_eq!(db.root(), initial_root);
assert_eq!(db.bounds().end, initial_size);
assert_eq!(db.get_metadata().await.unwrap(), None);
assert!(matches!(
db.get(Location::new(1)).await,
Err(Error::LocationOutOfBounds(_, size)) if size == initial_size
));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_rewind_pruned_target_errors<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let (mut db, first_range) = commit_appends(db, (0..16).map(V::Value::make), None).await;
let mut round = 0u64;
loop {
round += 1;
assert!(
round <= 64,
"failed to prune enough history for rewind test"
);
(db, _) =
commit_appends(db, (0..16).map(|i| V::Value::make(round * 100 + i)), None).await;
let last_commit = db.last_commit_loc();
db = db.prune(last_commit).await.unwrap();
if db.bounds().start > first_range.start {
break;
}
}
let oldest_retained = db.bounds().start;
let Err(boundary_err) = db.rewind(oldest_retained).await else {
panic!("expected rewind to fail");
};
assert!(
matches!(
boundary_err,
Error::Journal(crate::journal::Error::ItemPruned(_))
),
"unexpected rewind error at retained boundary: {boundary_err:?}"
);
let db = reopen(context.child("reopen_boundary")).await;
let Err(err) = db.rewind(first_range.start).await else {
panic!("expected rewind to fail");
};
assert!(
matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
"unexpected rewind error: {err:?}"
);
}
#[boxed]
pub(crate) async fn run_floor_tracking<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
let floor_a = Location::<F>::new(2);
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, floor_a)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.inactivity_floor_loc(), floor_a);
drop(db);
let db = reopen(context.child("reopen")).await;
assert_eq!(db.inactivity_floor_loc(), floor_a);
let merkleized = db
.new_batch()
.append(V::Value::make(3))
.merkleize(&db, None, floor_a)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), floor_a);
let floor_b = Location::<F>::new(5);
let merkleized = db
.new_batch()
.append(V::Value::make(4))
.merkleize(&db, None, floor_b)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), floor_b);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_floor_regression_rejected<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, Location::new(3))
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(3));
let root_before = db.root();
let last_commit_before = db.last_commit_loc();
let merkleized = db
.new_batch()
.append(V::Value::make(3))
.merkleize(&db, None, Location::new(1))
.await;
let Err(err) = db.apply_batch(merkleized).await else {
panic!("expected apply_batch to fail");
};
assert!(
matches!(err, Error::FloorRegressed(new, current) if *new == 1 && *current == 3),
"unexpected error: {err:?}"
);
let db = reopen(context.child("reopen")).await;
assert_eq!(db.inactivity_floor_loc(), Location::new(3));
assert_eq!(db.last_commit_loc(), last_commit_before);
assert_eq!(db.root(), root_before);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_floor_beyond_commit_loc_rejected<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let floor = db.inactivity_floor_loc();
let last_commit_loc = db.last_commit_loc();
let root = db.root();
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, Location::new(999))
.await;
let Err(err) = db.apply_batch(merkleized).await else {
panic!("expected apply_batch to fail");
};
assert!(
matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 999 && *commit == 3),
"unexpected error: {err:?}"
);
let db = reopen(context.child("reopen_boundary")).await;
assert_eq!(db.inactivity_floor_loc(), floor);
assert_eq!(db.last_commit_loc(), last_commit_loc);
assert_eq!(db.root(), root);
let merkleized = db
.new_batch()
.append(V::Value::make(3))
.append(V::Value::make(4))
.merkleize(&db, None, Location::new(4))
.await;
let Err(err) = db.apply_batch(merkleized).await else {
panic!("expected apply_batch to fail");
};
assert!(
matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 4 && *commit == 3),
"unexpected error: {err:?}"
);
}
#[boxed]
pub(crate) async fn run_rewind_restores_floor<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let floor_a = Location::<F>::new(3);
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, floor_a)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let rewind_target = db.last_commit_loc() + 1;
let floor_b = Location::<F>::new(6);
let merkleized = db
.new_batch()
.append(V::Value::make(3))
.append(V::Value::make(4))
.merkleize(&db, None, floor_b)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.inactivity_floor_loc(), floor_b);
let db = db.rewind(rewind_target).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), floor_a);
let db = db.prune(floor_a).await.unwrap();
let beyond = floor_a + 1;
let Err(err) = db.prune(beyond).await else {
panic!("expected prune to fail");
};
assert!(matches!(err, Error::PruneBeyondMinRequired(_, _)));
}
#[boxed]
pub(crate) async fn run_floor_changes_root<F: Family, V, C, H, S: Strategy>(
db_a: TestKeyless<F, V, C, H, S>,
db_b: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let appends = [V::Value::make(1), V::Value::make(2)];
let mut batch_a = db_a.new_batch();
for v in appends.iter() {
batch_a = batch_a.append(v.clone());
}
let merkleized = batch_a.merkleize(&db_a, None, Location::new(0)).await;
let (db_a, _) = db_a.apply_batch(merkleized).await.unwrap();
let mut batch_b = db_b.new_batch();
for v in appends.iter() {
batch_b = batch_b.append(v.clone());
}
let merkleized = batch_b.merkleize(&db_b, None, Location::new(3)).await;
let (db_b, _) = db_b.apply_batch(merkleized).await.unwrap();
assert_ne!(db_a.root(), db_b.root());
db_a.destroy().await.unwrap();
db_b.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_floor_at_commit_loc_accepted<F: Family, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let commit_loc = Location::<F>::new(3);
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, commit_loc)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), commit_loc);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_rewind_after_reopen_with_floor<F: Family, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let floor_a = Location::<F>::new(3);
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.merkleize(&db, None, floor_a)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let rewind_target = db.last_commit_loc() + 1;
let floor_b = Location::<F>::new(6);
let merkleized = db
.new_batch()
.append(V::Value::make(3))
.append(V::Value::make(4))
.merkleize(&db, None, floor_b)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
db.commit().await.unwrap();
let db = reopen(context.child("reopen")).await;
assert_eq!(db.inactivity_floor_loc(), floor_b);
let db = db.rewind(rewind_target).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), floor_a);
assert_eq!(db.last_commit_loc(), Location::new(3));
db.commit().await.unwrap();
let db = reopen(context.child("reopen").with_attribute("index", 2)).await;
assert_eq!(db.inactivity_floor_loc(), floor_a);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_ancestor_floor_regression_rejected<F, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
F: Family,
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let parent = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, None, Location::new(2))
.await;
let child = parent
.new_batch::<H>()
.append(V::Value::make(2))
.merkleize(&db, None, Location::new(1))
.await;
let root_before = db.root();
let last_commit_before = db.last_commit_loc();
let floor_before = db.inactivity_floor_loc();
let Err(err) = db.apply_batch(child).await else {
panic!("expected apply_batch to fail");
};
assert!(
matches!(err, Error::FloorRegressed(new, prev) if *new == 1 && *prev == 2),
"unexpected error: {err:?}"
);
let db = reopen(context.child("reopen")).await;
assert_eq!(db.root(), root_before);
assert_eq!(db.last_commit_loc(), last_commit_before);
assert_eq!(db.inactivity_floor_loc(), floor_before);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_ancestor_floor_beyond_commit_loc_rejected<F, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
F: Family,
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let parent = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, None, Location::new(3))
.await;
let child = parent
.new_batch::<H>()
.append(V::Value::make(2))
.merkleize(&db, None, Location::new(0))
.await;
let Err(err) = db.apply_batch(child).await else {
panic!("expected apply_batch to fail");
};
assert!(
matches!(err, Error::FloorBeyondSize(floor, commit) if *floor == 3 && *commit == 2),
"unexpected error: {err:?}"
);
}
#[boxed]
pub(crate) async fn run_single_commit_live_set<F, V, C, H, S: Strategy>(
context: deterministic::Context,
db: TestKeyless<F, V, C, H, S>,
reopen: Reopen<TestKeyless<F, V, C, H, S>>,
) where
F: Family,
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let metadata = V::Value::make(42);
let commit_loc = Location::<F>::new(4);
let merkleized = db
.new_batch()
.append(V::Value::make(1))
.append(V::Value::make(2))
.append(V::Value::make(3))
.merkleize(&db, Some(metadata.clone()), commit_loc)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.last_commit_loc(), commit_loc);
assert_eq!(db.inactivity_floor_loc(), commit_loc);
let root_after_commit = db.root();
let db = db.prune(commit_loc).await.unwrap();
let bounds = db.bounds();
assert!(
bounds.start <= commit_loc,
"prune must not advance bounds.start past the floor"
);
assert_eq!(bounds.end, commit_loc + 1);
assert_eq!(db.get(commit_loc).await.unwrap(), Some(metadata.clone()));
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));
assert_eq!(db.last_commit_loc(), commit_loc);
assert_eq!(db.inactivity_floor_loc(), commit_loc);
assert_eq!(db.root(), root_after_commit);
let db = db.sync().await.unwrap();
let Err(err) = db.prune(commit_loc + 1).await else {
panic!("expected prune to fail");
};
assert!(matches!(err, Error::PruneBeyondMinRequired(p, f)
if *p == *commit_loc + 1 && *f == *commit_loc));
let db = reopen(context.child("reopened")).await;
let reopened_bounds = db.bounds();
assert_eq!(reopened_bounds.end, commit_loc + 1);
assert_eq!(db.last_commit_loc(), commit_loc);
assert_eq!(db.inactivity_floor_loc(), commit_loc);
assert_eq!(db.root(), root_after_commit);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone()));
let next_commit_loc = Location::<F>::new(7);
let v5 = V::Value::make(5);
let v6 = V::Value::make(6);
let merkleized = db
.new_batch()
.append(v5.clone())
.append(v6.clone())
.merkleize(&db, None, next_commit_loc)
.await;
let (db, _) = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.last_commit_loc(), next_commit_loc);
assert_eq!(db.inactivity_floor_loc(), next_commit_loc);
assert_eq!(db.get(Location::new(5)).await.unwrap(), Some(v5));
assert_eq!(db.get(Location::new(6)).await.unwrap(), Some(v6));
assert_eq!(db.get(commit_loc).await.unwrap(), Some(metadata));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_chained_apply_with_valid_floors_succeeds<F, V, C, H, S: Strategy>(
db: TestKeyless<F, V, C, H, S>,
) where
F: Family,
V: ValueEncoding<Value: TestValue>,
C: Mutable<Item = Operation<F, V>>,
H: Hasher,
Operation<F, V>: EncodeShared,
{
let parent = db
.new_batch()
.append(V::Value::make(1))
.merkleize(&db, None, Location::new(2))
.await;
let child = parent
.new_batch::<H>()
.append(V::Value::make(2))
.merkleize(&db, None, Location::new(3))
.await;
let grandchild = child
.new_batch::<H>()
.append(V::Value::make(3))
.merkleize(&db, None, Location::new(5))
.await;
let (db, _) = db.apply_batch(grandchild).await.unwrap();
assert_eq!(db.last_commit_loc(), Location::new(6));
assert_eq!(db.inactivity_floor_loc(), Location::new(5));
db.destroy().await.unwrap();
}
}