use crate::{
Context,
index::immutable::Index,
journal::{
authenticated,
contiguous::{Contiguous, Mutable},
},
merkle::{Family, Location, Proof, full::Config as MerkleConfig},
qmdb::{
Error, any::ValueEncoding, chain, metrics::Metrics, operation::Key, single_operation_root,
sync::source,
},
translator::Translator,
};
use commonware_codec::EncodeShared;
use commonware_cryptography::Hasher;
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use commonware_runtime::{Handle, ReadOptions};
use core::num::{NonZeroU64, NonZeroUsize};
use futures::{StreamExt, pin_mut};
use std::{ops::Range, sync::Arc};
use tracing::warn;
pub mod batch;
mod compact;
pub mod fixed;
mod operation;
pub mod sync;
pub mod variable;
pub use compact::{
Config as CompactConfig, Db as CompactDb, MerkleizedBatch as CompactMerkleizedBatch,
UnmerkleizedBatch as CompactUnmerkleizedBatch,
};
pub use operation::Operation;
async fn build_snapshot<F, E, K, V, C, T>(
context: E,
translator: T,
inactivity_floor_loc: Location<F>,
log: &C,
init_buffer: NonZeroUsize,
) -> Result<Index<T>, Error<F>>
where
F: Family,
E: Context,
K: Key,
V: ValueEncoding,
C: Contiguous<Item = Operation<F, K, V>>,
T: Translator,
{
let end = log.bounds().end;
let mut snapshot = Index::new(
context,
translator,
*inactivity_floor_loc,
usize::try_from(end - *inactivity_floor_loc)
.expect("operations since the floor must fit in usize"),
);
let stream = log
.replay(
*inactivity_floor_loc..end,
init_buffer,
ReadOptions::DONT_CACHE,
)
.await?;
pin_mut!(stream);
while let Some(batch) = stream.next().await {
for (loc, op) in batch? {
if let Operation::Set(key, _) = op {
snapshot.insert(&key, loc);
}
}
}
Ok(snapshot)
}
pub fn initial_root<F, K, V, H>() -> H::Digest
where
F: Family,
K: Key,
V: ValueEncoding,
H: Hasher,
Operation<F, K, V>: EncodeShared,
{
single_operation_root::<F, H>(&Operation::<F, K, V>::Commit(None, Location::new(0)))
}
#[derive(Clone)]
pub struct Config<T: Translator, J, S: Strategy> {
pub merkle_config: MerkleConfig<S>,
pub log: J,
pub translator: T,
pub init_buffer: NonZeroUsize,
}
pub struct Immutable<
F: Family,
E: Context,
K: Key,
V: ValueEncoding,
C: Mutable<Item = Operation<F, K, V>>,
H: Hasher,
T: Translator,
S: Strategy,
> where
C::Item: EncodeShared,
{
journal: authenticated::Journal<F, E, C, H, S>,
pub(crate) root: H::Digest,
pub(crate) snapshot: Index<T>,
pub(crate) inactivity_floor_loc: Location<F>,
metrics: Metrics<E>,
}
impl<F, E, K, V, C, H, T, S> std::fmt::Debug for Immutable<F, E, K, V, C, H, T, S>
where
F: Family,
E: Context,
K: Key,
V: ValueEncoding,
C: Mutable<Item = Operation<F, K, V>>,
C::Item: EncodeShared,
H: Hasher,
T: Translator,
S: Strategy,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Immutable")
.field("bounds", &self.bounds())
.field("inactivity_floor_loc", &self.inactivity_floor_loc())
.finish_non_exhaustive()
}
}
impl<F, E, K, V, C, H, T, S> Immutable<F, E, K, V, C, H, T, S>
where
F: Family,
E: Context,
K: Key,
V: ValueEncoding,
C: Mutable<Item = Operation<F, K, V>>,
C::Item: EncodeShared,
H: Hasher,
T: Translator,
S: Strategy,
{
#[boxed]
pub async fn init(
context: E,
cfg: Config<T, C::Config, S>,
max_size: Option<Location<F>>,
) -> Result<Self, Error<F>>
where
C: authenticated::Backing<E>,
{
let mut journal = crate::qmdb::init_journal::<F, E, C, H, S>(
context.child("journal"),
cfg.merkle_config,
cfg.log,
max_size,
true,
)
.await?;
if journal.size() == 0 {
warn!("Authenticated log is empty, initialized new db.");
(journal, _) = journal
.append(&Operation::Commit(None, Location::new(0)))
.await?;
journal = journal.sync().await?;
}
let (last_commit_loc, inactivity_floor_loc, snapshot) = {
let bounds = journal.journal.bounds();
let last_commit_loc =
Location::new(bounds.end.checked_sub(1).expect("commit should exist"));
let last_op = journal.journal.read(*last_commit_loc).await?;
let inactivity_floor_loc = last_op
.has_floor()
.expect("last operation should be a commit with floor");
let snapshot = build_snapshot(
context.child("snapshot"),
cfg.translator,
inactivity_floor_loc,
&journal.journal,
cfg.init_buffer,
)
.await?;
(last_commit_loc, inactivity_floor_loc, snapshot)
};
let inactive_peaks = F::inactive_peaks(last_commit_loc + 1, inactivity_floor_loc);
let root = journal.root(inactive_peaks)?;
let metrics = Metrics::new(context);
let db = Self {
journal,
root,
snapshot,
inactivity_floor_loc,
metrics,
};
db.update_metrics();
Ok(db)
}
pub const fn inactivity_floor_loc(&self) -> Location<F> {
self.inactivity_floor_loc
}
pub fn size(&self) -> Location<F> {
self.bounds().end
}
pub fn bounds(&self) -> Range<Location<F>> {
Location::new(self.journal.bounds().start)..Location::new(self.journal.bounds().end)
}
fn update_metrics(&self) {
let bounds = self.journal.bounds();
self.metrics.update(
bounds.end,
bounds.start,
*self.inactivity_floor_loc,
bounds.end - 1,
);
}
pub const fn sync_boundary(&self) -> Location<F> {
self.inactivity_floor_loc
}
pub async fn get(&self, key: &K) -> Result<Option<V::Value>, Error<F>> {
self.get_from(key, Location::new(0)).await
}
pub(crate) async fn get_from(
&self,
key: &K,
floor: Location<F>,
) -> Result<Option<V::Value>, Error<F>> {
let _timer = self.metrics.get_timer();
self.metrics.get_calls.inc();
self.metrics.lookups_requested.inc();
for loc in self.snapshot.get(key) {
let loc = Location::new(loc);
if loc < floor {
continue;
}
if let Some(v) = Self::get_from_loc(&self.journal, key, loc).await? {
return Ok(Some(v));
}
}
Ok(None)
}
pub async fn get_many(&self, keys: &[&K]) -> Result<Vec<Option<V::Value>>, Error<F>> {
self.get_many_from(keys, Location::new(0)).await
}
pub(crate) async fn get_many_from(
&self,
keys: &[&K],
floor: Location<F>,
) -> Result<Vec<Option<V::Value>>, Error<F>> {
if keys.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(keys.len() as u64);
let mut candidates: Vec<(usize, u64)> = Vec::with_capacity(keys.len());
let mut results: Vec<Option<V::Value>> = vec![None; keys.len()];
for (key_idx, key) in keys.iter().enumerate() {
for loc in self.snapshot.get(key) {
if loc < *floor {
continue;
}
candidates.push((key_idx, loc));
}
}
if candidates.is_empty() {
return Ok(results);
}
candidates.sort_unstable_by_key(|&(_, pos)| pos);
let same_pos = |a: &(usize, u64), b: &(usize, u64)| a.1 == b.1;
let positions: Vec<u64> = candidates
.chunk_by(same_pos)
.map(|group| group[0].1)
.collect();
let ops = self.journal.read_many(&positions).await?;
assert_eq!(
ops.len(),
positions.len(),
"read_many returns one operation per position"
);
for (op, group) in ops.into_iter().zip(candidates.chunk_by(same_pos)) {
let Operation::Set(k, v) = op else {
return Err(Error::UnexpectedData(Location::new(group[0].1)));
};
let mut pending = None;
for &(key_idx, _) in group {
if results[key_idx].is_some() || k != *keys[key_idx] {
continue;
}
if let Some(prev) = pending.replace(key_idx) {
results[prev] = Some(v.clone());
}
}
if let Some(last) = pending {
results[last] = Some(v);
}
}
Ok(results)
}
async fn get_from_loc(
reader: &impl Contiguous<Item = Operation<F, K, V>>,
key: &K,
loc: Location<F>,
) -> Result<Option<V::Value>, Error<F>> {
if loc < reader.bounds().start {
return Err(Error::OperationPruned(loc));
}
let Operation::Set(k, v) = reader.read(*loc).await? else {
return Err(Error::UnexpectedData(loc));
};
if k != *key { Ok(None) } else { Ok(Some(v)) }
}
pub async fn get_metadata(&self) -> Result<Option<V::Value>, Error<F>> {
let last_commit_loc = self.size() - 1;
let Operation::Commit(metadata, _floor) =
self.journal.journal.read(*last_commit_loc).await?
else {
unreachable!("no commit operation at location of last commit {last_commit_loc}");
};
Ok(metadata)
}
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.immutable.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, K, 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 proof(
&self,
start_index: Location<F>,
max_ops: NonZeroU64,
) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, K, V>>), Error<F>> {
let op_count = self.bounds().end;
self.historical_proof(op_count, start_index, max_ops).await
}
#[tracing::instrument(name = "qmdb.immutable.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.snapshot.prune(*self.bounds().start);
self.update_metrics();
Ok(self)
}
pub const fn root(&self) -> H::Digest {
self.root
}
pub const fn strategy(&self) -> &S {
self.journal.strategy()
}
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.immutable.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.immutable.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.immutable.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) -> chain::Commitment<F, H::Digest> {
chain::Commitment::new(self.size(), self.root)
}
pub fn new_batch(&self) -> batch::UnmerkleizedBatch<F, H, K, V, S> {
batch::UnmerkleizedBatch::new(self, self.commitment())
}
pub fn validate_batch(
&self,
batch: &batch::MerkleizedBatch<F, H::Digest, K, V, S>,
) -> Result<(), Error<F>> {
batch.bounds.validate_apply_to(self.commitment())
}
#[tracing::instrument(name = "qmdb.immutable.db.apply_batch", level = "info", skip_all)]
pub async fn apply_batch(
mut self,
batch: Arc<batch::MerkleizedBatch<F, H::Digest, K, V, S>>,
) -> Result<Self, Error<F>> {
let _timer = self.metrics.apply_batch_timer();
self.metrics.apply_batch_calls.inc();
self.validate_batch(&batch)?;
let db_size = self.size();
let apply_snapshot = async {
for (key, entry) in batch.diff.iter() {
self.snapshot.insert(key, *entry.loc);
}
for (i, ancestor_diff) in batch.ancestor_diffs.iter().enumerate() {
if batch.bounds.ancestors[i].size <= db_size {
continue;
}
for (key, entry) in ancestor_diff.iter() {
self.snapshot.insert(key, *entry.loc);
}
}
};
let (journal, ()) = futures::join!(
self.journal.apply_batch(&batch.journal_batch),
apply_snapshot
);
self.journal = journal?;
self.inactivity_floor_loc = batch.bounds.inactivity_floor;
self.root = batch.root();
self.update_metrics();
self.metrics
.operations_applied
.inc_by(*batch.bounds.tip.size - *db_size);
Ok(self)
}
}
impl<F, E, K, V, C, H, T, S> crate::qmdb::sync::Source for Immutable<F, E, K, V, C, H, T, S>
where
F: Family,
E: Context,
K: Key,
V: ValueEncoding,
C: Mutable<Item = Operation<F, K, V>>,
Operation<F, K, V>: EncodeShared,
H: Hasher,
T: Translator,
S: Strategy,
{
type Family = F;
type Digest = H::Digest;
type Op = Operation<F, K, V>;
type Error = Error<F>;
async fn serve(&self, request: crate::qmdb::sync::Request<F>) -> source::Result<Self> {
self.journal.serve(request).await
}
}
#[cfg(test)]
pub(super) mod tests {
use super::*;
use crate::{
merkle::{Family, Location},
qmdb::{verify_proof, verify_proof_and_pinned_nodes},
translator::TwoCap,
};
use commonware_codec::EncodeShared;
use commonware_cryptography::{Sha256, sha256, sha256::Digest};
use commonware_runtime::{Supervisor as _, deterministic};
use commonware_utils::NZU64;
use core::{future::Future, pin::Pin};
const ITEMS_PER_SECTION: u64 = 5;
macro_rules! immutable_tests {
($($name:ident => $scenario:ident, $open:ident;)*) => {
$(
#[test_traced]
fn $name() {
deterministic::Runner::default().start(|ctx| async move {
immutable_tests!(@fixture $open, $scenario, mmr, ctx);
});
}
)*
paste::paste! {
$(
#[test_traced]
fn [<$name _mmb>]() {
deterministic::Runner::default().start(|ctx| async move {
immutable_tests!(@fixture $open, $scenario, mmb, ctx);
});
}
)*
}
};
(@fixture pair, $scenario:ident, $family:ident, $ctx:ident) => {
let db = Db::<$family::Family, _, Digest, Digest, Sha256, TwoCap, Sequential>::init(
$ctx.child("db"), config("db"), None,
).await.unwrap();
let foreign = Db::<$family::Family, _, Digest, Digest, Sha256, TwoCap, Sequential>::init(
$ctx.child("foreign"), config("foreign"), None,
).await.unwrap();
tests::$scenario(db, foreign).await;
};
(@fixture $open:ident, $scenario:ident, $family:ident, $ctx:ident) => {
tests::$scenario($ctx, $open::<$family::Family>).await;
};
}
pub(super) use immutable_tests;
type TestDb<F, V, C> = Immutable<
F,
deterministic::Context,
Digest,
V,
C,
Sha256,
TwoCap,
commonware_parallel::Sequential,
>;
#[boxed]
pub(crate) async fn run_dropped_ancestor_reads<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let untouched = Sha256::fill(0u8);
let untouched_value = Sha256::fill(100u8);
let seed = db
.new_batch()
.set(untouched, untouched_value)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap();
let a = db
.new_batch()
.set(Sha256::fill(1u8), Sha256::fill(101u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let b = a
.new_batch::<Sha256>()
.set(Sha256::fill(2u8), Sha256::fill(102u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let c = b
.new_batch::<Sha256>()
.set(Sha256::fill(3u8), Sha256::fill(103u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child = c.new_batch::<Sha256>();
drop(a);
drop(b);
for i in 0..4u8 {
let key = Sha256::fill(i);
let expected = Some(Sha256::fill(i.wrapping_add(100)));
assert_eq!(c.get(&key, &db).await.unwrap(), expected);
assert_eq!(child.get(&key, &db).await.unwrap(), expected);
}
let keys: Vec<_> = (0..4u8).map(Sha256::fill).collect();
let key_refs: Vec<_> = keys.iter().collect();
let expected: Vec<_> = (0..4u8)
.map(|i| Some(Sha256::fill(i.wrapping_add(100))))
.collect();
assert_eq!(c.get_many(&key_refs, &db).await.unwrap(), expected);
assert_eq!(child.get_many(&key_refs, &db).await.unwrap(), expected);
assert!(matches!(
child.merkleize(&db, None, Location::new(0)).await,
Err(Error::StaleBatch)
));
let db = db.apply_batch(c).await.unwrap();
for i in 0..4u8 {
assert_eq!(
db.get(&Sha256::fill(i)).await.unwrap(),
Some(Sha256::fill(i.wrapping_add(100)))
);
}
assert_eq!(db.get(&untouched).await.unwrap(), Some(untouched_value));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_multi_key_batch_reads<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let value = |i: u8| Sha256::fill(i.wrapping_add(100));
let mut batch = db.new_batch();
for i in (0..8u8).rev() {
batch = batch.set(Sha256::fill(i), value(i));
}
let parent = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
let keys: Vec<_> = (0..8u8).map(Sha256::fill).collect();
let key_refs: Vec<_> = keys.iter().collect();
let expected: Vec<_> = (0..8u8).map(|i| Some(value(i))).collect();
assert_eq!(parent.get_many(&key_refs, &db).await.unwrap(), expected);
let draft = parent.new_batch::<Sha256>();
assert_eq!(draft.get_many(&key_refs, &db).await.unwrap(), expected);
let child = draft.merkleize(&db, None, Location::new(0)).await.unwrap();
assert_eq!(child.get_many(&key_refs, &db).await.unwrap(), expected);
for i in 0..8u8 {
assert_eq!(
child.get(&Sha256::fill(i), &db).await.unwrap(),
Some(value(i))
);
}
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_proof_refused_after_sibling_apply<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let batch = db
.new_batch()
.set(Sha256::fill(1u8), Sha256::fill(101u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let sibling = db
.new_batch()
.set(Sha256::fill(1u8), Sha256::fill(102u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert!(batch.proof(&db).is_ok());
assert!(batch.pinned_nodes(&db).is_ok());
let db = db.apply_batch(sibling).await.unwrap();
assert!(matches!(batch.proof(&db), Err(Error::StaleRead)));
assert!(matches!(batch.pinned_nodes(&db), Err(Error::StaleRead)));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_reads_ignore_inactive_operations<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let (old, fresh, own) = (Sha256::fill(1u8), Sha256::fill(2u8), Sha256::fill(3u8));
let seed = db
.new_batch()
.set(old, Sha256::fill(101u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(seed).await.unwrap();
let mut filler = db.new_batch();
for i in 10..18u8 {
filler = filler.set(Sha256::fill(i), Sha256::fill(i));
}
let filler = filler.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(filler).await.unwrap();
let base = db.bounds().end;
let parent = db
.new_batch()
.set(fresh, Sha256::fill(102u8))
.merkleize(&db, None, base)
.await
.unwrap();
assert_eq!(parent.get(&old, &db).await.unwrap(), None);
assert_eq!(
parent.get(&fresh, &db).await.unwrap(),
Some(Sha256::fill(102u8))
);
let child = parent.new_batch::<Sha256>();
assert_eq!(child.get(&old, &db).await.unwrap(), None);
assert_eq!(
child.get_many(&[&old, &fresh], &db).await.unwrap(),
vec![None, Some(Sha256::fill(102u8))]
);
let grandchild = child
.set(own, Sha256::fill(103u8))
.merkleize(&db, None, base + 3)
.await
.unwrap();
assert_eq!(
grandchild
.get_many(&[&old, &fresh, &own], &db)
.await
.unwrap(),
vec![None, None, None]
);
drop(grandchild);
let db = db.apply_batch(parent).await.unwrap();
let db = db.commit().await.unwrap();
let expected = vec![None, Some(Sha256::fill(102u8))];
assert_eq!(
db.new_batch().get_many(&[&old, &fresh], &db).await.unwrap(),
expected
);
assert_eq!(
db.new_batch().get(&fresh, &db).await.unwrap(),
Some(Sha256::fill(102u8))
);
let mut twin = [1u8; 32];
twin[31] = 9;
let twin = Digest(twin);
let collision = db
.new_batch()
.set(twin, Sha256::fill(104u8))
.merkleize(&db, None, base)
.await
.unwrap();
let db = db.apply_batch(collision).await.unwrap();
let db = db.commit().await.unwrap();
let collided = vec![None, Some(Sha256::fill(104u8))];
assert_eq!(db.new_batch().get(&old, &db).await.unwrap(), None);
assert_eq!(
db.new_batch().get(&twin, &db).await.unwrap(),
Some(Sha256::fill(104u8))
);
assert_eq!(
db.new_batch().get_many(&[&old, &twin], &db).await.unwrap(),
collided
);
let db = db.prune(base).await.unwrap();
assert!(db.bounds().start > Location::new(1), "`old` must be pruned");
assert_eq!(
db.new_batch().get_many(&[&old, &fresh], &db).await.unwrap(),
expected
);
assert_eq!(
db.new_batch().get_many(&[&old, &twin], &db).await.unwrap(),
collided
);
drop(db);
let db = open_db(context.child("db")).await;
assert_eq!(
db.new_batch().get_many(&[&old, &fresh], &db).await.unwrap(),
expected
);
assert_eq!(db.new_batch().get(&old, &db).await.unwrap(), None);
assert_eq!(
db.new_batch().get_many(&[&old, &twin], &db).await.unwrap(),
collided
);
assert_eq!(
db.new_batch().get(&twin, &db).await.unwrap(),
Some(Sha256::fill(104u8))
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_merkleize_across_prune<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let mut seed = db.new_batch();
for i in 0..40u8 {
seed = seed.set(Sha256::fill(i), Sha256::fill(i.wrapping_add(100)));
}
let seed = seed.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(seed).await.unwrap();
let db = db.commit().await.unwrap();
let parent_floor = db.bounds().end + 1;
let parent = db
.new_batch()
.set(Sha256::fill(200u8), Sha256::fill(212u8))
.merkleize(&db, None, parent_floor)
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.set(Sha256::fill(201u8), Sha256::fill(213u8));
let twin = parent
.new_batch::<Sha256>()
.set(Sha256::fill(201u8), Sha256::fill(213u8));
let expected = twin
.merkleize(&db, None, parent_floor)
.await
.unwrap()
.root();
let db = db.apply_batch(Arc::clone(&parent)).await.unwrap();
let db = db.commit().await.unwrap();
let floor = db.inactivity_floor_loc();
let db = db.prune(floor).await.unwrap();
assert!(*db.bounds().start > 0, "the prune must remove history");
let merkleized = child.merkleize(&db, None, parent_floor).await.unwrap();
assert_eq!(merkleized.root(), expected);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_stale_fork_refuses<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let k1 = Sha256::fill(1u8);
let k2 = Sha256::fill(2u8);
let floor = db.inactivity_floor_loc();
let winner = db
.new_batch()
.set(k1, Sha256::fill(3u8))
.merkleize(&db, None, floor)
.await
.unwrap();
let loser = db
.new_batch()
.set(k1, Sha256::fill(4u8))
.merkleize(&db, None, floor)
.await
.unwrap();
assert_eq!(winner.bounds().tip.size, loser.bounds().tip.size);
assert_ne!(winner.root(), loser.root());
let child = loser.new_batch::<Sha256>().set(k2, Sha256::fill(5u8));
let direct = db.new_batch().set(k2, Sha256::fill(6u8));
let db = db.apply_batch(winner).await.unwrap();
assert!(matches!(direct.get(&k1, &db).await, Err(Error::StaleRead)));
assert!(matches!(
direct.get_many(&[&k1], &db).await,
Err(Error::StaleRead)
));
assert!(matches!(child.get(&k1, &db).await, Err(Error::StaleRead)));
assert!(matches!(
child.get_many(&[&k1], &db).await,
Err(Error::StaleRead)
));
assert!(matches!(loser.get(&k1, &db).await, Err(Error::StaleRead)));
let no_keys: [&Digest; 0] = [];
assert!(matches!(
child.get_many(&no_keys, &db).await,
Err(Error::StaleRead)
));
assert!(matches!(
loser.get_many(&no_keys, &db).await,
Err(Error::StaleRead)
));
assert!(matches!(
child.merkleize(&db, None, floor).await,
Err(Error::StaleBatch)
));
assert!(matches!(db.validate_batch(&loser), Err(Error::StaleBatch)));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_descendant_apply_makes_parent_reads_stale<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let (k1, k2) = (Sha256::fill(1u8), Sha256::fill(2u8));
let floor = db.inactivity_floor_loc();
let parent = db
.new_batch()
.set(k1, Sha256::fill(3u8))
.merkleize(&db, None, floor)
.await
.unwrap();
assert_eq!(parent.get(&k1, &db).await.unwrap(), Some(Sha256::fill(3u8)));
let child = parent
.new_batch::<Sha256>()
.set(k2, Sha256::fill(4u8))
.merkleize(&db, None, floor)
.await
.unwrap();
let db = db.apply_batch(child.clone()).await.unwrap();
assert_eq!(child.get(&k1, &db).await.unwrap(), Some(Sha256::fill(3u8)));
assert!(matches!(parent.get(&k1, &db).await, Err(Error::StaleRead)));
assert!(matches!(
parent.get_many(&[&k1], &db).await,
Err(Error::StaleRead)
));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_empty<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let bounds = db.bounds();
assert_eq!(bounds.end, 1);
assert_eq!(bounds.start, Location::new(0));
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
assert!(db.get_metadata().await.unwrap().is_none());
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
let root = db.root();
{
let _batch = db.new_batch().set(k1, v1);
}
drop(db);
let db = open_db(context.child("second")).await;
assert_eq!(db.root(), root);
assert_eq!(db.bounds().end, 1);
let merkleized = db
.new_batch()
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.bounds().end, 2); let root = db.root();
drop(db);
let db = open_db(context.child("third")).await;
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_commit_after_sync_recovery<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let k1 = Sha256::fill(1u8);
let k2 = Sha256::fill(2u8);
let v1 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
let db = commit_sets(db, [(k1, v1)], None).await;
let db = db.sync().await.unwrap();
let db = commit_sets(db, [(k2, v2)], None).await;
let committed_bounds = db.bounds();
let committed_root = db.root();
drop(db);
let db = open_db(context.child("second")).await;
assert_eq!(db.bounds(), committed_bounds);
assert_eq!(db.root(), committed_root);
assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_build_basic<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let k1 = Sha256::fill(1u8);
let k2 = Sha256::fill(2u8);
let v1 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
assert!(db.get(&k1).await.unwrap().is_none());
assert!(db.get(&k2).await.unwrap().is_none());
let metadata = Some(Sha256::fill(99u8));
let merkleized = db
.new_batch()
.set(k1, v1)
.merkleize(&db, metadata, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
assert!(db.get(&k2).await.unwrap().is_none());
assert_eq!(db.bounds().end, 3);
assert_eq!(db.get_metadata().await.unwrap(), Some(Sha256::fill(99u8)));
let merkleized = db
.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);
assert_eq!(db.bounds().end, 5);
assert_eq!(db.get_metadata().await.unwrap(), None);
let root = db.root();
let k3 = Sha256::fill(5u8);
let v3 = Sha256::fill(6u8);
{
let _batch = db.new_batch().set(k3, v3);
}
drop(db); let db = open_db(context.child("second")).await;
assert!(db.get(&k3).await.unwrap().is_none());
assert_eq!(db.root(), root);
assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1);
assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2);
assert_eq!(db.bounds().end, 5);
assert_eq!(db.get_metadata().await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_proof_verify<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(10u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let (proof, ops) = db.proof(Location::new(0), NZU64!(100)).await.unwrap();
let root = db.root();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(0),
&ops,
&root
));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_prune<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let mut db = open_db(context.child("first")).await;
for i in 0..20u8 {
let key = Sha256::fill(i);
let value = Sha256::fill(i.wrapping_add(100));
let floor = db.bounds().end;
let merkleized = db
.new_batch()
.set(key, value)
.merkleize(&db, None, floor)
.await
.unwrap();
db = db.apply_batch(merkleized).await.unwrap();
db = db.commit().await.unwrap();
}
let root_before = db.root();
let bounds_before = db.bounds();
let prune_loc = bounds_before.end - 5;
let db = db.prune(prune_loc).await.unwrap();
assert_eq!(db.root(), root_before);
let key_0 = Sha256::fill(0u8);
assert!(db.get(&key_0).await.unwrap().is_none());
let key_19 = Sha256::fill(19u8);
assert_eq!(
db.get(&key_19).await.unwrap(),
Some(Sha256::fill(19u8.wrapping_add(100)))
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_prune_after_uncommitted_apply_batch_recovery<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let mut batch = db.new_batch();
for i in 0..6u8 {
batch = batch.set(Sha256::fill(i), Sha256::fill(i.wrapping_add(10)));
}
let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.sync().await.unwrap();
let durable_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
let buffered_floor = db.bounds().end;
let key = Sha256::fill(100);
let value = Sha256::fill(101);
let merkleized = db
.new_batch()
.set(key, value)
.merkleize(&db, None, buffered_floor)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let buffered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
assert_ne!(buffered_state, durable_state);
let db = db.prune(buffered_floor).await.unwrap();
assert!(db.bounds().start > Location::new(0));
drop(db);
let db = open_db(context.child("second")).await;
assert!(db.bounds().start <= db.inactivity_floor_loc());
let recovered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
assert!(
recovered_state == durable_state || recovered_state == buffered_state,
"recovered state is neither the durable baseline nor the buffered state"
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_operations_match_applied_log<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared + PartialEq + core::fmt::Debug,
{
let db = open_db(context.child("db")).await;
let seed = db
.new_batch()
.set(Sha256::fill(1u8), Sha256::fill(11u8))
.set(Sha256::fill(2u8), Sha256::fill(12u8))
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
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 applied_start = db.bounds().end;
let db = db.apply_batch(seed).await.unwrap();
let seed_range = applied_start..db.bounds().end;
assert_eq!(seed_start, seed_range.start);
assert_eq!(*seed_start + seed_ops.len() as u64, *seed_range.end);
let parent = db
.new_batch()
.set(Sha256::fill(3u8), Sha256::fill(13u8))
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.set(Sha256::fill(4u8), Sha256::fill(14u8))
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
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 applied_start = db.bounds().end;
let db = db.apply_batch(parent).await.unwrap();
let parent_range = applied_start..db.bounds().end;
assert_eq!(child.proof(&db).unwrap(), child_proof);
assert_eq!(child.pinned_nodes(&db).unwrap(), child_pins);
let applied_start = db.bounds().end;
let db = db.apply_batch(child).await.unwrap();
let child_range = applied_start..db.bounds().end;
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
.unwrap();
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 applied_start = db.bounds().end;
let db = db.apply_batch(empty).await.unwrap();
let empty_range = applied_start..db.bounds().end;
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::<Sha256, _, _>(&proof, start, &ops, &root));
assert!(verify_proof_and_pinned_nodes::<Sha256, _, _>(
&proof, start, &ops, &pins, &root
));
}
let late = db
.new_batch()
.set(Sha256::fill(5u8), Sha256::fill(15u8))
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
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()
.set(Sha256::fill(6u8), Sha256::fill(16u8))
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
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::<Sha256, _, _>(
&flushed_proof,
flushed_start,
&flushed_ops,
&flushed_pins,
&flushed_root
));
let applied_start = db.bounds().end;
let db = db.apply_batch(flushed).await.unwrap();
let flushed_range = applied_start..db.bounds().end;
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_batch_chain<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let k1 = Sha256::fill(1u8);
let k2 = Sha256::fill(2u8);
let k3 = Sha256::fill(3u8);
let v1 = Sha256::fill(11u8);
let v2 = Sha256::fill(12u8);
let v3 = Sha256::fill(13u8);
let parent = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.set(k2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert_eq!(child.get(&k1, &db).await.unwrap(), Some(v1));
assert_eq!(child.get(&k2, &db).await.unwrap(), Some(v2));
assert!(child.get(&k3, &db).await.unwrap().is_none());
let db = db.apply_batch(child).await.unwrap().commit().await.unwrap();
assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
let merkleized = db
.new_batch()
.set(k3, v3)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_build_and_authenticate<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let mut batch = db.new_batch();
for i in 0u64..2_000 {
let k = Sha256::hash(&[&i.to_be_bytes()]);
let v = Sha256::fill(i as u8);
batch = batch.set(k, v);
}
let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.bounds().end, 2_000 + 2);
let root = db.root();
drop(db);
let db = open_db(context.child("second")).await;
assert_eq!(root, db.root());
assert_eq!(db.bounds().end, 2_000 + 2);
for i in 0u64..2_000 {
let k = Sha256::hash(&[&i.to_be_bytes()]);
let v = Sha256::fill(i as u8);
assert_eq!(db.get(&k).await.unwrap().unwrap(), v);
}
let max_ops = NZU64!(5);
for i in 0..*db.bounds().end {
let (proof, log) = db.proof(Location::new(i), max_ops).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(i),
&log,
&root
));
}
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_recovery_from_failed_merkle_sync<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
const ELEMENTS: u64 = 1000;
let db = open_db(context.child("first")).await;
let mut batch = db.new_batch();
for i in 0u64..ELEMENTS {
let k = Sha256::hash(&[&i.to_be_bytes()]);
let v = Sha256::fill(i as u8);
batch = batch.set(k, v);
}
let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.bounds().end, ELEMENTS + 2);
let db = db.sync().await.unwrap();
let halfway_root = db.root();
let mut batch = db.new_batch();
for i in ELEMENTS..ELEMENTS * 2 {
let k = Sha256::hash(&[&i.to_be_bytes()]);
let v = Sha256::fill(i as u8);
batch = batch.set(k, v);
}
let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
db.commit().await.unwrap();
let db = open_db(context.child("second")).await;
assert_eq!(db.bounds().end, 2003);
let root = db.root();
assert_ne!(root, halfway_root);
drop(db);
let db = open_db(context.child("third")).await;
assert_eq!(db.bounds().end, 2003);
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_recovery_from_failed_log_sync<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(3u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let first_commit_root = db.root();
drop(db);
let db = open_db(context.child("second")).await;
assert_eq!(db.bounds().end, 3);
let root = db.root();
assert_eq!(root, first_commit_root);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_pruning<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
const ELEMENTS: u64 = 2_000;
let db = open_db(context.child("first")).await;
let mut sorted_keys: Vec<sha256::Digest> = (1u64..ELEMENTS + 1)
.map(|i| Sha256::hash(&[&i.to_be_bytes()]))
.collect();
sorted_keys.sort();
let mut batch = db.new_batch();
for i in 1u64..ELEMENTS + 1 {
let k = Sha256::hash(&[&i.to_be_bytes()]);
let v = Sha256::fill(i as u8);
batch = batch.set(k, v);
}
let inactivity_floor = Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1);
let merkleized = batch.merkleize(&db, None, inactivity_floor).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.bounds().end, ELEMENTS + 2);
let db = db.prune(Location::new((ELEMENTS + 2) / 2)).await.unwrap();
let bounds = db.bounds();
assert_eq!(bounds.end, ELEMENTS + 2);
let oldest_retained_loc = bounds.start;
assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2));
let pruned_key = sorted_keys[*oldest_retained_loc as usize - 2];
assert!(db.get(&pruned_key).await.unwrap().is_none());
let unpruned_key = sorted_keys[*oldest_retained_loc as usize - 1];
assert!(db.get(&unpruned_key).await.unwrap().is_some());
let root = db.root();
db.sync().await.unwrap();
let db = open_db(context.child("second")).await;
assert_eq!(root, db.root());
let bounds = db.bounds();
assert_eq!(bounds.end, ELEMENTS + 2);
let oldest_retained_loc = bounds.start;
assert_eq!(oldest_retained_loc, Location::new(ELEMENTS / 2));
let loc = Location::new(ELEMENTS / 2 + (ITEMS_PER_SECTION * 2 - 1));
let db = db.prune(loc).await.unwrap();
let oldest_retained_loc = db.bounds().start;
assert_eq!(
oldest_retained_loc,
Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION)
);
db.sync().await.unwrap();
let db = open_db(context.child("third")).await;
let oldest_retained_loc = db.bounds().start;
assert_eq!(
oldest_retained_loc,
Location::new(ELEMENTS / 2 + ITEMS_PER_SECTION)
);
let floor_val = ELEMENTS / 2 + ITEMS_PER_SECTION * 2 - 1;
let inactive_key = sorted_keys[floor_val as usize - 2];
assert!(db.get(&inactive_key).await.unwrap().is_none());
let active_key = sorted_keys[floor_val as usize - 1];
assert!(db.get(&active_key).await.unwrap().is_some());
let pruned_pos = ELEMENTS / 2;
let proof_result = db
.proof(Location::new(pruned_pos), NZU64!(pruned_pos + 100))
.await;
assert!(
matches!(proof_result, Err(Error::Journal(crate::journal::Error::ItemPruned(pos))) if pos == pruned_pos)
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_prune_beyond_floor<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
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 = open_db(context.child("test")).await;
let k1 = Digest::from(*b"12345678901234567890123456789012");
let k2 = Digest::from(*b"abcdefghijklmnopqrstuvwxyz123456");
let k3 = Digest::from(*b"99999999999999999999999999999999");
let v1 = Sha256::fill(1u8);
let v2 = Sha256::fill(2u8);
let v3 = Sha256::fill(3u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.set(k2, v2)
.merkleize(&db, None, Location::new(3))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(*db.size() - 1, 3);
let merkleized = db
.new_batch()
.set(k3, v3)
.merkleize(&db, None, Location::new(5))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.prune(Location::new(3)).await.unwrap();
let floor = db.inactivity_floor_loc();
let beyond = floor + 1;
let result = db.prune(beyond).await;
assert!(
matches!(result, Err(Error::PruneBeyondMinRequired(prune_loc, f))
if prune_loc == beyond && f == floor)
);
}
async fn commit_sets<F: Family, V, C>(
db: TestDb<F, V, C>,
sets: impl IntoIterator<Item = (Digest, V::Value)>,
metadata: Option<V::Value>,
) -> TestDb<F, V, C>
where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
commit_sets_with_floor(db, sets, metadata, Location::new(0)).await
}
async fn commit_sets_with_floor<F: Family, V, C>(
db: TestDb<F, V, C>,
sets: impl IntoIterator<Item = (Digest, V::Value)>,
metadata: Option<V::Value>,
floor: Location<F>,
) -> TestDb<F, V, C>
where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let mut batch = db.new_batch();
for (key, value) in sets {
batch = batch.set(key, value);
}
let merkleized = batch.merkleize(&db, metadata, floor).await.unwrap();
let expected_end = merkleized.bounds().tip.size;
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.bounds().end, expected_end);
db.commit().await.unwrap()
}
#[boxed]
pub(crate) async fn run_bounded_initialization_recovery<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
Option<Location<F>>,
)
-> Pin<Box<dyn Future<Output = Result<TestDb<F, V, C>, Error<F>>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db"), None).await.unwrap();
let key1 = Sha256::hash(&[&1u64.to_be_bytes()]);
let key2 = Sha256::hash(&[&2u64.to_be_bytes()]);
let key3 = Sha256::hash(&[&3u64.to_be_bytes()]);
let key4 = Sha256::hash(&[&4u64.to_be_bytes()]);
let value1 = Sha256::fill(11u8);
let value2 = Sha256::fill(22u8);
let value3 = Sha256::fill(33u8);
let value4 = Sha256::fill(66u8);
let metadata_a = Sha256::fill(44u8);
let db = commit_sets(db, [(key1, value1), (key2, value2)], Some(metadata_a)).await;
let size_before = db.bounds().end;
let root_before = db.root();
let last_commit_before = db.size() - 1;
let metadata_b = Sha256::fill(55u8);
let db = commit_sets(db, [(key3, value3), (key4, value4)], Some(metadata_b)).await;
assert_ne!(db.root(), root_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_b));
assert_eq!(db.get(&key3).await.unwrap(), Some(value3));
assert_eq!(db.get(&key4).await.unwrap(), Some(value4));
_ = db.sync().await.unwrap();
let db = open_db(context.child("cap"), Some(size_before))
.await
.unwrap();
assert_eq!(db.root(), root_before);
assert_eq!(db.bounds().end, size_before);
assert_eq!(db.size() - 1, last_commit_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
assert_eq!(db.get(&key3).await.unwrap(), None);
assert_eq!(db.get(&key4).await.unwrap(), None);
db.commit().await.unwrap();
let db = open_db(context.child("reopen"), None).await.unwrap();
assert_eq!(db.root(), root_before);
assert_eq!(db.bounds().end, size_before);
assert_eq!(db.size() - 1, last_commit_before);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata_a));
assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
assert_eq!(db.get(&key3).await.unwrap(), None);
assert_eq!(db.get(&key4).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_bounded_initialization_preserves_collision_bucket<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
Option<Location<F>>,
)
-> Pin<Box<dyn Future<Output = Result<TestDb<F, V, C>, Error<F>>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db"), None).await.unwrap();
let mut k1_bytes = [0u8; 32];
let mut k2_bytes = [0u8; 32];
k1_bytes[0] = 0xAA;
k1_bytes[1] = 0xBB;
k2_bytes[0] = 0xAA;
k2_bytes[1] = 0xBB;
k1_bytes[31] = 0x01;
k2_bytes[31] = 0x02;
let key1 = Digest::from(k1_bytes);
let key2 = Digest::from(k2_bytes);
let value1 = Sha256::fill(11u8);
let value2 = Sha256::fill(22u8);
let db = commit_sets(db, [(key1, value1)], None).await;
let size_after_first = db.bounds().end;
let db = commit_sets(db, [(key2, value2)], None).await;
assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
_ = db.sync().await.unwrap();
let db = open_db(context.child("cap"), Some(size_after_first))
.await
.unwrap();
assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
assert_eq!(db.get(&key2).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_bounded_initialization_pruned_target_errors<F: Family, V, C>(
context: deterministic::Context,
open_small_sections_db: impl Fn(
deterministic::Context,
Option<Location<F>>,
) -> Pin<
Box<dyn Future<Output = Result<TestDb<F, V, C>, Error<F>>> + Send>,
>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_small_sections_db(context.child("db"), None)
.await
.unwrap();
let first_start = db.bounds().end;
let mut db = commit_sets(
db,
(0u64..16).map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8))),
None,
)
.await;
let mut round = 0u64;
loop {
round += 1;
assert!(
round <= 64,
"failed to prune enough history for the bounded initialization test"
);
let floor = db.bounds().end + 16;
db = commit_sets_with_floor(
db,
(0u64..16).map(|i| {
let seed = round * 100 + i;
(
Sha256::hash(&[&seed.to_be_bytes()]),
Sha256::fill(seed as u8),
)
}),
None,
floor,
)
.await;
let last_commit = db.size() - 1;
db = db.prune(last_commit).await.unwrap();
if db.bounds().start > first_start {
break;
}
}
let oldest_retained = db.bounds().start;
_ = db.sync().await.unwrap();
let Err(boundary_err) =
open_small_sections_db(context.child("cap_error"), Some(oldest_retained)).await
else {
panic!("expected bounded initialization to fail");
};
assert!(
matches!(
boundary_err,
Error::Journal(crate::journal::Error::ItemPruned(_))
),
"unexpected bounded initialization error at retained boundary: {boundary_err:?}"
);
let db = open_small_sections_db(context.child("db"), None)
.await
.unwrap();
_ = db.sync().await.unwrap();
let Err(err) = open_small_sections_db(context.child("cap_error"), Some(first_start)).await
else {
panic!("expected bounded initialization to fail");
};
assert!(
matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
"unexpected bounded initialization error: {err:?}"
);
}
#[boxed]
pub(crate) async fn run_batch_get_read_through<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
let val_a = Sha256::fill(1u8);
let merkleized = db
.new_batch()
.set(key_a, val_a)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let mut batch = db.new_batch();
assert_eq!(batch.get(&key_a, &db).await.unwrap(), Some(val_a));
let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
let val_b = Sha256::fill(2u8);
batch = batch.set(key_b, val_b);
assert_eq!(batch.get(&key_b, &db).await.unwrap(), Some(val_b));
let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
assert_eq!(batch.get(&key_c, &db).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_stacked_get<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
let val_a = Sha256::fill(10u8);
let parent = db.new_batch().set(key_a, val_a);
let parent_m = parent.merkleize(&db, None, Location::new(0)).await.unwrap();
let mut child = parent_m.new_batch::<Sha256>();
assert_eq!(child.get(&key_a, &db).await.unwrap(), Some(val_a));
let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
let val_b = Sha256::fill(20u8);
child = child.set(key_b, val_b);
assert_eq!(child.get(&key_b, &db).await.unwrap(), Some(val_b));
let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
assert_eq!(child.get(&key_c, &db).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_stacked_apply<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let mut kvs_first: Vec<(Digest, Digest)> = (0u64..5)
.map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8)))
.collect();
kvs_first.sort_by_key(|a| a.0);
let mut kvs_second: Vec<(Digest, Digest)> = (5u64..10)
.map(|i| (Sha256::hash(&[&i.to_be_bytes()]), Sha256::fill(i as u8)))
.collect();
kvs_second.sort_by_key(|a| a.0);
let mut parent = db.new_batch();
for (k, v) in &kvs_first {
parent = parent.set(*k, *v);
}
let parent_m = parent.merkleize(&db, None, Location::new(0)).await.unwrap();
let mut child = parent_m.new_batch::<Sha256>();
for (k, v) in &kvs_second {
child = child.set(*k, *v);
}
let child_m = child.merkleize(&db, None, Location::new(0)).await.unwrap();
let expected_root = child_m.root();
let db = db.apply_batch(child_m).await.unwrap();
assert_eq!(db.root(), expected_root);
for (k, v) in kvs_first.iter().chain(kvs_second.iter()) {
assert_eq!(db.get(k).await.unwrap(), Some(*v));
}
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_speculative_root<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let mut batch = db.new_batch();
for i in 0u8..10 {
let k = Sha256::hash(&[&[i]]);
batch = batch.set(k, Sha256::fill(i));
}
let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
let speculative = merkleized.root();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.root(), speculative);
let metadata = Some(Sha256::fill(55u8));
let mut batch = db.new_batch();
let k = Sha256::hash(&[&[0xAA]]);
batch = batch.set(k, Sha256::fill(0xAA));
let merkleized = batch
.merkleize(&db, metadata, Location::new(0))
.await
.unwrap();
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>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
let val_a = Sha256::fill(10u8);
let merkleized = db
.new_batch()
.set(key_a, val_a)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
let val_b = Sha256::fill(20u8);
let merkleized = db
.new_batch()
.set(key_b, val_b)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert_eq!(merkleized.get(&key_a, &db).await.unwrap(), Some(val_a));
assert_eq!(merkleized.get(&key_b, &db).await.unwrap(), Some(val_b));
let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
assert_eq!(merkleized.get(&key_c, &db).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_sequential_apply<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
let val_a = Sha256::fill(1u8);
let m = db
.new_batch()
.set(key_a, val_a)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let root1 = m.root();
let db = db.apply_batch(m).await.unwrap();
assert_eq!(db.root(), root1);
assert_eq!(db.get(&key_a).await.unwrap(), Some(val_a));
let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
let val_b = Sha256::fill(2u8);
let m = db
.new_batch()
.set(key_b, val_b)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let root2 = m.root();
let db = db.apply_batch(m).await.unwrap();
assert_eq!(db.root(), root2);
assert_eq!(db.get(&key_b).await.unwrap(), Some(val_b));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_many_sequential<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let mut db = open_db(context.child("db")).await;
const BATCHES: u64 = 20;
const KEYS_PER_BATCH: u64 = 5;
let mut all_kvs: Vec<(Digest, Digest)> = Vec::new();
for batch_idx in 0..BATCHES {
let mut batch = db.new_batch();
for j in 0..KEYS_PER_BATCH {
let seed = batch_idx * 100 + j;
let k = Sha256::hash(&[&seed.to_be_bytes()]);
let v = Sha256::fill(seed as u8);
batch = batch.set(k, v);
all_kvs.push((k, v));
}
let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
db = db.apply_batch(merkleized).await.unwrap();
}
for (k, v) in &all_kvs {
assert_eq!(db.get(k).await.unwrap(), Some(*v));
}
let root = db.root();
let (proof, ops) = db.proof(Location::new(0), NZU64!(10000)).await.unwrap();
assert!(verify_proof::<Sha256, _, _>(
&proof,
Location::new(0),
&ops,
&root
));
let expected = 1 + BATCHES * (KEYS_PER_BATCH + 1);
assert_eq!(db.bounds().end, expected);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_empty_batch<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let k = Sha256::hash(&[&[1u8]]);
let merkleized = db
.new_batch()
.set(k, Sha256::fill(1u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
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, Location::new(0))
.await
.unwrap();
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>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&[&0u64.to_be_bytes()]);
let val_a = Sha256::fill(10u8);
let merkleized = db
.new_batch()
.set(key_a, val_a)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let key_b = Sha256::hash(&[&1u64.to_be_bytes()]);
let val_b = Sha256::fill(1u8);
let parent_m = db
.new_batch()
.set(key_b, val_b)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let key_c = Sha256::hash(&[&2u64.to_be_bytes()]);
let val_c = Sha256::fill(2u8);
let child_m = parent_m
.new_batch::<Sha256>()
.set(key_c, val_c)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert_eq!(child_m.get(&key_a, &db).await.unwrap(), Some(val_a));
assert_eq!(child_m.get(&key_b, &db).await.unwrap(), Some(val_b));
assert_eq!(child_m.get(&key_c, &db).await.unwrap(), Some(val_c));
let key_d = Sha256::hash(&[&3u64.to_be_bytes()]);
assert_eq!(child_m.get(&key_d, &db).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_large<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
const N: u64 = 500;
let mut kvs: Vec<(Digest, Digest)> = Vec::new();
let mut batch = db.new_batch();
for i in 0..N {
let k = Sha256::hash(&[&i.to_be_bytes()]);
let v = Sha256::fill((i % 256) as u8);
batch = batch.set(k, v);
kvs.push((k, v));
}
let merkleized = batch.merkleize(&db, None, Location::new(0)).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
for (k, v) in &kvs {
assert_eq!(db.get(k).await.unwrap(), Some(*v));
}
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_prune_collision_bucket<F: Family, V, C>(
context: deterministic::Context,
open_db_small_sections: impl Fn(
deterministic::Context,
)
-> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db_small_sections(context.child("db")).await;
let mut k1_bytes = [0u8; 32];
let mut k2_bytes = [0u8; 32];
k1_bytes[0] = 0xAA;
k1_bytes[1] = 0xBB;
k2_bytes[0] = 0xAA;
k2_bytes[1] = 0xBB;
k1_bytes[31] = 0x01;
k2_bytes[31] = 0x02;
let key1 = Digest::from(k1_bytes);
let key2 = Digest::from(k2_bytes);
let key3 = Sha256::fill(3u8);
let v1 = Sha256::fill(1u8);
let v2 = Sha256::fill(2u8);
let v3 = Sha256::fill(3u8);
let merkleized = db
.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let merkleized = db
.new_batch()
.set(key2, v2)
.set(key3, v3)
.merkleize(&db, None, Location::new(4))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
assert_eq!(db.snapshot.len(), 3);
let db = db.prune(Location::new(2)).await.unwrap();
assert_eq!(db.bounds().start, Location::new(2));
assert_eq!(db.snapshot.len(), 2);
assert_eq!(db.get(&key1).await.unwrap(), None);
assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_prune_retires_pruned_sets<F: Family, V, C>(
context: deterministic::Context,
open_db_small_sections: impl Fn(
deterministic::Context,
)
-> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let mut db = open_db_small_sections(context.child("db")).await;
for i in 0..10u8 {
let floor = db.bounds().end;
let merkleized = db
.new_batch()
.set(Sha256::fill(i), Sha256::fill(i))
.merkleize(&db, None, floor)
.await
.unwrap();
db = db.apply_batch(merkleized).await.unwrap();
}
assert_eq!(db.bounds(), Location::new(0)..Location::new(21));
assert_eq!(db.snapshot.len(), 10);
for loc in [2u64, 4, 6] {
db = db.prune(Location::new(loc)).await.unwrap();
assert_eq!(db.bounds().start, Location::new(loc));
assert_eq!(db.snapshot.len(), 10 - loc as usize / 2);
}
assert!(db.get(&Sha256::fill(0)).await.unwrap().is_none());
db = db.prune(Location::new(9)).await.unwrap();
assert_eq!(db.snapshot.len(), 6);
assert_eq!(
db.get(&Sha256::fill(4)).await.unwrap(),
Some(Sha256::fill(4))
);
db = db.prune(Location::new(10)).await.unwrap();
assert_eq!(db.snapshot.len(), 5);
assert!(db.get(&Sha256::fill(4)).await.unwrap().is_none());
assert_eq!(
db.get(&Sha256::fill(5)).await.unwrap(),
Some(Sha256::fill(5))
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_batch_metadata<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let metadata = Sha256::fill(42u8);
let k = Sha256::hash(&[&[1u8]]);
let merkleized = db
.new_batch()
.set(k, Sha256::fill(1u8))
.merkleize(&db, Some(metadata), Location::new(0))
.await
.unwrap();
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, Location::new(0))
.await
.unwrap();
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_merkleize_foreign_db<F: Family, V, C>(
db: TestDb<F, V, C>,
foreign: TestDb<F, V, C>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let key = Sha256::fill(1);
let batch = db
.new_batch()
.set(key, Sha256::fill(11))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch).await.unwrap();
let batch = foreign
.new_batch()
.set(key, Sha256::fill(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 direct = db.new_batch().set(Sha256::fill(2), Sha256::fill(22));
let direct = direct.merkleize(&foreign, None, Location::new(0)).await;
let parent = db
.new_batch()
.set(Sha256::fill(3), Sha256::fill(33))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.set(Sha256::fill(4), Sha256::fill(44));
let child = child.merkleize(&foreign, None, Location::new(0)).await;
assert!(matches!(direct, Err(Error::StaleBatch)));
assert!(matches!(child, Err(Error::StaleBatch)));
db.destroy().await.unwrap();
foreign.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_merkleize_stale_sibling<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let direct = db.new_batch().set(Sha256::fill(1), Sha256::fill(11));
let sibling = db
.new_batch()
.set(Sha256::fill(2), Sha256::fill(22))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(sibling).await.unwrap();
let direct = direct.merkleize(&db, None, Location::new(0)).await;
assert!(matches!(direct, Err(Error::StaleBatch)));
let parent = db
.new_batch()
.set(Sha256::fill(3), Sha256::fill(33))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.set(Sha256::fill(4), Sha256::fill(44));
let sibling = db
.new_batch()
.set(Sha256::fill(5), Sha256::fill(55))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(sibling).await.unwrap();
let child = child.merkleize(&db, None, Location::new(0)).await;
assert!(matches!(child, Err(Error::StaleBatch)));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_stale_batch_rejected<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let v1 = Sha256::fill(10u8);
let v2 = Sha256::fill(20u8);
let batch_a = db
.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let batch_b = db
.new_batch()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(batch_a).await.unwrap();
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), None);
assert_eq!(db.get_metadata().await.unwrap(), None);
let db = db.commit().await.unwrap();
let root = db.root();
let size = db.size();
let result = db.apply_batch(batch_b).await;
assert!(matches!(result, Err(Error::StaleBatch)));
let db = open_db(context.child("reopen")).await;
assert_eq!(db.root(), root);
assert_eq!(db.size(), size);
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_stale_batch_chained<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let key3 = Sha256::hash(&[&[3]]);
let common_parent = db
.new_batch()
.set(Sha256::hash(&[&[10]]), Sha256::fill(10u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let sibling_a = common_parent
.new_batch::<Sha256>()
.set(Sha256::hash(&[&[11]]), Sha256::fill(11u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let sibling_b = common_parent
.new_batch::<Sha256>()
.set(Sha256::hash(&[&[12]]), Sha256::fill(12u8))
.merkleize(&db, None, Location::new(0))
.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()
.set(key1, Sha256::fill(1u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let parent_b = db
.new_batch()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child_b = parent_b
.new_batch::<Sha256>()
.set(key3, Sha256::fill(3u8))
.merkleize(&db, None, Location::new(0))
.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();
}
#[boxed]
pub(crate) async fn run_partial_ancestor_commit<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let key3 = Sha256::hash(&[&[3]]);
let v1 = Sha256::fill(1u8);
let v2 = Sha256::fill(2u8);
let v3 = Sha256::fill(3u8);
let a = db
.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let e = a
.new_batch::<Sha256>()
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let b = e
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let c = b
.new_batch::<Sha256>()
.set(key3, v3)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let expected_root = c.root();
let db = db.apply_batch(a).await.unwrap();
let db = db.apply_batch(c).await.unwrap();
assert_eq!(db.root(), expected_root);
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
let keys = [key1, key2, key3];
let locations = |db: &TestDb<F, V, C>| {
keys.iter()
.map(|key| {
let mut locations: Vec<_> = db.snapshot.get(key).collect();
locations.sort();
locations
})
.collect::<Vec<_>>()
};
assert_eq!(db.snapshot.len(), keys.len());
let live = locations(&db);
db.sync().await.unwrap();
let db = open_db(context.child("reopen")).await;
assert_eq!(db.snapshot.len(), keys.len());
assert_eq!(locations(&db), live);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_delayed_merkleize_after_ancestor_apply<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let key3 = Sha256::hash(&[&[3]]);
let v1 = Sha256::fill(1u8);
let v2 = Sha256::fill(2u8);
let v3 = Sha256::fill(3u8);
let a = db
.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let b = a
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let pending = b
.new_batch::<Sha256>()
.set(key3, v3)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let expected_root = pending.root();
let c = b.new_batch::<Sha256>().set(key3, v3);
let db = db.apply_batch(Arc::clone(&a)).await.unwrap();
let live = b
.new_batch::<Sha256>()
.set(key3, v3)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert_eq!(live.root(), expected_root);
drop(a);
let retired = c.merkleize(&db, None, Location::new(0)).await.unwrap();
assert_eq!(retired.bounds().db, db.commitment());
assert_eq!(retired.root(), expected_root);
let c = b.new_batch::<Sha256>().set(key3, v3);
let db = db.apply_batch(b).await.unwrap();
let c = c.merkleize(&db, None, Location::new(0)).await.unwrap();
assert_eq!(c.root(), expected_root);
let db = db.apply_batch(c).await.unwrap();
assert_eq!(db.root(), expected_root);
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_sequential_commit_parent_then_child<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let v1 = Sha256::fill(1u8);
let v2 = Sha256::fill(2u8);
let parent_m = db
.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child_m = parent_m
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(parent_m).await.unwrap();
let db = db.apply_batch(child_m).await.unwrap();
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_child_root_matches_pending_and_committed<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let parent = db
.new_batch()
.set(key1, Sha256::fill(1u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let pending_child = parent
.new_batch::<Sha256>()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(parent).await.unwrap();
let db = db.commit().await.unwrap();
let committed_child = db
.new_batch()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
assert_eq!(pending_child.root(), committed_child.root());
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_stale_batch_child_applied_before_parent<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let parent_m = db
.new_batch()
.set(key1, Sha256::fill(1u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let child_m = parent_m
.new_batch::<Sha256>()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(child_m).await.unwrap();
let result = db.apply_batch(parent_m).await;
assert!(matches!(result, Err(Error::StaleBatch)));
}
#[boxed]
pub(crate) async fn run_to_batch<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let v1 = Sha256::fill(10u8);
let merkleized = db
.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let snapshot = db.to_batch();
assert_eq!(snapshot.root(), db.root());
let key2 = Sha256::hash(&[&[2]]);
let v2 = Sha256::fill(20u8);
let child = snapshot
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(child).await.unwrap();
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_apply_after_ancestor_dropped<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[&[1]]);
let key2 = Sha256::hash(&[&[2]]);
let key3 = Sha256::hash(&[&[3]]);
let v1 = Sha256::fill(1u8);
let v2 = Sha256::fill(2u8);
let v3 = Sha256::fill(3u8);
let a = db
.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let b = a
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let c = b
.new_batch::<Sha256>()
.set(key3, v3)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
drop(a);
drop(b);
let db = db.apply_batch(c).await.unwrap();
assert_eq!(db.get(&key1).await.unwrap(), Some(v1));
assert_eq!(db.get(&key2).await.unwrap(), Some(v2));
assert_eq!(db.get(&key3).await.unwrap(), Some(v3));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_inactivity_floor_tracking<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(0))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
let merkleized = db
.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(3))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(3));
let db = db.commit().await.unwrap();
db.sync().await.unwrap();
let db = open_db(context.child("reopen")).await;
assert_eq!(db.inactivity_floor_loc(), Location::new(3));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_floor_monotonicity<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(2));
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
let merkleized = db
.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(2));
let k3 = Sha256::fill(5u8);
let v3 = Sha256::fill(6u8);
let merkleized = db
.new_batch()
.set(k3, v3)
.merkleize(&db, None, Location::new(5))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(5));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_bounded_initialization_restores_floor<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
Option<Location<F>>,
)
-> Pin<Box<dyn Future<Output = Result<TestDb<F, V, C>, Error<F>>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test"), None).await.unwrap();
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let first_size = db.bounds().end;
assert_eq!(db.inactivity_floor_loc(), Location::new(2));
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
let merkleized = db
.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(4))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(4));
_ = db.sync().await.unwrap();
let db = open_db(context.child("cap"), Some(first_size))
.await
.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_floor_monotonicity_violation<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
let result = db
.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(1))
.await;
assert!(matches!(result, Err(Error::FloorRegressed(new, current))
if new == Location::new(1) && current == Location::new(2)));
}
#[boxed]
pub(crate) async fn run_floor_beyond_size<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
let result = db
.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(100))
.await;
assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit))
if floor == Location::new(100) && commit == Location::new(2)));
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
let result = db
.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(3))
.await;
assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit))
if floor == Location::new(3) && commit == Location::new(2)));
let merkleized = db
.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_chained_floor_regression<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
let a = db
.new_batch()
.set(Sha256::fill(1u8), Sha256::fill(2u8))
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let Err(err) = a
.new_batch::<Sha256>()
.set(Sha256::fill(3u8), Sha256::fill(4u8))
.merkleize(&db, None, Location::new(1))
.await
else {
panic!("merkleize must reject the invalid floor");
};
assert!(
matches!(err, Error::FloorRegressed(new, prev)
if new == Location::new(1) && prev == Location::new(2)),
"unexpected error: {err:?}"
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_chained_floor_beyond_commit<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
let a = db
.new_batch()
.set(Sha256::fill(1u8), Sha256::fill(2u8))
.merkleize(&db, None, Location::new(2))
.await
.unwrap();
let Err(err) = a
.new_batch::<Sha256>()
.set(Sha256::fill(3u8), Sha256::fill(4u8))
.merkleize(&db, None, Location::new(5))
.await
else {
panic!("merkleize must reject the invalid floor");
};
assert!(
matches!(err, Error::FloorBeyondSize(floor, commit)
if floor == Location::new(5) && commit == Location::new(4)),
"unexpected error: {err:?}"
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_bounded_initialization_after_reopen_with_floor_change<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
Option<Location<F>>,
)
-> Pin<Box<dyn Future<Output = Result<TestDb<F, V, C>, Error<F>>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first"), None).await.unwrap();
let k1 = Sha256::fill(1u8);
let k2 = Sha256::fill(2u8);
let k3 = Sha256::fill(3u8);
let v1 = Sha256::fill(11u8);
let v2 = Sha256::fill(12u8);
let v3 = Sha256::fill(13u8);
let db = commit_sets(db, [(k1, v1), (k2, v2), (k3, v3)], None).await;
let first_size = db.bounds().end;
let first_root = db.root();
let k4 = Sha256::fill(4u8);
let k5 = Sha256::fill(5u8);
let k6 = Sha256::fill(6u8);
let v4 = Sha256::fill(14u8);
let v5 = Sha256::fill(15u8);
let v6 = Sha256::fill(16u8);
let db = commit_sets_with_floor(db, [(k4, v4), (k5, v5), (k6, v6)], None, first_size).await;
db.sync().await.unwrap();
let db = open_db(context.child("second"), None).await.unwrap();
assert!(db.get(&k1).await.unwrap().is_none());
_ = db.sync().await.unwrap();
let db = open_db(context.child("cap"), Some(first_size))
.await
.unwrap();
assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
assert_eq!(db.root(), first_root);
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
assert!(db.get(&k4).await.unwrap().is_none());
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_bounded_initialization_after_reopen_partial_floor_gap<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
Option<Location<F>>,
)
-> Pin<Box<dyn Future<Output = Result<TestDb<F, V, C>, Error<F>>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("first"), None).await.unwrap();
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(11u8);
let db = commit_sets(db, [(k1, v1)], None).await;
let first_size = db.bounds().end;
let first_root = db.root();
let k2 = Sha256::fill(2u8);
let v2 = Sha256::fill(12u8);
let db = commit_sets_with_floor(db, [(k2, v2)], None, first_size).await;
let second_size = db.bounds().end;
let k3 = Sha256::fill(3u8);
let v3 = Sha256::fill(13u8);
let db = commit_sets_with_floor(db, [(k3, v3)], None, second_size).await;
db.sync().await.unwrap();
let db = open_db(context.child("second"), None).await.unwrap();
assert!(db.get(&k1).await.unwrap().is_none());
assert!(db.get(&k2).await.unwrap().is_none());
assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
_ = db.sync().await.unwrap();
let db = open_db(context.child("cap"), Some(second_size))
.await
.unwrap();
assert!(db.get(&k1).await.unwrap().is_none()); assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
assert!(db.get(&k3).await.unwrap().is_none());
_ = db.sync().await.unwrap();
let db = open_db(context.child("cap"), Some(first_size))
.await
.unwrap();
assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
assert!(db.get(&k2).await.unwrap().is_none()); assert_eq!(db.root(), first_root);
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_single_commit_live_set<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("test")).await;
let metadata = Sha256::fill(42u8);
let commit_loc = Location::<F>::new(4);
let k1 = Sha256::fill(1u8);
let k2 = Sha256::fill(2u8);
let k3 = Sha256::fill(3u8);
let v1 = Sha256::fill(11u8);
let v2 = Sha256::fill(12u8);
let v3 = Sha256::fill(13u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.set(k2, v2)
.set(k3, v3)
.merkleize(&db, Some(metadata), commit_loc)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.size() - 1, commit_loc);
assert_eq!(db.inactivity_floor_loc(), commit_loc);
let root_after_commit = db.root();
assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
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.size() - 1, 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));
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 = open_db(context.child("reopened")).await;
assert_eq!(db.size() - 1, 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));
assert!(db.get(&k1).await.unwrap().is_none());
assert!(db.get(&k2).await.unwrap().is_none());
assert!(db.get(&k3).await.unwrap().is_none());
let k4 = Sha256::fill(4u8);
let v4 = Sha256::fill(14u8);
let next_commit_loc = Location::<F>::new(6);
let merkleized = db
.new_batch()
.set(k4, v4)
.merkleize(&db, None, next_commit_loc)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.size() - 1, next_commit_loc);
assert_eq!(db.inactivity_floor_loc(), next_commit_loc);
assert_eq!(db.get(&k4).await.unwrap(), Some(v4));
assert!(db.get(&k1).await.unwrap().is_none());
assert_eq!(db.get_metadata().await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_get_many<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let k1 = Sha256::fill(1u8);
let k2 = Sha256::fill(2u8);
let k3 = Sha256::fill(3u8);
let k_missing = Sha256::fill(99u8);
let v1 = Sha256::fill(11u8);
let v2 = Sha256::fill(12u8);
let v3 = Sha256::fill(13u8);
let merkleized = db
.new_batch()
.set(k1, v1)
.set(k2, v2)
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let results = db.get_many(&[&k1, &k2, &k_missing]).await.unwrap();
assert_eq!(results, vec![Some(v1), Some(v2), None]);
let results = db.get_many(&([] as [&Digest; 0])).await.unwrap();
assert!(results.is_empty());
let batch = db.new_batch().set(k3, v3);
let results = batch.get_many(&[&k3, &k1, &k_missing], &db).await.unwrap();
assert_eq!(results, vec![Some(v3), Some(v1), None]);
let parent = db
.new_batch()
.set(k3, v3)
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
let results = parent.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap();
assert_eq!(results, vec![Some(v1), Some(v3), None]);
let k4 = Sha256::fill(4u8);
let v4 = Sha256::fill(14u8);
let child = parent.new_batch::<Sha256>().set(k4, v4);
let results = child
.get_many(&[&k1, &k3, &k4, &k_missing], &db)
.await
.unwrap();
assert_eq!(results, vec![Some(v1), Some(v3), Some(v4), None]);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_get_many_duplicate_keys<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key = Sha256::fill(1u8);
let value = Sha256::fill(11u8);
let k_missing = Sha256::fill(99u8);
let merkleized = db
.new_batch()
.set(key, value)
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let db = db.commit().await.unwrap();
let results = db.get_many(&[&key, &k_missing, &key, &key]).await.unwrap();
assert_eq!(results, vec![Some(value), None, Some(value), Some(value)]);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn run_get_many_unexpected_data<F: Family, V, C>(
context: deterministic::Context,
open_db: impl Fn(
deterministic::Context,
) -> Pin<Box<dyn Future<Output = TestDb<F, V, C>> + Send>>,
) where
V: ValueEncoding<Value = Digest>,
C: Mutable<Item = Operation<F, Digest, V>>,
C::Item: EncodeShared,
{
let db = open_db(context.child("db")).await;
let key = Sha256::fill(1u8);
let value = Sha256::fill(11u8);
let merkleized = db
.new_batch()
.set(key, value)
.merkleize(&db, None, db.inactivity_floor_loc())
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let mut db = db.commit().await.unwrap();
let bad_key = Sha256::fill(99u8);
let bad_loc = db.size() - 1;
db.snapshot.insert(&bad_key, *bad_loc);
let err = db.get(&bad_key).await.unwrap_err();
assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc));
let err = db.get_many(&[&bad_key]).await.unwrap_err();
assert!(matches!(err, Error::UnexpectedData(loc) if loc == bad_loc));
db.destroy().await.unwrap();
}
}