use crate::{
index::{unordered::Index, Unordered as _},
journal::{
authenticated,
contiguous::{Contiguous, Mutable},
},
merkle::{full::Config as MerkleConfig, Family, Location, Proof},
qmdb::{
any::ValueEncoding, build_snapshot_from_log, metrics::Metrics, operation::Key,
single_operation_root, Error,
},
translator::Translator,
Context,
};
use ahash::AHashSet;
use commonware_codec::EncodeShared;
use commonware_cryptography::Hasher;
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use core::num::{NonZeroU64, NonZeroUsize};
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;
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_cache_size: Option<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,
{
pub(crate) journal: authenticated::Journal<F, E, C, H, S>,
pub(crate) root: H::Digest,
pub(crate) snapshot: Index<T, Location<F>>,
pub(crate) last_commit_loc: Location<F>,
pub(crate) inactivity_floor_loc: Location<F>,
metrics: Metrics<E>,
}
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(crate) async fn init_from_journal(
mut journal: authenticated::Journal<F, E, C, H, S>,
context: E,
translator: T,
cache_size: Option<NonZeroUsize>,
) -> Result<Self, Error<F>> {
if journal.size() == 0 {
warn!("Authenticated log is empty, initialized new db.");
journal
.append(&Operation::Commit(None, Location::new(0)))
.await?;
journal.sync().await?;
}
let mut snapshot = Index::new(context.child("snapshot"), translator);
let (last_commit_loc, inactivity_floor_loc) = {
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");
if inactivity_floor_loc > last_commit_loc {
return Err(Error::DataCorrupted("inactivity floor exceeds last commit"));
}
build_snapshot_from_log::<F, _, _, _>(
inactivity_floor_loc,
&journal.journal,
&mut snapshot,
cache_size,
|_, _| {},
)
.await?;
(last_commit_loc, inactivity_floor_loc)
};
let inactive_peaks = F::inactive_peaks(
F::location_to_position(Location::new(*last_commit_loc + 1)),
inactivity_floor_loc,
);
let root = journal.root(inactive_peaks)?;
let metrics = Metrics::new(context);
let db = Self {
journal,
root,
snapshot,
last_commit_loc,
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,
*self.last_commit_loc,
);
}
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>> {
let _timer = self.metrics.get_timer();
self.metrics.get_calls.inc();
self.metrics.lookups_requested.inc();
let iter = self.snapshot.get(key);
let oldest = self.journal.bounds().start;
let mut result = None;
for &loc in iter {
if loc < oldest {
continue;
}
if let Some(v) = Self::get_from_loc(&self.journal, key, loc).await? {
result = Some(v);
break;
}
}
Ok(result)
}
pub async fn get_many(&self, keys: &[&K]) -> 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()];
let oldest = self.journal.bounds().start;
for (key_idx, key) in keys.iter().enumerate() {
for &loc in self.snapshot.get(key) {
if loc < oldest {
continue;
}
candidates.push((key_idx, *loc));
}
}
if candidates.is_empty() {
return Ok(results);
}
candidates.sort_unstable_by_key(|&(_, pos)| pos);
let mut positions: Vec<u64> = Vec::with_capacity(candidates.len());
for &(_, pos) in &candidates {
if positions.last() != Some(&pos) {
positions.push(pos);
}
}
let ops = self.journal.read_many(&positions).await?;
for &(key_idx, pos) in &candidates {
if results[key_idx].is_some() {
continue;
}
let op_idx = positions
.binary_search(&pos)
.expect("position was deduped from candidates");
let Operation::Set(k, v) = &ops[op_idx] else {
return Err(Error::UnexpectedData(Location::new(pos)));
};
if k == keys[key_idx] {
results[key_idx] = Some(v.clone());
}
}
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.last_commit_loc;
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, |op| op.has_floor())
.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)]
pub async fn prune(&mut self, loc: Location<F>) -> Result<(), 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.prune(loc).await?;
self.update_metrics();
Ok(())
}
#[tracing::instrument(name = "qmdb.immutable.db.rewind", level = "info", skip_all)]
pub async fn rewind(&mut self, size: Location<F>) -> Result<(), Error<F>> {
let rewind_size = *size;
let current_size = *self.last_commit_loc + 1;
if rewind_size == current_size {
return Ok(());
}
if rewind_size == 0 || rewind_size > current_size {
return Err(Error::Journal(crate::journal::Error::InvalidRewind(
rewind_size,
)));
}
let (rewind_last_loc, rewind_floor, rewound_keys) = {
let bounds = self.journal.bounds();
let rewind_last_loc = Location::new(rewind_size - 1);
if rewind_size <= bounds.start {
return Err(Error::Journal(crate::journal::Error::ItemPruned(
*rewind_last_loc,
)));
}
let rewind_last_op = self.journal.read(*rewind_last_loc).await?;
let Operation::Commit(_, rewind_floor) = &rewind_last_op else {
return Err(Error::UnexpectedData(rewind_last_loc));
};
let rewind_floor = *rewind_floor;
if *rewind_floor < bounds.start {
return Err(Error::Journal(crate::journal::Error::ItemPruned(
*rewind_floor,
)));
}
let mut rewound_keys = Vec::new();
for loc in rewind_size..current_size {
if let Operation::Set(key, _) = self.journal.read(loc).await? {
rewound_keys.push(key);
}
}
(rewind_last_loc, rewind_floor, rewound_keys)
};
let old_floor = self.inactivity_floor_loc;
self.journal.rewind(rewind_size).await?;
let rewind_loc = Location::<F>::new(rewind_size);
for key in &rewound_keys {
self.snapshot.retain(key, |loc| *loc < rewind_loc);
}
if rewind_floor < old_floor {
let gap_end = core::cmp::min(*old_floor, rewind_size);
for loc in *rewind_floor..gap_end {
if let Operation::Set(key, _) = self.journal.journal.read(loc).await? {
self.snapshot.insert(&key, Location::new(loc));
}
}
}
self.last_commit_loc = rewind_last_loc;
self.inactivity_floor_loc = rewind_floor;
let inactive_peaks = F::inactive_peaks(F::location_to_position(size), rewind_floor);
self.root = self.journal.root(inactive_peaks)?;
self.update_metrics();
Ok(())
}
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<(), Error<F>> {
let _timer = self.metrics.sync_timer();
self.metrics.sync_calls.inc();
self.journal.sync().await?;
Ok(())
}
#[tracing::instrument(name = "qmdb.immutable.db.commit", level = "info", skip_all)]
pub async fn commit(&mut self) -> Result<(), Error<F>> {
let _timer = self.metrics.commit_timer();
self.metrics.commit_calls.inc();
self.journal.commit().await?;
Ok(())
}
#[boxed]
pub async fn destroy(self) -> Result<(), Error<F>> {
Ok(self.journal.destroy().await?)
}
#[allow(clippy::type_complexity)]
pub fn new_batch(&self) -> batch::UnmerkleizedBatch<F, H, K, V, S> {
let journal_size = *self.last_commit_loc + 1;
batch::UnmerkleizedBatch::new(self, journal_size)
}
#[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<Range<Location<F>>, Error<F>> {
let _timer = self.metrics.apply_batch_timer();
self.metrics.apply_batch_calls.inc();
let db_size = *self.last_commit_loc + 1;
batch
.bounds
.validate_apply_to(db_size, self.inactivity_floor_loc)?;
let start_loc = Location::new(db_size);
self.journal.apply_batch(&batch.journal_batch).await?;
let bounds = self.journal.bounds();
let track_shadow = batch.bounds.ancestors.iter().any(|a| a.end > db_size);
let seen_cap = if track_shadow {
batch.diff.len()
+ batch
.bounds
.ancestors
.iter()
.zip(&batch.ancestor_diffs)
.filter(|(a, _)| a.end > db_size)
.map(|(_, d)| d.len())
.sum::<usize>()
} else {
0
};
let mut seen: AHashSet<&K> = AHashSet::with_capacity(seen_cap);
for (key, entry) in batch.diff.iter() {
if track_shadow {
seen.insert(key);
}
self.snapshot
.insert_and_retain(key, entry.loc, |v| *v >= bounds.start);
}
for (i, ancestor_diff) in batch.ancestor_diffs.iter().enumerate() {
if batch.bounds.ancestors[i].end <= db_size {
continue;
}
for (key, entry) in ancestor_diff.iter() {
if seen.insert(key) {
self.snapshot
.insert_and_retain(key, entry.loc, |v| *v >= bounds.start);
}
}
}
self.last_commit_loc = Location::new(batch.bounds.total_size - 1);
self.inactivity_floor_loc = batch.bounds.inactivity_floor;
self.root = batch.root;
let range = start_loc..Location::new(batch.bounds.total_size);
self.update_metrics();
self.metrics
.operations_applied
.inc_by(*range.end - *range.start);
Ok(range)
}
}
#[cfg(test)]
pub(super) mod test {
use super::*;
use crate::{
merkle::{Family, Location},
qmdb::verify_proof,
translator::TwoCap,
};
use commonware_codec::EncodeShared;
use commonware_cryptography::{sha256, sha256::Digest, Sha256};
use commonware_runtime::{deterministic, Supervisor as _};
use commonware_utils::NZU64;
use core::{future::Future, pin::Pin};
use std::ops::Range;
const ITEMS_PER_SECTION: u64 = 5;
type TestDb<F, V, C> = Immutable<
F,
deterministic::Context,
Digest,
V,
C,
Sha256,
TwoCap,
commonware_parallel::Sequential,
>;
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("second")).await;
assert_eq!(db.root(), root);
assert_eq!(db.bounds().end, 1);
db.apply_batch(db.new_batch().merkleize(&db, None, Location::new(0)).await)
.await
.unwrap();
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 test_immutable_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 mut 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);
commit_sets(&mut db, [(k1, v1)], None).await;
db.sync().await.unwrap();
commit_sets(&mut 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 test_immutable_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 mut 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));
db.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, metadata, Location::new(0))
.await,
)
.await
.unwrap();
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)));
db.apply_batch(
db.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
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 test_immutable_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 mut db = open_db(context.child("first")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(10u8);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
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 test_immutable_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;
db.apply_batch(
db.new_batch()
.set(key, value)
.merkleize(&db, None, floor)
.await,
)
.await
.unwrap();
db.commit().await.unwrap();
}
let root_before = db.root();
let bounds_before = db.bounds();
let prune_loc = Location::new(*bounds_before.end - 5);
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 test_immutable_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 mut 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)));
}
db.apply_batch(batch.merkleize(&db, None, Location::new(0)).await)
.await
.unwrap();
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);
db.apply_batch(
db.new_batch()
.set(key, value)
.merkleize(&db, None, buffered_floor)
.await,
)
.await
.unwrap();
let buffered_state = (db.root(), db.inactivity_floor_loc(), db.bounds().end);
assert_ne!(buffered_state, durable_state);
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 test_immutable_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 mut 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;
let child = parent
.new_batch::<Sha256>()
.set(k2, v2)
.merkleize(&db, None, Location::new(0))
.await;
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());
db.apply_batch(child).await.unwrap();
db.commit().await.unwrap();
assert_eq!(db.get(&k1).await.unwrap(), Some(v1));
assert_eq!(db.get(&k2).await.unwrap(), Some(v2));
db.apply_batch(
db.new_batch()
.set(k3, v3)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
db.commit().await.unwrap();
assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut 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;
db.apply_batch(merkleized).await.unwrap();
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 test_immutable_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 mut 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;
db.apply_batch(merkleized).await.unwrap();
db.commit().await.unwrap();
assert_eq!(db.bounds().end, ELEMENTS + 2);
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;
db.apply_batch(merkleized).await.unwrap();
db.commit().await.unwrap();
drop(db);
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 test_immutable_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 mut db = open_db(context.child("first")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(3u8);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
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 test_immutable_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 mut 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;
db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.bounds().end, ELEMENTS + 2);
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();
drop(db);
let mut 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));
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();
drop(db);
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 test_immutable_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 mut 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 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);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.set(k2, v2)
.merkleize(&db, None, Location::new(3))
.await,
)
.await
.unwrap();
assert_eq!(*db.last_commit_loc, 3);
db.apply_batch(
db.new_batch()
.set(k3, v3)
.merkleize(&db, None, Location::new(5))
.await,
)
.await
.unwrap();
assert!(db.prune(Location::new(3)).await.is_ok());
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)
);
db.destroy().await.unwrap();
}
async fn commit_sets<F: Family, V, C>(
db: &mut TestDb<F, V, C>,
sets: impl IntoIterator<Item = (Digest, V::Value)>,
metadata: Option<V::Value>,
) -> Range<Location<F>>
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: &mut TestDb<F, V, C>,
sets: impl IntoIterator<Item = (Digest, V::Value)>,
metadata: Option<V::Value>,
floor: Location<F>,
) -> Range<Location<F>>
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 range = db
.apply_batch(batch.merkleize(db, metadata, floor).await)
.await
.unwrap();
db.commit().await.unwrap();
range
}
#[boxed]
pub(crate) async fn test_immutable_rewind_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 mut db = open_db(context.child("db")).await;
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 first_range =
commit_sets(&mut 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.last_commit_loc;
assert_eq!(size_before, first_range.end);
let metadata_b = Sha256::fill(55u8);
let second_range =
commit_sets(&mut db, [(key3, value3), (key4, value4)], Some(metadata_b)).await;
assert_eq!(second_range.start, size_before);
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.rewind(size_before).await.unwrap();
assert_eq!(db.root(), root_before);
assert_eq!(db.bounds().end, size_before);
assert_eq!(db.last_commit_loc, 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();
drop(db);
let db = open_db(context.child("reopen")).await;
assert_eq!(db.root(), root_before);
assert_eq!(db.bounds().end, size_before);
assert_eq!(db.last_commit_loc, 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 test_immutable_rewind_preserves_collision_bucket<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;
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);
commit_sets(&mut db, [(key1, value1)], None).await;
let size_after_first = db.bounds().end;
commit_sets(&mut db, [(key2, value2)], None).await;
assert_eq!(db.get(&key1).await.unwrap(), Some(value1));
assert_eq!(db.get(&key2).await.unwrap(), Some(value2));
db.rewind(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 test_immutable_rewind_pruned_target_errors<F: Family, V, C>(
context: deterministic::Context,
open_small_sections_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_small_sections_db(context.child("db")).await;
let first_range = commit_sets(
&mut 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 rewind test"
);
let floor = Location::new(*db.bounds().end + 16);
commit_sets_with_floor(
&mut db,
(0u64..16).map(|i| {
let seed = round * 100 + i;
(Sha256::hash(&seed.to_be_bytes()), Sha256::fill(seed as u8))
}),
None,
floor,
)
.await;
db.prune(db.last_commit_loc).await.unwrap();
if db.bounds().start > first_range.start {
break;
}
}
let oldest_retained = db.bounds().start;
let boundary_err = db.rewind(oldest_retained).await.unwrap_err();
assert!(
matches!(
boundary_err,
Error::Journal(crate::journal::Error::ItemPruned(_))
),
"unexpected rewind error at retained boundary: {boundary_err:?}"
);
let err = db.rewind(first_range.start).await.unwrap_err();
assert!(
matches!(err, Error::Journal(crate::journal::Error::ItemPruned(_))),
"unexpected rewind error: {err:?}"
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&0u64.to_be_bytes());
let val_a = Sha256::fill(1u8);
db.apply_batch(
db.new_batch()
.set(key_a, val_a)
.merkleize(&db, None, Location::new(0))
.await,
)
.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 test_immutable_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;
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 test_immutable_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 mut 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;
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;
let expected_root = child_m.root();
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 test_immutable_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 mut 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;
let speculative = merkleized.root();
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;
let speculative = merkleized.root();
db.apply_batch(merkleized).await.unwrap();
assert_eq!(db.root(), speculative);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&0u64.to_be_bytes());
let val_a = Sha256::fill(10u8);
db.apply_batch(
db.new_batch()
.set(key_a, val_a)
.merkleize(&db, None, Location::new(0))
.await,
)
.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;
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 test_immutable_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 mut 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;
let root1 = m.root();
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;
let root2 = m.root();
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 test_immutable_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;
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 test_immutable_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 mut db = open_db(context.child("db")).await;
let k = Sha256::hash(&[1u8]);
db.apply_batch(
db.new_batch()
.set(k, Sha256::fill(1u8))
.merkleize(&db, None, Location::new(0))
.await,
)
.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;
let speculative = merkleized.root();
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 test_immutable_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 mut db = open_db(context.child("db")).await;
let key_a = Sha256::hash(&0u64.to_be_bytes());
let val_a = Sha256::fill(10u8);
db.apply_batch(
db.new_batch()
.set(key_a, val_a)
.merkleize(&db, None, Location::new(0))
.await,
)
.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;
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;
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 test_immutable_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 mut 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;
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 test_immutable_batch_chained_key_override<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;
let key = Sha256::hash(&0u64.to_be_bytes());
let val_parent = Sha256::fill(1u8);
let val_child = Sha256::fill(2u8);
let parent_m = db
.new_batch()
.set(key, val_parent)
.merkleize(&db, None, Location::new(0))
.await;
let mut child = parent_m.new_batch::<Sha256>();
child = child.set(key, val_child);
assert_eq!(child.get(&key, &db).await.unwrap(), Some(val_child));
let child_m = child.merkleize(&db, None, Location::new(0)).await;
assert_eq!(child_m.get(&key, &db).await.unwrap(), Some(val_child));
db.apply_batch(child_m).await.unwrap();
assert_eq!(db.get(&key).await.unwrap(), Some(val_child));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_batch_sequential_key_override<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;
let key = Sha256::hash(&0u64.to_be_bytes());
let v1 = Sha256::fill(1u8);
let v2 = Sha256::fill(2u8);
db.apply_batch(
db.new_batch()
.set(key, v1)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
assert_eq!(db.get(&key).await.unwrap(), Some(v1));
db.apply_batch(
db.new_batch()
.set(key, v2)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
let live = db.get(&key).await.unwrap().unwrap();
assert!(live == v1 || live == v2);
db.commit().await.unwrap();
drop(db);
let db = open_db_small_sections(context.child("reopen")).await;
let reopened = db.get(&key).await.unwrap().unwrap();
assert!(reopened == v1 || reopened == v2);
db.destroy().await.unwrap();
let mut db = open_db_small_sections(context.child("prune")).await;
db.apply_batch(
db.new_batch()
.set(key, v1)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
db.apply_batch(
db.new_batch()
.set(key, v2)
.merkleize(&db, None, Location::new(4))
.await,
)
.await
.unwrap();
db.prune(Location::new(2)).await.unwrap();
assert_eq!(db.get(&key).await.unwrap(), Some(v2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("db")).await;
let metadata = Sha256::fill(42u8);
let k = Sha256::hash(&[1u8]);
db.apply_batch(
db.new_batch()
.set(k, Sha256::fill(1u8))
.merkleize(&db, Some(metadata), Location::new(0))
.await,
)
.await
.unwrap();
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
db.apply_batch(db.new_batch().merkleize(&db, None, Location::new(0)).await)
.await
.unwrap();
assert_eq!(db.get_metadata().await.unwrap(), None);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut 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;
let batch_b = db
.new_batch()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await;
db.apply_batch(batch_a).await.unwrap();
let expected_root = db.root();
let expected_bounds = db.bounds();
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 result = db.apply_batch(batch_b).await;
assert!(
matches!(result, Err(Error::StaleBatch { .. })),
"expected StaleBatch error, got {result:?}"
);
assert_eq!(db.root(), expected_root);
assert_eq!(db.bounds(), expected_bounds);
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);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[1]);
let key2 = Sha256::hash(&[2]);
let key3 = Sha256::hash(&[3]);
let parent_m = db
.new_batch()
.set(key1, Sha256::fill(1u8))
.merkleize(&db, None, Location::new(0))
.await;
let child_a = parent_m
.new_batch::<Sha256>()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await;
let child_b = parent_m
.new_batch::<Sha256>()
.set(key3, Sha256::fill(3u8))
.merkleize(&db, None, Location::new(0))
.await;
db.apply_batch(child_a).await.unwrap();
let result = db.apply_batch(child_b).await;
assert!(
matches!(result, Err(Error::StaleBatch { .. })),
"expected StaleBatch error, got {result:?}"
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut 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;
let b = a
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await;
let c = b
.new_batch::<Sha256>()
.set(key3, v3)
.merkleize(&db, None, Location::new(0))
.await;
let expected_root = c.root();
db.apply_batch(a).await.unwrap();
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 test_immutable_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 mut 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;
let child_m = parent_m
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await;
db.apply_batch(parent_m).await.unwrap();
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 test_immutable_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 mut 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;
let pending_child = parent
.new_batch::<Sha256>()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await;
db.apply_batch(parent).await.unwrap();
db.commit().await.unwrap();
let committed_child = db
.new_batch()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await;
assert_eq!(pending_child.root(), committed_child.root());
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut 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;
let child_m = parent_m
.new_batch::<Sha256>()
.set(key2, Sha256::fill(2u8))
.merkleize(&db, None, Location::new(0))
.await;
db.apply_batch(child_m).await.unwrap();
let result = db.apply_batch(parent_m).await;
assert!(
matches!(result, Err(Error::StaleBatch { .. })),
"expected StaleBatch error, got {result:?}"
);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("db")).await;
let key1 = Sha256::hash(&[1]);
let v1 = Sha256::fill(10u8);
db.apply_batch(
db.new_batch()
.set(key1, v1)
.merkleize(&db, None, Location::new(0))
.await,
)
.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;
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 test_immutable_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 mut 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;
let b = a
.new_batch::<Sha256>()
.set(key2, v2)
.merkleize(&db, None, Location::new(0))
.await;
let c = b
.new_batch::<Sha256>()
.set(key3, v3)
.merkleize(&db, None, Location::new(0))
.await;
drop(a);
drop(b);
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 test_immutable_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 mut 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);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(0))
.await,
)
.await
.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(0));
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
db.apply_batch(
db.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(3))
.await,
)
.await
.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(3));
db.commit().await.unwrap();
db.sync().await.unwrap();
drop(db);
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 test_immutable_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 mut db = open_db(context.child("test")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(2))
.await,
)
.await
.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(2));
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
db.apply_batch(
db.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(2))
.await,
)
.await
.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(2));
let k3 = Sha256::fill(5u8);
let v3 = Sha256::fill(6u8);
db.apply_batch(
db.new_batch()
.set(k3, v3)
.merkleize(&db, None, Location::new(5))
.await,
)
.await
.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(5));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_rewind_restores_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 mut db = open_db(context.child("test")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(2))
.await,
)
.await
.unwrap();
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);
db.apply_batch(
db.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(4))
.await,
)
.await
.unwrap();
db.commit().await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(4));
db.rewind(first_size).await.unwrap();
assert_eq!(db.inactivity_floor_loc(), Location::new(2));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("test")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(2))
.await,
)
.await
.unwrap();
let k2 = Sha256::fill(3u8);
let v2 = Sha256::fill(4u8);
let result = db
.apply_batch(
db.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(1))
.await,
)
.await;
assert!(matches!(result, Err(Error::FloorRegressed(new, current))
if new == Location::new(1) && current == Location::new(2)));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("test")).await;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(2u8);
let result = db
.apply_batch(
db.new_batch()
.set(k1, v1)
.merkleize(&db, None, Location::new(100))
.await,
)
.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
.apply_batch(
db.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(3))
.await,
)
.await;
assert!(matches!(result, Err(Error::FloorBeyondSize(floor, commit))
if floor == Location::new(3) && commit == Location::new(2)));
db.apply_batch(
db.new_batch()
.set(k2, v2)
.merkleize(&db, None, Location::new(2))
.await,
)
.await
.unwrap();
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_chained_ancestor_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 mut 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;
let b = a
.new_batch::<Sha256>()
.set(Sha256::fill(3u8), Sha256::fill(4u8))
.merkleize(&db, None, Location::new(1))
.await;
let c = b
.new_batch::<Sha256>()
.set(Sha256::fill(5u8), Sha256::fill(6u8))
.merkleize(&db, None, Location::new(2))
.await;
let root_before = db.root();
let last_commit_before = db.last_commit_loc;
let floor_before = db.inactivity_floor_loc();
let err = db.apply_batch(c).await.unwrap_err();
assert!(
matches!(err, Error::FloorRegressed(new, prev)
if new == Location::new(1) && prev == Location::new(2)),
"unexpected error: {err:?}"
);
assert_eq!(db.root(), root_before);
assert_eq!(db.last_commit_loc, last_commit_before);
assert_eq!(db.inactivity_floor_loc(), floor_before);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_chained_ancestor_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 mut db = open_db(context.child("test")).await;
let a = db
.new_batch()
.set(Sha256::fill(1u8), Sha256::fill(2u8))
.merkleize(&db, None, Location::new(3))
.await;
let b = a
.new_batch::<Sha256>()
.set(Sha256::fill(3u8), Sha256::fill(4u8))
.merkleize(&db, None, Location::new(0))
.await;
let root_before = db.root();
let last_commit_before = db.last_commit_loc;
let floor_before = db.inactivity_floor_loc();
let err = db.apply_batch(b).await.unwrap_err();
assert!(
matches!(err, Error::FloorBeyondSize(floor, commit)
if floor == Location::new(3) && commit == Location::new(2)),
"unexpected error: {err:?}"
);
assert_eq!(db.root(), root_before);
assert_eq!(db.last_commit_loc, last_commit_before);
assert_eq!(db.inactivity_floor_loc(), floor_before);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_rewind_after_reopen_with_floor_change<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;
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);
commit_sets(&mut 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);
commit_sets_with_floor(&mut db, [(k4, v4), (k5, v5), (k6, v6)], None, first_size).await;
db.sync().await.unwrap();
drop(db);
let mut db = open_db(context.child("second")).await;
assert!(db.get(&k1).await.unwrap().is_none());
db.rewind(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 test_immutable_rewind_after_reopen_partial_floor_gap<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;
let k1 = Sha256::fill(1u8);
let v1 = Sha256::fill(11u8);
commit_sets(&mut 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);
commit_sets_with_floor(&mut db, [(k2, v2)], None, first_size).await;
let second_size = db.bounds().end;
let k3 = Sha256::fill(3u8);
let v3 = Sha256::fill(13u8);
commit_sets_with_floor(&mut db, [(k3, v3)], None, second_size).await;
db.sync().await.unwrap();
drop(db);
let mut db = open_db(context.child("second")).await;
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.rewind(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.rewind(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 test_immutable_rewind_after_reopen_repeated_key_gap<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;
let key = Sha256::fill(7u8);
let v1 = Sha256::fill(17u8);
let v2 = Sha256::fill(18u8);
let k3 = Sha256::fill(8u8);
let v3 = Sha256::fill(19u8);
commit_sets(&mut db, [(key, v1)], None).await;
let first_size = db.bounds().end;
commit_sets(&mut db, [(key, v2)], None).await;
let second_size = db.bounds().end;
let live = db.get(&key).await.unwrap().unwrap();
assert!(live == v1 || live == v2);
commit_sets_with_floor(&mut db, [(k3, v3)], None, second_size).await;
db.sync().await.unwrap();
drop(db);
let mut db = open_db(context.child("second")).await;
assert!(db.get(&key).await.unwrap().is_none());
assert_eq!(db.get(&k3).await.unwrap(), Some(v3));
db.rewind(second_size).await.unwrap();
let rewound = db.get(&key).await.unwrap().unwrap();
assert!(rewound == v1 || rewound == v2);
db.rewind(first_size).await.unwrap();
assert_eq!(db.get(&key).await.unwrap(), Some(v1));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_rewind_after_reopen_mixed_gap_retained<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;
let key = Sha256::fill(7u8);
let v1 = Sha256::fill(17u8);
let v2 = Sha256::fill(18u8);
let k3 = Sha256::fill(8u8);
let v3 = Sha256::fill(19u8);
commit_sets(&mut db, [(key, v1)], None).await;
let first_size = db.bounds().end;
commit_sets(&mut db, [(key, v2)], None).await;
let second_size = db.bounds().end;
let live = db.get(&key).await.unwrap().unwrap();
assert!(live == v1 || live == v2);
commit_sets_with_floor(&mut db, [(k3, v3)], None, first_size).await;
db.sync().await.unwrap();
drop(db);
let mut db = open_db(context.child("second")).await;
assert_eq!(db.get(&key).await.unwrap(), Some(v2));
db.rewind(second_size).await.unwrap();
let rewound = db.get(&key).await.unwrap().unwrap();
assert!(rewound == v1 || rewound == v2);
db.rewind(first_size).await.unwrap();
assert_eq!(db.get(&key).await.unwrap(), Some(v1));
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut 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);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.set(k2, v2)
.set(k3, v3)
.merkleize(&db, Some(metadata), commit_loc)
.await,
)
.await
.unwrap();
db.commit().await.unwrap();
assert_eq!(db.last_commit_loc, commit_loc);
assert_eq!(db.inactivity_floor_loc(), commit_loc);
let root_after_commit = db.root();
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));
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, Location::new(*commit_loc + 1));
let err = db.prune(Location::new(*commit_loc + 1)).await.unwrap_err();
assert!(matches!(err, Error::PruneBeyondMinRequired(p, f)
if *p == *commit_loc + 1 && *f == *commit_loc));
assert_eq!(db.last_commit_loc, commit_loc);
assert_eq!(db.inactivity_floor_loc(), commit_loc);
assert_eq!(db.root(), root_after_commit);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
db.sync().await.unwrap();
drop(db);
let mut db = open_db(context.child("reopened")).await;
assert_eq!(db.last_commit_loc, commit_loc);
assert_eq!(db.inactivity_floor_loc(), commit_loc);
assert_eq!(db.root(), root_after_commit);
assert_eq!(db.get_metadata().await.unwrap(), Some(metadata));
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);
db.apply_batch(
db.new_batch()
.set(k4, v4)
.merkleize(&db, None, next_commit_loc)
.await,
)
.await
.unwrap();
db.commit().await.unwrap();
assert_eq!(db.last_commit_loc, next_commit_loc);
assert_eq!(db.inactivity_floor_loc(), next_commit_loc);
assert_eq!(db.get(&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 test_immutable_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 mut 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);
db.apply_batch(
db.new_batch()
.set(k1, v1)
.set(k2, v2)
.merkleize(&db, None, db.inactivity_floor_loc())
.await,
)
.await
.unwrap();
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;
let results = parent.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap();
assert_eq!(results, vec![Some(v1), Some(v3), None]);
let v3_new = Sha256::fill(30u8);
let child = parent.new_batch::<Sha256>().set(k3, v3_new);
let results = child.get_many(&[&k1, &k3, &k_missing], &db).await.unwrap();
assert_eq!(results, vec![Some(v1), Some(v3_new), None]);
db.destroy().await.unwrap();
}
#[boxed]
pub(crate) async fn test_immutable_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 mut db = open_db(context.child("db")).await;
let key = Sha256::fill(1u8);
let value = Sha256::fill(11u8);
db.apply_batch(
db.new_batch()
.set(key, value)
.merkleize(&db, None, db.inactivity_floor_loc())
.await,
)
.await
.unwrap();
db.commit().await.unwrap();
let bad_key = Sha256::fill(99u8);
let bad_loc = db.last_commit_loc;
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();
}
}