use crate::{
Context,
index::Unordered as UnorderedIndex,
journal::contiguous::{Contiguous, Mutable},
merkle::{
self, Graftable, Location, Position, Readable, hasher::Hasher as _, mem::Mem,
storage::Storage as MerkleStorage,
},
metadata::{Config as MConfig, Metadata},
qmdb::{
self, Error,
any::{
self,
operation::{Operation, update::Update},
},
current::{
batch::BitmapBatch,
grafting,
proof::{OpsRootWitness, RangeProof, RangeProofSpec, constant::OperationProof},
},
},
};
use commonware_codec::{Codec, CodecShared, Copying, DecodeExt};
use commonware_cryptography::{Digest, DigestOf, Hasher};
use commonware_macros::boxed;
use commonware_parallel::Strategy;
use commonware_runtime::{
Handle,
telemetry::metrics::{
Counter, Gauge, GaugeExt as _, MetricsExt as _,
histogram::{ScopedTimer, Timed},
},
};
use commonware_utils::{
Widen,
bitmap::{self, Readable as _},
sequence::prefixed_u64::U64,
};
use core::num::NonZeroU64;
use std::{
collections::{BTreeMap, VecDeque},
sync::Arc,
};
use tracing::{error, warn};
pub(super) struct GraftRoots<D> {
start: usize,
roots: VecDeque<D>,
}
impl<D: Copy> GraftRoots<D> {
fn get(&self, chunk: usize) -> Option<D> {
self.roots.get(chunk.checked_sub(self.start)?).copied()
}
pub(super) fn get_or_read<F: Graftable, const N: usize>(
&self,
chunk: usize,
ops_tree: &impl Readable<Family = F, Digest = D>,
) -> Result<D, merkle::Error<F>> {
if let Some(digest) = self.get(chunk) {
return Ok(digest);
}
assert!(
chunk >= self.end(),
"graft root {chunk} must be retained or not yet grafted by the database ({}..{} retained)",
self.start,
self.end()
);
let pos = graft_root_position::<F, N>(chunk);
ops_tree
.get_node(pos)
.ok_or(merkle::Error::ElementPruned(pos))
}
fn end(&self) -> usize {
self.start + self.roots.len()
}
fn extend_to<F: Graftable, const N: usize>(
&mut self,
leaves: usize,
ops_tree: &impl Readable<Family = F, Digest = D>,
) {
let chunks = self.end()..leaves;
self.roots.extend(chunks.map(|chunk| {
ops_tree
.get_node(graft_root_position::<F, N>(chunk))
.expect("newly graftable root must be in the in-memory ops tree")
}));
}
fn release_below_floor<const N: usize>(&mut self, floor: Location<impl Graftable>) {
let chunk = usize::try_from(*floor / bitmap::Prunable::<N>::CHUNK_SIZE_BITS)
.unwrap_or(usize::MAX)
.min(self.end());
if chunk <= self.start {
return;
}
self.roots.drain(..chunk - self.start);
self.start = chunk;
if self.roots.len() < self.roots.capacity() / 4 {
self.roots.shrink_to(self.roots.len() * 2);
}
}
}
fn graft_root_position<F: Graftable, const N: usize>(chunk: usize) -> Position<F> {
let height = grafting::height::<N>();
F::subtree_root_position(Location::new((chunk as u64) << height), height)
}
const NODE_PREFIX: u8 = 0;
const PRUNED_CHUNKS_PREFIX: u8 = 1;
type GraftedPinnedNodes<F, D> = Vec<(Position<F>, D)>;
pub(crate) struct Metrics<E: Context> {
pruned_chunks: Gauge,
sync_boundary: Gauge,
pub apply_batch_calls: Counter,
apply_batch_duration: Timed,
pub sync_calls: Counter,
sync_duration: Timed,
pub prune_calls: Counter,
prune_duration: Timed,
clock: Arc<E>,
}
impl<E: Context> Metrics<E> {
pub fn new(context: E) -> Self {
Self {
pruned_chunks: context.gauge("pruned_chunks", "Number of pruned bitmap chunks"),
sync_boundary: context
.gauge("sync_boundary", "Most recent safe sync boundary location"),
apply_batch_calls: context.counter("apply_batch_calls", "Number of apply-batch calls"),
apply_batch_duration: Timed::register(
&context,
"apply_batch_duration",
"Duration of apply-batch calls",
),
sync_calls: context.counter("sync_calls", "Number of sync calls"),
sync_duration: Timed::register(&context, "sync_duration", "Duration of sync calls"),
prune_calls: context.counter("prune_calls", "Number of prune calls"),
prune_duration: Timed::register(&context, "prune_duration", "Duration of prune calls"),
clock: Arc::new(context),
}
}
pub fn apply_batch_timer(&self) -> ScopedTimer<E> {
self.apply_batch_duration.scoped(&self.clock)
}
pub fn sync_timer(&self) -> ScopedTimer<E> {
self.sync_duration.scoped(&self.clock)
}
pub fn prune_timer(&self) -> ScopedTimer<E> {
self.prune_duration.scoped(&self.clock)
}
pub fn update(&self, pruned_chunks: u64, sync_boundary: u64) {
let _ = self.pruned_chunks.try_set(pruned_chunks);
let _ = self.sync_boundary.try_set(sync_boundary);
}
}
pub struct Db<
F: merkle::Graftable,
E: Context,
C: Contiguous<Item: CodecShared>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
U: Send + Sync,
const N: usize,
S: Strategy,
> {
pub(super) any: any::db::Db<F, E, C, I, H, U, N, S>,
pub(super) grafted_tree: Arc<Mem<F, H::Digest>>,
pub(super) graft_roots: GraftRoots<H::Digest>,
pub(super) metadata: Metadata<E, U64, Vec<u8>>,
pub(super) strategy: S,
pub(super) root: DigestOf<H>,
pub(super) metrics: Metrics<E>,
#[cfg(test)]
pub(super) halt_before_prune_log: bool,
}
impl<F, E, C, I, H, U, const N: usize, S> std::fmt::Debug for Db<F, E, C, I, H, U, N, S>
where
F: merkle::Graftable,
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
U: Update,
S: Strategy,
Operation<F, U>: Codec,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Db")
.field("bounds", &self.bounds())
.field("inactivity_floor_loc", &self.any.inactivity_floor_loc)
.finish_non_exhaustive()
}
}
impl<F, E, C, I, H, U, const N: usize, S> Db<F, E, C, I, H, U, N, S>
where
F: merkle::Graftable,
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
U: Update,
S: Strategy,
Operation<F, U>: Codec,
{
#[cfg(any(test, feature = "test-traits"))]
pub(crate) const fn inactivity_floor_loc(&self) -> Location<F> {
self.any.inactivity_floor_loc()
}
pub const fn is_empty(&self) -> bool {
self.any.is_empty()
}
pub async fn get_metadata(&self) -> Result<Option<U::Value>, Error<F>> {
self.any.get_metadata().await
}
pub async fn get_many(&self, keys: &[&U::Key]) -> Result<Vec<Option<U::Value>>, Error<F>> {
self.any.get_many(keys).await
}
pub fn bounds(&self) -> std::ops::Range<Location<F>> {
self.any.bounds()
}
pub fn bitmap(&self) -> &impl bitmap::Readable<N> {
self.any.bitmap.as_ref()
}
pub fn verify_range_proof(
proof: &RangeProof<F, H::Digest>,
start_loc: Location<F>,
ops: &[Operation<F, U>],
chunks: &[[u8; N]],
root: &H::Digest,
) -> bool {
proof.verify::<H, _, N>(start_loc, ops, chunks, root)
}
pub fn grafted_storage(&self) -> impl MerkleStorage<F, Digest = H::Digest> + '_ {
grafting::Storage::<F, H, _, _>::new(
&self.grafted_tree,
grafting::height::<N>(),
&self.any.log.merkle,
)
}
pub const fn root(&self) -> H::Digest {
self.root
}
pub const fn strategy(&self) -> &S {
&self.strategy
}
pub const fn ops_root(&self) -> H::Digest {
self.any.root()
}
pub async fn ops_root_witness(&self) -> Result<OpsRootWitness<F, H::Digest>, Error<F>> {
let storage = self.grafted_storage();
let ops_size = storage.size();
let ops_leaves = Location::<F>::try_from(ops_size)?;
compute_ops_root_witness::<F, H, _, _, N>(
self.any.bitmap.as_ref(),
&storage,
ops_leaves,
self.any.inactivity_floor_loc,
)
.await
}
pub(super) fn grafted_snapshot(&self) -> Arc<merkle::batch::MerkleizedBatch<F, H::Digest, S>> {
merkle::batch::MerkleizedBatch::from_mem_with_strategy(
&self.grafted_tree,
self.strategy.clone(),
)
}
pub fn new_batch(&self) -> super::batch::UnmerkleizedBatch<F, H, U, N, S> {
super::batch::UnmerkleizedBatch::new(
self.any.new_batch(),
self.grafted_snapshot(),
BitmapBatch::Base(Arc::clone(&self.any.bitmap)),
)
}
pub(super) async fn operation_proof(
&self,
loc: Location<F>,
) -> Result<OperationProof<F, H::Digest, N>, Error<F>> {
let storage = self.grafted_storage();
let ops_root = self.any.root();
OperationProof::new::<H, _>(
self.any.bitmap.as_ref(),
&storage,
self.any.inactivity_floor_loc,
loc,
ops_root,
)
.await
}
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.current.db.range_proof",
level = "info",
skip_all,
fields(
start_loc = *start_loc,
max_ops = max_ops.get(),
),
)]
pub async fn range_proof(
&self,
start_loc: Location<F>,
max_ops: NonZeroU64,
) -> Result<(RangeProof<F, H::Digest>, Vec<Operation<F, U>>, Vec<[u8; N]>), Error<F>> {
let storage = self.grafted_storage();
let ops_root = self.any.root();
RangeProof::new_with_ops::<H, _, _, N>(
self.any.bitmap.as_ref(),
&storage,
&self.any.log,
RangeProofSpec {
start_loc,
max_ops,
inactivity_floor: self.any.inactivity_floor_loc,
ops_root,
},
)
.await
}
}
impl<F, E, C, I, H, U, const N: usize, S> Db<F, E, C, I, H, U, N, S>
where
F: merkle::Graftable,
E: Context,
C: Mutable<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
U: Update,
S: Strategy,
Operation<F, U>: Codec,
{
pub async fn ops_historical_proof(
&self,
historical_size: Location<F>,
start_loc: Location<F>,
max_ops: NonZeroU64,
) -> Result<(merkle::Proof<F, H::Digest>, Vec<Operation<F, U>>), Error<F>> {
self.any
.historical_proof(historical_size, start_loc, max_ops)
.await
}
pub async fn pinned_nodes_at(&self, loc: Location<F>) -> Result<Vec<H::Digest>, Error<F>> {
self.any.pinned_nodes_at(loc).await
}
pub fn sync_boundary(&self) -> Location<F> {
sync_boundary::<F, N>(
*self.any.inactivity_floor_loc / bitmap::Prunable::<N>::CHUNK_SIZE_BITS,
*self.any.log.size(),
)
}
pub(super) fn update_metrics(&self) {
self.metrics.update(
self.any.bitmap.pruned_chunks() as u64,
*self.sync_boundary(),
);
}
fn grafted_pinned_nodes(
&self,
loc: Location<F>,
) -> Result<GraftedPinnedNodes<F, H::Digest>, Error<F>> {
F::nodes_to_pin(loc)
.map(|pos| {
let digest = self
.grafted_tree
.get_node(pos)
.ok_or(Error::<F>::DataCorrupted("missing grafted pinned node"))?;
Ok((pos, digest))
})
.collect()
}
fn prune_grafted_tree_to_bitmap(&mut self) -> Result<(), Error<F>> {
let pruned_chunks = self.any.bitmap.pruned_chunks() as u64;
if pruned_chunks == 0 {
return Ok(());
}
let prune_loc = Location::<F>::new(pruned_chunks);
if prune_loc <= self.grafted_tree.bounds().start {
return Ok(());
}
let prune_pos = Position::try_from(prune_loc)
.map_err(|_| Error::<F>::DataCorrupted("prune location overflow"))?;
let size = self.grafted_tree.size();
let pinned: BTreeMap<_, _> = self.grafted_pinned_nodes(prune_loc)?.into_iter().collect();
let mut retained = Vec::with_capacity((*size - *prune_pos) as usize);
for p in *prune_pos..*size {
let digest = self
.grafted_tree
.get_node(Position::new(p))
.ok_or(Error::<F>::DataCorrupted("missing retained grafted node"))?;
retained.push(digest);
}
self.grafted_tree = Arc::new(Mem::from_pruned_with_retained(prune_pos, pinned, retained));
Ok(())
}
#[tracing::instrument(name = "qmdb.current.db.prune", level = "info", skip_all)]
#[boxed]
pub async fn prune(mut self, prune_loc: Location<F>) -> Result<Self, Error<F>> {
let _timer = self.metrics.prune_timer();
self.metrics.prune_calls.inc();
let sync_boundary = self.sync_boundary();
if prune_loc > sync_boundary {
return Err(Error::PruneBeyondMinRequired(prune_loc, sync_boundary));
}
self.any.log = self.any.log.commit().await?;
self.any.prune_bitmap(prune_loc);
self.prune_grafted_tree_to_bitmap()?;
self = self.sync_metadata().await?;
#[cfg(test)]
if self.halt_before_prune_log {
std::future::pending::<()>().await;
}
self.any = self.any.prune_log(prune_loc).await?;
self.any.update_metrics();
self.update_metrics();
Ok(self)
}
pub(crate) async fn sync_metadata(mut self) -> Result<Self, Error<F>> {
self.metadata.clear();
let pruned_chunks_u64 = self.any.bitmap.pruned_chunks() as u64;
let key = U64::new(PRUNED_CHUNKS_PREFIX, 0);
self.metadata
.put(key, pruned_chunks_u64.to_be_bytes().to_vec());
let pruned_chunks = Location::<F>::new(pruned_chunks_u64);
for (i, (_, digest)) in self
.grafted_pinned_nodes(pruned_chunks)?
.into_iter()
.enumerate()
{
let key = U64::new(NODE_PREFIX, i as u64);
self.metadata.put(key, digest.to_vec());
}
self.metadata = self.metadata.sync().await?;
Ok(self)
}
#[tracing::instrument(name = "qmdb.current.db.start_sync", level = "info", skip_all)]
#[boxed]
pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error<F>> {
let (any, handle) = self.any.start_sync().await?;
self.any = any;
Ok((self, handle))
}
#[tracing::instrument(name = "qmdb.current.db.commit", level = "info", skip_all)]
#[boxed]
pub async fn commit(mut self) -> Result<Self, Error<F>> {
self.any = self.any.commit().await?;
Ok(self)
}
#[tracing::instrument(name = "qmdb.current.db.sync", level = "info", skip_all)]
#[boxed]
pub async fn sync(mut self) -> Result<Self, Error<F>> {
let _timer = self.metrics.sync_timer();
self.metrics.sync_calls.inc();
self.any = self.any.sync().await?;
self = self.sync_metadata().await?;
self.update_metrics();
Ok(self)
}
#[boxed]
pub async fn destroy(self) -> Result<(), Error<F>> {
let Self { any, metadata, .. } = self;
metadata.destroy().await?;
any.destroy().await
}
pub fn validate_batch(
&self,
batch: &super::batch::MerkleizedBatch<F, H::Digest, U, N, S>,
) -> Result<(), Error<F>> {
self.any.validate_batch(&batch.inner)
}
#[tracing::instrument(name = "qmdb.current.db.apply_batch", level = "info", skip_all)]
#[boxed]
pub async fn apply_batch(
mut self,
batch: Arc<super::batch::MerkleizedBatch<F, H::Digest, U, N, S>>,
) -> Result<Self, Error<F>>
where
I: 'static,
{
let _timer = self.metrics.apply_batch_timer();
self.metrics.apply_batch_calls.inc();
self.any = self.any.apply_batch(Arc::clone(&batch.inner)).await?;
self.graft_roots
.extend_to::<F, N>(*batch.grafted.leaves() as usize, self.any.log.merkle.mem());
self.graft_roots
.release_below_floor::<N>(self.any.inactivity_floor_loc);
Arc::make_mut(&mut self.grafted_tree).apply_batch(&batch.grafted)?;
self.root = batch.canonical_root;
self.update_metrics();
Ok(self)
}
}
pub(crate) fn sync_boundary<F: Graftable, const N: usize>(
mut floor_chunks: u64,
ops_leaves: u64,
) -> Location<F> {
let chunk_bits = bitmap::Prunable::<N>::CHUNK_SIZE_BITS;
let grafting_height = grafting::height::<N>();
while floor_chunks > 0 {
let required_ops = pair_absorption_threshold::<F, N>(floor_chunks).unwrap_or_else(|| {
let youngest_start = (floor_chunks - 1) * chunk_bits;
let pos = F::subtree_root_position(Location::<F>::new(youngest_start), grafting_height);
F::peak_birth_size(pos, grafting_height)
});
if ops_leaves >= required_ops {
break;
}
floor_chunks -= 1;
}
Location::new(floor_chunks * chunk_bits)
}
pub(super) fn pair_absorption_threshold<F: Graftable, const N: usize>(
chunk_count: u64,
) -> Option<u64> {
if chunk_count == 0 {
return None;
}
let grafting_height = grafting::height::<N>();
let youngest = chunk_count - 1;
let youngest_start = youngest << grafting_height;
let youngest_end = (youngest + 1) << grafting_height;
let youngest_pos =
F::subtree_root_position(Location::<F>::new(youngest_start), grafting_height);
if F::peak_birth_size(youngest_pos, grafting_height) <= youngest_end {
return None;
}
let pair_chunk = youngest & !1;
let pair_start = pair_chunk << grafting_height;
let pair_pos = F::subtree_root_position(Location::<F>::new(pair_start), grafting_height + 1);
Some(F::peak_birth_size(pair_pos, grafting_height + 1))
}
pub(super) fn partial_chunk<B: bitmap::Readable<N>, const N: usize>(
bitmap: &B,
) -> Option<([u8; N], u64)> {
let next_bit = bitmap.len() % bitmap::Prunable::<N>::CHUNK_SIZE_BITS;
if next_bit == 0 {
return None;
}
let (last_chunk, _) = bitmap.last_chunk();
Some((last_chunk, next_bit))
}
pub(super) fn graftable_chunk_window<F: merkle::Graftable>(
ops_leaves: Location<F>,
complete: u64,
pruned: u64,
grafting_height: u32,
) -> Result<u64, Error<F>> {
let graftable = grafting::graftable_chunks::<F>(*ops_leaves, grafting_height).min(complete);
let pending = complete - graftable;
if pending > 1 {
return Err(Error::DataCorrupted("multiple pending bitmap chunks"));
}
if pruned > graftable {
return Err(Error::DataCorrupted(
"pruned chunks exceed graftable chunks",
));
}
Ok(graftable)
}
pub(super) fn pending_chunk<F: merkle::Graftable, B: bitmap::Readable<N>, const N: usize>(
bitmap: &B,
ops_leaves: Location<F>,
grafting_height: u32,
) -> Result<Option<[u8; N]>, Error<F>> {
let complete = Widen::widen(bitmap.complete_chunks());
let graftable = graftable_chunk_window(
ops_leaves,
complete,
Widen::widen(bitmap.pruned_chunks()),
grafting_height,
)?;
if complete - graftable != 1 {
return Ok(None);
}
Ok(Some(bitmap.get_chunk(graftable as usize)))
}
pub(super) fn combine_roots<H: Hasher>(
ops_root: &H::Digest,
grafted_root: &H::Digest,
pending: Option<&H::Digest>,
partial: Option<(u64, &H::Digest)>,
) -> H::Digest {
let hasher = qmdb::hasher::<H>();
match (pending, partial) {
(None, None) => hasher.hash(&[ops_root.as_ref(), grafted_root.as_ref()]),
(Some(pe), None) => hasher.hash(&[ops_root.as_ref(), grafted_root.as_ref(), pe.as_ref()]),
(None, Some((nb, p))) => {
let nb_bytes = nb.to_be_bytes();
hasher.hash(&[
ops_root.as_ref(),
grafted_root.as_ref(),
nb_bytes.as_slice(),
p.as_ref(),
])
}
(Some(pe), Some((nb, p))) => {
let nb_bytes = nb.to_be_bytes();
hasher.hash(&[
ops_root.as_ref(),
grafted_root.as_ref(),
pe.as_ref(),
nb_bytes.as_slice(),
p.as_ref(),
])
}
}
}
pub(super) async fn compute_ops_root_witness<
F: merkle::Graftable,
H: Hasher,
B: bitmap::Readable<N>,
S: MerkleStorage<F, Digest = H::Digest>,
const N: usize,
>(
status: &B,
storage: &S,
ops_leaves: Location<F>,
inactivity_floor: Location<F>,
) -> Result<OpsRootWitness<F, H::Digest>, Error<F>> {
let grafted_root =
compute_grafted_root::<F, H, B, S, N>(status, storage, ops_leaves, inactivity_floor)
.await?;
let hasher = qmdb::hasher::<H>();
let pending_chunk_digest =
pending_chunk::<F, B, N>(status, ops_leaves, grafting::height::<N>())?
.map(|chunk| hasher.digest(&chunk))
.try_into()
.expect("pending_chunk must be consistent with family");
let partial_chunk =
partial_chunk::<B, N>(status).map(|(chunk, next_bit)| (next_bit, hasher.digest(&chunk)));
Ok(OpsRootWitness {
grafted_root,
pending_chunk_digest,
partial_chunk,
})
}
pub(super) async fn rebuild_grafted_tree<F, H, S, const N: usize>(
bitmap: &impl bitmap::Readable<N>,
pinned_nodes: &[H::Digest],
ops_tree: &impl MerkleStorage<F, Digest = H::Digest>,
inactivity_floor: Location<F>,
ops_root: H::Digest,
strategy: &S,
) -> Result<(Mem<F, H::Digest>, GraftRoots<H::Digest>, H::Digest), Error<F>>
where
F: merkle::Graftable,
H: Hasher,
S: Strategy,
{
let ops_leaves = Location::<F>::try_from(ops_tree.size())?;
let (grafted_tree, mut graft_roots) =
build_grafted_tree::<F, H, S, N>(bitmap, pinned_nodes, ops_tree, ops_leaves, strategy)
.await?;
graft_roots.release_below_floor::<N>(inactivity_floor);
let storage =
grafting::Storage::<F, H, _, _>::new(&grafted_tree, grafting::height::<N>(), ops_tree);
let root =
compute_ops_root_witness::<F, H, _, _, N>(bitmap, &storage, ops_leaves, inactivity_floor)
.await?
.root::<H>(&ops_root);
Ok((grafted_tree, graft_roots, root))
}
pub(super) async fn compute_grafted_root<
F: merkle::Graftable,
H: Hasher,
B: bitmap::Readable<N>,
S: MerkleStorage<F, Digest = H::Digest>,
const N: usize,
>(
status: &B,
storage: &S,
ops_leaves: Location<F>,
inactivity_floor: Location<F>,
) -> Result<H::Digest, Error<F>> {
let size = storage.size();
let leaves = Location::try_from(size)?;
let mut peaks: Vec<H::Digest> = Vec::new();
for (peak_pos, _) in F::peaks(size) {
let digest = storage
.get_node(peak_pos)
.await?
.ok_or_else(|| merkle::Error::<F>::MissingNode(peak_pos))?;
peaks.push(digest);
}
let grafting_height = grafting::height::<N>();
graftable_chunk_window(
ops_leaves,
Widen::widen(status.complete_chunks()),
Widen::widen(status.pruned_chunks()),
grafting_height,
)?;
let inactive_peaks =
grafting::chunk_aligned_inactive_peaks::<F>(leaves, inactivity_floor, grafting_height)?;
let hasher = qmdb::hasher::<H>();
Ok(hasher.root(leaves, inactive_peaks, peaks.iter())?)
}
pub(super) async fn read_graft_inputs<F: merkle::Graftable, D: Digest, const N: usize>(
ops_tree: &impl MerkleStorage<F, Digest = D>,
chunks: impl IntoIterator<Item = (usize, [u8; N])>,
) -> Result<Vec<(usize, D, [u8; N])>, Error<F>> {
let chunks: Vec<(usize, [u8; N])> = chunks.into_iter().collect();
let positions: Vec<Position<F>> = chunks
.iter()
.map(|&(chunk_idx, _)| graft_root_position::<F, N>(chunk_idx))
.collect();
let nodes = ops_tree.get_nodes(&positions).await?;
Ok(chunks
.into_iter()
.zip(nodes)
.map(|((chunk_idx, chunk), digest)| (chunk_idx, digest, chunk))
.collect())
}
pub(super) async fn build_grafted_tree<
F: merkle::Graftable,
H: Hasher,
S: Strategy,
const N: usize,
>(
bitmap: &impl bitmap::Readable<N>,
pinned_nodes: &[H::Digest],
ops_tree: &impl MerkleStorage<F, Digest = H::Digest>,
ops_leaves: Location<F>,
strategy: &S,
) -> Result<(Mem<F, H::Digest>, GraftRoots<H::Digest>), Error<F>> {
let grafting_height = grafting::height::<N>();
let pruned_chunks = bitmap.pruned_chunks();
let complete_chunks = bitmap.complete_chunks();
let graftable_chunks = grafting::graftable_chunks::<F>(*ops_leaves, grafting_height)
.min(complete_chunks as u64) as usize;
assert!(
pruned_chunks <= graftable_chunks && graftable_chunks <= complete_chunks,
"invariant violated: pruned={pruned_chunks} graftable={graftable_chunks} complete={complete_chunks}"
);
let inputs = read_graft_inputs::<F, _, N>(
ops_tree,
(pruned_chunks..graftable_chunks).map(|chunk_idx| (chunk_idx, bitmap.get_chunk(chunk_idx))),
)
.await?;
let graft_roots = GraftRoots {
start: pruned_chunks,
roots: inputs.iter().map(|(_, digest, _)| *digest).collect(),
};
let leaves = grafting::graft_chunk_digests::<H, _, N>(strategy, inputs);
let mut grafted_tree = if pruned_chunks > 0 {
let grafted_pruning_boundary = Location::<F>::new(pruned_chunks as u64);
Mem::from_components(Vec::new(), grafted_pruning_boundary, pinned_nodes.to_vec())
.map_err(|_| Error::<F>::DataCorrupted("grafted tree rebuild failed"))?
} else {
Mem::new()
};
if !leaves.is_empty() {
let batch = {
let batch = grafted_tree.new_batch_with_strategy(strategy.clone());
let batch = batch.add_leaf_digests(leaves.iter().map(|&(_, digest)| digest));
let grafted_hasher = grafting::hasher::<F, H>(grafting_height);
batch.merkleize(&grafted_tree, &grafted_hasher)
};
grafted_tree.apply_batch(&batch)?;
}
Ok((grafted_tree, graft_roots))
}
pub(super) async fn init_metadata<F: merkle::Graftable, E: Context, D: Digest>(
context: E,
partition: &str,
) -> Result<(Metadata<E, U64, Vec<u8>>, usize, Vec<D>), Error<F>> {
let metadata_cfg = MConfig {
partition: partition.into(),
codec_config: ((0..).into(), ()),
};
let metadata =
Metadata::<_, U64, Vec<u8>>::init(context.child("metadata"), metadata_cfg).await?;
let key = U64::new(PRUNED_CHUNKS_PREFIX, 0);
let pruned_chunks = match metadata.get(&key) {
Some(bytes) => u64::from_be_bytes(bytes.as_slice().try_into().map_err(|_| {
error!("pruned chunks value not a valid u64");
Error::<F>::DataCorrupted("pruned chunks value not a valid u64")
})?),
None => {
warn!("bitmap metadata does not contain pruned chunks, initializing as empty");
0
}
} as usize;
let pinned_nodes = if pruned_chunks > 0 {
let pruned_loc = Location::<F>::new(pruned_chunks as u64);
if !pruned_loc.is_valid() {
return Err(Error::DataCorrupted("pruned chunks exceeds MAX_LEAVES"));
}
let mut pinned = Vec::new();
for (index, _pos) in F::nodes_to_pin(pruned_loc).enumerate() {
let metadata_key = U64::new(NODE_PREFIX, index as u64);
let Some(bytes) = metadata.get(&metadata_key) else {
return Err(Error::DataCorrupted(
"missing pinned node in grafted tree metadata",
));
};
let digest = D::decode(Copying(bytes))
.map_err(|_| Error::<F>::DataCorrupted("invalid pinned node digest"))?;
pinned.push(digest);
}
pinned
} else {
Vec::new()
};
Ok((metadata, pruned_chunks, pinned_nodes))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
merkle::{Bagging::ForwardFold, hasher::Standard as StandardHasher, mmb, mmr},
qmdb::{
any::{
test::{Choice, Inspect as _, Script},
traits::{DbAny, UnmerkleizedBatch as _},
},
current::{tests::fixed_config, unordered::fixed},
floor::Proportional,
},
translator::OneCap,
};
use commonware_codec::FixedSize;
use commonware_cryptography::{Sha256, sha256};
use commonware_macros::test_traced;
use commonware_runtime::{Runner as _, Supervisor as _, deterministic};
use commonware_utils::bitmap::Prunable as PrunableBitMap;
const N: usize = sha256::Digest::SIZE;
#[test]
fn partial_chunk_single_bit() {
let mut bm = PrunableBitMap::<N>::new();
bm.push(true);
let result = partial_chunk::<PrunableBitMap<N>, N>(&bm);
assert!(result.is_some());
let (chunk, next_bit) = result.unwrap();
assert_eq!(next_bit, 1);
assert_eq!(chunk[0], 1); }
#[test]
fn partial_chunk_aligned() {
let mut bm = PrunableBitMap::<N>::new();
for _ in 0..PrunableBitMap::<N>::CHUNK_SIZE_BITS {
bm.push(true);
}
let result = partial_chunk::<PrunableBitMap<N>, N>(&bm);
assert!(result.is_none());
}
#[test]
fn partial_chunk_partial() {
let mut bm = PrunableBitMap::<N>::new();
for _ in 0..(PrunableBitMap::<N>::CHUNK_SIZE_BITS + 5) {
bm.push(true);
}
let result = partial_chunk::<PrunableBitMap<N>, N>(&bm);
assert!(result.is_some());
let (_chunk, next_bit) = result.unwrap();
assert_eq!(next_bit, 5);
}
#[test]
fn partial_chunk_empty() {
let bm = PrunableBitMap::<N>::new();
assert!(partial_chunk::<PrunableBitMap<N>, N>(&bm).is_none());
}
#[test]
fn partial_chunk_fully_pruned() {
let bm = PrunableBitMap::<N>::new_with_pruned_chunks(1).unwrap();
assert!(partial_chunk::<PrunableBitMap<N>, N>(&bm).is_none());
}
#[test]
fn combine_roots_deterministic() {
let ops = Sha256::hash(&[b"ops"]);
let grafted = Sha256::hash(&[b"grafted"]);
let r1 = combine_roots::<Sha256>(&ops, &grafted, None, None);
let r2 = combine_roots::<Sha256>(&ops, &grafted, None, None);
assert_eq!(r1, r2);
}
#[test]
fn combine_roots_with_partial_differs() {
let ops = Sha256::hash(&[b"ops"]);
let grafted = Sha256::hash(&[b"grafted"]);
let partial_digest = Sha256::hash(&[b"partial"]);
let without = combine_roots::<Sha256>(&ops, &grafted, None, None);
let with = combine_roots::<Sha256>(&ops, &grafted, None, Some((5, &partial_digest)));
assert_ne!(without, with);
}
#[test]
fn combine_roots_with_pending_differs() {
let ops = Sha256::hash(&[b"ops"]);
let grafted = Sha256::hash(&[b"grafted"]);
let pending_digest = Sha256::hash(&[b"pending"]);
let without = combine_roots::<Sha256>(&ops, &grafted, None, None);
let with = combine_roots::<Sha256>(&ops, &grafted, Some(&pending_digest), None);
assert_ne!(without, with);
}
#[test]
fn combine_roots_pending_and_partial_independent() {
let ops = Sha256::hash(&[b"ops"]);
let grafted = Sha256::hash(&[b"grafted"]);
let pending_digest = Sha256::hash(&[b"pending"]);
let partial_digest = Sha256::hash(&[b"partial"]);
let only_pending = combine_roots::<Sha256>(&ops, &grafted, Some(&pending_digest), None);
let only_partial =
combine_roots::<Sha256>(&ops, &grafted, None, Some((5, &partial_digest)));
let both = combine_roots::<Sha256>(
&ops,
&grafted,
Some(&pending_digest),
Some((5, &partial_digest)),
);
assert_ne!(only_pending, only_partial);
assert_ne!(only_pending, both);
assert_ne!(only_partial, both);
}
#[test]
fn combine_roots_different_ops_root() {
let ops_a = Sha256::hash(&[b"ops_a"]);
let ops_b = Sha256::hash(&[b"ops_b"]);
let grafted = Sha256::hash(&[b"grafted"]);
let r1 = combine_roots::<Sha256>(&ops_a, &grafted, None, None);
let r2 = combine_roots::<Sha256>(&ops_b, &grafted, None, None);
assert_ne!(r1, r2);
}
#[test]
fn combine_roots_format_golden() {
let hasher = StandardHasher::<Sha256>::new(ForwardFold);
let ops = Sha256::hash(&[b"ops"]);
let grafted = Sha256::hash(&[b"grafted"]);
let pending = Sha256::hash(&[b"pending"]);
let partial = Sha256::hash(&[b"partial"]);
let next_bit: u64 = 0x1122_3344_5566_7788;
assert_eq!(
combine_roots::<Sha256>(&ops, &grafted, None, None),
hasher.hash(&[ops.as_ref(), grafted.as_ref()])
);
assert_eq!(
combine_roots::<Sha256>(&ops, &grafted, Some(&pending), None),
hasher.hash(&[ops.as_ref(), grafted.as_ref(), pending.as_ref()])
);
assert_eq!(
combine_roots::<Sha256>(&ops, &grafted, None, Some((next_bit, &partial))),
hasher.hash(&[
ops.as_ref(),
grafted.as_ref(),
next_bit.to_be_bytes().as_slice(),
partial.as_ref(),
])
);
assert_eq!(
combine_roots::<Sha256>(&ops, &grafted, Some(&pending), Some((next_bit, &partial))),
hasher.hash(&[
ops.as_ref(),
grafted.as_ref(),
pending.as_ref(),
next_bit.to_be_bytes().as_slice(),
partial.as_ref(),
])
);
}
type MmrDb = fixed::Db<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
32,
commonware_parallel::Sequential,
>;
type MmbDb = fixed::Db<
mmb::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
32,
commonware_parallel::Sequential,
>;
#[boxed]
async fn populate_fixed_db<F, DB>(db: DB, start: u64, count: u64) -> DB
where
F: merkle::Graftable,
DB: DbAny<F, Key = sha256::Digest, Value = sha256::Digest>,
{
let mut batch = db.new_batch();
for idx in start..start + count {
let key = Sha256::hash(&[&idx.to_be_bytes()]);
let value = Sha256::hash(&[&(idx + count).to_be_bytes()]);
batch = batch.write(key, Some(value));
}
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
db.commit().await.unwrap()
}
#[test_traced]
fn test_merkleize_after_compatible_ancestor_apply_matches() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let db = MmrDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("resume-after-apply"),
None,
)
.await
.unwrap();
let db = populate_fixed_db::<mmr::Family, _>(db, 0, 40).await;
let db = populate_fixed_db::<mmr::Family, _>(db, 0, 40).await;
let hot = Sha256::hash(&[&0u64.to_be_bytes()]);
let cold = Sha256::hash(&[&1u64.to_be_bytes()]);
let untouched = Sha256::hash(&[&2u64.to_be_bytes()]);
let parent_write = Sha256::hash(&[b"parent-write"]);
let parent = db
.new_batch()
.write(hot, Some(parent_write))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let child_write = Sha256::hash(&[b"child-write"]);
let build =
|parent: &Arc<crate::qmdb::current::batch::MerkleizedBatch<_, _, _, 32, _>>| {
parent
.new_batch::<Sha256>()
.write(cold, Some(child_write))
.write(hot, Some(child_write))
};
let child_pre = build(&parent);
let read_pre = child_pre.get(&untouched, &db).await.unwrap();
assert_eq!(read_pre, Some(Sha256::hash(&[&42u64.to_be_bytes()])));
let root_pre = child_pre
.merkleize(&db, None, &mut Proportional)
.await
.unwrap()
.root();
let db = db.apply_batch(parent.clone()).await.unwrap();
let child_post = build(&parent);
let read_post = child_post.get(&untouched, &db).await.unwrap();
assert_eq!(
read_pre, read_post,
"a database read must not change when the parent is applied"
);
let child_post = child_post
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
assert_eq!(
root_pre,
child_post.root(),
"canonical root must not depend on whether the parent is applied",
);
let db = db.apply_batch(child_post).await.unwrap();
assert_eq!(db.root(), root_pre);
assert_eq!(db.get(&hot).await.unwrap(), Some(child_write));
assert_eq!(db.get(&cold).await.unwrap(), Some(child_write));
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_stale_fork_reads_and_merkleizes_refuse() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let db = MmrDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("stale-fork-refuse"),
None,
)
.await
.unwrap();
let db = populate_fixed_db::<mmr::Family, _>(db, 0, 8).await;
let hot = Sha256::hash(&[&0u64.to_be_bytes()]);
let winner = db
.new_batch()
.write(hot, Some(Sha256::hash(&[b"winner"])))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let loser = db
.new_batch()
.write(hot, Some(Sha256::hash(&[b"loser"])))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
assert_ne!(winner.root(), loser.root());
let child = loser
.new_batch::<Sha256>()
.write(hot, Some(Sha256::hash(&[b"child"])));
let db = db.apply_batch(winner).await.unwrap();
assert!(matches!(child.get(&hot, &db).await, Err(Error::StaleRead)));
assert!(matches!(loser.get(&hot, &db).await, Err(Error::StaleRead)));
assert!(matches!(
child.merkleize(&db, None, &mut Proportional).await,
Err(Error::StaleBatch)
));
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_current_merkleize_across_prune() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let db = MmrDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("across-prune"),
None,
)
.await
.unwrap();
let db = populate_fixed_db::<mmr::Family, _>(db, 0, 140).await;
let base = db.bounds().end;
let mut parent = db.new_batch();
for idx in 0..140u64 {
let key = Sha256::hash(&[&idx.to_be_bytes()]);
parent = parent.write(key, Some(Sha256::hash(&[b"rewrite"])));
}
let parent = parent
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let build =
|parent: &Arc<crate::qmdb::current::batch::MerkleizedBatch<_, _, _, 32, _>>| {
let mut child = parent.new_batch::<Sha256>();
for idx in 100..140u64 {
let key = Sha256::hash(&[&idx.to_be_bytes()]);
child = child.write(key, Some(Sha256::fill(1)));
}
child
};
let child = build(&parent);
let expected = build(&parent)
.merkleize(&db, None, &mut Proportional)
.await
.unwrap()
.root();
let db = db.apply_batch(parent).await.unwrap();
let db = db.commit().await.unwrap();
let boundary = db.sync_boundary();
assert!(boundary > base);
let db = db.prune(boundary).await.unwrap();
assert!(*db.bounds().start > 0, "the prune must remove history");
let merkleized = child.merkleize(&db, None, &mut Proportional).await.unwrap();
assert_eq!(merkleized.root(), expected);
db.destroy().await.unwrap();
});
}
#[test]
fn test_retained_graft_reads() {
struct NoReads;
impl Readable for NoReads {
type Family = mmr::Family;
type Digest = sha256::Digest;
fn size(&self) -> Position<mmr::Family> {
unreachable!()
}
fn get_node(&self, _: Position<mmr::Family>) -> Option<Self::Digest> {
panic!("retained graft reads must not reach the fallback tree")
}
}
let digest = Sha256::hash(&[b"ops"]);
let mut graft_roots = GraftRoots {
start: 4,
roots: VecDeque::from([digest; 3]),
};
graft_roots.release_below_floor::<32>(Location::<mmr::Family>::new(5 * 256));
assert!(graft_roots.get(4).is_none());
for chunk in 5..7 {
assert_eq!(
graft_roots
.get_or_read::<mmr::Family, 32>(chunk, &NoReads)
.unwrap(),
digest
);
}
}
type GraftRootsDb<F> = fixed::Db<
F,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
32,
commonware_parallel::Sequential,
>;
async fn graft_roots_lifecycle<F: Graftable>(ctx: deterministic::Context) {
fn retained_start<F: Graftable>(db: &GraftRootsDb<F>) -> usize {
let floor_chunk = (*db.any.inactivity_floor_loc / 256) as usize;
floor_chunk.min(db.graft_roots.end())
}
let config = fixed_config::<OneCap>("graft-roots");
let mut db = GraftRootsDb::<F>::init(ctx.child("initial"), config.clone(), None)
.await
.unwrap();
assert_eq!(db.graft_roots.end(), 0);
for round in 0u64..4 {
let mut batch = db.new_batch();
for idx in 0u64..1024 {
batch = batch.write(
Sha256::hash(&[&idx.to_be_bytes()]),
Some(Sha256::hash(&[&round.to_be_bytes()])),
);
}
let parent = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let child = parent
.new_batch::<Sha256>()
.write(
Sha256::hash(&[b"child"]),
Some(Sha256::hash(&[&round.to_be_bytes()])),
)
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
drop(parent);
db = db.apply_batch(child).await.unwrap();
db = db.commit().await.unwrap();
assert_eq!(db.graft_roots.end(), *db.grafted_tree.leaves() as usize);
assert_eq!(db.graft_roots.start, retained_start(&db));
for chunk in db.graft_roots.start..db.graft_roots.end() {
let pos = graft_root_position::<F, 32>(chunk);
assert_eq!(
db.graft_roots.get(chunk),
db.any.log.merkle.get_node(pos).await.unwrap()
);
}
}
assert!(
db.graft_roots.start > db.any.bitmap.pruned_chunks(),
"the advancing floor must release roots before any prune"
);
let (start, roots, root) = (
db.graft_roots.start,
db.graft_roots.roots.clone(),
db.root(),
);
drop(db);
let mut db = GraftRootsDb::<F>::init(ctx.child("unpruned"), config.clone(), None)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.graft_roots.start, start);
assert_eq!(db.graft_roots.roots, roots);
let boundary = db.sync_boundary();
assert!(*boundary >= 256);
db = db.prune(boundary).await.unwrap();
assert_eq!(db.graft_roots.start, start);
assert_eq!(db.graft_roots.roots, roots);
assert!(db.graft_roots.start >= db.any.bitmap.pruned_chunks());
let root = db.root();
drop(db);
let db = GraftRootsDb::<F>::init(ctx.child("reopened"), config, None)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.graft_roots.start, start);
assert_eq!(db.graft_roots.roots, roots);
db.destroy().await.unwrap();
}
#[test]
fn test_graft_roots_lifecycle_mmr() {
deterministic::Runner::default().start(graft_roots_lifecycle::<mmr::Family>);
}
#[test]
fn test_graft_roots_lifecycle_mmb() {
deterministic::Runner::default().start(graft_roots_lifecycle::<mmb::Family>);
}
async fn graft_roots_from_unapplied_parent<F: Graftable>(ctx: deterministic::Context) {
let config = fixed_config::<OneCap>("graft-roots-unapplied-parent");
let db = GraftRootsDb::<F>::init(ctx.child("initial"), config.clone(), None)
.await
.unwrap();
let keys: Vec<_> = (0u64..1024)
.map(|idx| Sha256::hash(&[&idx.to_be_bytes()]))
.collect();
let mut parent = db.new_batch();
for &key in &keys {
parent = parent.write(key, Some(Sha256::hash(&[b"parent"])));
}
let parent = parent
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let mut child = parent.new_batch::<Sha256>();
for &key in &keys {
child = child.write(key, Some(Sha256::hash(&[b"child"])));
}
assert_eq!(db.graft_roots.end(), 0);
let child = child.merkleize(&db, None, &mut Proportional).await.unwrap();
let root = child.root();
let db = db.apply_batch(parent).await.unwrap();
assert!(
db.graft_roots.end() > 0,
"the parent must make chunks graftable"
);
let db = db.apply_batch(child).await.unwrap();
let db = db.commit().await.unwrap();
assert_eq!(db.root(), root);
drop(db);
let db = GraftRootsDb::<F>::init(ctx.child("reopened"), config, None)
.await
.unwrap();
assert_eq!(db.root(), root);
db.destroy().await.unwrap();
}
#[test]
fn test_graft_roots_from_unapplied_parent_mmr() {
deterministic::Runner::default().start(graft_roots_from_unapplied_parent::<mmr::Family>);
}
#[test]
fn test_graft_roots_from_unapplied_parent_mmb() {
deterministic::Runner::default().start(graft_roots_from_unapplied_parent::<mmb::Family>);
}
#[test_traced]
fn test_operations_match_applied_range() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let db = MmrDb::init(
ctx.child("db"),
fixed_config::<OneCap>("operations-match-applied-range"),
None,
)
.await
.unwrap();
let db = populate_fixed_db::<mmr::Family, _>(db, 0, 8).await;
let mut batch = db.new_batch();
for idx in 0..4u64 {
let key = Sha256::hash(&[&idx.to_be_bytes()]);
let value = Sha256::hash(&[&(idx + 100).to_be_bytes()]);
batch = batch.write(key, Some(value));
}
let merkleized = batch.merkleize(&db, None, &mut Proportional).await.unwrap();
let (start, ops) = merkleized.operations();
let applied_start = db.bounds().end;
let db = db.apply_batch(merkleized).await.unwrap();
let range = applied_start..db.bounds().end;
assert_eq!(start, range.start);
assert_eq!(*start + ops.len() as u64, *range.end);
db.destroy().await.unwrap();
});
}
#[test_traced]
fn test_start_sync_recovery() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let db = MmrDb::init(
ctx.child("first"),
fixed_config::<OneCap>("start-sync-recovery"),
None,
)
.await
.unwrap();
let key = Sha256::hash(&[&0u64.to_be_bytes()]);
let value = Sha256::hash(&[&1u64.to_be_bytes()]);
let merkleized = db
.new_batch()
.write(key, Some(value))
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
let db = db.apply_batch(merkleized).await.unwrap();
let (db, handle) = db.start_sync().await.unwrap();
handle.await.unwrap();
let root = db.root();
drop(db);
let db = MmrDb::init(
ctx.child("second"),
fixed_config::<OneCap>("start-sync-recovery"),
None,
)
.await
.unwrap();
assert_eq!(db.root(), root);
assert_eq!(db.get(&key).await.unwrap(), Some(value));
db.destroy().await.unwrap();
});
}
async fn prune_dropped_before_log_prune(ctx: deterministic::Context, fixed: bool) {
let partition = if fixed {
"prune-park-fixed"
} else {
"prune-park"
};
let db = MmrDb::init(
ctx.child("storage"),
fixed_config::<OneCap>(partition),
None,
)
.await
.unwrap();
let db = populate_fixed_db::<mmr::Family, _>(db, 0, 512).await;
let durable_floor = db.inactivity_floor_loc();
let keys: Vec<_> = (0..512u64)
.map(|idx| Sha256::hash(&[&idx.to_be_bytes()]))
.collect();
let (merkleized, expected) = if fixed {
let replacement = Sha256::hash(&[b"fixed-prune-value"]);
let value = |key: &sha256::Digest| (key.as_ref()[0] % 2 == 0).then_some(replacement);
let mut policy = Script::new(usize::MAX, u64::MAX, |key: &sha256::Digest| {
value(key).map_or(Choice::Evict, Choice::Replace)
});
let merkleized = db
.new_batch()
.merkleize(&db, None, &mut policy)
.await
.unwrap();
let mut visited: Vec<_> = policy.visited.iter().map(|(_, key, _)| *key).collect();
visited.sort();
let mut sorted = keys.clone();
sorted.sort();
assert_eq!(visited, sorted);
let expected: Vec<_> = keys.iter().map(|key| (*key, value(key))).collect();
(merkleized, expected)
} else {
let expected: Vec<_> = (0..512u64)
.zip(&keys)
.map(|(idx, key)| (*key, Some(Sha256::hash(&[&(idx + 1024).to_be_bytes()]))))
.collect();
let merkleized = expected
.iter()
.fold(db.new_batch(), |batch, (key, value)| {
batch.write(*key, *value)
})
.merkleize(&db, None, &mut Proportional)
.await
.unwrap();
(merkleized, expected)
};
assert!(expected.iter().any(|(_, value)| value.is_some()));
let mut db = db.apply_batch(merkleized).await.unwrap();
assert!(db.sync_boundary() > durable_floor);
for (key, value) in &expected {
assert_eq!(db.get(key).await.unwrap(), *value);
}
db.assert_exact().await;
let bounds = db.bounds();
let floor = db.inactivity_floor_loc();
let root = db.root();
db.halt_before_prune_log = true;
let boundary = db.sync_boundary();
{
let fut = db.prune(boundary);
futures::pin_mut!(fut);
assert!(
futures::poll!(fut.as_mut()).is_pending(),
"prune must park before the log prune"
);
}
let db = MmrDb::init(ctx.child("reopen"), fixed_config::<OneCap>(partition), None)
.await
.expect("prune crash must leave the db recoverable");
assert_eq!(db.bounds(), bounds);
assert_eq!(db.inactivity_floor_loc(), floor);
assert_eq!(db.root(), root);
assert!(db.any.bitmap.pruned_bits() > *durable_floor);
for (key, value) in &expected {
assert_eq!(db.get(key).await.unwrap(), *value);
}
db.assert_exact().await;
db.destroy().await.unwrap();
}
#[test_traced]
fn test_current_prune_dropped_before_log_prune() {
deterministic::Runner::default()
.start(|ctx| async move { prune_dropped_before_log_prune(ctx, false).await });
}
#[test_traced]
fn test_current_fixed_policy_prune_dropped_before_log_prune() {
deterministic::Runner::default()
.start(|ctx| async move { prune_dropped_before_log_prune(ctx, true).await });
}
#[test_traced]
fn test_ops_root_witness_verifies_without_partial_chunk() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let mut db = MmrDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("ops-root-witness-full"),
None,
)
.await
.unwrap();
let mut next_idx = 0;
db = populate_fixed_db::<mmr::Family, _>(db, next_idx, 256).await;
next_idx += 256;
while partial_chunk::<_, 32>(db.any.bitmap.as_ref()).is_some() {
db = populate_fixed_db::<mmr::Family, _>(db, next_idx, 1).await;
next_idx += 1;
}
let witness = db.ops_root_witness().await.unwrap();
let ops_root = db.ops_root();
let canonical_root = db.root();
assert!(witness.partial_chunk.is_none());
assert!(witness.verify::<Sha256>(&ops_root, &canonical_root));
let wrong_ops_root = Sha256::hash(&[b"wrong ops root"]);
assert!(!witness.verify::<Sha256>(&wrong_ops_root, &canonical_root));
let wrong_canonical_root = Sha256::hash(&[b"wrong canonical root"]);
assert!(!witness.verify::<Sha256>(&ops_root, &wrong_canonical_root));
let mut tampered = witness;
tampered.grafted_root = Sha256::hash(&[b"wrong grafted root"]);
assert!(!tampered.verify::<Sha256>(&ops_root, &canonical_root));
});
}
#[test_traced]
fn test_ops_root_witness_verifies_with_partial_chunk() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let db = MmbDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("ops-root-witness-partial"),
None,
)
.await
.unwrap();
let db = populate_fixed_db::<mmb::Family, _>(db, 0, 260).await;
let witness = db.ops_root_witness().await.unwrap();
let ops_root = db.ops_root();
let canonical_root = db.root();
assert!(witness.partial_chunk.is_some());
assert!(witness.verify::<Sha256>(&ops_root, &canonical_root));
let wrong_ops_root = Sha256::hash(&[b"wrong ops root"]);
assert!(!witness.verify::<Sha256>(&wrong_ops_root, &canonical_root));
let wrong_canonical_root = Sha256::hash(&[b"wrong canonical root"]);
assert!(!witness.verify::<Sha256>(&ops_root, &wrong_canonical_root));
let mut tampered = witness.clone();
tampered.grafted_root = Sha256::hash(&[b"wrong grafted root"]);
assert!(!tampered.verify::<Sha256>(&ops_root, &canonical_root));
let mut tampered = witness.clone();
tampered.partial_chunk.as_mut().unwrap().0 += 1;
assert!(!tampered.verify::<Sha256>(&ops_root, &canonical_root));
let mut tampered = witness;
tampered.partial_chunk.as_mut().unwrap().1 = Sha256::hash(&[b"wrong partial chunk"]);
assert!(!tampered.verify::<Sha256>(&ops_root, &canonical_root));
});
}
#[test_traced]
fn test_ops_root_witness_verifies_with_pruned_db() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let mut db = MmrDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("ops-root-witness-pruned"),
None,
)
.await
.unwrap();
for _ in 0..5 {
db = populate_fixed_db::<mmr::Family, _>(db, 0, 512).await;
}
let boundary = db.sync_boundary();
let db = db.prune(boundary).await.unwrap();
assert!(
db.any.bitmap.pruned_chunks() > 0,
"test requires at least one pruned chunk to exercise the zero-chunk path"
);
let witness = db.ops_root_witness().await.unwrap();
let ops_root = db.ops_root();
let canonical_root = db.root();
assert!(witness.verify::<Sha256>(&ops_root, &canonical_root));
let wrong_canonical_root = Sha256::hash(&[b"wrong canonical root"]);
assert!(!witness.verify::<Sha256>(&ops_root, &wrong_canonical_root));
let mut tampered = witness;
tampered.grafted_root = Sha256::hash(&[b"wrong grafted root"]);
assert!(!tampered.verify::<Sha256>(&ops_root, &canonical_root));
});
}
#[test_traced]
fn test_ops_root_witness_verifies_on_fresh_db() {
let executor = deterministic::Runner::default();
executor.start(|ctx| async move {
let db = MmrDb::init(
ctx.child("storage"),
fixed_config::<OneCap>("ops-root-witness-fresh"),
None,
)
.await
.unwrap();
let witness = db.ops_root_witness().await.unwrap();
let ops_root = db.ops_root();
let canonical_root = db.root();
assert!(witness.verify::<Sha256>(&ops_root, &canonical_root));
});
}
}