use crate::{
index::{Ordered as OrderedIndex, Unordered as UnorderedIndex},
journal::{
authenticated,
contiguous::{Contiguous, Mutable},
},
merkle::{Family, Location},
qmdb::{
any::{
db::Db,
operation::{update, Operation},
ordered::{find_next_key, find_next_key_ascending, find_prev_key},
ValueEncoding,
},
batch_chain::{self, Bounds},
bitmap::Shared,
delete_known_loc,
operation::{Key, Operation as OperationTrait},
update_known_loc,
},
Context,
};
use ahash::{AHashMap, AHashSet};
use commonware_codec::{Codec, CodecShared};
use commonware_cryptography::{Digest, Hasher};
use commonware_parallel::Strategy;
use commonware_utils::bitmap;
use core::{cmp::Ordering, ops::Range};
use std::{
collections::BTreeMap,
iter, mem,
sync::{Arc, Weak},
};
use tracing::debug;
type DiffVec<K, F, V> = Vec<(K, DiffEntry<F, V>)>;
type DiffSlice<K, F, V> = [(K, DiffEntry<F, V>)];
type CandidateChunk<'a, F, U> = (&'a [Location<F>], &'a [Operation<F, U>]);
pub(crate) struct PrefetchedCandidates<F: Family, U: update::Update + Send + Sync>
where
Operation<F, U>: Codec,
{
locs: Vec<Location<F>>,
shards: Vec<Vec<Operation<F, U>>>,
next_scan: Location<F>,
}
type PrevCandidates<K, F, V> = Vec<(K, (Option<V>, Location<F>))>;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum StagedLoc<F: Family> {
Committed(Location<F>),
Ancestor {
loc: Location<F>,
base_old_loc: Option<Location<F>>,
},
}
impl<F: Family> StagedLoc<F> {
const fn loc(&self) -> Location<F> {
match self {
Self::Committed(loc) | Self::Ancestor { loc, .. } => *loc,
}
}
}
type StagedUpdate<F, U> = (
<U as update::Update>::Key,
StagedLoc<F>,
<U as update::Update>::Cached,
Option<<U as update::Update>::Value>,
);
type PendingRead<'a, K> = (usize, &'a K);
type UncommittedReadResolution<'a, K, V> = (Vec<Option<V>>, Vec<PendingRead<'a, K>>);
#[derive(Clone)]
pub(crate) enum DiffEntry<F: Family, V> {
Active {
value: V,
loc: Location<F>,
base_old_loc: Option<Location<F>>,
},
Deleted {
base_old_loc: Option<Location<F>>,
},
}
impl<F: Family, V> DiffEntry<F, V> {
pub(crate) const fn base_old_loc(&self) -> Option<Location<F>> {
match self {
Self::Active { base_old_loc, .. } | Self::Deleted { base_old_loc } => *base_old_loc,
}
}
pub(crate) const fn loc(&self) -> Option<Location<F>> {
match self {
Self::Active { loc, .. } => Some(*loc),
Self::Deleted { .. } => None,
}
}
pub(crate) const fn value(&self) -> Option<&V> {
match self {
Self::Active { value, .. } => Some(value),
Self::Deleted { .. } => None,
}
}
}
pub(crate) fn lookup_sorted<'a, K: Ord, V>(entries: &'a [(K, V)], key: &K) -> Option<&'a V> {
entries
.binary_search_by(|(candidate, _)| candidate.cmp(key))
.ok()
.map(|idx| &entries[idx].1)
}
fn sorted_contains<T: Ord>(items: &[T], cursor: &mut usize, target: &T) -> bool {
while items.get(*cursor).is_some_and(|item| item < target) {
*cursor += 1;
}
items.get(*cursor) == Some(target)
}
fn merge_sorted_diffs<K: Ord, F: Family, V>(
a: DiffVec<K, F, V>,
b: DiffVec<K, F, V>,
) -> DiffVec<K, F, V> {
let mut merged = Vec::with_capacity(a.len() + b.len());
let mut a = a.into_iter().peekable();
let mut b = b.into_iter().peekable();
while let (Some(x), Some(y)) = (a.peek(), b.peek()) {
if x.0 < y.0 {
merged.push(a.next().expect("peeked"));
} else {
merged.push(b.next().expect("peeked"));
}
}
merged.extend(a);
merged.extend(b);
merged
}
enum Base<F: Family, D: Digest, U: update::Update + Send + Sync, S: Strategy>
where
Operation<F, U>: Send + Sync,
{
Db {
db_size: u64,
inactivity_floor_loc: Location<F>,
active_keys: usize,
},
Child(Arc<MerkleizedBatch<F, D, U, S>>),
}
impl<F: Family, D: Digest, U: update::Update + Send + Sync, S: Strategy> Base<F, D, U, S>
where
Operation<F, U>: Send + Sync,
{
fn base_size(&self) -> u64 {
match self {
Self::Db { db_size, .. } => *db_size,
Self::Child(parent) => parent.bounds.total_size,
}
}
fn db_size(&self) -> u64 {
match self {
Self::Db { db_size, .. } => *db_size,
Self::Child(parent) => parent.bounds.db_size,
}
}
fn inactivity_floor_loc(&self) -> Location<F> {
match self {
Self::Db {
inactivity_floor_loc,
..
} => *inactivity_floor_loc,
Self::Child(parent) => parent.bounds.inactivity_floor,
}
}
fn active_keys(&self) -> usize {
match self {
Self::Db { active_keys, .. } => *active_keys,
Self::Child(parent) => parent.total_active_keys,
}
}
const fn parent(&self) -> Option<&Arc<MerkleizedBatch<F, D, U, S>>> {
match self {
Self::Db { .. } => None,
Self::Child(parent) => Some(parent),
}
}
}
pub struct UnmerkleizedBatch<F: Family, H, U, S: Strategy>
where
U: update::Update + Send + Sync,
H: Hasher,
Operation<F, U>: Codec,
{
journal_batch: authenticated::UnmerkleizedBatch<F, H, Operation<F, U>, S>,
mutations: BTreeMap<U::Key, Option<U::Value>>,
base: Base<F, H::Digest, U, S>,
}
pub(crate) type StagedUpdates<F, U> = Vec<StagedUpdate<F, U>>;
type StagedResolution<F, U> = Option<(StagedLoc<F>, <U as update::Update>::Cached)>;
pub struct Staged<F: Family, H, U, S: Strategy>
where
U: update::Update + Send + Sync,
H: Hasher,
Operation<F, U>: Codec,
{
batch: UnmerkleizedBatch<F, H, U, S>,
keys: StagedKeys<U::Key>,
resolutions: Vec<StagedResolution<F, U>>,
}
struct StagedKeys<K> {
keys: Vec<K>,
slots: Vec<usize>,
ids: AHashMap<K, usize>,
}
impl<K: Clone + Eq + core::hash::Hash> StagedKeys<K> {
fn new(keys: Vec<K>) -> Self {
let mut staged = Self {
keys: Vec::new(),
slots: Vec::new(),
ids: AHashMap::with_capacity(keys.len()),
};
staged.append(keys);
staged
}
fn append(&mut self, mut keys: Vec<K>) {
self.slots.reserve(keys.len());
for key in &keys {
let next = self.ids.len();
let id = *self.ids.entry(key.clone()).or_insert(next);
self.slots.push(id);
}
self.keys.append(&mut keys);
}
const fn len(&self) -> usize {
self.keys.len()
}
fn key(&self, slot: usize) -> &K {
&self.keys[slot]
}
fn id(&self, slot: usize) -> usize {
self.slots[slot]
}
fn distinct(&self) -> usize {
self.ids.len()
}
}
#[allow(clippy::type_complexity)]
#[derive(Clone)]
pub struct MerkleizedBatch<F: Family, D: Digest, U: update::Update + Send + Sync, S: Strategy>
where
Operation<F, U>: Send + Sync,
{
pub(crate) journal_batch: Arc<authenticated::MerkleizedBatch<F, D, Operation<F, U>, S>>,
pub(crate) root: D,
pub(crate) diff: Arc<DiffVec<U::Key, F, U::Value>>,
parent: Option<Weak<Self>>,
pub(crate) total_active_keys: usize,
pub(crate) ancestor_diffs: Vec<Arc<DiffVec<U::Key, F, U::Value>>>,
pub(crate) bounds: batch_chain::Bounds<F>,
}
type AncestorBatch<F, D, U, S> = Arc<MerkleizedBatch<F, D, U, S>>;
struct Merkleizer<F: Family, H, U, S: Strategy>
where
U: update::Update + Send + Sync,
H: Hasher,
Operation<F, U>: Codec,
{
journal_batch: authenticated::UnmerkleizedBatch<F, H, Operation<F, U>, S>,
ancestors: Vec<AncestorBatch<F, H::Digest, U, S>>,
base_size: u64,
db_size: u64,
base_inactivity_floor_loc: Location<F>,
base_active_keys: usize,
}
fn resolve_in_ancestors<'a, F: Family, D: Digest, U: update::Update + Send + Sync, S: Strategy>(
ancestors: &'a [Arc<MerkleizedBatch<F, D, U, S>>],
key: &U::Key,
) -> Option<&'a DiffEntry<F, U::Value>>
where
Operation<F, U>: Send + Sync,
{
for batch in ancestors {
if let Some(entry) = lookup_sorted(batch.diff.as_slice(), key) {
return Some(entry);
}
}
None
}
enum FloorOutcome<F: Family> {
Inactive,
MoveExisting {
idx: usize,
base_old_loc: Option<Location<F>>,
},
MoveNew { base_old_loc: Option<Location<F>> },
}
pub(crate) struct DiffCursors<'a, K, F: Family, V> {
diffs: Vec<(&'a DiffSlice<K, F, V>, usize)>,
}
impl<'a, K: Ord, F: Family, V> DiffCursors<'a, K, F, V> {
pub(crate) fn new(diffs: impl IntoIterator<Item = &'a DiffSlice<K, F, V>>) -> Self {
Self {
diffs: diffs.into_iter().map(|diff| (diff, 0)).collect(),
}
}
pub(crate) fn resolve(&mut self, key: &K) -> Option<&'a DiffEntry<F, V>> {
for (diff, cursor) in &mut self.diffs {
assert!(
*cursor == 0 || diff[*cursor - 1].0 < *key,
"queries must be non-decreasing"
);
while *cursor < diff.len() && diff[*cursor].0 < *key {
*cursor += 1;
}
if let Some((k, entry)) = diff.get(*cursor) {
if k == key {
return Some(entry);
}
}
}
None
}
}
fn resolve_pending_from_diffs<'a, K, F: Family, V: Clone + Send + Sync + 'a, S: Strategy>(
pending: &[PendingRead<'a, K>],
diffs: &[&'a DiffSlice<K, F, V>],
strategy: &S,
resolved: &mut [bool],
results: &mut [Option<V>],
mut on_hit: impl FnMut(usize, &DiffEntry<F, V>),
) where
K: Ord + Sync,
{
if pending.is_empty() || diffs.is_empty() {
return;
}
let resolve = |chunk: &[PendingRead<'a, K>]| -> Vec<(usize, &'a DiffEntry<F, V>)> {
chunk
.iter()
.filter_map(|(slot, key)| {
diffs
.iter()
.find_map(|diff| lookup_sorted(diff, key))
.map(|entry| (*slot, entry))
})
.collect()
};
let hits: Vec<(usize, &'a DiffEntry<F, V>)> = strategy.run(
pending.len(),
|| resolve(pending),
|| {
let manual = strategy.manual();
let chunk_len = pending.len().div_ceil(manual.parallelism());
let chunks: Vec<_> = pending.chunks(chunk_len).collect();
manual
.map_collect_vec(chunks, &resolve)
.into_iter()
.flatten()
.collect()
},
);
for (slot, entry) in hits {
resolved[slot] = true;
results[slot] = entry.value().cloned();
on_hit(slot, entry);
}
}
fn resolve_reads<'a, K, F: Family, V, S: Strategy>(
keys: &[&'a K],
local: impl Fn(&K) -> Option<Option<V>>,
diffs: &[&DiffSlice<K, F, V>],
strategy: &S,
on_diff_hit: impl FnMut(usize, &DiffEntry<F, V>),
) -> UncommittedReadResolution<'a, K, V>
where
K: Ord + Sync,
V: Clone + Send + Sync,
{
let mut results = vec![None; keys.len()];
let mut resolved = vec![false; keys.len()];
let mut pending = Vec::new();
for (i, key) in keys.iter().enumerate() {
if let Some(value) = local(key) {
results[i] = value;
resolved[i] = true;
} else {
pending.push((i, *key));
}
}
resolve_pending_from_diffs(
&pending,
diffs,
strategy,
&mut resolved,
&mut results,
on_diff_hit,
);
let unresolved = pending.into_iter().filter(|(i, _)| !resolved[*i]).collect();
(results, unresolved)
}
fn apply_diff<F: Family, V, I: UnorderedIndex<Value = Location<F>>, const N: usize>(
snapshot: &mut I,
bitmap: &mut bitmap::Prunable<N>,
key: &impl Key,
entry: &DiffEntry<F, V>,
base_old_loc: Option<Location<F>>,
) {
match entry {
DiffEntry::Active { loc, .. } => match base_old_loc {
Some(old) => update_known_loc::<F, _>(snapshot, key, old, *loc),
None => snapshot.insert(key, *loc),
},
DiffEntry::Deleted { .. } => {
if let Some(old) = base_old_loc {
delete_known_loc::<F, _>(snapshot, key, old);
}
}
}
if let Some(loc) = entry.loc() {
bitmap.set_bit(*loc, true);
}
if let Some(loc) = base_old_loc {
bitmap.set_bit(*loc, false);
}
}
struct DiffMerge<'a, K, F: Family, V> {
cursors: Vec<(&'a DiffSlice<K, F, V>, usize)>,
}
impl<'a, K: Ord, F: Family, V> DiffMerge<'a, K, F, V> {
fn new(streams: impl IntoIterator<Item = &'a DiffSlice<K, F, V>>) -> Self {
Self {
cursors: streams.into_iter().map(|s| (s, 0)).collect(),
}
}
fn peek_key(cursor: &(&'a DiffSlice<K, F, V>, usize)) -> Option<&'a K> {
cursor.0.get(cursor.1).map(|(k, _)| k)
}
fn next_general(&mut self) -> Option<(&'a K, &'a DiffEntry<F, V>)> {
let n = self.cursors.len();
let mut winner: Option<usize> = None;
for level in 0..n {
let Some(k) = Self::peek_key(&self.cursors[level]) else {
continue;
};
let better = match winner {
None => true,
Some(w) => *k < *Self::peek_key(&self.cursors[w]).unwrap(),
};
if better {
winner = Some(level);
}
}
let level = winner?;
let (slice, pos) = self.cursors[level];
let winning_key = &slice[pos].0;
for cursor in &mut self.cursors {
if Self::peek_key(cursor).is_some_and(|k| k == winning_key) {
cursor.1 += 1;
}
}
Some((&slice[pos].0, &slice[pos].1))
}
}
impl<'a, K: Ord, F: Family, V> Iterator for DiffMerge<'a, K, F, V> {
type Item = (&'a K, &'a DiffEntry<F, V>);
fn next(&mut self) -> Option<Self::Item> {
match self.cursors.len() {
0 => None,
1 => {
let (slice, pos) = &mut self.cursors[0];
let (k, entry) = slice.get(*pos)?;
*pos += 1;
Some((k, entry))
}
2 => {
let ka = Self::peek_key(&self.cursors[0]);
let kb = Self::peek_key(&self.cursors[1]);
let winner = match (ka, kb) {
(Some(a), Some(b)) => match a.cmp(b) {
Ordering::Less => 0,
Ordering::Greater => 1,
Ordering::Equal => {
self.cursors[1].1 += 1;
0
}
},
(Some(_), None) => 0,
(None, Some(_)) => 1,
(None, None) => return None,
};
let (slice, pos) = &mut self.cursors[winner];
let (k, entry) = &slice[*pos];
*pos += 1;
Some((k, entry))
}
_ => self.next_general(),
}
}
}
fn fill_candidates<F: Family, const N: usize>(
bitmap: &Shared<N>,
floor: Location<F>,
tip: u64,
limit: usize,
out: &mut Vec<Location<F>>,
) -> Location<F> {
let mut raw: Vec<u64> = Vec::with_capacity(limit);
let next = bitmap.fill_candidates(*floor, tip, limit, &mut raw);
out.extend(raw.into_iter().map(Location::new));
Location::new(next)
}
fn read_op_from_ancestors<F: Family, D: Digest, U: update::Update + Send + Sync, S: Strategy>(
ancestors: &[Arc<MerkleizedBatch<F, D, U, S>>],
loc: u64,
db_size: u64,
) -> &Operation<F, U>
where
Operation<F, U>: Send + Sync,
{
for (i, batch) in ancestors.iter().enumerate() {
let batch_base = ancestors
.get(i + 1)
.map_or(db_size, |b| b.journal_batch.size());
let batch_end = batch.journal_batch.size();
if loc >= batch_base && loc < batch_end {
return &batch.journal_batch.items()[(loc - batch_base) as usize];
}
}
unreachable!("location {loc} not found in ancestor chain (db_size={db_size})")
}
impl<F: Family, H, U, S: Strategy> Merkleizer<F, H, U, S>
where
U: update::Update + Send + Sync,
H: Hasher,
Operation<F, U>: Codec,
{
fn try_read_op_from_uncommitted(
&self,
loc: Location<F>,
batch_ops: &[Operation<F, U>],
) -> Option<Operation<F, U>> {
let loc = *loc;
if loc >= self.base_size {
return Some(batch_ops[(loc - self.base_size) as usize].clone());
}
if loc >= self.db_size {
return Some(read_op_from_ancestors(&self.ancestors, loc, self.db_size).clone());
}
None
}
fn all_committed_ascending(&self, locations: &[Location<F>]) -> bool {
locations.is_sorted_by(|a, b| a < b)
&& locations.last().is_some_and(|last| **last < self.db_size)
}
async fn read_ops<R: Contiguous<Item = Operation<F, U>>>(
&self,
locations: &[Location<F>],
batch_ops: &[Operation<F, U>],
reader: &R,
) -> Result<Vec<Operation<F, U>>, crate::qmdb::Error<F>> {
if self.all_committed_ascending(locations) {
let positions: Vec<u64> = locations.iter().map(|loc| **loc).collect();
return Ok(reader.read_many(&positions).await?);
}
let mut results: Vec<Option<Operation<F, U>>> = locations
.iter()
.map(|loc| self.try_read_op_from_uncommitted(*loc, batch_ops))
.collect();
let committed: Vec<(usize, u64)> = locations
.iter()
.zip(results.iter())
.enumerate()
.filter_map(|(idx, (loc, resolved))| resolved.is_none().then_some((idx, **loc)))
.collect();
if committed.is_empty() {
return Ok(results.into_iter().map(Option::unwrap).collect());
}
let mut positions: Vec<u64> = committed.iter().map(|(_, loc)| *loc).collect();
let presorted = positions.is_sorted_by(|a, b| a < b);
if !presorted {
positions.sort_unstable();
positions.dedup();
}
let read = reader.read_many(&positions).await?;
for (idx, loc) in committed {
let result_idx = positions
.binary_search(&loc)
.expect("read result missing for requested location");
results[idx] = Some(read[result_idx].clone());
}
Ok(results
.into_iter()
.map(|r| r.expect("operation should be resolved"))
.collect())
}
async fn read_ops_sharded<E, C>(
&self,
locations: &[Location<F>],
batch_ops: &[Operation<F, U>],
reader: &authenticated::Journal<F, E, C, H, S>,
) -> Result<Vec<Vec<Operation<F, U>>>, crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
Operation<F, U>: CodecShared,
{
if self.all_committed_ascending(locations) {
let positions: Vec<u64> = locations.iter().map(|loc| **loc).collect();
return Ok(reader.read_many_sharded(&positions).await?);
}
Ok(vec![self.read_ops(locations, batch_ops, reader).await?])
}
fn gather_existing_locations<E, C, I, const N: usize>(
&self,
mutations: &BTreeMap<U::Key, Option<U::Value>>,
db: &Db<F, E, C, I, H, U, N, S>,
include_active_collision_siblings: bool,
) -> Vec<Location<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
{
let mut locations = Vec::with_capacity(mutations.len() * 3 / 2);
if self.ancestors.is_empty() {
for key in mutations.keys() {
locations.extend(db.snapshot.get(key).copied());
}
} else {
let mut ancestors = DiffCursors::new(self.ancestors.iter().map(|a| a.diff.as_slice()));
for key in mutations.keys() {
match ancestors.resolve(key) {
Some(DiffEntry::Deleted { .. }) => {
}
Some(DiffEntry::Active {
loc, base_old_loc, ..
}) => {
locations.push(*loc);
if include_active_collision_siblings {
locations.extend(
db.snapshot
.get(key)
.copied()
.filter(move |loc| Some(*loc) != *base_old_loc),
);
}
}
None => {
locations.extend(db.snapshot.get(key).copied());
}
}
}
}
db.strategy().sort_by(&mut locations, |a, b| a.cmp(b));
locations.dedup();
locations
}
#[allow(clippy::type_complexity)]
fn extract_parent_deleted_creates(
&self,
mutations: &mut BTreeMap<U::Key, Option<U::Value>>,
) -> Vec<(U::Key, U::Value, Option<Location<F>>)> {
if self.ancestors.is_empty() {
return Vec::new();
}
let mut ancestors = DiffCursors::new(self.ancestors.iter().map(|a| a.diff.as_slice()));
let mut creates = Vec::new();
mutations.retain(|key, value| {
if let Some(DiffEntry::Deleted { base_old_loc }) = ancestors.resolve(key) {
if let Some(v) = value.take() {
creates.push((key.clone(), v, *base_old_loc));
return false;
}
}
true
});
creates
}
#[allow(clippy::too_many_arguments)]
async fn finish<E, C, I, const N: usize>(
self,
mut ops: Vec<Operation<F, U>>,
mut diff: DiffVec<U::Key, F, U::Value>,
mut superseded_locs: Vec<Location<F>>,
active_keys_delta: isize,
user_steps: u64,
metadata: Option<U::Value>,
mut prefetched: Option<PrefetchedCandidates<F, U>>,
mut fill_candidates: impl FnMut(Location<F>, u64, usize, &mut Vec<Location<F>>) -> Location<F>,
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<Arc<MerkleizedBatch<F, H::Digest, U, S>>, crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
{
let total_steps = user_steps + 1;
let total_active_keys = self.base_active_keys as isize + active_keys_delta;
let mut floor = self.base_inactivity_floor_loc;
let mut diff_sort = None;
if !diff.is_empty() {
let unsorted = mem::take(&mut diff);
diff_sort = Some(db.strategy().spawn(move |strategy| {
let mut diff = unsorted;
strategy.sort_by(&mut diff, |a, b| a.0.cmp(&b.0));
diff
}));
}
let mut floor_diff = Vec::new();
if total_active_keys > 0 {
let strategy = db.strategy();
let fixed_tip = self.base_size + ops.len() as u64;
let mut moved = 0u64;
let mut scan_from = floor;
floor_diff.reserve(total_steps as usize);
if !superseded_locs.is_sorted_by(|a, b| a < b) {
strategy.sort_by(&mut superseded_locs, |a, b| a.cmp(b));
superseded_locs.dedup();
}
ops.reserve(total_steps as usize + 1);
let mut superseded_cursor = 0;
while moved < total_steps {
let limit = (total_steps - moved) as usize;
let (mut candidates, pf_shards) = match prefetched.take() {
Some(pf) => {
scan_from = pf.next_scan;
(pf.locs, pf.shards)
}
None => (Vec::with_capacity(limit), Vec::new()),
};
if candidates.len() < limit {
scan_from = fill_candidates(
scan_from,
fixed_tip,
limit - candidates.len(),
&mut candidates,
);
}
if candidates.is_empty() {
break;
}
assert!(candidates[0] >= floor);
assert!(candidates.is_sorted_by(|a, b| a < b));
let pf_count: usize = pf_shards.iter().map(Vec::len).sum();
assert!(pf_count <= candidates.len());
let mut read_candidates: Vec<Location<F>> = Vec::with_capacity(candidates.len());
read_candidates.extend_from_slice(&candidates[..pf_count]);
for candidate in &candidates[pf_count..] {
if !sorted_contains(&superseded_locs, &mut superseded_cursor, candidate) {
read_candidates.push(*candidate);
}
}
let (resolved, outcomes): (_, Vec<Vec<FloorOutcome<F>>>) =
if read_candidates.is_empty() {
(Vec::new(), Vec::new())
} else {
let live = &read_candidates[pf_count..];
let mut resolved = pf_shards;
if !live.is_empty() {
resolved.extend(self.read_ops_sharded(live, &ops, &db.log).await?);
}
if let Some(job) = diff_sort.take() {
diff = job.await;
}
let classify = |candidate: Location<F>, op: &Operation<F, U>| {
let Some(key) = op.key() else {
return FloorOutcome::Inactive; };
match diff.binary_search_by(|(k, _)| k.cmp(key)) {
Ok(idx) => {
let entry = &diff[idx].1;
if entry.loc() == Some(candidate) {
FloorOutcome::MoveExisting {
idx,
base_old_loc: entry.base_old_loc(),
}
} else {
FloorOutcome::Inactive
}
}
Err(_) => resolve_in_ancestors(&self.ancestors, key).map_or_else(
|| {
if db.snapshot.get(key).any(|&l| l == candidate) {
FloorOutcome::MoveNew {
base_old_loc: Some(candidate),
}
} else {
FloorOutcome::Inactive
}
},
|entry| {
if entry.loc() == Some(candidate) {
FloorOutcome::MoveNew {
base_old_loc: entry.base_old_loc(),
}
} else {
FloorOutcome::Inactive
}
},
),
}
};
let manual = strategy.manual();
let target = read_candidates
.len()
.div_ceil(manual.parallelism() * 4)
.max(1);
let mut chunks: Vec<CandidateChunk<'_, F, U>> = Vec::new();
let mut offset = 0;
for chunk in &resolved {
let locs = &read_candidates[offset..offset + chunk.len()];
offset += chunk.len();
chunks.extend(locs.chunks(target).zip(chunk.chunks(target)));
}
let outcomes = manual.map_collect_vec(chunks, |(chunk_locs, chunk_ops)| {
chunk_locs
.iter()
.zip(chunk_ops)
.map(|(loc, op)| classify(*loc, op))
.collect()
});
(resolved, outcomes)
};
let mut outcomes = outcomes.into_iter().flatten();
let mut reads = resolved.into_iter().flatten();
let mut pending = read_candidates.iter().peekable();
for candidate in candidates {
floor = Location::new(*candidate + 1);
if pending.next_if(|&&pending| pending == candidate).is_none() {
continue;
}
let op = reads.next().expect("one read per candidate");
let outcome = outcomes.next().expect("one outcome per read candidate");
match outcome {
FloorOutcome::Inactive => continue,
FloorOutcome::MoveExisting { idx, base_old_loc } => {
let new_loc = Location::new(self.base_size + ops.len() as u64);
let value = extract_update_value(&op);
ops.push(op);
diff[idx].1 = DiffEntry::Active {
value,
loc: new_loc,
base_old_loc,
};
}
FloorOutcome::MoveNew { base_old_loc } => {
let key = op.key().cloned().expect("moved op has a key");
let new_loc = Location::new(self.base_size + ops.len() as u64);
let value = extract_update_value(&op);
ops.push(op);
floor_diff.push((
key,
DiffEntry::Active {
value,
loc: new_loc,
base_old_loc,
},
));
}
}
moved += 1;
if moved >= total_steps {
break;
}
}
}
} else {
floor = Location::new(self.base_size + ops.len() as u64);
debug!(tip = ?floor, "db is empty, raising floor to tip");
}
if let Some(job) = diff_sort.take() {
diff = job.await;
}
let mut diff_merge = None;
if !floor_diff.is_empty() {
diff_merge = Some(db.strategy().spawn(move |strategy| {
let mut floor_diff = floor_diff;
strategy.sort_by(&mut floor_diff, |a, b| a.0.cmp(&b.0));
let diff = merge_sorted_diffs(diff, floor_diff);
assert!(diff.is_sorted_by(|a, b| a.0 < b.0));
diff
}));
diff = Vec::new();
}
let commit_loc = Location::<F>::new(self.base_size + ops.len() as u64);
ops.push(Operation::CommitFloor(metadata, floor));
let leaves = Location::new(self.base_size + ops.len() as u64);
let inactive_peaks = db.inactive_peaks(leaves, floor);
let (journal, root) = db
.log
.merkleize(self.journal_batch, ops, inactive_peaks)
.await?;
if let Some(job) = diff_merge.take() {
diff = job.await;
}
let ancestor_diffs: Vec<_> = self.ancestors.iter().map(|a| Arc::clone(&a.diff)).collect();
let ancestors: Vec<_> = self
.ancestors
.iter()
.map(|a| batch_chain::AncestorBounds {
floor: a.bounds.inactivity_floor,
end: a.bounds.total_size,
})
.collect();
assert!(total_active_keys >= 0, "active_keys underflow");
Ok(Arc::new(MerkleizedBatch {
journal_batch: journal,
root,
diff: Arc::new(diff),
parent: self.ancestors.first().map(Arc::downgrade),
total_active_keys: total_active_keys as usize,
ancestor_diffs,
bounds: batch_chain::Bounds {
base_size: self.base_size,
db_size: self.db_size,
total_size: *commit_loc + 1,
ancestors,
inactivity_floor: floor,
},
}))
}
}
impl<F: Family, H, U, S: Strategy> UnmerkleizedBatch<F, H, U, S>
where
U: update::Update + Send + Sync,
H: Hasher,
Operation<F, U>: Codec,
{
pub fn write(mut self, key: U::Key, value: Option<U::Value>) -> Self {
self.mutations.insert(key, value);
self
}
#[allow(clippy::type_complexity)]
fn into_parts(self) -> (BTreeMap<U::Key, Option<U::Value>>, Merkleizer<F, H, U, S>) {
let ancestors: Vec<_> = self.base.parent().map_or_else(Vec::new, |parent| {
let mut v = vec![Arc::clone(parent)];
v.extend(parent.ancestors());
v
});
let db_size = self.base.db_size();
let effective_db_size = ancestors.last().map_or(db_size, |oldest| {
let oldest_base =
oldest.journal_batch.size() - oldest.journal_batch.items().len() as u64;
db_size.max(oldest_base)
});
let m = Merkleizer {
journal_batch: self.journal_batch,
ancestors,
base_size: self.base.base_size(),
db_size: effective_db_size,
base_inactivity_floor_loc: self.base.inactivity_floor_loc(),
base_active_keys: self.base.active_keys(),
};
(self.mutations, m)
}
}
impl<F: Family, H, U, S: Strategy> Staged<F, H, U, S>
where
U: update::Update + Send + Sync,
H: Hasher,
Operation<F, U>: Codec,
{
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.any.batch.expand",
level = "info",
skip_all,
fields(keys = keys.len() as u64, staged = self.keys.len() as u64),
)]
pub async fn expand<E, C, I, const N: usize>(
mut self,
keys: &[&U::Key],
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<(Range<usize>, Vec<Option<U::Value>>, Self), crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
{
let start = self.keys.len();
let end = start
.checked_add(keys.len())
.expect("staged read index overflow");
let (values, keys, mut resolutions) = self.batch.stage_reads(keys, db).await?;
self.keys.append(keys);
self.resolutions.append(&mut resolutions);
Ok((start..end, values, self))
}
fn apply_upserts(
mut batch: UnmerkleizedBatch<F, H, U, S>,
upserts: Vec<(U::Key, Option<U::Value>)>,
) -> UnmerkleizedBatch<F, H, U, S> {
for (key, value) in upserts {
batch = batch.write(key, value);
}
batch
}
pub(crate) fn resolve_updates(
self,
updates: Vec<(usize, Option<U::Value>)>,
upserts: Vec<(U::Key, Option<U::Value>)>,
strategy: &S,
) -> (UnmerkleizedBatch<F, H, U, S>, StagedUpdates<F, U>) {
let Self {
mut batch,
keys,
mut resolutions,
} = self;
let mut staged_updates = StagedUpdates::<F, U>::new();
if updates.is_empty() {
return (Self::apply_upserts(batch, upserts), staged_updates);
}
let upsert_keys: AHashSet<&U::Key> = upserts.iter().map(|(key, _)| key).collect();
let mut winners: Vec<Option<(usize, Option<U::Value>)>> = vec![None; keys.distinct()];
let mut touched: Vec<usize> = Vec::with_capacity(updates.len());
for (slot, value) in updates {
assert!(slot < keys.len(), "update index out of staged read range");
if !upsert_keys.is_empty() && upsert_keys.contains(keys.key(slot)) {
continue;
}
let id = keys.id(slot);
if winners[id].is_none() {
touched.push(id);
}
winners[id] = Some((slot, value));
}
let had_mutations = !batch.mutations.is_empty();
let mut order: Vec<(Location<F>, usize)> = Vec::with_capacity(touched.len());
for &id in &touched {
let winner = &mut winners[id];
let Some((slot, value)) = winner else {
unreachable!("touched ids hold a winner");
};
let key = keys.key(*slot);
match &resolutions[*slot] {
Some((sloc, _)) if value.is_some() || U::STAGES_DELETES => {
if had_mutations {
batch.mutations.remove(key);
}
order.push((sloc.loc(), *slot));
}
_ => {
let (_, value) = winner.take().expect("winner checked above");
batch.mutations.insert(key.clone(), value);
}
}
}
strategy.sort_by(&mut order, |a, b| a.0.cmp(&b.0));
staged_updates = order
.iter()
.map(|&(_, slot)| {
let (_, value) = winners[keys.id(slot)]
.take()
.expect("winner recorded for staged slot");
let (sloc, payload) = resolutions[slot].take().expect("resolution checked above");
(keys.key(slot).clone(), sloc, payload, value)
})
.collect();
(Self::apply_upserts(batch, upserts), staged_updates)
}
}
impl<F: Family, K, V, H, S: Strategy> Staged<F, H, update::Unordered<K, V>, S>
where
K: Key,
V: ValueEncoding,
H: Hasher,
Operation<F, update::Unordered<K, V>>: Codec,
{
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.any.unordered.batch.merkleize.staged",
level = "info",
skip_all,
fields(updates = updates.len() as u64, upserts = upserts.len() as u64),
)]
pub async fn merkleize<E, C, I, const N: usize>(
self,
updates: Vec<(usize, Option<V::Value>)>,
upserts: Vec<(K, Option<V::Value>)>,
metadata: Option<V::Value>,
db: &Db<F, E, C, I, H, update::Unordered<K, V>, N, S>,
) -> Result<Arc<MerkleizedBatch<F, H::Digest, update::Unordered<K, V>, S>>, crate::qmdb::Error<F>>
where
E: Context,
C: Mutable<Item = Operation<F, update::Unordered<K, V>>>,
I: UnorderedIndex<Value = Location<F>>,
{
let (batch, staged_updates, prefetched) = self
.resolve_updates_prefetched(updates, upserts, db, |floor, tip, limit, out| {
fill_candidates(&db.bitmap, floor, tip, limit, out)
})
.await?;
batch
.merkleize_with_floor_scan(
db,
metadata,
staged_updates,
Some(prefetched),
|floor, tip, limit, out| fill_candidates(&db.bitmap, floor, tip, limit, out),
)
.await
}
#[allow(clippy::type_complexity)]
pub(crate) async fn resolve_updates_prefetched<E, C, I, const N: usize>(
self,
updates: Vec<(usize, Option<V::Value>)>,
upserts: Vec<(K, Option<V::Value>)>,
db: &Db<F, E, C, I, H, update::Unordered<K, V>, N, S>,
mut fill_candidates: impl FnMut(Location<F>, u64, usize, &mut Vec<Location<F>>) -> Location<F>,
) -> Result<
(
UnmerkleizedBatch<F, H, update::Unordered<K, V>, S>,
StagedUpdates<F, update::Unordered<K, V>>,
PrefetchedCandidates<F, update::Unordered<K, V>>,
),
crate::qmdb::Error<F>,
>
where
E: Context,
C: Contiguous<Item = Operation<F, update::Unordered<K, V>>>,
I: UnorderedIndex<Value = Location<F>>,
{
let resolved_updates = updates
.iter()
.filter(|(slot, _)| self.resolutions.get(*slot).is_some_and(Option::is_some))
.count()
.min(self.keys.distinct());
let existing_writes = upserts
.iter()
.map(|(key, _)| key)
.chain(self.batch.mutations.keys())
.filter(|&key| db.snapshot.get(key).next().is_some())
.count();
let steps_bound = resolved_updates + existing_writes + 1;
let scan_from = self.batch.base.inactivity_floor_loc();
let resolve = db
.strategy()
.spawn(move |strategy| self.resolve_updates(updates, upserts, &strategy));
let committed_tip = bitmap::Readable::<N>::len(&*db.bitmap);
let mut locs: Vec<Location<F>> = Vec::with_capacity(steps_bound);
let next_scan = fill_candidates(scan_from, committed_tip, steps_bound, &mut locs);
let raw: Vec<u64> = locs.iter().map(|loc| **loc).collect();
let read = db.log.read_many_sharded(&raw).await;
let (batch, staged_updates) = resolve.await;
let prefetched = PrefetchedCandidates {
locs,
shards: read?,
next_scan,
};
Ok((batch, staged_updates, prefetched))
}
}
impl<F: Family, K, V, H, S: Strategy> Staged<F, H, update::Ordered<K, V>, S>
where
K: Key,
V: ValueEncoding,
H: Hasher,
Operation<F, update::Ordered<K, V>>: Codec,
{
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.any.ordered.batch.merkleize.staged",
level = "info",
skip_all,
fields(updates = updates.len() as u64, upserts = upserts.len() as u64),
)]
pub async fn merkleize<E, C, I, const N: usize>(
self,
updates: Vec<(usize, Option<V::Value>)>,
upserts: Vec<(K, Option<V::Value>)>,
metadata: Option<V::Value>,
db: &Db<F, E, C, I, H, update::Ordered<K, V>, N, S>,
) -> Result<Arc<MerkleizedBatch<F, H::Digest, update::Ordered<K, V>, S>>, crate::qmdb::Error<F>>
where
E: Context,
C: Mutable<Item = Operation<F, update::Ordered<K, V>>>,
I: OrderedIndex<Value = Location<F>>,
{
let (batch, staged_updates) = self.resolve_updates(updates, upserts, db.strategy());
batch
.merkleize_with_floor_scan(db, metadata, staged_updates, |floor, tip, limit, out| {
fill_candidates(&db.bitmap, floor, tip, limit, out)
})
.await
}
}
impl<F: Family, H, U, S: Strategy> UnmerkleizedBatch<F, H, U, S>
where
U: update::Update + Send + Sync,
H: Hasher,
Operation<F, U>: Codec,
{
fn reads_committed_only(&self) -> bool {
self.mutations.is_empty() && self.base.parent().is_none()
}
fn resolve_uncommitted_reads<'a>(
&self,
keys: &[&'a U::Key],
strategy: &S,
on_diff_hit: impl FnMut(usize, &DiffEntry<F, U::Value>),
) -> UncommittedReadResolution<'a, U::Key, U::Value>
where
U::Value: Send + Sync,
{
let ancestors = self.base.parent().map(|parent| {
let mut ancestors = vec![Arc::clone(parent)];
ancestors.extend(parent.ancestors());
ancestors
});
let diffs: Vec<_> = ancestors
.iter()
.flatten()
.map(|batch| batch.diff.as_slice())
.collect();
resolve_reads(
keys,
|key| self.mutations.get(key).cloned(),
&diffs,
strategy,
on_diff_hit,
)
}
async fn fill_committed_reads<E, C, I, T: Send, const N: usize>(
unresolved: Vec<PendingRead<'_, U::Key>>,
db: &Db<F, E, C, I, H, U, N, S>,
results: &mut [Option<U::Value>],
map: impl Fn(&U, Location<F>) -> T + Send + Sync,
mut apply: impl FnMut(usize, T) -> U::Value,
) -> Result<(), crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
{
if unresolved.is_empty() {
return Ok(());
}
let db_keys: Vec<_> = unresolved.iter().map(|(_, key)| *key).collect();
let db_results = db.get_many_map(&db_keys, map).await?;
for ((slot, _), result) in unresolved.into_iter().zip(db_results) {
results[slot] = result.map(|value| apply(slot, value));
}
Ok(())
}
pub async fn get<E, C, I, const N: usize>(
&self,
key: &U::Key,
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<Option<U::Value>, crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
{
let mut values = self.get_many(&[key], db).await?;
Ok(values.pop().expect("one result per key"))
}
pub async fn get_many<E, C, I, const N: usize>(
&self,
keys: &[&U::Key],
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<Vec<Option<U::Value>>, crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
{
if keys.is_empty() {
return Ok(Vec::new());
}
if self.reads_committed_only() {
return db.get_many(keys).await;
}
let (mut results, unresolved) =
self.resolve_uncommitted_reads(keys, db.strategy(), |_, _| {});
Self::fill_committed_reads(
unresolved,
db,
&mut results,
|data, _| data.value().clone(),
|_, value| value,
)
.await?;
Ok(results)
}
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.any.batch.stage",
level = "info",
skip_all,
fields(keys = keys.len() as u64),
)]
pub async fn stage<E, C, I, const N: usize>(
self,
keys: &[&U::Key],
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<(Vec<Option<U::Value>>, Staged<F, H, U, S>), crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
{
let (results, keys, resolutions) = self.stage_reads(keys, db).await?;
Ok((
results,
Staged {
batch: self,
keys: StagedKeys::new(keys),
resolutions,
},
))
}
#[allow(clippy::type_complexity)]
async fn stage_reads<E, C, I, const N: usize>(
&self,
keys: &[&U::Key],
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<
(
Vec<Option<U::Value>>,
Vec<U::Key>,
Vec<StagedResolution<F, U>>,
),
crate::qmdb::Error<F>,
>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
{
let mut resolutions: Vec<StagedResolution<F, U>> =
iter::repeat_with(|| None).take(keys.len()).collect();
let (mut results, unresolved) =
self.resolve_uncommitted_reads(keys, db.strategy(), |slot, entry| {
let Some(cached) = U::STAGES_ANCESTORS else {
return;
};
if let DiffEntry::Active {
loc, base_old_loc, ..
} = entry
{
resolutions[slot] = Some((
StagedLoc::Ancestor {
loc: *loc,
base_old_loc: *base_old_loc,
},
cached,
));
}
});
Self::fill_committed_reads(
unresolved,
db,
&mut results,
|data, loc| (data.value().clone(), loc, data.cached()),
|slot, (value, loc, payload)| {
resolutions[slot] = Some((StagedLoc::Committed(loc), payload));
value
},
)
.await?;
Ok((
results,
keys.iter().map(|key| (*key).to_owned()).collect(),
resolutions,
))
}
}
impl<F: Family, K, V, H, S: Strategy> UnmerkleizedBatch<F, H, update::Unordered<K, V>, S>
where
K: Key,
V: ValueEncoding,
H: Hasher,
Operation<F, update::Unordered<K, V>>: Codec,
{
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.any.unordered.batch.merkleize",
level = "info",
skip_all,
fields(mutations = self.mutations.len() as u64),
)]
pub async fn merkleize<E, C, I, const N: usize>(
self,
db: &Db<F, E, C, I, H, update::Unordered<K, V>, N, S>,
metadata: Option<V::Value>,
) -> Result<Arc<MerkleizedBatch<F, H::Digest, update::Unordered<K, V>, S>>, crate::qmdb::Error<F>>
where
E: Context,
C: Mutable<Item = Operation<F, update::Unordered<K, V>>>,
I: UnorderedIndex<Value = Location<F>>,
{
self.merkleize_with_floor_scan(
db,
metadata,
StagedUpdates::<F, update::Unordered<K, V>>::new(),
None,
|floor, tip, limit, out| fill_candidates(&db.bitmap, floor, tip, limit, out),
)
.await
}
pub(crate) async fn merkleize_with_floor_scan<E, C, I, const N: usize>(
self,
db: &Db<F, E, C, I, H, update::Unordered<K, V>, N, S>,
metadata: Option<V::Value>,
staged_updates: StagedUpdates<F, update::Unordered<K, V>>,
prefetched: Option<PrefetchedCandidates<F, update::Unordered<K, V>>>,
fill_candidates: impl FnMut(Location<F>, u64, usize, &mut Vec<Location<F>>) -> Location<F>,
) -> Result<Arc<MerkleizedBatch<F, H::Digest, update::Unordered<K, V>, S>>, crate::qmdb::Error<F>>
where
E: Context,
C: Mutable<Item = Operation<F, update::Unordered<K, V>>>,
I: UnorderedIndex<Value = Location<F>>,
{
let (mut mutations, m) = self.into_parts();
let locations = m.gather_existing_locations(&mutations, db, false);
let results = m.read_ops(&locations, &[], &db.log).await?;
let mut ops: Vec<Operation<F, update::Unordered<K, V>>> =
Vec::with_capacity(mutations.len() + staged_updates.len() + 1);
let mut diff: DiffVec<K, F, V::Value> =
Vec::with_capacity(mutations.len() + staged_updates.len());
let mut superseded_locs: Vec<Location<F>> = Vec::with_capacity(diff.capacity());
let mut active_keys_delta: isize = 0;
let mut user_steps: u64 = 0;
let mut emit = |key: K, base_old_loc: Option<Location<F>>, mutation: Option<V::Value>| {
let new_loc = Location::new(m.base_size + ops.len() as u64);
superseded_locs.extend(base_old_loc);
match mutation {
Some(value) => {
ops.push(Operation::Update(update::Unordered(
key.clone(),
value.clone(),
)));
diff.push((
key,
DiffEntry::Active {
value,
loc: new_loc,
base_old_loc,
},
));
}
None => {
ops.push(Operation::Delete(key.clone()));
diff.push((key, DiffEntry::Deleted { base_old_loc }));
active_keys_delta -= 1;
}
}
user_steps += 1;
};
let staged_base_old_loc = |sloc: StagedLoc<F>| match sloc {
StagedLoc::Committed(loc) => Some(loc),
StagedLoc::Ancestor { loc, .. } if *loc < m.db_size => Some(loc),
StagedLoc::Ancestor { base_old_loc, .. } => base_old_loc,
};
let mut cached = staged_updates.into_iter().peekable();
for (op, &old_loc) in results.iter().zip(&locations) {
while cached
.peek()
.is_some_and(|&(_, sloc, (), _)| sloc.loc() < old_loc)
{
let (key, sloc, (), mutation) = cached.next().expect("peeked entry exists");
emit(key, staged_base_old_loc(sloc), mutation);
}
let key = op.key().expect("updates should have a key");
let base_old_loc = if let Some(entry) = resolve_in_ancestors(&m.ancestors, key) {
if entry.loc() != Some(old_loc) {
continue;
}
entry.base_old_loc()
} else {
Some(old_loc)
};
let Some(mutation) = mutations.remove(key) else {
continue;
};
emit(key.clone(), base_old_loc, mutation);
}
for (key, sloc, (), mutation) in cached {
emit(key, staged_base_old_loc(sloc), mutation);
}
let parent_deleted_creates = m.extract_parent_deleted_creates(&mut mutations);
let mut creates: Vec<(K, V::Value, Option<Location<F>>)> =
Vec::with_capacity(mutations.len() + parent_deleted_creates.len());
for (key, value) in mutations {
if let Some(value) = value {
creates.push((key, value, None));
}
}
creates.extend(parent_deleted_creates);
db.strategy()
.sort_by(&mut creates, |(a, _, _), (b, _, _)| a.cmp(b));
for (key, value, base_old_loc) in creates {
let new_loc = Location::new(m.base_size + ops.len() as u64);
superseded_locs.extend(base_old_loc);
ops.push(Operation::Update(update::Unordered(
key.clone(),
value.clone(),
)));
diff.push((
key,
DiffEntry::Active {
value,
loc: new_loc,
base_old_loc,
},
));
active_keys_delta += 1;
}
m.finish(
ops,
diff,
superseded_locs,
active_keys_delta,
user_steps,
metadata,
prefetched,
fill_candidates,
db,
)
.await
}
}
impl<F: Family, K, V, H, S: Strategy> UnmerkleizedBatch<F, H, update::Ordered<K, V>, S>
where
K: Key,
V: ValueEncoding,
H: Hasher,
Operation<F, update::Ordered<K, V>>: Codec,
{
#[allow(clippy::type_complexity)]
#[tracing::instrument(
name = "qmdb.any.ordered.batch.merkleize",
level = "info",
skip_all,
fields(mutations = self.mutations.len() as u64),
)]
pub async fn merkleize<E, C, I, const N: usize>(
self,
db: &Db<F, E, C, I, H, update::Ordered<K, V>, N, S>,
metadata: Option<V::Value>,
) -> Result<Arc<MerkleizedBatch<F, H::Digest, update::Ordered<K, V>, S>>, crate::qmdb::Error<F>>
where
E: Context,
C: Mutable<Item = Operation<F, update::Ordered<K, V>>>,
I: OrderedIndex<Value = Location<F>>,
{
self.merkleize_with_floor_scan(
db,
metadata,
StagedUpdates::<F, update::Ordered<K, V>>::new(),
|floor, tip, limit, out| fill_candidates(&db.bitmap, floor, tip, limit, out),
)
.await
}
pub(crate) async fn merkleize_with_floor_scan<E, C, I, const N: usize>(
self,
db: &Db<F, E, C, I, H, update::Ordered<K, V>, N, S>,
metadata: Option<V::Value>,
staged_updates: StagedUpdates<F, update::Ordered<K, V>>,
fill_candidates: impl FnMut(Location<F>, u64, usize, &mut Vec<Location<F>>) -> Location<F>,
) -> Result<Arc<MerkleizedBatch<F, H::Digest, update::Ordered<K, V>, S>>, crate::qmdb::Error<F>>
where
E: Context,
C: Mutable<Item = Operation<F, update::Ordered<K, V>>>,
I: OrderedIndex<Value = Location<F>>,
{
let (mut mutations, m) = self.into_parts();
let locations = m.gather_existing_locations(&mutations, db, true);
let mut next_candidates: Vec<K> = Vec::new();
let mut prev_candidates: PrevCandidates<K, F, V::Value> = Vec::new();
let mut deleted: Vec<(K, Location<F>)> = Vec::new();
let mut updated: Vec<(K, V::Value, Location<F>)> = Vec::new();
for (op, &old_loc) in m
.read_ops(&locations, &[], &db.log)
.await?
.into_iter()
.zip(&locations)
{
let update::Ordered {
key,
value,
next_key,
} = match op {
Operation::Update(data) => data,
_ => unreachable!("snapshot should only reference Update operations"),
};
next_candidates.push(next_key);
let mutation = mutations.remove(&key);
prev_candidates.push((key.clone(), (Some(value), old_loc)));
let Some(mutation) = mutation else {
continue;
};
if let Some(new_value) = mutation {
updated.push((key, new_value, old_loc));
} else {
deleted.push((key, old_loc));
}
}
for (key, sloc, old_next, value) in staged_updates {
let value = value.expect("ordered path never stages deletes");
let StagedLoc::Committed(loc) = sloc else {
unreachable!("ordered path never stages ancestor resolutions")
};
next_candidates.push(old_next);
prev_candidates.push((key.clone(), (None, loc)));
updated.push((key, value, loc));
}
db.strategy().sort_by(&mut deleted, |a, b| a.0.cmp(&b.0));
db.strategy().sort_by(&mut updated, |a, b| a.0.cmp(&b.0));
let parent_deleted_creates = m.extract_parent_deleted_creates(&mut mutations);
let mut created: Vec<(K, V::Value, Option<Location<F>>)> =
Vec::with_capacity(mutations.len() + parent_deleted_creates.len());
for (key, value) in mutations {
let Some(value) = value else {
continue; };
next_candidates.push(key.clone());
created.push((key, value, None));
}
for (key, value, base_old_loc) in parent_deleted_creates {
next_candidates.push(key.clone());
created.push((key, value, base_old_loc));
}
db.strategy()
.sort_by(&mut created, |(a, _, _), (b, _, _)| a.cmp(b));
let mut prev_locations = Vec::new();
for key in deleted
.iter()
.map(|(k, _)| k)
.chain(created.iter().map(|(k, _, _)| k))
{
let Some((iter, _)) = db.snapshot.prev_translated_key(key) else {
continue;
};
prev_locations.extend(iter.copied());
}
prev_locations.sort();
prev_locations.dedup();
let prev_results = m.read_ops(&prev_locations, &[], &db.log).await?;
for (op, &old_loc) in prev_results.into_iter().zip(&prev_locations) {
let data = match op {
Operation::Update(data) => data,
_ => unreachable!("expected update operation"),
};
next_candidates.push(data.next_key);
prev_candidates.push((data.key, (Some(data.value), old_loc)));
}
let track_shadow = m.ancestors.len() > 1;
let seen_cap = if track_shadow {
m.ancestors.iter().map(|a| a.diff.len()).sum()
} else {
0
};
let mut seen: AHashSet<&K> = AHashSet::with_capacity(seen_cap);
let mut ancestor_deleted: Vec<K> = Vec::new();
let mut ancestor_active: Vec<(&K, &V::Value, Location<F>)> = Vec::new();
for batch in m.ancestors.iter() {
let (mut ui, mut ci, mut di) = (0, 0, 0);
for (key, entry) in batch.diff.iter() {
if track_shadow && !seen.insert(key) {
continue;
}
while ui < updated.len() && updated[ui].0 < *key {
ui += 1;
}
while ci < created.len() && created[ci].0 < *key {
ci += 1;
}
while di < deleted.len() && deleted[di].0 < *key {
di += 1;
}
if updated.get(ui).is_some_and(|(k, ..)| k == key)
|| created.get(ci).is_some_and(|(k, ..)| k == key)
|| deleted.get(di).is_some_and(|(k, _)| k == key)
{
continue;
}
match entry {
DiffEntry::Active { value, loc, .. } => {
ancestor_active.push((key, value, *loc));
}
DiffEntry::Deleted { .. } => {
ancestor_deleted.push(key.clone());
}
}
}
}
ancestor_deleted.sort();
ancestor_deleted.dedup();
let ancestor_locs: Vec<Location<F>> =
ancestor_active.iter().map(|&(_, _, loc)| loc).collect();
for (op, (key, value, loc)) in m
.read_ops(&ancestor_locs, &[], &db.log)
.await?
.into_iter()
.zip(ancestor_active)
{
let data = match op {
Operation::Update(data) => data,
_ => unreachable!("ancestor diff Active should reference Update op"),
};
next_candidates.push(key.clone());
next_candidates.push(data.next_key);
prev_candidates.push((key.clone(), (Some(value.clone()), loc)));
}
db.strategy().sort_by(&mut next_candidates, |a, b| a.cmp(b));
next_candidates.dedup();
prev_candidates.sort_by(|a, b| a.0.cmp(&b.0));
prev_candidates.dedup_by(|a, b| {
if a.0 == b.0 {
std::mem::swap(a, b);
true
} else {
false
}
});
let is_deleted = |k: &K| -> bool {
deleted.binary_search_by(|(dk, _)| dk.cmp(k)).is_ok()
|| (ancestor_deleted.binary_search(k).is_ok()
&& created.binary_search_by(|(ck, _, _)| ck.cmp(k)).is_err())
};
next_candidates.retain(|k| !is_deleted(k));
prev_candidates.retain(|(k, _)| !is_deleted(k));
let mut ops: Vec<Operation<F, update::Ordered<K, V>>> =
Vec::with_capacity(deleted.len() + updated.len() + created.len() + 1);
let mut diff: DiffVec<K, F, V::Value> =
Vec::with_capacity(deleted.len() + updated.len() + created.len());
let mut active_keys_delta: isize = 0;
let mut user_steps: u64 = 0;
let mut ancestors = DiffCursors::new(m.ancestors.iter().map(|a| a.diff.as_slice()));
for (key, old_loc) in &deleted {
ops.push(Operation::Delete(key.clone()));
let base_old_loc = ancestors
.resolve(key)
.map_or(Some(*old_loc), DiffEntry::base_old_loc);
diff.push((key.clone(), DiffEntry::Deleted { base_old_loc }));
active_keys_delta -= 1;
user_steps += 1;
}
let mut ancestors = DiffCursors::new(m.ancestors.iter().map(|a| a.diff.as_slice()));
let mut next_idx = 0;
for (key, value, old_loc) in &updated {
let new_loc = Location::new(m.base_size + ops.len() as u64);
let next_key = find_next_key_ascending(key, &next_candidates, &mut next_idx);
ops.push(Operation::Update(update::Ordered {
key: key.clone(),
value: value.clone(),
next_key,
}));
let base_old_loc = ancestors
.resolve(key)
.map_or(Some(*old_loc), DiffEntry::base_old_loc);
diff.push((
key.clone(),
DiffEntry::Active {
value: value.clone(),
loc: new_loc,
base_old_loc,
},
));
user_steps += 1;
}
let mut next_idx = 0;
for (key, value, base_old_loc) in &created {
let new_loc = Location::new(m.base_size + ops.len() as u64);
let next_key = find_next_key_ascending(key, &next_candidates, &mut next_idx);
ops.push(Operation::Update(update::Ordered {
key: key.clone(),
value: value.clone(),
next_key,
}));
diff.push((
key.clone(),
DiffEntry::Active {
value: value.clone(),
loc: new_loc,
base_old_loc: *base_old_loc,
},
));
active_keys_delta += 1;
}
if !prev_candidates.is_empty() {
let mut rewritten_predecessors = AHashSet::with_capacity(created.len() + deleted.len());
for key in created
.iter()
.map(|(k, _, _)| k)
.chain(deleted.iter().map(|(k, _)| k))
{
let (prev_key, (prev_value, prev_loc)) = find_prev_key(key, &prev_candidates);
if deleted.binary_search_by(|(k, _)| k.cmp(prev_key)).is_ok()
|| updated
.binary_search_by(|(k, _, _)| k.cmp(prev_key))
.is_ok()
|| created
.binary_search_by(|(k, _, _)| k.cmp(prev_key))
.is_ok()
{
continue;
}
if !rewritten_predecessors.insert(prev_key.clone()) {
continue;
}
let prev_value = prev_value
.as_ref()
.expect("staged-resolved keys are skipped as updated");
let prev_new_loc = Location::new(m.base_size + ops.len() as u64);
let prev_next_key = find_next_key(prev_key, &next_candidates);
ops.push(Operation::Update(update::Ordered {
key: prev_key.clone(),
value: prev_value.clone(),
next_key: prev_next_key,
}));
let prev_base_old_loc = resolve_in_ancestors(&m.ancestors, prev_key)
.map_or(Some(*prev_loc), DiffEntry::base_old_loc);
diff.push((
prev_key.clone(),
DiffEntry::Active {
value: prev_value.clone(),
loc: prev_new_loc,
base_old_loc: prev_base_old_loc,
},
));
user_steps += 1;
}
}
let superseded_locs: Vec<_> = diff
.iter()
.filter_map(|(_, entry)| entry.base_old_loc())
.collect();
m.finish(
ops,
diff,
superseded_locs,
active_keys_delta,
user_steps,
metadata,
None,
fill_candidates,
db,
)
.await
}
}
impl<F: Family, D: Digest, U: update::Update + Send + Sync, S: Strategy> MerkleizedBatch<F, D, U, S>
where
Operation<F, U>: Send + Sync,
{
pub const fn root(&self) -> D {
self.root
}
pub const fn bounds(&self) -> &Bounds<F> {
&self.bounds
}
pub(crate) fn ancestors(&self) -> impl Iterator<Item = Arc<Self>> {
batch_chain::ancestors(self.parent.clone(), |batch| batch.parent.as_ref())
}
}
impl<F: Family, D: Digest, U: update::Update + Send + Sync, S: Strategy> MerkleizedBatch<F, D, U, S>
where
Operation<F, U>: Codec,
{
#[tracing::instrument(
name = "qmdb.any.batch.new.from_batch",
level = "debug",
skip_all,
fields(
base_size = self.bounds.base_size,
total_size = self.bounds.total_size,
ancestor_batches = self.ancestor_diffs.len() as u64,
),
)]
pub fn new_batch<H>(self: &Arc<Self>) -> UnmerkleizedBatch<F, H, U, S>
where
H: Hasher<Digest = D>,
{
UnmerkleizedBatch {
journal_batch: self.journal_batch.new_batch::<H>(),
mutations: BTreeMap::new(),
base: Base::Child(Arc::clone(self)),
}
}
pub async fn get<E, C, I, H, const N: usize>(
&self,
key: &U::Key,
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<Option<U::Value>, crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
H: Hasher<Digest = D>,
{
if let Some(entry) = lookup_sorted(self.diff.as_slice(), key) {
return Ok(entry.value().cloned());
}
for batch in self.ancestors() {
if let Some(entry) = lookup_sorted(batch.diff.as_slice(), key) {
return Ok(entry.value().cloned());
}
}
db.get(key).await
}
pub async fn get_many<E, C, I, H, const N: usize>(
&self,
keys: &[&U::Key],
db: &Db<F, E, C, I, H, U, N, S>,
) -> Result<Vec<Option<U::Value>>, crate::qmdb::Error<F>>
where
E: Context,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>> + 'static,
H: Hasher<Digest = D>,
{
if keys.is_empty() {
return Ok(Vec::new());
}
let ancestors: Vec<_> = self.ancestors().collect();
let diffs: Vec<_> = ancestors
.iter()
.map(|batch| batch.diff.as_slice())
.collect();
let (mut results, unresolved) = resolve_reads(
keys,
|key| lookup_sorted(self.diff.as_slice(), key).map(|entry| entry.value().cloned()),
&diffs,
db.strategy(),
|_, _| {},
);
if !unresolved.is_empty() {
let db_keys: Vec<_> = unresolved.iter().map(|(_, key)| *key).collect();
let db_results = db.get_many(&db_keys).await?;
for ((slot, _), value) in unresolved.into_iter().zip(db_results) {
results[slot] = value;
}
}
Ok(results)
}
}
impl<F, E, C, I, H, U, const N: usize, S> Db<F, E, C, I, H, U, N, S>
where
F: Family,
E: Context,
U: update::Update + Send + Sync,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
S: Strategy,
Operation<F, U>: Codec,
{
#[tracing::instrument(
name = "qmdb.any.batch.new.from_db",
level = "debug",
skip_all,
fields(
base_size = *self.last_commit_loc + 1,
inactivity_floor = *self.inactivity_floor_loc,
active_keys = self.active_keys as u64,
),
)]
pub fn new_batch(&self) -> UnmerkleizedBatch<F, H, U, S> {
let journal_size = *self.last_commit_loc + 1;
UnmerkleizedBatch {
journal_batch: self.log.new_batch(),
mutations: BTreeMap::new(),
base: Base::Db {
db_size: journal_size,
inactivity_floor_loc: self.inactivity_floor_loc,
active_keys: self.active_keys,
},
}
}
}
impl<F, E, C, I, H, U, const N: usize, S> Db<F, E, C, I, H, U, N, S>
where
F: Family,
E: Context,
U: update::Update + Send + Sync + 'static,
C: Mutable<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
S: Strategy,
Operation<F, U>: Codec,
{
#[tracing::instrument(
name = "qmdb.any.db.apply_batch",
level = "info",
skip_all,
fields(
batch_total_size = batch.bounds.total_size,
batch_base_size = batch.bounds.base_size,
db_size = *self.last_commit_loc + 1,
ancestor_batches = batch.ancestor_diffs.len() as u64,
),
)]
pub async fn apply_batch(
&mut self,
batch: Arc<MerkleizedBatch<F, H::Digest, U, S>>,
) -> Result<Range<Location<F>>, crate::qmdb::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.log.apply_batch(&batch.journal_batch).await?;
{
let mut bitmap = self.bitmap.write();
bitmap.extend_to(batch.bounds.total_size);
if batch.ancestor_diffs.is_empty() {
for (key, entry) in batch.diff.iter() {
apply_diff(
&mut self.snapshot,
&mut bitmap,
key,
entry,
entry.base_old_loc(),
);
}
} else {
let mut applied = Vec::with_capacity(batch.ancestor_diffs.len());
let mut pending = Vec::with_capacity(batch.ancestor_diffs.len());
for (i, ancestor_diff) in batch.ancestor_diffs.iter().enumerate() {
if batch.bounds.ancestors[i].end <= db_size {
applied.push(ancestor_diff.as_slice());
} else {
pending.push(ancestor_diff.as_slice());
}
}
let mut resolver = DiffCursors::new(applied);
let merge = DiffMerge::new(
iter::once(batch.diff.as_slice()).chain(pending.iter().copied()),
);
for (key, entry) in merge {
let old = resolver
.resolve(key)
.map(DiffEntry::loc)
.unwrap_or_else(|| entry.base_old_loc());
apply_diff(&mut self.snapshot, &mut bitmap, key, entry, old);
}
}
bitmap.set_bit(*self.last_commit_loc, false);
bitmap.set_bit(batch.bounds.total_size - 1, true);
}
self.active_keys = batch.total_active_keys;
self.inactivity_floor_loc = batch.bounds.inactivity_floor;
self.last_commit_loc = Location::new(batch.bounds.total_size - 1);
self.root = batch.root;
let end_loc = Location::new(*self.last_commit_loc + 1);
let range = start_loc..end_loc;
self.update_metrics();
self.metrics
.operations_applied
.inc_by(*range.end - *range.start);
Ok(range)
}
}
impl<F: Family, E, C, I, H, U, const N: usize, S> Db<F, E, C, I, H, U, N, S>
where
E: Context,
U: update::Update + Send + Sync,
C: Contiguous<Item = Operation<F, U>>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
S: Strategy,
Operation<F, U>: Codec,
{
#[tracing::instrument(
name = "qmdb.any.db.to_batch",
level = "info",
skip_all,
fields(
db_size = *self.last_commit_loc + 1,
inactivity_floor = *self.inactivity_floor_loc,
active_keys = self.active_keys as u64,
),
)]
pub fn to_batch(&self) -> Arc<MerkleizedBatch<F, H::Digest, U, S>> {
let journal_size = *self.last_commit_loc + 1;
Arc::new(MerkleizedBatch {
journal_batch: self.log.to_merkleized_batch(),
root: self.root,
diff: Arc::new(Vec::new()),
parent: None,
total_active_keys: self.active_keys,
ancestor_diffs: Vec::new(),
bounds: batch_chain::Bounds {
base_size: journal_size,
db_size: journal_size,
total_size: journal_size,
ancestors: Vec::new(),
inactivity_floor: self.inactivity_floor_loc,
},
})
}
}
fn extract_update_value<F: Family, U: update::Update>(op: &Operation<F, U>) -> U::Value {
match op {
Operation::Update(update) => update.value().clone(),
_ => unreachable!("floor raise should only re-append Update operations"),
}
}
#[cfg(any(test, feature = "test-traits"))]
mod trait_impls {
use super::*;
use crate::qmdb::any::traits::{
BatchableDb, MerkleizedBatch as MerkleizedBatchTrait,
UnmerkleizedBatch as UnmerkleizedBatchTrait,
};
use std::future::Future;
impl<F, K, V, H, E, C, I, const N: usize, S>
UnmerkleizedBatchTrait<Db<F, E, C, I, H, update::Unordered<K, V>, N, S>>
for UnmerkleizedBatch<F, H, update::Unordered<K, V>, S>
where
F: Family,
K: Key,
V: ValueEncoding + 'static,
H: Hasher,
E: Context,
C: Mutable<Item = Operation<F, update::Unordered<K, V>>>,
I: UnorderedIndex<Value = Location<F>>,
S: Strategy,
Operation<F, update::Unordered<K, V>>: Codec,
{
type Family = F;
type K = K;
type V = V::Value;
type Metadata = V::Value;
type Merkleized = Arc<MerkleizedBatch<F, H::Digest, update::Unordered<K, V>, S>>;
fn write(self, key: K, value: Option<V::Value>) -> Self {
Self::write(self, key, value)
}
fn merkleize(
self,
db: &Db<F, E, C, I, H, update::Unordered<K, V>, N, S>,
metadata: Option<V::Value>,
) -> impl Future<Output = Result<Self::Merkleized, crate::qmdb::Error<F>>> {
self.merkleize(db, metadata)
}
}
impl<F, K, V, H, E, C, I, const N: usize, S>
UnmerkleizedBatchTrait<Db<F, E, C, I, H, update::Ordered<K, V>, N, S>>
for UnmerkleizedBatch<F, H, update::Ordered<K, V>, S>
where
F: Family,
K: Key,
V: ValueEncoding + 'static,
H: Hasher,
E: Context,
C: Mutable<Item = Operation<F, update::Ordered<K, V>>>,
I: OrderedIndex<Value = Location<F>>,
S: Strategy,
Operation<F, update::Ordered<K, V>>: Codec,
{
type Family = F;
type K = K;
type V = V::Value;
type Metadata = V::Value;
type Merkleized = Arc<MerkleizedBatch<F, H::Digest, update::Ordered<K, V>, S>>;
fn write(self, key: K, value: Option<V::Value>) -> Self {
Self::write(self, key, value)
}
fn merkleize(
self,
db: &Db<F, E, C, I, H, update::Ordered<K, V>, N, S>,
metadata: Option<V::Value>,
) -> impl Future<Output = Result<Self::Merkleized, crate::qmdb::Error<F>>> {
self.merkleize(db, metadata)
}
}
impl<F: Family, D: Digest, U: update::Update + Send + Sync + 'static, S: Strategy>
MerkleizedBatchTrait for Arc<MerkleizedBatch<F, D, U, S>>
where
Operation<F, U>: Codec,
{
type Digest = D;
fn root(&self) -> D {
MerkleizedBatch::root(self)
}
}
impl<F, E, K, V, C, I, H, const N: usize, S> BatchableDb
for Db<F, E, C, I, H, update::Unordered<K, V>, N, S>
where
F: Family,
E: Context,
K: Key,
V: ValueEncoding + 'static,
C: Mutable<Item = Operation<F, update::Unordered<K, V>>>,
I: UnorderedIndex<Value = Location<F>>,
H: Hasher,
S: Strategy,
Operation<F, update::Unordered<K, V>>: Codec,
{
type Family = F;
type K = K;
type V = V::Value;
type Merkleized = Arc<MerkleizedBatch<F, H::Digest, update::Unordered<K, V>, S>>;
type Batch = UnmerkleizedBatch<F, H, update::Unordered<K, V>, S>;
fn new_batch(&self) -> Self::Batch {
self.new_batch()
}
fn apply_batch(
&mut self,
batch: Self::Merkleized,
) -> impl Future<Output = Result<Range<Location<F>>, crate::qmdb::Error<F>>> {
self.apply_batch(batch)
}
}
impl<F, E, K, V, C, I, H, const N: usize, S> BatchableDb
for Db<F, E, C, I, H, update::Ordered<K, V>, N, S>
where
F: Family,
E: Context,
K: Key,
V: ValueEncoding + 'static,
C: Mutable<Item = Operation<F, update::Ordered<K, V>>>,
I: OrderedIndex<Value = Location<F>>,
H: Hasher,
S: Strategy,
Operation<F, update::Ordered<K, V>>: Codec,
{
type Family = F;
type K = K;
type V = V::Value;
type Merkleized = Arc<MerkleizedBatch<F, H::Digest, update::Ordered<K, V>, S>>;
type Batch = UnmerkleizedBatch<F, H, update::Ordered<K, V>, S>;
fn new_batch(&self) -> Self::Batch {
self.new_batch()
}
fn apply_batch(
&mut self,
batch: Self::Merkleized,
) -> impl Future<Output = Result<Range<Location<F>>, crate::qmdb::Error<F>>> {
self.apply_batch(batch)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
mmr,
qmdb::any::{
ordered::fixed::Db as OrderedFixedDb,
test::{colliding_digest, fixed_db_config},
unordered::fixed::Db as UnorderedFixedDb,
value::FixedEncoding,
BITMAP_CHUNK_BYTES,
},
translator::OneCap,
};
use commonware_cryptography::{sha256, Sha256};
use commonware_parallel::Sequential;
use commonware_runtime::{deterministic, Runner as _, Supervisor as _};
use commonware_utils::test_rng;
use rand::RngExt as _;
const BITMAP_CHUNK_BITS: u64 = bitmap::Prunable::<BITMAP_CHUNK_BYTES>::CHUNK_SIZE_BITS;
fn loc(n: u64) -> Location<mmr::Family> {
Location::new(n)
}
fn committed(n: u64) -> StagedLoc<mmr::Family> {
StagedLoc::Committed(loc(n))
}
fn shared_with<F>(build: F) -> Shared<BITMAP_CHUNK_BYTES>
where
F: FnOnce(&mut bitmap::Prunable<BITMAP_CHUNK_BYTES>),
{
let mut bm = bitmap::Prunable::<BITMAP_CHUNK_BYTES>::new();
build(&mut bm);
Shared::new(bm)
}
#[test]
fn diff_cursors_matches_lookup_sorted() {
let mut rng = test_rng();
for _ in 0..50 {
let num_diffs = rng.random_range(1..=4);
let diffs: Vec<DiffVec<u64, mmr::Family, u64>> = (0..num_diffs)
.map(|d| {
let mut keys: Vec<u64> = (0..rng.random_range(0..30))
.map(|_| rng.random_range(0..50u64))
.collect();
keys.sort_unstable();
keys.dedup();
keys.into_iter()
.map(|k| {
(
k,
DiffEntry::Active {
value: k * 1000 + d,
loc: loc(k * 1000 + d),
base_old_loc: None,
},
)
})
.collect()
})
.collect();
let mut queries: Vec<u64> = (0..rng.random_range(1..60))
.map(|_| rng.random_range(0..55u64))
.collect();
queries.sort_unstable();
let mut cursors = DiffCursors::new(diffs.iter().map(|d| d.as_slice()));
for q in queries {
let expected = diffs.iter().find_map(|d| lookup_sorted(d.as_slice(), &q));
let actual = cursors.resolve(&q);
assert_eq!(
expected.map(DiffEntry::loc),
actual.map(DiffEntry::loc),
"query {q} diverged"
);
}
}
}
#[test]
#[should_panic(expected = "queries must be non-decreasing")]
fn diff_cursors_rejects_out_of_order_query() {
let diff: DiffVec<u64, mmr::Family, u64> = vec![1, 5]
.into_iter()
.map(|k| {
(
k,
DiffEntry::Active {
value: k,
loc: loc(k),
base_old_loc: None,
},
)
})
.collect();
let mut cursors = DiffCursors::new([diff.as_slice()]);
assert!(cursors.resolve(&5).is_some());
cursors.resolve(&1);
}
#[test]
fn sorted_contains_matches_binary_search() {
let mut rng = test_rng();
for _ in 0..50 {
let mut items: Vec<u64> = (0..rng.random_range(0..40))
.map(|_| rng.random_range(0..100u64))
.collect();
items.sort_unstable();
items.dedup();
let mut queries: Vec<u64> = (0..rng.random_range(1..80))
.map(|_| rng.random_range(0..110u64))
.collect();
queries.sort_unstable();
let mut cursor = 0;
for q in queries {
assert_eq!(
sorted_contains(&items, &mut cursor, &q),
items.binary_search(&q).is_ok(),
"query {q} diverged"
);
}
}
}
#[test]
fn merge_sorted_diffs_matches_sort() {
let mut rng = test_rng();
for _ in 0..50 {
let mut build = |offset: u64| -> DiffVec<u64, mmr::Family, u64> {
let mut keys: Vec<u64> = (0..rng.random_range(0..30))
.map(|_| rng.random_range(0..50u64) * 2 + offset)
.collect();
keys.sort_unstable();
keys.dedup();
keys.into_iter().map(|k| (k, active(k, k))).collect()
};
let a = build(0);
let b = build(1);
let mut reference = a.clone();
reference.extend(b.clone());
reference.sort_by_key(|x| x.0);
let merged = merge_sorted_diffs(a, b);
assert_eq!(merged.len(), reference.len());
for ((mk, me), (rk, re)) in merged.iter().zip(&reference) {
assert_eq!(mk, rk);
assert_eq!(me.loc(), re.loc());
assert_eq!(me.value(), re.value());
}
}
}
fn next_candidate<F: Family, const N: usize>(
bitmap: &Shared<N>,
floor: Location<F>,
tip: u64,
) -> Option<Location<F>> {
let floor = *floor;
let bitmap_len = bitmap::Readable::<N>::len(bitmap);
let committed_end = bitmap_len.min(tip);
if floor < committed_end {
if let Some(idx) = bitmap.next_one_from(floor) {
if idx < committed_end {
return Some(Location::new(idx));
}
}
}
let candidate = floor.max(bitmap_len);
(candidate < tip).then(|| Location::new(candidate))
}
fn active(value: u64, location: u64) -> DiffEntry<mmr::Family, u64> {
DiffEntry::Active {
value,
loc: loc(location),
base_old_loc: None,
}
}
fn deleted(base_old_loc: Option<u64>) -> DiffEntry<mmr::Family, u64> {
DiffEntry::Deleted {
base_old_loc: base_old_loc.map(loc),
}
}
#[test]
fn diff_merge_returns_sorted_newest_entries() {
let child = vec![(2, active(20, 20)), (5, active(50, 50))];
let parent = vec![
(1, active(11, 11)),
(2, active(12, 12)),
(4, deleted(Some(4))),
(7, active(17, 17)),
];
let grandparent = vec![
(2, active(102, 102)),
(3, active(103, 103)),
(4, active(104, 104)),
(6, active(106, 106)),
];
let merged: Vec<_> =
DiffMerge::new([child.as_slice(), parent.as_slice(), grandparent.as_slice()])
.map(|(key, entry)| (*key, entry.value().copied(), entry.loc()))
.collect();
assert_eq!(
merged,
vec![
(1, Some(11), Some(loc(11))),
(2, Some(20), Some(loc(20))),
(3, Some(103), Some(loc(103))),
(4, None, None),
(5, Some(50), Some(loc(50))),
(6, Some(106), Some(loc(106))),
(7, Some(17), Some(loc(17))),
]
);
}
#[test]
fn diff_merge_two_way_priority() {
let a = vec![
(1, active(10, 10)),
(3, active(30, 30)),
(5, deleted(Some(5))),
];
let b = vec![
(2, active(20, 20)),
(3, active(300, 300)),
(4, active(40, 40)),
(5, active(50, 50)),
];
let merged: Vec<_> = DiffMerge::new([a.as_slice(), b.as_slice()])
.map(|(key, entry)| (*key, entry.value().copied(), entry.loc()))
.collect();
assert_eq!(
merged,
vec![
(1, Some(10), Some(loc(10))),
(2, Some(20), Some(loc(20))),
(3, Some(30), Some(loc(30))),
(4, Some(40), Some(loc(40))),
(5, None, None),
]
);
}
#[test]
fn diff_merge_single_stream() {
let a = vec![(1, active(10, 10)), (3, active(30, 30))];
let merged: Vec<_> = DiffMerge::new([a.as_slice()])
.map(|(key, entry)| (*key, entry.value().copied()))
.collect();
assert_eq!(merged, vec![(1, Some(10)), (3, Some(30))]);
}
#[test]
fn diff_cursors_use_nearest_touch() {
let parent = vec![(2, active(20, 20)), (5, deleted(Some(5)))];
let grandparent = vec![
(2, active(200, 200)),
(4, active(40, 40)),
(5, active(50, 50)),
];
let mut cursors = DiffCursors::new([parent.as_slice(), grandparent.as_slice()]);
assert_eq!(cursors.resolve(&1).map(DiffEntry::loc), None);
assert_eq!(cursors.resolve(&2).map(DiffEntry::loc), Some(Some(loc(20))));
assert_eq!(cursors.resolve(&4).map(DiffEntry::loc), Some(Some(loc(40))));
assert_eq!(cursors.resolve(&5).map(DiffEntry::loc), Some(None));
assert_eq!(cursors.resolve(&9).map(DiffEntry::loc), None);
}
#[test]
fn bitmap_scan_empty() {
let bitmap = shared_with(|_| {});
assert_eq!(next_candidate(&bitmap, loc(0), 0), None);
}
#[test]
fn bitmap_scan_uncommitted_tail() {
let bitmap = shared_with(|_| {});
assert_eq!(next_candidate(&bitmap, loc(0), 3), Some(loc(0)));
assert_eq!(next_candidate(&bitmap, loc(1), 3), Some(loc(1)));
assert_eq!(next_candidate(&bitmap, loc(2), 3), Some(loc(2)));
assert_eq!(next_candidate(&bitmap, loc(3), 3), None);
}
#[test]
fn bitmap_scan_committed_region() {
let bitmap = shared_with(|bm| {
bm.extend_to(10);
bm.set_bit(*loc(3), true);
bm.set_bit(*loc(7), true);
});
assert_eq!(next_candidate(&bitmap, loc(0), 10), Some(loc(3)));
assert_eq!(next_candidate(&bitmap, loc(4), 10), Some(loc(7)));
assert_eq!(next_candidate(&bitmap, loc(8), 10), None);
assert_eq!(next_candidate(&bitmap, loc(0), 5), Some(loc(3)));
assert_eq!(next_candidate(&bitmap, loc(4), 5), None);
}
#[test]
fn bitmap_scan_transitions_into_tail() {
let bitmap = shared_with(|bm| {
bm.extend_to(5);
bm.set_bit(*loc(2), true);
});
assert_eq!(next_candidate(&bitmap, loc(0), 8), Some(loc(2)));
assert_eq!(next_candidate(&bitmap, loc(3), 8), Some(loc(5)));
assert_eq!(next_candidate(&bitmap, loc(6), 8), Some(loc(6)));
assert_eq!(next_candidate(&bitmap, loc(8), 8), None);
}
#[test]
fn bitmap_scan_after_prune() {
let bitmap = shared_with(|bm| {
bm.extend_to(BITMAP_CHUNK_BITS * 3);
bm.set_bit(*loc(BITMAP_CHUNK_BITS * 2 + 5), true);
bm.prune_to_bit(BITMAP_CHUNK_BITS * 2);
});
assert_eq!(
commonware_utils::bitmap::Readable::pruned_chunks(&bitmap),
2
);
assert_eq!(
next_candidate(&bitmap, loc(BITMAP_CHUNK_BITS * 2), BITMAP_CHUNK_BITS * 3),
Some(loc(BITMAP_CHUNK_BITS * 2 + 5))
);
}
#[test]
fn bitmap_scan_after_truncate() {
let bitmap = shared_with(|bm| {
bm.extend_to(BITMAP_CHUNK_BITS * 2);
bm.set_bit(*loc(BITMAP_CHUNK_BITS + 3), true);
bm.truncate(BITMAP_CHUNK_BITS);
});
assert_eq!(
commonware_utils::bitmap::Readable::<BITMAP_CHUNK_BYTES>::len(&bitmap),
BITMAP_CHUNK_BITS
);
assert_eq!(next_candidate(&bitmap, loc(0), BITMAP_CHUNK_BITS), None);
}
#[test]
fn bitmap_fill_candidates_matches_oracle() {
let shapes: Vec<(&str, Shared<BITMAP_CHUNK_BYTES>)> = vec![
("empty", shared_with(|_| {})),
(
"committed_bits",
shared_with(|bm| {
bm.extend_to(10);
bm.set_bit(3, true);
bm.set_bit(7, true);
}),
),
(
"transition_into_tail",
shared_with(|bm| {
bm.extend_to(5);
bm.set_bit(2, true);
}),
),
(
"pruned",
shared_with(|bm| {
bm.extend_to(BITMAP_CHUNK_BITS * 3);
bm.set_bit(BITMAP_CHUNK_BITS * 2 + 5, true);
bm.prune_to_bit(BITMAP_CHUNK_BITS * 2);
}),
),
(
"truncated",
shared_with(|bm| {
bm.extend_to(BITMAP_CHUNK_BITS * 2);
bm.set_bit(BITMAP_CHUNK_BITS + 3, true);
bm.truncate(BITMAP_CHUNK_BITS);
}),
),
];
for (name, bitmap) in shapes {
let bitmap_len = bitmap::Readable::<BITMAP_CHUNK_BYTES>::len(&bitmap);
let start = commonware_utils::bitmap::Readable::pruned_chunks(&bitmap) as u64
* BITMAP_CHUNK_BITS;
for tip in [
start,
bitmap_len.saturating_sub(2),
bitmap_len,
bitmap_len + 6,
] {
let mut expected = Vec::new();
let mut floor = loc(start);
while let Some(candidate) = next_candidate(&bitmap, floor, tip) {
expected.push(candidate);
floor = loc(*candidate + 1);
}
for limit in 1..=expected.len().max(1) + 1 {
let mut actual = Vec::new();
let mut scan = loc(start);
loop {
let mut batch = Vec::new();
scan = fill_candidates(&bitmap, scan, tip, limit, &mut batch);
if batch.is_empty() {
break;
}
actual.extend(batch);
}
assert_eq!(
actual, expected,
"shape={name} tip={tip} limit={limit} diverged from oracle"
);
}
}
}
}
fn extract_parent_deleted_creates<K: Ord + Clone, V: Clone>(
mutations: &mut BTreeMap<K, Option<V>>,
base_diff: &[(K, DiffEntry<mmr::Family, V>)],
) -> Vec<(K, V, Option<crate::mmr::Location>)> {
let creates: Vec<_> = mutations
.iter()
.filter_map(|(key, value)| {
if let Some(DiffEntry::Deleted { base_old_loc }) = lookup_sorted(base_diff, key) {
if let Some(value) = value {
return Some((key.clone(), value.clone(), *base_old_loc));
}
}
None
})
.collect();
for (key, _, _) in &creates {
mutations.remove(key);
}
creates
}
#[test]
fn extract_parent_deleted_creates_basic() {
let mut mutations: BTreeMap<u64, Option<u64>> = BTreeMap::new();
mutations.insert(1, Some(100)); mutations.insert(2, None); mutations.insert(3, Some(300));
let mut base_diff: Vec<(u64, DiffEntry<mmr::Family, u64>)> = vec![
(
1,
DiffEntry::Deleted {
base_old_loc: Some(crate::mmr::Location::new(5)),
},
),
(
4,
DiffEntry::Active {
value: 400,
loc: crate::mmr::Location::new(10),
base_old_loc: None,
},
),
];
base_diff.sort_by_key(|a| a.0);
let creates = extract_parent_deleted_creates(&mut mutations, &base_diff);
assert_eq!(creates.len(), 1);
let (key, value, base_old_loc) = creates.first().unwrap();
assert_eq!(*key, 1);
assert_eq!(*value, 100);
assert_eq!(*base_old_loc, Some(crate::mmr::Location::new(5)));
assert_eq!(mutations.len(), 2);
assert!(mutations.contains_key(&2));
assert!(mutations.contains_key(&3));
}
#[test]
fn extract_parent_deleted_creates_delete_not_extracted() {
let mut mutations: BTreeMap<u64, Option<u64>> = BTreeMap::new();
mutations.insert(1, None);
let base_diff: Vec<(u64, DiffEntry<mmr::Family, u64>)> = vec![(
1,
DiffEntry::Deleted {
base_old_loc: Some(crate::mmr::Location::new(5)),
},
)];
let creates = extract_parent_deleted_creates(&mut mutations, &base_diff);
assert!(creates.is_empty());
assert_eq!(mutations.len(), 1);
assert!(mutations.contains_key(&1));
}
#[test]
fn apply_batch_merges_committed_and_uncommitted_overlaps() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("mixed-ancestor-overlaps", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let key_update = Sha256::hash(b"update-through-all-layers");
let key_recreate_then_delete = Sha256::hash(b"recreate-then-delete");
let key_delete_from_uncommitted = Sha256::hash(b"delete-from-uncommitted");
let key_uncommitted_create = Sha256::hash(b"uncommitted-create");
let seed = db
.new_batch()
.write(key_update, Some(Sha256::hash(b"seed-update")))
.write(
key_recreate_then_delete,
Some(Sha256::hash(b"seed-recreate")),
)
.write(
key_delete_from_uncommitted,
Some(Sha256::hash(b"seed-delete")),
)
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(seed).await.unwrap();
let applied = db
.new_batch()
.write(key_update, Some(Sha256::hash(b"committed-update")))
.write(key_recreate_then_delete, None)
.write(
key_delete_from_uncommitted,
Some(Sha256::hash(b"committed-delete-base")),
)
.merkleize(&db, None)
.await
.unwrap();
let pending = applied
.new_batch::<Sha256>()
.write(key_update, Some(Sha256::hash(b"uncommitted-update")))
.write(
key_recreate_then_delete,
Some(Sha256::hash(b"uncommitted-recreate")),
)
.write(key_delete_from_uncommitted, None)
.write(
key_uncommitted_create,
Some(Sha256::hash(b"uncommitted-create")),
)
.merkleize(&db, None)
.await
.unwrap();
let final_update = Sha256::hash(b"child-update");
let child = pending
.new_batch::<Sha256>()
.write(key_update, Some(final_update))
.write(key_recreate_then_delete, None)
.merkleize(&db, None)
.await
.unwrap();
let expected_root = child.root();
db.apply_batch(applied).await.unwrap();
db.apply_batch(child).await.unwrap();
assert_eq!(db.root(), expected_root);
assert_eq!(db.get(&key_update).await.unwrap(), Some(final_update));
assert_eq!(db.get(&key_recreate_then_delete).await.unwrap(), None);
assert_eq!(db.get(&key_delete_from_uncommitted).await.unwrap(), None);
assert_eq!(
db.get(&key_uncommitted_create).await.unwrap(),
Some(Sha256::hash(b"uncommitted-create"))
);
db.destroy().await.unwrap();
});
}
macro_rules! bulk_update_paths_match_explicit_writes_test {
($name:ident, $db:ident, $partition:literal, $shift:literal) => {
#[test]
fn $name() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = $db<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>($partition, &context);
let mut db = TestDb::init(context, config).await.unwrap();
let k0 = colliding_digest(0x40 + $shift, 0);
let k1 = colliding_digest(0x40 + $shift, 1);
let k2 = colliding_digest(0x41 + $shift, 0);
let missing = colliding_digest(0x40 + $shift, 9);
let read_only = colliding_digest(0x41 + $shift, 1);
let unread_existing = colliding_digest(0x41 + $shift, 2);
let unread_missing = colliding_digest(0x40 + $shift, 10);
let del_read = colliding_digest(0x41 + $shift, 3);
let del_unread = colliding_digest(0x41 + $shift, 4);
let v0 = colliding_digest(0x50 + $shift, 0);
let v1 = colliding_digest(0x50 + $shift, 1);
let v2 = colliding_digest(0x51 + $shift, 0);
let read_only_value = colliding_digest(0x51 + $shift, 1);
let unread_existing_value = colliding_digest(0x51 + $shift, 2);
let del_read_value = colliding_digest(0x51 + $shift, 3);
let del_unread_value = colliding_digest(0x51 + $shift, 4);
let seed = db
.new_batch()
.write(k0, Some(v0))
.write(k1, Some(v1))
.write(k2, Some(v2))
.write(read_only, Some(read_only_value))
.write(unread_existing, Some(unread_existing_value))
.write(del_read, Some(del_read_value))
.write(del_unread, Some(del_unread_value))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let read_keys = [k0, read_only, missing, k1, k0, missing, k2, del_read];
let keys: Vec<_> = read_keys.iter().collect();
let indexed_updates = vec![
(0, Some(colliding_digest(0x60 + $shift, 0))),
(2, Some(colliding_digest(0x60 + $shift, 1))),
(3, Some(colliding_digest(0x60 + $shift, 2))),
(4, Some(colliding_digest(0x60 + $shift, 3))),
(5, Some(colliding_digest(0x60 + $shift, 4))),
(6, Some(colliding_digest(0x60 + $shift, 5))),
(7, None),
];
let upserts = vec![
(unread_existing, Some(colliding_digest(0x60 + $shift, 6))),
(unread_missing, Some(colliding_digest(0x60 + $shift, 7))),
(k0, Some(colliding_digest(0x60 + $shift, 8))),
(del_unread, None),
];
let loaded_values = vec![
Some(v0),
Some(read_only_value),
None,
Some(v1),
Some(v0),
None,
Some(v2),
Some(del_read_value),
];
let mut explicit = db.new_batch();
let explicit_values = explicit.get_many(&keys, &db).await.unwrap();
for (slot, value) in &indexed_updates {
explicit = explicit.write(read_keys[*slot], *value);
}
for (key, value) in &upserts {
explicit = explicit.write(*key, *value);
}
let explicit = explicit.merkleize(&db, None).await.unwrap();
let (staged_values, staged) = db.new_batch().stage(&keys, &db).await.unwrap();
let staged_merkleized = staged
.merkleize(indexed_updates.clone(), upserts.clone(), None, &db)
.await
.unwrap();
let split = 3;
let (mut expanded_values, staged) =
db.new_batch().stage(&keys[..split], &db).await.unwrap();
let (range, suffix_values, staged) =
staged.expand(&keys[split..], &db).await.unwrap();
assert_eq!(range, split..keys.len());
expanded_values.extend(suffix_values);
let expanded = staged
.merkleize(indexed_updates.clone(), upserts.clone(), None, &db)
.await
.unwrap();
assert_eq!(explicit_values, loaded_values);
assert_eq!(explicit_values, staged_values);
assert_eq!(explicit_values, expanded_values);
assert_eq!(explicit.root(), staged_merkleized.root());
assert_eq!(explicit.root(), expanded.root());
db.apply_batch(expanded).await.unwrap();
assert_eq!(db.get(&k0).await.unwrap(), upserts[2].1);
assert_eq!(db.get(&missing).await.unwrap(), indexed_updates[4].1);
assert_eq!(db.get(&k1).await.unwrap(), indexed_updates[2].1);
assert_eq!(db.get(&k2).await.unwrap(), indexed_updates[5].1);
assert_eq!(db.get(&read_only).await.unwrap(), Some(read_only_value));
assert_eq!(db.get(&unread_existing).await.unwrap(), upserts[0].1);
assert_eq!(db.get(&unread_missing).await.unwrap(), upserts[1].1);
assert_eq!(db.get(&del_read).await.unwrap(), None);
assert_eq!(db.get(&del_unread).await.unwrap(), None);
db.destroy().await.unwrap();
});
}
};
}
bulk_update_paths_match_explicit_writes_test!(
unordered_bulk_update_paths_match_explicit_writes,
UnorderedFixedDb,
"unordered-bulk-load-update",
0
);
bulk_update_paths_match_explicit_writes_test!(
ordered_bulk_update_paths_match_explicit_writes,
OrderedFixedDb,
"ordered-bulk-load-update",
2
);
fn staged_with<F: Family, H: Hasher, U: update::Update + Send + Sync, S: Strategy>(
batch: UnmerkleizedBatch<F, H, U, S>,
keys: Vec<U::Key>,
resolutions: Vec<StagedResolution<F, U>>,
) -> Staged<F, H, U, S>
where
Operation<F, U>: Codec,
{
Staged {
batch,
keys: StagedKeys::new(keys),
resolutions,
}
}
#[test]
fn unordered_staged_resolve_updates_collapses_duplicates_before_sorting() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
type TestUpdate = update::Unordered<sha256::Digest, FixedEncoding<sha256::Digest>>;
let config = fixed_db_config::<OneCap>("unordered-staged-resolve-updates", &context);
let db = TestDb::init(context, config).await.unwrap();
let k0 = colliding_digest(0x90, 0);
let k1 = colliding_digest(0x90, 1);
let k2 = colliding_digest(0x90, 2);
let k3 = colliding_digest(0x90, 3);
let old0 = colliding_digest(0x91, 0);
let old1 = colliding_digest(0x91, 1);
let new0 = colliding_digest(0x91, 2);
let staged_k2 = colliding_digest(0x91, 3);
let fallback = colliding_digest(0x91, 4);
let upsert = colliding_digest(0x91, 5);
let staged = staged_with::<mmr::Family, Sha256, TestUpdate, Sequential>(
db.new_batch(),
vec![k0, k1, k0, k2, k1, k3],
vec![
Some((committed(30), ())),
Some((committed(10), ())),
Some((committed(30), ())),
Some((committed(40), ())),
Some((committed(10), ())),
None,
],
);
let (batch, staged_updates) = staged.resolve_updates(
vec![
(0, Some(old0)),
(1, Some(old1)),
(2, Some(new0)),
(3, Some(staged_k2)),
(4, None),
(5, Some(fallback)),
],
vec![(k2, Some(upsert))],
&Sequential,
);
assert_eq!(
staged_updates,
vec![
(k1, committed(10), (), None),
(k0, committed(30), (), Some(new0))
]
);
assert_eq!(batch.mutations.len(), 2);
assert_eq!(batch.mutations.get(&k2), Some(&Some(upsert)));
assert_eq!(batch.mutations.get(&k3), Some(&Some(fallback)));
assert!(!batch.mutations.contains_key(&k0));
assert!(!batch.mutations.contains_key(&k1));
db.destroy().await.unwrap();
});
}
#[test]
fn unordered_staged_resolve_updates_collapses_duplicates_at_scale() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
type TestUpdate = update::Unordered<sha256::Digest, FixedEncoding<sha256::Digest>>;
let config =
fixed_db_config::<OneCap>("unordered-staged-resolve-updates-scale", &context);
let db = TestDb::init(context, config).await.unwrap();
let n: usize = 512;
let keys: Vec<_> = (0..n).map(|i| colliding_digest(0xA0, i as u64)).collect();
let old_values: Vec<_> = (0..n).map(|i| colliding_digest(0xB0, i as u64)).collect();
let new_values: Vec<_> = (0..n).map(|i| colliding_digest(0xB1, i as u64)).collect();
let mut_newer = colliding_digest(0xB2, 0);
let staged_newer = colliding_digest(0xB2, 1);
let overlapped = colliding_digest(0xB2, 2);
let mut_old = colliding_digest(0xB3, 0);
let mut_new = colliding_digest(0xB3, 1);
let staged_old = colliding_digest(0xB3, 2);
let staged_new = colliding_digest(0xB3, 3);
let overlapped_write = colliding_digest(0xB3, 4);
let upsert = colliding_digest(0xB3, 5);
let mut staged_keys = keys.clone();
staged_keys.extend(keys.iter().cloned());
staged_keys.extend([mut_newer, mut_newer, staged_newer, staged_newer, overlapped]);
let mut resolutions: Vec<Option<(StagedLoc<mmr::Family>, ())>> = (0..2 * n)
.map(|slot| Some((committed(1_000 + (slot % n) as u64), ())))
.collect();
resolutions.extend([
Some((committed(500), ())),
None,
None,
Some((committed(501), ())),
Some((committed(502), ())),
]);
let staged = staged_with::<mmr::Family, Sha256, TestUpdate, Sequential>(
db.new_batch(),
staged_keys,
resolutions,
);
let mut updates: Vec<(usize, Option<sha256::Digest>)> =
vec![(2 * n, Some(mut_old)), (2 * n + 2, Some(staged_old))];
updates.extend((0..n).map(|i| (i, Some(old_values[i]))));
updates.push((2 * n + 4, Some(overlapped_write)));
updates.extend((0..n).map(|i| (n + i, Some(new_values[i]))));
updates.extend([(2 * n + 1, Some(mut_new)), (2 * n + 3, Some(staged_new))]);
let (batch, staged_updates) =
staged.resolve_updates(updates, vec![(overlapped, Some(upsert))], &Sequential);
let mut expected = vec![(staged_newer, committed(501), (), Some(staged_new))];
expected.extend((0..n).map(|i| {
(
keys[i],
committed(1_000 + i as u64),
(),
Some(new_values[i]),
)
}));
assert_eq!(staged_updates, expected);
assert_eq!(batch.mutations.len(), 2);
assert_eq!(batch.mutations.get(&mut_newer), Some(&Some(mut_new)));
assert_eq!(batch.mutations.get(&overlapped), Some(&Some(upsert)));
db.destroy().await.unwrap();
});
}
#[test]
fn unordered_staged_merkleize_discards_prior_mutation_for_cached_update() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
type TestUpdate = update::Unordered<sha256::Digest, FixedEncoding<sha256::Digest>>;
let config = fixed_db_config::<OneCap>("unordered-staged-prior-mutation", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let key = colliding_digest(0x95, 0);
let old = colliding_digest(0x95, 1);
let prior = colliding_digest(0x95, 2);
let replacement = colliding_digest(0x95, 3);
let seed = db
.new_batch()
.write(key, Some(old))
.merkleize(&db, None)
.await
.unwrap();
let old_loc = lookup_sorted(seed.diff.as_slice(), &key)
.and_then(DiffEntry::loc)
.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let explicit = db
.new_batch()
.write(key, Some(prior))
.write(key, Some(replacement))
.merkleize(&db, None)
.await
.unwrap();
let staged = staged_with::<mmr::Family, Sha256, TestUpdate, Sequential>(
db.new_batch().write(key, Some(prior)),
vec![key],
vec![Some((StagedLoc::Committed(old_loc), ()))],
);
let staged = staged
.merkleize(vec![(0, Some(replacement))], Vec::new(), None, &db)
.await
.unwrap();
assert_eq!(explicit.root(), staged.root());
db.destroy().await.unwrap();
});
}
#[test]
fn ordered_staged_resolve_updates_keeps_deletes_as_mutations() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = OrderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
type TestUpdate = update::Ordered<sha256::Digest, FixedEncoding<sha256::Digest>>;
let config = fixed_db_config::<OneCap>("ordered-staged-resolve-updates", &context);
let db = TestDb::init(context, config).await.unwrap();
let delete_key = colliding_digest(0x92, 0);
let update_a = colliding_digest(0x92, 1);
let update_b = colliding_digest(0x92, 2);
let next_delete = colliding_digest(0x93, 0);
let next_a = colliding_digest(0x93, 1);
let next_b = colliding_digest(0x93, 2);
let value_a = colliding_digest(0x94, 0);
let value_b = colliding_digest(0x94, 1);
let staged = staged_with::<mmr::Family, Sha256, TestUpdate, Sequential>(
db.new_batch(),
vec![delete_key, update_a, update_b],
vec![
Some((committed(11), next_delete)),
Some((committed(30), next_a)),
Some((committed(7), next_b)),
],
);
let (batch, staged_updates) = staged.resolve_updates(
vec![(0, None), (1, Some(value_a)), (2, Some(value_b))],
Vec::new(),
&Sequential,
);
assert_eq!(
staged_updates,
vec![
(update_b, committed(7), next_b, Some(value_b)),
(update_a, committed(30), next_a, Some(value_a)),
]
);
assert_eq!(batch.mutations.len(), 1);
assert_eq!(batch.mutations.get(&delete_key), Some(&None));
assert!(!batch.mutations.contains_key(&update_a));
assert!(!batch.mutations.contains_key(&update_b));
db.destroy().await.unwrap();
});
}
#[test]
#[should_panic(expected = "update index out of staged read range")]
fn staged_merkleize_rejects_out_of_range_update_index() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("staged-bad-index", &context);
let db = TestDb::init(context, config).await.unwrap();
let k0 = colliding_digest(0x40, 0);
let keys = vec![&k0];
let (_values, staged) = db.new_batch().stage(&keys, &db).await.unwrap();
let _ = staged
.merkleize(
vec![(1, Some(colliding_digest(0x50, 0)))],
Vec::new(),
None,
&db,
)
.await;
});
}
macro_rules! staged_updates_survive_ancestor_commit_test {
(
$name:ident, $db:ident, $key_prefix:literal, $val_prefix:literal,
$read_label:literal, $write_label:literal
) => {
#[test]
fn $name() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = $db<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let key = |i| colliding_digest($key_prefix, i);
let val = |i| colliding_digest($val_prefix, i);
let suffixes: Vec<u64> = (1..10).chain(20..30).chain([0]).collect();
let indexed_updates: Vec<_> = suffixes
.iter()
.enumerate()
.map(|(slot, suffix)| (slot, Some(val(suffix + 3_000))))
.collect();
let mut roots = Vec::new();
for staged_read in [false, true] {
let label = if staged_read {
$read_label
} else {
$write_label
};
let context = context.child(label);
let config = fixed_db_config::<OneCap>(label, &context);
let mut db = TestDb::init(context, config).await.unwrap();
let mut seed = db.new_batch();
for i in 0..100u64 {
seed = seed.write(key(i), Some(val(i)));
}
let seed = seed.merkleize(&db, None).await.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let mut grandparent = db.new_batch();
for i in 0..10u64 {
grandparent = grandparent.write(key(i), Some(val(i + 1_000)));
}
let grandparent = grandparent.merkleize(&db, None).await.unwrap();
let mut parent = grandparent.new_batch::<Sha256>();
for i in 50..60u64 {
parent = parent.write(key(i), Some(val(i + 2_000)));
}
let parent = parent.merkleize(&db, None).await.unwrap();
let child = if staged_read {
let read_keys: Vec<_> =
suffixes.iter().map(|suffix| key(*suffix)).collect();
let keys: Vec<_> = read_keys.iter().collect();
let child = parent.new_batch::<Sha256>();
let split = 15;
let (mut values, staged) =
child.stage(&keys[..split], &db).await.unwrap();
db.apply_batch(grandparent).await.unwrap();
db.commit().await.unwrap();
let (range, suffix_values, staged) =
staged.expand(&keys[split..], &db).await.unwrap();
assert_eq!(range, split..keys.len());
values.extend(suffix_values);
for (slot, suffix) in suffixes.iter().enumerate() {
let expected = if *suffix < 10 {
val(suffix + 1_000)
} else {
val(*suffix)
};
assert_eq!(values[slot], Some(expected));
}
staged
.merkleize(indexed_updates.clone(), Vec::new(), None, &db)
.await
.unwrap()
} else {
let mut child = parent.new_batch::<Sha256>();
db.apply_batch(grandparent).await.unwrap();
db.commit().await.unwrap();
for suffix in &suffixes {
child = child.write(key(*suffix), Some(val(suffix + 3_000)));
}
child.merkleize(&db, None).await.unwrap()
};
db.apply_batch(parent).await.unwrap();
db.apply_batch(child).await.unwrap();
db.commit().await.unwrap();
for suffix in &suffixes {
assert_eq!(
db.get(&key(*suffix)).await.unwrap(),
Some(val(suffix + 3_000))
);
}
roots.push(db.root());
db.destroy().await.unwrap();
}
assert_eq!(roots[0], roots[1]);
});
}
};
}
staged_updates_survive_ancestor_commit_test!(
unordered_staged_updates_survive_ancestor_commit,
UnorderedFixedDb,
0x80,
0x81,
"unordered_staged_ancestor_read",
"unordered_staged_ancestor_write"
);
staged_updates_survive_ancestor_commit_test!(
ordered_staged_updates_survive_ancestor_commit,
OrderedFixedDb,
0x82,
0x83,
"ordered_staged_ancestor_read",
"ordered_staged_ancestor_write"
);
#[test]
fn read_ops_resolves_committed_ancestor_and_current_sources() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("read-locations-all-sources", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let key_db = colliding_digest(0x30, 0);
let value_db = colliding_digest(0x30, 1);
let key_parent = colliding_digest(0x31, 0);
let value_parent = colliding_digest(0x31, 1);
let key_current = colliding_digest(0x32, 0);
let value_current = colliding_digest(0x32, 1);
let seed = db
.new_batch()
.write(key_db, Some(value_db))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let committed_loc = db.snapshot.get(&key_db).next().copied().unwrap();
let parent = db
.new_batch()
.write(key_parent, Some(value_parent))
.merkleize(&db, None)
.await
.unwrap();
let parent_loc = lookup_sorted(parent.diff.as_slice(), &key_parent)
.unwrap()
.loc()
.unwrap();
let child = parent
.new_batch::<Sha256>()
.write(key_current, Some(value_current));
let (_mutations, merkleizer) = child.into_parts();
let current_loc = Location::new(merkleizer.base_size);
let batch_ops = vec![Operation::Update(update::Unordered(
key_current,
value_current,
))];
let ops = merkleizer
.read_ops(
&[current_loc, committed_loc, parent_loc, committed_loc],
&batch_ops,
&db.log,
)
.await
.unwrap();
assert_eq!(
ops,
vec![
Operation::Update(update::Unordered(key_current, value_current)),
Operation::Update(update::Unordered(key_db, value_db)),
Operation::Update(update::Unordered(key_parent, value_parent)),
Operation::Update(update::Unordered(key_db, value_db)),
]
);
db.destroy().await.unwrap();
});
}
#[test]
fn child_root_matches_between_pending_and_committed_paths_under_collisions() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("batch-collision-regression", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let key_a = colliding_digest(0xAA, 1);
let key_b = colliding_digest(0xAA, 0);
let mut initial = db.new_batch();
for i in 0..4 {
initial = initial.write(colliding_digest(0xAA, i), Some(colliding_digest(0xBB, i)));
}
let initial = initial.merkleize(&db, None).await.unwrap();
db.apply_batch(initial).await.unwrap();
db.commit().await.unwrap();
let parent = db
.new_batch()
.write(key_a, Some(colliding_digest(0xCC, 1)))
.merkleize(&db, None)
.await
.unwrap();
assert!(
!parent.diff.iter().any(|(k, _)| k == &key_b),
"regression requires a sibling collision to remain only in the committed snapshot"
);
let pending_child = parent
.new_batch::<Sha256>()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None)
.await
.unwrap();
let pending_root = pending_child.root();
db.apply_batch(parent).await.unwrap();
db.commit().await.unwrap();
let committed_child = db
.new_batch()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None)
.await
.unwrap();
assert_eq!(pending_root, committed_child.root());
db.apply_batch(pending_child).await.unwrap();
assert_eq!(db.root(), committed_child.root());
db.destroy().await.unwrap();
});
}
#[test]
fn ordered_child_root_matches_between_pending_and_committed_paths_under_collisions() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = OrderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("ordered-batch-collision-regression", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let key_a = colliding_digest(0xAA, 1);
let key_b = colliding_digest(0xAA, 0);
let mut initial = db.new_batch();
for i in 0..4 {
initial = initial.write(colliding_digest(0xAA, i), Some(colliding_digest(0xBB, i)));
}
let initial = initial.merkleize(&db, None).await.unwrap();
db.apply_batch(initial).await.unwrap();
db.commit().await.unwrap();
let parent = db
.new_batch()
.write(key_a, Some(colliding_digest(0xCC, 1)))
.merkleize(&db, None)
.await
.unwrap();
assert!(
!parent.diff.iter().any(|(k, _)| k == &key_b),
"ordered regression requires a sibling collision to remain only in the committed snapshot"
);
let pending_child = parent
.new_batch::<Sha256>()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None)
.await
.unwrap();
let pending_root = pending_child.root();
db.apply_batch(parent).await.unwrap();
db.commit().await.unwrap();
let committed_child = db
.new_batch()
.write(key_a, Some(colliding_digest(0xDD, 1)))
.write(key_b, Some(colliding_digest(0xDD, 0)))
.merkleize(&db, None)
.await
.unwrap();
assert_eq!(pending_root, committed_child.root());
db.apply_batch(pending_child).await.unwrap();
assert_eq!(db.root(), committed_child.root());
db.destroy().await.unwrap();
});
}
#[test]
fn sequential_commit_basic() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("seq-commit-basic", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let seed = db
.new_batch()
.write(colliding_digest(0x01, 0), Some(colliding_digest(0x01, 1)))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let key_a = colliding_digest(0x02, 0);
let val_a = colliding_digest(0x02, 1);
let batch_a = db
.new_batch()
.write(key_a, Some(val_a))
.merkleize(&db, None)
.await
.unwrap();
let key_b = colliding_digest(0x03, 0);
let val_b = colliding_digest(0x03, 1);
let batch_b = batch_a
.new_batch::<Sha256>()
.write(key_b, Some(val_b))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(batch_a).await.unwrap();
db.commit().await.unwrap();
let committed_b = db
.new_batch()
.write(key_b, Some(val_b))
.merkleize(&db, None)
.await
.unwrap();
assert_eq!(batch_b.root(), committed_b.root());
db.apply_batch(batch_b).await.unwrap();
assert_eq!(db.root(), committed_b.root());
db.destroy().await.unwrap();
});
}
#[test]
fn sequential_commit_fixes_base_old_loc() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("seq-commit-base-old-loc", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let key = colliding_digest(0x10, 0);
let seed = db
.new_batch()
.write(key, Some(colliding_digest(0x10, 1)))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let val_a = colliding_digest(0x10, 2);
let batch_a = db
.new_batch()
.write(key, Some(val_a))
.merkleize(&db, None)
.await
.unwrap();
let a_entry = lookup_sorted(batch_a.diff.as_slice(), &key).unwrap();
let a_loc = a_entry.loc();
assert!(a_loc.is_some());
let val_b = colliding_digest(0x10, 3);
let batch_b = batch_a
.new_batch::<Sha256>()
.write(key, Some(val_b))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(batch_a).await.unwrap();
db.commit().await.unwrap();
let committed_b = db
.new_batch()
.write(key, Some(val_b))
.merkleize(&db, None)
.await
.unwrap();
assert_eq!(batch_b.root(), committed_b.root());
db.apply_batch(batch_b).await.unwrap();
assert_eq!(db.root(), committed_b.root());
db.destroy().await.unwrap();
});
}
#[test]
fn fork_apply_after_parent_committed() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("fork-after-commit", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let seed = db
.new_batch()
.write(colliding_digest(0x20, 0), Some(colliding_digest(0x20, 1)))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let key_a = colliding_digest(0x21, 0);
let val_a = colliding_digest(0x21, 1);
let batch_a = db
.new_batch()
.write(key_a, Some(val_a))
.merkleize(&db, None)
.await
.unwrap();
let key_b = colliding_digest(0x22, 0);
let val_b = colliding_digest(0x22, 1);
let batch_b = batch_a
.new_batch::<Sha256>()
.write(key_b, Some(val_b))
.merkleize(&db, None)
.await
.unwrap();
let key_c = colliding_digest(0x23, 0);
let val_c = colliding_digest(0x23, 1);
let batch_c = batch_a
.new_batch::<Sha256>()
.write(key_c, Some(val_c))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(batch_a).await.unwrap();
db.commit().await.unwrap();
let committed_b = db
.new_batch()
.write(key_b, Some(val_b))
.merkleize(&db, None)
.await
.unwrap();
assert_eq!(batch_b.root(), committed_b.root());
let committed_c = db
.new_batch()
.write(key_c, Some(val_c))
.merkleize(&db, None)
.await
.unwrap();
assert_eq!(batch_c.root(), committed_c.root());
db.destroy().await.unwrap();
});
}
#[test]
fn sequential_commit_three_deep() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("ff-cross", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let grandparent = db
.new_batch()
.write(colliding_digest(0x01, 0), Some(colliding_digest(0x01, 1)))
.write(colliding_digest(0x02, 0), Some(colliding_digest(0x02, 1)))
.merkleize(&db, None)
.await
.unwrap();
let parent = grandparent
.new_batch::<Sha256>()
.write(colliding_digest(0x03, 0), Some(colliding_digest(0x03, 1)))
.merkleize(&db, None)
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.write(colliding_digest(0x04, 0), Some(colliding_digest(0x04, 1)))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(grandparent).await.unwrap();
db.commit().await.unwrap();
db.apply_batch(parent).await.unwrap();
db.commit().await.unwrap();
db.apply_batch(child).await.unwrap();
for i in 1..=4 {
assert_eq!(
db.get(&colliding_digest(i, 0)).await.unwrap(),
Some(colliding_digest(i, 1))
);
}
db.destroy().await.unwrap();
});
}
#[test]
fn recreate_deleted_key_with_collision_sibling_root_matches() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("recreate-deleted-collision", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let k0 = colliding_digest(0xAA, 0);
let k6 = colliding_digest(0xAA, 6);
let initial = db
.new_batch()
.write(k0, Some(colliding_digest(0xBB, 0)))
.write(k6, Some(colliding_digest(0xBB, 6)))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(initial).await.unwrap();
db.commit().await.unwrap();
let parent = db
.new_batch()
.write(k0, None)
.merkleize(&db, None)
.await
.unwrap();
let k29 = colliding_digest(0xAA, 29);
let pending_child = parent
.new_batch::<Sha256>()
.write(k0, Some(colliding_digest(0xCC, 0)))
.write(k29, Some(colliding_digest(0xCC, 29)))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(parent).await.unwrap();
db.commit().await.unwrap();
let committed_child = db
.new_batch()
.write(k0, Some(colliding_digest(0xCC, 0)))
.write(k29, Some(colliding_digest(0xCC, 29)))
.merkleize(&db, None)
.await
.unwrap();
assert_eq!(
pending_child.root(),
committed_child.root(),
"root depended on pending-vs-committed parent path \
when re-creating a deleted key with collision siblings"
);
db.destroy().await.unwrap();
});
}
#[test]
fn get_many_resolves_mutation_parent_and_db() {
let runner = deterministic::Runner::default();
runner.start(|context| async move {
type TestDb = UnorderedFixedDb<
mmr::Family,
deterministic::Context,
sha256::Digest,
sha256::Digest,
Sha256,
OneCap,
Sequential,
>;
let config = fixed_db_config::<OneCap>("get-many-basic", &context);
let mut db = TestDb::init(context, config).await.unwrap();
let key_db = colliding_digest(0x40, 0);
let val_db = colliding_digest(0x40, 1);
let key_parent = colliding_digest(0x41, 0);
let val_parent = colliding_digest(0x41, 1);
let key_batch = colliding_digest(0x42, 0);
let val_batch = colliding_digest(0x42, 1);
let key_missing = colliding_digest(0x43, 0);
let seed = db
.new_batch()
.write(key_db, Some(val_db))
.merkleize(&db, None)
.await
.unwrap();
db.apply_batch(seed).await.unwrap();
db.commit().await.unwrap();
let results = db.get_many(&[&key_db, &key_missing]).await.unwrap();
assert_eq!(results, vec![Some(val_db), None]);
let batch = db.new_batch().write(key_batch, Some(val_batch));
let results = batch
.get_many(&[&key_batch, &key_db, &key_missing], &db)
.await
.unwrap();
assert_eq!(results, vec![Some(val_batch), Some(val_db), None]);
let parent = db
.new_batch()
.write(key_parent, Some(val_parent))
.merkleize(&db, None)
.await
.unwrap();
let child = parent
.new_batch::<Sha256>()
.write(key_batch, Some(val_batch));
let results = child
.get_many(&[&key_batch, &key_parent, &key_db, &key_missing], &db)
.await
.unwrap();
assert_eq!(
results,
vec![Some(val_batch), Some(val_parent), Some(val_db), None]
);
let results = parent
.get_many(&[&key_parent, &key_db, &key_missing], &db)
.await
.unwrap();
assert_eq!(results, vec![Some(val_parent), Some(val_db), None]);
let results: Vec<Option<sha256::Digest>> =
db.get_many(&([] as [&sha256::Digest; 0])).await.unwrap();
assert!(results.is_empty());
db.destroy().await.unwrap();
});
}
}