//! A mutable key-value database that supports variable-sized values, but without authentication. //! //! # Example //! //! ```rust //! use commonware_storage::{ //! journal::contiguous::variable::Config as JournalConfig, //! qmdb::{ //! floor::Proportional, //! store::db::{Config, Db}, //! }, //! translator::TwoCap, //! }; //! use commonware_utils::{NZUsize, NZU16, NZU64}; //! use commonware_cryptography::{blake3::Digest, Digest as _}; //! use commonware_math::algebra::Random; //! use commonware_runtime::{ //! buffer::paged::CacheRef, deterministic::Runner, Metrics, Runner as _, Supervisor as _, //! }; //! //! use std::num::NonZeroU16; //! const PAGE_SIZE: NonZeroU16 = NZU16!(8192); //! const PAGE_CACHE_SIZE: usize = 100; //! //! let executor = Runner::default(); //! executor.start(|mut ctx| async move { //! let config = Config { //! log: JournalConfig { //! partition: "test-partition".into(), //! write_buffer: NZUsize!(64 * 1024), //! replay_buffer: NZUsize!(64 * 1024), //! compression: None, //! codec_config: ((), ()), //! items_per_section: NZU64!(4), //! page_cache: CacheRef::new(PAGE_SIZE, NZUsize!(PAGE_CACHE_SIZE)), //! }, //! translator: TwoCap, //! init_cache: Some(NZUsize!(1 << 16)), //! init_buffer: NZUsize!(1 << 21), //! }; //! let db = //! Db::<_, Digest, Digest, TwoCap>::init(ctx.child("store"), config, None) //! .await //! .unwrap(); //! //! // Insert a key-value pair //! let k = Digest::random(&mut ctx); //! let v = Digest::random(&mut ctx); //! let metadata = Some(Digest::random(&mut ctx)); //! let batch = db.new_batch().update(k, v).finalize(metadata); //! let db = db.apply_batch(batch, &mut Proportional).await.unwrap().commit().await.unwrap(); //! //! // Fetch the value //! let fetched_value = db.get(&k).await.unwrap(); //! assert_eq!(fetched_value.unwrap(), v); //! //! // Delete the key's value //! let batch = db.new_batch().delete(k).finalize(None); //! let db = db.apply_batch(batch, &mut Proportional).await.unwrap().commit().await.unwrap(); //! //! // Fetch the value //! let fetched_value = db.get(&k).await.unwrap(); //! assert!(fetched_value.is_none()); //! //! // Destroy the store //! db.destroy().await.unwrap(); //! }); //! ``` //! //! ```ignore //! // Apply a batch and commit it, then build a child batch from the newly published state //! // and apply it. Each mutation takes the database and returns it, so committing and //! // building run in sequence on the threaded handle. //! let batch = db.new_batch().update(key_a, value_a).finalize(None); //! let db = db.apply_batch(batch, &mut Proportional).await?.commit().await?; //! //! let child = db.new_batch().update(key_b, value_b).finalize(None); //! let db = db.apply_batch(child, &mut Proportional).await?.commit().await?; //! ``` use crate::{ Context, index::{Cursor as _, Unordered as _, unordered::Index}, journal::{ authenticated::{Backing as _, BackingRecovery as _}, contiguous::{ Contiguous, Many, variable::{Config as JournalConfig, Journal}, }, }, merkle::mmr::Location, qmdb::{ any::{ BITMAP_CHUNK_BYTES, VariableValue, unordered::{Update, variable::Operation}, }, bitmap::fill_from, build_snapshot_from_log, delete_known_loc, floor::{Action, Entry, Limits, Policy, Walk}, operation::{Committable as _, Floored as _, Key, Operation as _}, update_known_loc, }, translator::Translator, }; use commonware_codec::{CodecShared, Read}; use commonware_macros::boxed; use commonware_runtime::Handle; use commonware_utils::{Widen, bitmap}; use core::num::NonZeroUsize; use std::collections::{BTreeMap, HashMap}; use tracing::{debug, warn}; type Error = crate::qmdb::Error; /// Configuration for initializing a [Db]. #[derive(Clone)] pub struct Config { /// Configuration for the variable-size operations log journal. pub log: JournalConfig, /// The [Translator] used by the [Index]. pub translator: T, /// Maximum number of entries in the `(location -> key)` cache used during init to resolve /// snapshot collisions without re-reading the log; `None` disables it. pub init_cache: Option, /// Size (in bytes) of the read buffer used to replay the log during init. pub init_buffer: NonZeroUsize, } /// A finalized batch of writes and deletes ready to be applied to the store. pub struct Changeset { diff: BTreeMap>, metadata: Option, } impl FromIterator<(K, Option)> for Changeset { fn from_iter)>>(iter: TIter) -> Self { Self { diff: iter.into_iter().collect(), metadata: None, } } } impl From<[(K, Option); N]> for Changeset { fn from(items: [(K, Option); N]) -> Self { items.into_iter().collect() } } /// A mutable batch of writes and deletes staged against the current store state. pub struct Batch<'a, E, K, V, T> where E: Context, K: Key, V: VariableValue, T: Translator, { db: &'a Db, diff: BTreeMap>, } impl<'a, E, K, V, T> Batch<'a, E, K, V, T> where E: Context, K: Key, V: VariableValue, T: Translator, { const fn new(db: &'a Db) -> Self { Self { db, diff: BTreeMap::new(), } } /// Finalize the batch into a changeset that can be applied to the store. pub fn finalize(self, metadata: Option) -> Changeset { Changeset { diff: self.diff, metadata, } } /// Get the value of `key` in the batch, or the value in the store if it has /// not been modified by the batch. pub async fn get(&self, key: &K) -> Result, Error> { if let Some(value) = self.diff.get(key) { return Ok(value.clone()); } self.db.get(key).await } /// Update the value of `key` in the batch. pub fn update(mut self, key: K, value: V) -> Self { self.diff.insert(key, Some(value)); self } /// Delete the value of `key` in the batch. pub fn delete(mut self, key: K) -> Self { self.diff.insert(key, None); self } } /// An unauthenticated key-value database based off of an append-only [Journal] of operations. pub struct Db where E: Context, K: Key, V: VariableValue, T: Translator, { /// A log of all [Operation]s that have been applied to the store. /// /// # Invariants /// /// - There is always at least one commit operation in the log. /// - The log is never pruned beyond the inactivity floor. log: Journal>, /// A snapshot of all currently active operations in the form of a map from each key to the /// location containing its most recent update. /// /// # Invariant /// /// Only references operations of type [Operation::Update]. snapshot: Index, /// The number of active keys in the store. active_keys: usize, /// Activity status of each retained operation. /// /// # Invariants /// /// - `bitmap.len() == log.size()`. /// - Bit `i` is set iff location `i` holds the current update of an active key, or is the /// last commit. bitmap: bitmap::Prunable, /// A location before which all operations are "inactive" (that is, operations before this point /// are over keys that have been updated by some operation at or after this point). inactivity_floor_loc: Location, } impl std::fmt::Debug for Db where E: Context, K: Key, V: VariableValue, T: Translator, { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Db") .field("bounds", &self.bounds()) .field("inactivity_floor_loc", &self.inactivity_floor_loc()) .finish_non_exhaustive() } } impl Db where E: Context, K: Key, V: VariableValue, T: Translator, { /// Get the value of `key` in the db, or None if it has no value. pub async fn get(&self, key: &K) -> Result, Error> { for &loc in self.snapshot.get(key) { let Operation::Update(Update(k, v)) = self.get_op(loc).await? else { unreachable!("location ({loc}) does not reference update operation"); }; if &k == key { return Ok(Some(v)); } } Ok(None) } /// Returns a new empty batch of changes. pub const fn new_batch(&self) -> Batch<'_, E, K, V, T> { Batch::new(self) } /// Whether the db currently has no active keys. pub const fn is_empty(&self) -> bool { self.active_keys == 0 } /// Gets a [Operation] from the log at the given location. Returns [Error::OperationPruned] /// if the location precedes the oldest retained location. The location is otherwise assumed /// valid. async fn get_op(&self, loc: Location) -> Result, Error> { assert!(*loc < self.log.bounds().end); self.log.read(*loc).await.map_err(|e| match e { crate::journal::Error::ItemPruned(_) => Error::OperationPruned(loc), e => Error::Journal(e), }) } /// Return [start, end) where `start` and `end - 1` are the Locations of the oldest and newest /// retained operations respectively. pub fn bounds(&self) -> std::ops::Range { let bounds = self.log.bounds(); Location::new(bounds.start)..Location::new(bounds.end) } /// Return the Location of the next operation appended to this db. pub fn size(&self) -> Location { Location::new(self.log.size()) } /// Return the inactivity floor location. This is the location before which all operations are /// known to be inactive. Operations before this point can be safely pruned. pub const fn inactivity_floor_loc(&self) -> Location { self.inactivity_floor_loc } /// Get the metadata associated with the last commit. pub async fn get_metadata(&self) -> Result, Error> { // The log always ends with a commit operation. let Operation::CommitFloor(metadata, _) = self.log.read(*self.size() - 1).await? else { unreachable!("last commit should be a commit floor operation"); }; Ok(metadata) } /// Prune historical operations prior to `prune_loc`. This does not affect the db's root /// or current snapshot. /// /// `prune` requires no prior commit. After a crash, the database remains recoverable; /// uncommitted operations are not guaranteed to survive. #[boxed] pub async fn prune(mut self, prune_loc: Location) -> Result { if prune_loc > self.inactivity_floor_loc { return Err(Error::PruneBeyondMinRequired( prune_loc, self.inactivity_floor_loc, )); } // The floor justifying the boundary may exist only in buffered operations (it // advances before its batch is durable), and pruning does not guarantee buffered // appends are durable. Commit so the justification survives the prune. self.log = self.log.commit().await?; // Prune the log. The log will prune at section boundaries, so the actual oldest retained // location may be less than requested. let pruned; (self.log, pruned) = self.log.prune(*prune_loc).await?; if !pruned { return Ok(self); } let bounds = self.log.bounds(); self.bitmap.prune_to_bit(bounds.start); let log_size = Location::new(bounds.end); let oldest_retained_loc = Location::new(bounds.start); debug!( ?log_size, ?oldest_retained_loc, ?prune_loc, "pruned inactive ops" ); Ok(self) } /// Initializes a new [Db] with the given configuration. /// `Some(max_size)` selects the latest retained commit with at most `max_size` operations. /// `None` selects the latest retained state. #[boxed] pub async fn init( context: E, cfg: Config as Read>::Cfg>, max_size: Option, ) -> Result { // Variable-journal recovery rebuilds item offsets before selecting a commit. Keep the // recovery owner unpublished until that commit and its replay floor are validated. crate::qmdb::validate_initialization_bound(max_size)?; let pending = Journal::>::recover( context.child("log"), cfg.log, max_size.map(|size| *size), ) .await?; let size = pending .last_matching(max_size.map_or(u64::MAX, |size| *size), |op| op.is_commit()) .await?; let bounds = pending.bounds(); let commit = if size == 0 { None } else { Some(pending.read(size - 1).await?) }; crate::qmdb::validate_initialization_commit( bounds.start, size, bounds == (0..0), commit.as_ref(), true, )?; // Finishing recovery publishes the selected offset prefix before releasing later // value bytes. if size < bounds.end { warn!( journal_size = bounds.end, rewound_items = bounds.end - size, "rewinding journal items" ); } let mut log = pending.finish(size).await?; if size == 0 { warn!("Log is empty, initializing new db"); (log, _) = log .append(&Operation::CommitFloor(None, Location::new(0))) .await?; } // Persist recovery repairs and any genesis commit so the next startup need not repeat them. let log = log.sync().await?; let last_commit_loc = Location::new(log.size().checked_sub(1).expect("commit should exist")); // Build the snapshot only from the durable selected prefix. let cache_size = cfg.init_cache; let init_buffer = cfg.init_buffer; let mut snapshot = Index::new(context.child("snapshot"), cfg.translator); let op = log.read(*last_commit_loc).await?; let inactivity_floor_loc = op.has_floor().expect("last op should be a commit"); // Seed the bitmap so its pruned prefix matches the retained log boundary. Operations // below the inactivity floor are inactive. let bounds = log.bounds(); let pruned_chunks = (bounds.start / bitmap::Prunable::::CHUNK_SIZE_BITS) as usize; let mut bitmap = bitmap::Prunable::new_with_pruned_chunks(pruned_chunks) .expect("pruned chunk count fits in u64 bits"); bitmap.extend_to(*inactivity_floor_loc); // Replay the log from the floor, appending each operation's status and clearing the bit // of any location it supersedes. The state after the last operation is each location's // final status. let active_keys = build_snapshot_from_log( inactivity_floor_loc, &log, &mut snapshot, init_buffer, cache_size, |is_active, old_loc| { bitmap.push(is_active); if let Some(loc) = old_loc { bitmap.set_bit(*loc, false); } }, ) .await?; assert_eq!(bitmap.len(), bounds.end); Ok(Self { log, snapshot, active_keys, bitmap, inactivity_floor_loc, }) } /// Sync all database state to disk. While this isn't necessary to ensure durability of /// committed operations, periodic invocation may reduce memory usage and the time required to /// recover the database on restart. #[boxed] pub async fn sync(mut self) -> Result { self.log = self.log.sync().await?; Ok(self) } /// Destroy the db, removing all data from disk. #[boxed] pub async fn destroy(self) -> Result<(), Error> { self.log.destroy().await.map_err(Into::into) } /// Applies a finalized batch to the in-memory database state and appends its operations to the /// journal. /// /// This publishes the batch to the in-memory database state and appends it to the journal. /// Call [`Db::commit`] or [`Db::sync`], or await the handle returned by [`Db::start_sync`], to /// make the applied state durable. /// /// After its writes and before its commit operation, the batch walks the inactivity floor /// with [`Policy`]. If the batch leaves the store empty, the floor moves to its commit. /// /// When every batch since the initial commit uses /// [`Proportional`](crate::qmdb::floor::Proportional), each batch leaves the database's size at /// most `3 * n + 1` operations past its floor, for `n` active keys. #[boxed] pub async fn apply_batch

( mut self, batch: Changeset, policy: &mut P, ) -> Result where P: Policy, { let start_loc = self.size(); let Changeset { diff, metadata } = batch; let mut resolved = { // Resolve each written key to its applied update, if any, by reading its // translated-key bucket in rounds sized by the bucket's unresolved batch keys: a // sparse write stops at its matching update, while a batch covering the bucket reads // all its updates together. // // Keys sharing a translated key see the same bucket in the same order, so the // bucket's first location identifies it. Each source pairs a bucket's unread // locations with the number of its batch keys still unresolved. A key with an empty // bucket is absent and needs no read. let mut buckets = HashMap::::new(); let mut sources: Vec<(_, usize)> = Vec::new(); for key in diff.keys() { let mut locations = self.snapshot.get(key); let Some(&first) = locations.next() else { continue; }; if let Some(&index) = buckets.get(&first) { sources[index].1 += 1; } else { buckets.insert(first, sources.len()); sources.push((core::iter::once(first).chain(locations.copied()), 1)); } } // A round takes from every live source as many unread locations as it has // unresolved keys and reads them in one ascending batch. A source that yields // nothing has run out, so its remaining keys are absent, and it leaves at the next // round. let mut resolved = HashMap::new(); let mut candidates = Vec::new(); loop { candidates.clear(); sources.retain(|(_, pending)| *pending > 0); for (index, (locations, pending)) in sources.iter_mut().enumerate() { let before = candidates.len(); candidates.extend(locations.by_ref().take(*pending).map(|loc| (loc, index))); if candidates.len() == before { *pending = 0; } } if candidates.is_empty() { break; } candidates.sort_unstable_by_key(|(loc, _)| *loc); let positions: Vec<_> = candidates.iter().map(|(loc, _)| **loc).collect(); let read = self.log.read_many(&positions).await?; // A read update resolves its key when the batch writes that key; a collision // sibling the batch leaves alone is passed over. for ((loc, index), op) in candidates.iter().zip(read) { let key = op.into_key().expect("snapshot operation has key"); if diff.contains_key(&key) { resolved.insert(key, *loc); sources[*index].1 -= 1; } } } resolved }; // Generate operations in key order. Earlier writes can change a bucket's layout, but // each surviving original update still has its resolved location. let mut ops: Vec> = Vec::with_capacity(diff.len() + 1); let mut made_inactive = 0usize; for (key, value) in diff { let new_loc = Location::new(*start_loc + Widen::widen(ops.len())); let old_loc = resolved.remove(&key); let matches = |loc: &Location| Some(*loc) == old_loc; if let Some(value) = value { if let Some(mut cursor) = self.snapshot.get_mut_or_insert(&key, new_loc) { if cursor.find(matches) { cursor.update(new_loc); } else { cursor.insert(new_loc); } } if let Some(old_loc) = old_loc { self.bitmap.set_bit(*old_loc, false); made_inactive += 1; } else { self.active_keys += 1; } self.bitmap.push(true); ops.push(Operation::Update(Update(key, value))); } else if let Some(old_loc) = old_loc { delete_known_loc(&mut self.snapshot, &key, old_loc); self.bitmap.set_bit(*old_loc, false); self.bitmap.push(false); ops.push(Operation::Delete(key)); made_inactive += 2; self.active_keys -= 1; } } // The previous commit becomes inactive. self.bitmap.set_bit(*start_loc - 1, false); // Walk the floor toward the tip the writes reached, under the limits the policy sets for // the operations they made inactive. let tip = Location::new(*start_loc + Widen::widen(ops.len())); let Limits { entries, skips } = policy.limits(made_inactive); let mut walk = Walk::new(self.inactivity_floor_loc, tip, entries, skips); self.walk(&mut walk, policy, &mut ops).await?; self.inactivity_floor_loc = walk.floor; // The writes or the policy's evictions may leave the store empty. if self.is_empty() { self.inactivity_floor_loc = Location::new(*start_loc + Widen::widen(ops.len())); debug!(tip = ?self.inactivity_floor_loc, "db is empty, raising floor to tip"); } // Append the batch's operations and its commit with the new inactivity floor. self.bitmap.push(true); ops.push(Operation::CommitFloor(metadata, self.inactivity_floor_loc)); (self.log, _) = self.log.append_many(Many::Flat(&ops)).await?; Ok(self) } /// Advance `walk` over the active updates below its end, deciding each one with `policy`. /// Each decision applies at once: the snapshot, the bitmap, and the key count change with it, /// and the update it writes or the delete it appends joins `ops`. /// /// `ops` holds the unappended operations starting at `log.size()`, and the bitmap covers them /// with the previous commit inactive. async fn walk

( &mut self, walk: &mut Walk, policy: &mut P, ops: &mut Vec>, ) -> Result<(), Error> where P: Policy, { // The bitmap is exact after the writes, so the set bits from the floor are the active // updates the walk may decide, and which of them its limits let it reach is arithmetic // over their locations. Only those are read, though a policy may stop the walk before it // decides them all. Candidates in the batch's own region already have their operations // in memory. let start = self.log.size(); let mut candidates = Vec::::with_capacity(walk.entries.min(self.active_keys)); fill_from( &self.bitmap, *walk.floor, *walk.end, walk.entries, &mut candidates, ); let reachable = walk.reachable(&candidates); // Each reached update appends at most one operation, and the commit follows. ops.reserve(reachable + 1); // Candidates ascend, so the reachable ones below `start` form a prefix the log serves in // one read; the rest are this batch's own writes, already in `ops`. let logged = candidates[..reachable].partition_point(|&loc| loc < start); let mut reads = self.log.read_many(&candidates[..logged]).await?.into_iter(); // Decide the reachable candidates in location order, taking each update from the read // or from `ops`. A write relocates the key to the tip and an eviction removes it; either // way the candidate's own location becomes inactive. A stop leaves the update active at // the floor, where a later walk starts. for &loc in &candidates[..reachable] { let reached = walk.reach(Location::new(loc)); assert!(reached, "candidate within the walk's reach"); let op = if loc < start { reads.next().expect("one read per logged candidate") } else { ops[(loc - start) as usize].clone() }; let Operation::Update(Update(key, value)) = op else { unreachable!("active candidate must be an update"); }; let new_loc = Location::new(start + Widen::widen(ops.len())); let action = policy .decide(Entry::new(Location::new(loc), &key, value)) .into_action(); let op = match action { Action::Write(value) => { update_known_loc(&mut self.snapshot, &key, Location::new(loc), new_loc); self.bitmap.push(true); Operation::Update(Update(key, value)) } Action::Evict => { delete_known_loc(&mut self.snapshot, &key, Location::new(loc)); self.active_keys -= 1; self.bitmap.push(false); Operation::Delete(key) } Action::Stop => return Ok(()), }; self.bitmap.set_bit(loc, false); ops.push(op); walk.decide(); } // The first candidate the walk cannot reach ends it with the floor advanced by the // remaining skips. Without one, the window holds no further active update. match candidates.get(reachable) { Some(&beyond) => { let reached = walk.reach(Location::new(beyond)); assert!(!reached, "candidate beyond the walk's reach"); } None if walk.entries > 0 => walk.exhaust(), None => {} } Ok(()) } /// Begin durably persisting the journal state published by prior [`Db::apply_batch`] calls. /// /// Awaiting the returned [Handle] provides the same durability guarantee as [Self::commit], /// plus a best-effort attempt to bound the recovery needed on startup. Use [Self::sync] to /// guarantee none is needed. A new sync waits for the prior sync before starting. Failures /// of the deferred durability work surface on the returned handle. A failed data sync also /// fails the next durability operation. A failed offsets or recovery-watermark sync is not /// observed by [Self::commit] and resurfaces on the next [Self::sync]. #[boxed] pub async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error> { let handle; (self.log, handle) = self.log.start_sync().await?; Ok((self, handle)) } /// Durably commit the journal state published by prior [`Db::apply_batch`] calls. #[boxed] pub async fn commit(mut self) -> Result { self.log = self.log.commit().await?; Ok(self) } } #[cfg(test)] mod test { use super::*; use crate::{ qmdb::{ any::{ batch::tests::{CountedValue, Evict, Growing}, test::{ Choice, Script, assert_bits, assert_bound, churn, colliding_digest, counter, keep, keys_after, randomized_churn, replay, walk_model, }, }, floor::{Bounded, Hold, Proportional}, }, translator::{OneCap, TwoCap}, }; use commonware_cryptography::{ Hasher as _, Sha256, blake3::{Blake3, Digest}, sha256, }; use commonware_macros::{test_collect_traces, test_traced}; use commonware_math::algebra::Random; use commonware_runtime::{ Runner, Spawner as _, Supervisor as _, buffer::paged::CacheRef, deterministic, mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs}, reschedule, telemetry::traces::collector::TraceStorage, }; use commonware_utils::{NZU16, NZU64, NZUsize, test_rng}; use core::{future::Future, hash::BuildHasher, ops::Range}; use futures::FutureExt as _; use rand::RngExt as _; use rstest::rstest; use std::{ cell::Cell, collections::BTreeSet, num::{NonZeroU16, NonZeroUsize}, sync::{ Arc, atomic::{AtomicUsize, Ordering}, }, }; const PAGE_SIZE: NonZeroU16 = NZU16!(77); const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9); /// The type of the store used in tests. type TestStore = Db, TwoCap>; fn test_config() -> Config> as Read>::Cfg> { Config { log: JournalConfig { partition: "journal".into(), write_buffer: NZUsize!(64 * 1024), replay_buffer: NZUsize!(64 * 1024), compression: None, codec_config: ((), ((0..=10000).into(), ())), items_per_section: NZU64!(7), page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE), }, translator: TwoCap, init_cache: Some(NZUsize!(1024)), init_buffer: NZUsize!(1 << 21), } } async fn create_test_store(context: deterministic::Context) -> TestStore { let cfg = test_config(); TestStore::init(context, cfg, None).await.unwrap() } async fn apply_entries( db: TestStore, iter: impl IntoIterator>)> + Send, ) -> (TestStore, Range) { let applied_start = db.size(); let db = db .apply_batch(iter.into_iter().collect(), &mut Proportional) .await .unwrap(); let applied_end = db.size(); (db, applied_start..applied_end) } #[test_traced] fn test_store_bounded_initialization_commit_selection() { for cap_case in 0..4 { deterministic::Runner::default().start(move |context| async move { // Persist two commits with distinct metadata and key state. let cfg = test_config(); let db = TestStore::init(context.child("seed"), cfg.clone(), None) .await .unwrap(); let a = Blake3::hash(&[b"a"]); let b = Blake3::hash(&[b"b"]); let c = Blake3::hash(&[b"c"]); let batch = db .new_batch() .update(a, vec![1]) .update(b, vec![2]) .finalize(Some(vec![10])); let db = db.apply_batch(batch, &mut Proportional).await.unwrap(); let first_size = db.size(); let batch = db .new_batch() .update(a, vec![3]) .delete(b) .update(c, vec![4]) .finalize(Some(vec![20])); let db = db.apply_batch(batch, &mut Proportional).await.unwrap(); let latest_size = db.size(); _ = db.sync().await.unwrap(); // Exact, in-between, equal-tip, and above-tip caps must select the latest commit at // or below the bound. let cap = match cap_case { 0 => first_size, 1 => first_size + 1, 2 => latest_size, _ => Location::new(u64::MAX), }; assert!(first_size + 1 < latest_size); let old = cap_case < 2; let expected_size = if old { first_size } else { latest_size }; let db = TestStore::init(context.child("cap"), cfg.clone(), Some(cap)) .await .unwrap(); assert_eq!(db.size(), expected_size); assert_bitmap_consistent(&db).await; assert_eq!( db.get_metadata().await.unwrap(), Some(vec![if old { 10 } else { 20 }]) ); assert_eq!( db.get(&a).await.unwrap(), Some(vec![if old { 1 } else { 3 }]) ); assert_eq!(db.get(&b).await.unwrap(), old.then(|| vec![2])); assert_eq!(db.get(&c).await.unwrap(), (!old).then(|| vec![4])); // Selection is durable, and future appends continue from the selected commit. drop(db); let db = TestStore::init(context.child("reopen"), cfg.clone(), None) .await .unwrap(); assert_eq!(db.size(), expected_size); assert_bitmap_consistent(&db).await; assert_eq!( db.get(&a).await.unwrap(), Some(vec![if old { 1 } else { 3 }]) ); assert_eq!(db.get(&b).await.unwrap(), old.then(|| vec![2])); assert_eq!(db.get(&c).await.unwrap(), (!old).then(|| vec![4])); let batch = db .new_batch() .update(a, vec![5]) .update(c, vec![6]) .finalize(None); let db = db.apply_batch(batch, &mut Proportional).await.unwrap(); let appended_size = db.size(); assert!(appended_size > expected_size); drop(db.commit().await.unwrap()); let db = TestStore::init(context.child("after_append"), cfg, None) .await .unwrap(); assert_eq!(db.size(), appended_size); assert_bitmap_consistent(&db).await; assert_eq!(db.get(&a).await.unwrap(), Some(vec![5])); assert_eq!(db.get(&b).await.unwrap(), old.then(|| vec![2])); assert_eq!(db.get(&c).await.unwrap(), Some(vec![6])); }); } } #[test_traced] fn test_store_bounded_initialization_rejects_zero() { deterministic::Runner::default().start(|context| async move { let cfg = test_config(); let db = TestStore::init(context.child("seed"), cfg.clone(), None) .await .unwrap(); let key = Blake3::hash(&[b"key"]); let (db, _) = apply_entries(db, [(key, Some(vec![1]))]).await; let size = db.size(); _ = db.sync().await.unwrap(); assert!(matches!( TestStore::init(context.child("zero"), cfg.clone(), Some(Location::new(0))).await, Err(Error::InvalidInitializationBound) )); let db = TestStore::init(context.child("unchanged"), cfg, None) .await .unwrap(); assert_eq!(db.size(), size); assert_eq!(db.get(&key).await.unwrap(), Some(vec![1])); }); } #[test_traced] fn test_store_bounded_initialization_rejects_pruned_floor() { deterministic::Runner::default().start(|context| async move { const KEYS: u64 = 64; let cfg = test_config(); let mut db = TestStore::init(context.child("seed"), cfg.clone(), None) .await .unwrap(); let key = |i: u64| Blake3::hash(&[&i.to_be_bytes()]); for value in [1, 2] { (db, _) = apply_entries(db, (0..KEYS).map(|i| (key(i), Some(vec![value])))).await; } let target = db.size(); let target_floor = db.inactivity_floor_loc(); let prune_loc = target_floor + KEYS / 2; assert!(prune_loc < target); let mut value = 2; while db.inactivity_floor_loc() < prune_loc { value += 1; assert!(value <= 10); (db, _) = apply_entries(db, (0..KEYS).map(|i| (key(i), Some(vec![value])))).await; } let db = db.prune(prune_loc).await.unwrap(); let bounds = db.bounds(); assert!(bounds.start > *target_floor && bounds.start < *target); _ = db.sync().await.unwrap(); assert!(matches!( TestStore::init(context.child("cap"), cfg.clone(), Some(target)).await, Err(Error::HistoricalFloorPruned(size)) if size == target )); let db = TestStore::init(context.child("unchanged"), cfg, None) .await .unwrap(); assert_eq!(db.bounds(), bounds); for i in 0..KEYS { assert_eq!(db.get(&key(i)).await.unwrap(), Some(vec![value])); } }); } /// A store over a delayed-sync storage backend. type DelayedStore = Db, Digest, Vec, TwoCap>; /// Open a [DelayedStore] whose blob syncs park on `pending`. /// /// Init durably persists the recovered database, so while syncs park the returned future /// must be driven with [drive_pending_syncs] (or the mock unblocked first). The journal /// uses large pages and sections: an apply that fills the write buffer or rolls the blob /// over waits for the in-flight sync, so mid-sync applies must stay clear of both. fn open_delayed_store( context: deterministic::Context, suffix: &str, pending: &PendingSyncs, ) -> impl Future> { let cfg = Config { log: JournalConfig { partition: format!("journal-{suffix}"), write_buffer: NZUsize!(64 * 1024), replay_buffer: NZUsize!(64 * 1024), compression: None, codec_config: ((), ((0..=10000).into(), ())), items_per_section: NZU64!(1000), page_cache: CacheRef::new(NZU16!(1024), NZUsize!(8)), }, translator: TwoCap, init_cache: Some(NZUsize!(1024)), init_buffer: NZUsize!(1 << 21), }; DelayedStore::init( DelayedSyncContext { inner: context, pending: pending.clone(), }, cfg, None, ) } /// Apply a single-key batch writing `key -> value`. async fn apply_write(db: DelayedStore, key: Digest, value: Vec) -> DelayedStore { db.apply_batch([(key, Some(value))].into(), &mut Proportional) .await .unwrap() } /// A sync handle must not block database use while the backend sync is pending. #[test_traced] fn test_store_start_sync_overlaps_work() { deterministic::Runner::default().start(|ctx| async move { let pending = PendingSyncs::default(); let open = open_delayed_store(ctx.child("delayed"), "start-sync-overlap", &pending); let mut db = drive_pending_syncs(&pending, open).await.unwrap(); let key0 = Blake3::hash(&[&0u64.to_be_bytes()]); let value0 = vec![1u8; 8]; db = apply_write(db, key0, value0.clone()).await; let starts_before = pending.starts(); let entered_before = pending.entered(); let completions_before = pending.completions(); let handle; (db, handle) = db.start_sync().await.unwrap(); assert!(pending.starts() > starts_before); assert_eq!(pending.completions(), completions_before); // Observe the sync while the database keeps working. let waiter = ctx .child("await_sync") .spawn(|_| async move { handle.await.unwrap() }); while pending.entered() == entered_before { reschedule().await; } // Reads and applies complete before the sync does. assert_eq!(db.get(&key0).await.unwrap(), Some(value0)); let key1 = Blake3::hash(&[&1u64.to_be_bytes()]); let value1 = vec![2u8; 8]; db = apply_write(db, key1, value1.clone()).await; assert_eq!( pending.completions(), completions_before, "the database made progress while the sync was still in flight" ); pending.unblock(); waiter.await.unwrap(); // The mid-sync batch is durable after the next start_sync completes. let handle; (db, handle) = db.start_sync().await.unwrap(); handle.await.unwrap(); let size = db.size(); drop(db); let db = open_delayed_store(ctx.child("reopen"), "start-sync-overlap", &pending) .await .unwrap(); assert_eq!(db.size(), size); assert_eq!(db.get(&key1).await.unwrap(), Some(value1)); db.destroy().await.unwrap(); }); } /// A sync begun by `start_sync` that fails in flight surfaces the error through both the /// returned handle and the next durability operation. #[test_traced] fn test_store_start_sync_failure_propagates() { deterministic::Runner::default().start(|ctx| async move { // Pass syncs through so opening the database doesn't park. let pending = PendingSyncs::default(); pending.unblock(); let mut db = open_delayed_store(ctx.child("delayed"), "start-sync-fail", &pending) .await .unwrap(); db = apply_write(db, Blake3::hash(&[&0u64.to_be_bytes()]), vec![1u8; 8]).await; // Arm all future syncs to resolve to an injected error. pending.arm_fail(); let handle; (db, handle) = db.start_sync().await.unwrap(); assert!( handle.await.is_err(), "the sync handle surfaces the failure" ); let starts_before = pending.starts(); // A failed mutable method consumes the database per the failures-are-fatal contract. assert!( db.commit().await.is_err(), "the next durability op surfaces the failed in-flight sync" ); assert_eq!( pending.starts(), starts_before, "the surfaced error is the retained failure, not a fresh sync's" ); }); } #[test_collect_traces("WARN")] fn test_store_recovery_warns_when_discarding_uncommitted_suffix(traces: TraceStorage) { deterministic::Runner::default().start(|context| async move { let mut db = create_test_store(context.child("seed")).await; let key = Blake3::hash(&[b"uncommitted"]); // A crash during apply_batch can persist an update before its trailing commit. (db.log, _) = db .log .append(&Operation::Update(Update(key, vec![7]))) .await .unwrap(); db.log = db.log.sync().await.unwrap(); drop(db); let db = create_test_store(context.child("recover")).await; assert_eq!(*db.size(), 1); assert_eq!(db.get(&key).await.unwrap(), None); }); traces .get_by_level(tracing::Level::WARN) .expect_event(|event| { let metadata = &event.metadata; metadata.content == "rewinding journal items" && metadata.expect_field_exact("journal_size", "2").is_ok() && metadata.expect_field_exact("rewound_items", "1").is_ok() }) .unwrap(); } /// State persisted via an awaited start_sync handle is recovered on reopen. #[test_traced] fn test_store_start_sync_recovery() { deterministic::Runner::default().start(|ctx| async move { let pending = PendingSyncs::default(); pending.unblock(); let mut db = open_delayed_store(ctx.child("delayed"), "start-sync-recovery", &pending) .await .unwrap(); let key = Blake3::hash(&[&0u64.to_be_bytes()]); let value = vec![1u8; 8]; db = apply_write(db, key, value.clone()).await; let handle; (db, handle) = db.start_sync().await.unwrap(); handle.await.unwrap(); let size = db.size(); drop(db); let db = open_delayed_store(ctx.child("reopen"), "start-sync-recovery", &pending) .await .unwrap(); assert_eq!(db.size(), size); assert_eq!(db.get(&key).await.unwrap(), Some(value)); db.destroy().await.unwrap(); }); } /// Pruning drains the in-flight sync before mutating storage. #[test_traced] fn test_store_start_sync_prune_waits() { deterministic::Runner::default().start(|ctx| async move { let pending = PendingSyncs::default(); let open = open_delayed_store(ctx.child("delayed"), "start-sync-prune", &pending); let mut db = drive_pending_syncs(&pending, open).await.unwrap(); // Two batches so the floor walks leave a non-trivial prune target. db = apply_write(db, Blake3::hash(&[&0u64.to_be_bytes()]), vec![1u8; 8]).await; db = apply_write(db, Blake3::hash(&[&1u64.to_be_bytes()]), vec![2u8; 8]).await; let starts_before = pending.starts(); let handle; (db, handle) = db.start_sync().await.unwrap(); assert!(pending.starts() > starts_before); let floor = db.inactivity_floor_loc(); assert!(*floor > 0); let db = { let mut prune = std::pin::pin!(db.prune(floor)); assert!( prune.as_mut().now_or_never().is_none(), "prune proceeded while the started sync was pending" ); pending.unblock(); prune.await.unwrap() }; handle.await.unwrap(); db.destroy().await.unwrap(); }); } #[test_traced("DEBUG")] pub fn test_store_construct_empty() { let executor = deterministic::Runner::default(); executor.start(|mut context| async move { let db = create_test_store(context.child("store").with_attribute("index", 0)).await; assert_eq!(db.bounds().end, 1); assert_eq!(db.log.bounds().start, 0); assert_eq!(db.inactivity_floor_loc(), 0); assert!(db.get_metadata().await.unwrap().is_none()); let floor = db.inactivity_floor_loc(); let db = db.prune(floor).await.unwrap(); assert!(matches!( db.prune(Location::new(1)).await, Err(Error::PruneBeyondMinRequired(_, _)) )); let db = create_test_store(context.child("store").with_attribute("index", 3)).await; // Make sure closing/reopening gets us back to the same state, even after adding an uncommitted op. let d1 = Digest::random(&mut context); let v1 = vec![1, 2, 3]; let (db, _) = apply_entries(db, [(d1, Some(v1))]).await; drop(db); let db = create_test_store(context.child("store").with_attribute("index", 1)).await; assert_eq!(db.bounds().end, 1); // Test calling commit on an empty db which should make it (durably) non-empty. let metadata = vec![1, 2, 3]; let batch = db.new_batch().finalize(Some(metadata.clone())); let applied_start = db.size(); let db = db.apply_batch(batch, &mut Proportional).await.unwrap(); let range = applied_start..db.size(); assert_eq!(range.start, 1); assert_eq!(range.end, 2); let db = db.commit().await.unwrap(); assert_eq!(db.bounds().end, 2); let floor = db.inactivity_floor_loc(); let db = db.prune(floor).await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone())); drop(db); let db = create_test_store(context.child("store").with_attribute("index", 2)).await; assert_eq!(db.get_metadata().await.unwrap(), Some(metadata)); // Confirm the inactivity floor doesn't fall endlessly behind with multiple commits on a // non-empty db. let (db, _) = apply_entries(db, [(Digest::random(&mut context), Some(vec![1, 2, 3]))]).await; let mut db = db.commit().await.unwrap(); for _ in 1..100 { let merkleized = db.new_batch().finalize(None); db = db.apply_batch(merkleized, &mut Proportional).await.unwrap(); db = db.commit().await.unwrap(); // Distance should equal 3 after the second commit, with inactivity_floor // referencing the previous commit operation. assert!(db.bounds().end - db.inactivity_floor_loc <= 3); assert!(db.get_metadata().await.unwrap().is_none()); } db.destroy().await.unwrap(); }); } #[test_traced("DEBUG")] fn test_store_construct_basic() { let executor = deterministic::Runner::default(); executor.start(|mut ctx| async move { let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await; // Ensure the store is empty assert_eq!(db.bounds().end, 1); assert_eq!(db.inactivity_floor_loc, 0); let key = Digest::random(&mut ctx); let value = vec![2, 3, 4, 5]; // Attempt to get a key that does not exist let result = db.get(&key).await; assert!(result.unwrap().is_none()); // Insert a key-value pair. apply_batch writes the Update, the floor walk's move, and // a CommitFloor: 3 new ops on top of the initial commit. let (db, _) = apply_entries(db, [(key, Some(value.clone()))]).await; assert_eq!(*db.bounds().end, 4); assert_eq!(*db.inactivity_floor_loc, 2); // Fetch the value let fetched_value = db.get(&key).await.unwrap(); assert_eq!(fetched_value.unwrap(), value); // Simulate commit failure: drop without commit. drop(db); // Re-open the store let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await; // Ensure the re-opened store removed the uncommitted operations assert_eq!(*db.bounds().end, 1); assert_eq!(*db.inactivity_floor_loc, 0); assert!(db.get_metadata().await.unwrap().is_none()); // Insert a key-value pair and persist with metadata. let metadata = vec![99, 100]; let batch = db .new_batch() .update(key, value.clone()) .finalize(Some(metadata.clone())); let applied_start = db.size(); let db = db.apply_batch(batch, &mut Proportional).await.unwrap(); let range = applied_start..db.size(); assert_eq!(*range.start, 1); assert_eq!(*range.end, 4); let db = db.commit().await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone())); assert_eq!(*db.bounds().end, 4); assert_eq!(*db.inactivity_floor_loc, 2); // Re-open the store drop(db); let db = create_test_store(ctx.child("store").with_attribute("index", 2)).await; // Ensure the re-opened store retained the committed operations assert_eq!(*db.bounds().end, 4); assert_eq!(*db.inactivity_floor_loc, 2); // Fetch the value, ensuring it is still present let fetched_value = db.get(&key).await.unwrap(); assert_eq!(fetched_value.unwrap(), value); // Insert two new k/v pairs to force pruning of the first section. let (k1, v1) = (Digest::random(&mut ctx), vec![2, 3, 4, 5, 6]); let (k2, v2) = (Digest::random(&mut ctx), vec![6, 7, 8]); let (db, _) = apply_entries(db, [(k1, Some(v1.clone()))]).await; let (db, _) = apply_entries(db, [(k2, Some(v2.clone()))]).await; assert_eq!(*db.bounds().end, 10); assert_eq!(*db.inactivity_floor_loc, 5); // Each apply_entries writes a CommitFloor with None metadata, replacing // the previously committed metadata. assert_eq!(db.get_metadata().await.unwrap(), None); let db = db.commit().await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), None); // commit() is just an fsync now, so bounds and floor are unchanged. assert_eq!(*db.bounds().end, 10); assert_eq!(*db.inactivity_floor_loc, 5); // Ensure all keys can be accessed, despite the first section being pruned. assert_eq!(db.get(&key).await.unwrap().unwrap(), value); assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1); assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2); // Update existing key with modified value. let mut v1_updated = db.get(&k1).await.unwrap().unwrap(); v1_updated.push(7); let (db, _) = apply_entries(db, [(k1, Some(v1_updated))]).await; let db = db.commit().await.unwrap(); assert_eq!(db.get(&k1).await.unwrap().unwrap(), vec![2, 3, 4, 5, 6, 7]); // Create new key. let k3 = Digest::random(&mut ctx); let (db, _) = apply_entries(db, [(k3, Some(vec![8]))]).await; let db = db.commit().await.unwrap(); assert_eq!(db.get(&k3).await.unwrap().unwrap(), vec![8]); // Destroy the store db.destroy().await.unwrap(); }); } #[test_traced("DEBUG")] fn test_store_log_replay() { let executor = deterministic::Runner::default(); executor.start(|mut ctx| async move { let mut db = create_test_store(ctx.child("store").with_attribute("index", 0)).await; // Update the same key many times. const UPDATES: u64 = 100; let k = Digest::random(&mut ctx); for _ in 0..UPDATES { let v = vec![1, 2, 3, 4, 5]; (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await; } let iter = db.snapshot.get(&k); assert_eq!(iter.count(), 1); let db = db.commit().await.unwrap(); db.sync().await.unwrap(); // Re-open the store, prune it, then ensure it replays the log correctly. let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await; let floor = db.inactivity_floor_loc(); let db = db.prune(floor).await.unwrap(); let iter = db.snapshot.get(&k); assert_eq!(iter.count(), 1); // Each apply_entries appends the Update, one move of it, and a CommitFloor. The walk // ends at the tip the writes reached, so it does not reach the moved Update. Total: // 1 (init) + 100 * 3 = 301. assert_eq!(*db.bounds().end, 301); // Only the last moved Update and CommitFloor are active, so the floor is 299. assert_eq!(*db.inactivity_floor_loc, 299); let floor = db.inactivity_floor_loc; // All blobs prior to the inactivity floor are pruned, so the oldest retained location // is the first in the last retained blob. assert_eq!(db.log.bounds().start, *floor - *floor % 7); db.destroy().await.unwrap(); }); } /// Each round of a batch's key resolution reads as many locations from a collision bucket, in /// bucket order, as the bucket has unresolved keys, independent of the bucket's size. The /// proportional walk then reads the updates it reaches. /// /// Each case rewrites the keys at `positions` of a bucket of `count` keys and expects the /// bucket's reads, the walk's reads, and the floor after the batch. #[rstest] // A head-key match reads one update; the proportional walk reads its two moves. #[case::head_of_8(8, &[0], 1, 2, 4, 4096)] #[case::head_of_64(64, &[0], 1, 2, 4, 4096)] // A key at position 5 resolves in the sixth round, each reading one location. #[case::sixth(8, &[5], 6, 2, 4, 1)] // The first round reads positions 0..3 and resolves two keys. The second reads position 3 // and resolves the last. #[case::two_rounds(8, &[1, 2, 3], 4, 4, 9, 1)] // Rounds of three, three, and two locations read the whole bucket. #[case::whole_bucket(8, &[5, 6, 7], 8, 4, 6, 1)] #[test_traced("WARN")] fn test_store_collision_reads( #[case] count: u64, #[case] positions: &'static [usize], #[case] bucket_reads: u64, #[case] walk_reads: u64, #[case] floor: u64, #[case] value_len: usize, ) { deterministic::Runner::default().start(move |context| async move { let db = create_test_store(context.child("store")).await; let keys: Vec<_> = (0..count) .map(|i| { let mut bytes = [0u8; 32]; bytes[..2].copy_from_slice(&[0xa5, 0x5a]); bytes[24..].copy_from_slice(&i.to_be_bytes()); Digest::from(bytes) }) .collect(); let writes = keys .iter() .zip(0u8..) .map(|(key, i)| (*key, Some(vec![i; value_len]))); let (db, _) = apply_entries(db, writes).await; // The keys lie at 1..count + 1 in key order. The previous commit's entry moves the // first to count + 1 in place, so the bucket lists it first and the rest by position. let bucket: Vec<_> = db.snapshot.get(&keys[0]).map(|loc| **loc).collect(); let expected: Vec<_> = std::iter::once(count + 1).chain(2..=count).collect(); assert_eq!(bucket, expected); assert_eq!(*db.inactivity_floor_loc(), 2); // The walk has one entry per rewrite plus one for the previous commit, and every // update it reaches lies below the batch. assert_eq!(walk_reads, Widen::widen(positions.len()) + 1); let before = counter(&context, "log_items_read_total"); let writes = positions .iter() .map(|i| (keys[*i], Some(vec![0xff; value_len]))); let (db, _) = apply_entries(db, writes).await; let reads = counter(&context, "log_items_read_total") - before; assert_eq!(reads, bucket_reads + walk_reads); assert_eq!(*db.inactivity_floor_loc(), floor); assert_eq!(db.active_keys, count as usize); for (i, key) in keys.iter().enumerate() { let value = if positions.contains(&i) { 0xff } else { i as u8 }; assert_eq!(db.get(key).await.unwrap(), Some(vec![value; value_len])); } assert_bitmap_consistent(&db).await; drop(db.commit().await.unwrap()); let db = create_test_store(context.child("reopened")).await; assert_eq!( db.get(&keys[positions[0]]).await.unwrap(), Some(vec![0xff; value_len]) ); db.destroy().await.unwrap(); }); } #[test_traced("DEBUG")] fn test_store_build_snapshot_keys_with_shared_prefix() { let executor = deterministic::Runner::default(); executor.start(|mut ctx| async move { let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await; let (k1, v1) = (Digest::random(&mut ctx), vec![1, 2, 3, 4, 5]); let (mut k2, v2) = (Digest::random(&mut ctx), vec![6, 7, 8, 9, 10]); // Ensure k2 shares 2 bytes with k1 (test DB uses `TwoCap` translator.) k2.0[0..2].copy_from_slice(&k1.0[0..2]); let (db, _) = apply_entries(db, [(k1, Some(v1.clone()))]).await; let (db, _) = apply_entries(db, [(k2, Some(v2.clone()))]).await; assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1); assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2); let db = db.commit().await.unwrap(); db.sync().await.unwrap(); // Re-open the store to ensure it builds the snapshot for the conflicting // keys correctly. let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await; assert_eq!(db.get(&k1).await.unwrap().unwrap(), v1); assert_eq!(db.get(&k2).await.unwrap().unwrap(), v2); db.destroy().await.unwrap(); }); } #[test_traced("DEBUG")] fn test_store_delete() { let executor = deterministic::Runner::default(); executor.start(|mut ctx| async move { let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await; // Insert a key-value pair let k = Digest::random(&mut ctx); let v = vec![1, 2, 3, 4, 5]; let (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await; let db = db.commit().await.unwrap(); // Fetch the value let fetched_value = db.get(&k).await.unwrap(); assert_eq!(fetched_value.unwrap(), v); // Delete the key assert!(db.get(&k).await.unwrap().is_some()); let (db, _) = apply_entries(db, [(k, None)]).await; // Ensure the key is no longer present let fetched_value = db.get(&k).await.unwrap(); assert!(fetched_value.is_none()); assert!(db.get(&k).await.unwrap().is_none()); // Commit the changes db.commit().await.unwrap(); // Re-open the store and ensure the key is still deleted let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await; let fetched_value = db.get(&k).await.unwrap(); assert!(fetched_value.is_none()); // Re-insert the key let (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await; let fetched_value = db.get(&k).await.unwrap(); assert_eq!(fetched_value.unwrap(), v); // Commit the changes db.commit().await.unwrap(); // Re-open the store and ensure the snapshot restores the key, after processing // the delete and the subsequent set. let db = create_test_store(ctx.child("store").with_attribute("index", 2)).await; let fetched_value = db.get(&k).await.unwrap(); assert_eq!(fetched_value.unwrap(), v); // Delete a non-existent key (no-op) let k_n = Digest::random(&mut ctx); let (db, range) = apply_entries(db, [(k_n, None)]).await; assert_eq!(range.start, 9); assert_eq!(range.end, 11); let db = db.commit().await.unwrap(); assert!(db.get(&k_n).await.unwrap().is_none()); // Make sure k is still there assert!(db.get(&k).await.unwrap().is_some()); db.destroy().await.unwrap(); }); } /// Tests the pruning example in the module documentation. #[test_traced("DEBUG")] fn test_store_pruning() { let executor = deterministic::Runner::default(); executor.start(|mut ctx| async move { let db = create_test_store(ctx.child("store")).await; let k_a = Digest::random(&mut ctx); let k_b = Digest::random(&mut ctx); let v_a = vec![1]; let v_b = vec![]; let v_c = vec![4, 5, 6]; let (db, _) = apply_entries(db, [(k_a, Some(v_a.clone()))]).await; let (db, _) = apply_entries(db, [(k_b, Some(v_b.clone()))]).await; let db = db.commit().await.unwrap(); assert_eq!(*db.bounds().end, 7); assert_eq!(*db.inactivity_floor_loc, 3); assert_eq!(db.get(&k_a).await.unwrap().unwrap(), v_a); let (db, _) = apply_entries(db, [(k_b, Some(v_a.clone()))]).await; let (db, _) = apply_entries(db, [(k_a, Some(v_c.clone()))]).await; let db = db.commit().await.unwrap(); assert_eq!(*db.bounds().end, 15); assert_eq!(*db.inactivity_floor_loc, 12); assert_eq!(db.get(&k_a).await.unwrap().unwrap(), v_c); assert_eq!(db.get(&k_b).await.unwrap().unwrap(), v_a); db.destroy().await.unwrap(); }); } /// Apply `writes` as one proportional batch and replay its operations into `live`. /// /// Asserts that the batch moves each update at most once. A key the batch writes may appear /// twice: its write and one move of that write. /// /// Asserts that the replayed keys are the keys of `live` with `writes` applied, and that the /// applied state satisfies [`assert_bound`]. Every batch since the initial commit must go /// through this function. async fn bounded( db: TestStore, live: &mut BTreeMap, writes: &[(Digest, Option>)], ) -> TestStore { let expected = keys_after(live, writes); let (db, range) = apply_entries(db, writes.iter().cloned()).await; let written: BTreeSet<_> = writes.iter().map(|(key, _)| *key).collect(); let mut updated = BTreeMap::new(); for loc in *range.start..*range.end { let loc = Location::new(loc); match db.get_op(loc).await.unwrap() { Operation::Update(Update(key, _)) => { let count = updated.entry(key).or_insert(0usize); *count += 1; let limit = 1 + usize::from(written.contains(&key)); assert!(*count <= limit, "the batch moves an update of {key} twice"); live.insert(key, loc); } Operation::Delete(key) => { live.remove(&key); } _ => {} } } assert_eq!(live.keys().copied().collect::>(), expected); assert_eq!(live.len(), db.active_keys); assert_bound(db.inactivity_floor_loc(), db.size(), live); db } /// Every batch of [`churn`], or of [`randomized_churn`] under `seed`, keeps the size at most /// `3 * n + 1` operations past the floor for `n` live keys. #[rstest] #[case::churn(None)] #[case::seed_0(Some(0))] #[case::seed_1(Some(1))] #[case::seed_7(Some(7))] #[case::seed_5133(Some(0x5133))] #[test_traced("WARN")] fn test_store_floor_bound(#[case] seed: Option) { let executor = seed.map_or_else( deterministic::Runner::default, deterministic::Runner::seeded, ); executor.start(move |mut context| async move { let key = |i: u64| Blake3::hash(&[&i.to_be_bytes()]); let value = |i: u64| i.to_be_bytes().to_vec(); let db = create_test_store(context.child("store")).await; let db = match seed { None => churn(db, key, value, bounded).await, Some(_) => randomized_churn(&mut context, db, key, value, bounded).await, }; assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// A proportional batch has one entry for each operation it makes inactive: one for each /// update it supersedes, two for each delete (itself and the update it supersedes), and one /// for its previous commit. Its walk moves updates, including the batch's own written updates, /// and passes its own deletes up to the tip its writes reached. A custom policy with /// proportional limits (`scripted`) moves the same updates without deciding one. /// /// Each case seeds `n` keys with a held or proportional floor, writes `writes` (a seed index /// and a new value, or `None` to delete), and expects the batch's range, its floor, and the /// seed indices of the updates the walk moves, in order. #[rstest] // The deletes lie at 23..28 in key order. Their 2 * 5 entries and the previous commit's entry // move the updates at 2..13 to 28..39, so the commit at 39 records the floor 13. #[case::delete_first_and_last_four( 20, false, vec![(0, None), (16, None), (17, None), (18, None), (19, None)], 23..40, 13, (1..12).collect(), )] // The 19 deletes lie at 23..42. The walk keeps the survivor, moving it to 42, and passes every // delete, so the commit at 43 records the floor 42, 2 operations below the size. #[case::delete_all_but_first(20, false, (1..20).map(|i| (i, None)).collect(), 23..44, 42, vec![0])] // Over six held updates at 1..7, two updates and a delete make four operations inactive, so // with the previous commit the walk moves five updates between the writes and the commit: // the three remaining seed updates and the two written updates. The floor ends past the last // written update, at 11. #[case::update_delete_update( 6, true, vec![(0, Some(200)), (1, None), (2, Some(202))], 8..17, 11, vec![3, 4, 5, 0, 2], )] #[test_traced("WARN")] fn test_store_proportional_entries( #[case] n: u64, #[case] held: bool, #[case] writes: Vec<(usize, Option)>, #[case] range: Range, #[case] floor: u64, #[case] moved: Vec, #[values(false, true)] scripted: bool, ) { deterministic::Runner::default().start(move |context| async move { let db = open(context.child("store"), "proportional").await; let seed = seed(n); let db = if held { apply(db, seed.clone(), &mut Hold).await.0 } else { // Seed 20 keys in key order at 1..21. The previous commit's entry moves the first // update to 21, and the commit lies at 22. let (db, _) = apply(db, seed.clone(), &mut Proportional).await; assert_eq!((*db.inactivity_floor_loc(), *db.size()), (2, 23)); db }; let writes: Vec<_> = writes .iter() .map(|&(i, value)| (seed[i].0, value.map(digest))) .collect(); let (db, r) = if scripted { let mut policy = Script::proportional(|_: &sha256::Digest| Choice::Keep); apply(db, writes.clone(), &mut policy).await } else { apply(db, writes.clone(), &mut Proportional).await }; assert_eq!(*r.start..*r.end, range); assert_eq!(*db.inactivity_floor_loc(), floor); // The batch appends its writes in key order, the updates its walk moves with their // latest values, and its commit. let value = |i: usize| { writes .iter() .find(|(key, _)| *key == seed[i].0) .map_or(seed[i].1, |(_, value)| *value) }; let expected: Vec<_> = writes .iter() .map(|&(key, value)| { value.map_or(Operation::Delete(key), |value| { Operation::Update(Update(key, value)) }) }) .chain( moved .iter() .map(|&i| Operation::Update(Update(seed[i].0, value(i).unwrap()))), ) .chain([Operation::CommitFloor(None, Location::new(floor))]) .collect(); assert_eq!(Widen::widen(expected.len()), range.end - range.start); for (loc, op) in range.zip(expected) { assert_eq!(db.get_op(Location::new(loc)).await.unwrap(), op); } // The survivors keep their latest values. let mut active = 0; for (i, (key, _)) in seed.iter().enumerate() { let value = value(i); active += usize::from(value.is_some()); assert_eq!(db.get(key).await.unwrap(), value); } assert_eq!(db.active_keys, active); assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// Pruning to a floor advanced by applied-but-uncommitted entries must not durably outrun /// the last durable commit: after a crash, the recovered floor would lie below the pruned /// boundary and the store could never reopen. #[test_traced("WARN")] pub fn test_store_db_prune_after_unsynced_floor_recovery() { let executor = deterministic::Runner::default(); const ELEMENTS: u64 = 1000; executor.start(|context| async move { let mut db = create_test_store(context.child("store").with_attribute("index", 0)).await; // Establish a durable state whose last commit declares an early inactivity floor. for i in 0u64..ELEMENTS { let k = Blake3::hash(&[&i.to_be_bytes()]); let v = vec![(i % 255) as u8; ((i % 13) + 7) as usize]; (db, _) = apply_entries(db, [(k, Some(v))]).await; } let mut db = db.commit().await.unwrap(); let durable_floor = db.inactivity_floor_loc; // Apply (but do not commit) entries that advance the in-memory floor past the // durable commit's floor. for i in 0u64..ELEMENTS { let k = Blake3::hash(&[&i.to_be_bytes()]); let v = vec![((i + 1) % 255) as u8; ((i % 13) + 8) as usize]; (db, _) = apply_entries(db, [(k, Some(v))]).await; } let unsynced_floor = db.inactivity_floor_loc; assert!(unsynced_floor > durable_floor); // Prune to the in-memory floor, then crash before any further commit. let db = db.prune(unsynced_floor).await.unwrap(); let op_count = db.bounds().end; drop(db); // Reopening must succeed: prune committed the buffered operations first, so the // replayed log reproduces the advanced floor. let db = create_test_store(context.child("store").with_attribute("index", 1)).await; assert_eq!(db.bounds().end, op_count); assert_eq!(db.inactivity_floor_loc, unsynced_floor); db.destroy().await.unwrap(); }); } #[test_traced("WARN")] pub fn test_store_db_recovery() { let executor = deterministic::Runner::default(); // Build a db with 1000 keys, some of which we update and some of which we delete. const ELEMENTS: u64 = 1000; executor.start(|context| async move { let db = create_test_store(context.child("store").with_attribute("index", 0)).await; // Simulate building batches but not applying them (data is not persisted). { let mut batch = db.new_batch(); for i in 0u64..ELEMENTS { let k = Blake3::hash(&[&i.to_be_bytes()]); let v = vec![(i % 255) as u8; ((i % 13) + 7) as usize]; batch = batch.update(k, v); } // Drop the batch without applying it. This simulates a failure before apply. } drop(db); let mut db = create_test_store(context.child("store").with_attribute("index", 1)).await; assert_eq!(*db.bounds().end, 1); // Apply the updates and commit them. for i in 0u64..ELEMENTS { let k = Blake3::hash(&[&i.to_be_bytes()]); let v = vec![(i % 255) as u8; ((i % 13) + 7) as usize]; (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await; } let mut db = db.commit().await.unwrap(); // Update every 3rd key and commit. for i in 0u64..ELEMENTS { if i % 3 != 0 { continue; } let k = Blake3::hash(&[&i.to_be_bytes()]); let v = vec![((i + 1) % 255) as u8; ((i % 13) + 8) as usize]; (db, _) = apply_entries(db, [(k, Some(v.clone()))]).await; } let mut db = db.commit().await.unwrap(); assert_eq!(db.snapshot.items(), 1000); // Delete every 7th key and commit. for i in 0u64..ELEMENTS { if i % 7 != 1 { continue; } let k = Blake3::hash(&[&i.to_be_bytes()]); (db, _) = apply_entries(db, [(k, None)]).await; } let db = db.commit().await.unwrap(); let final_count = db.bounds().end; let final_floor = db.inactivity_floor_loc; // Sync and reopen the store to ensure the state is preserved. db.sync().await.unwrap(); let db = create_test_store(context.child("store").with_attribute("index", 2)).await; assert_eq!(db.bounds().end, final_count); assert_eq!(db.inactivity_floor_loc, final_floor); let floor = db.inactivity_floor_loc(); let db = db.prune(floor).await.unwrap(); assert_eq!(db.log.bounds().start, *final_floor - *final_floor % 7); assert_eq!(db.snapshot.items(), 857); db.destroy().await.unwrap(); }); } #[test_traced("WARN")] pub fn test_store_commit_after_sync_recovers_without_second_sync() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let db = create_test_store(context.child("store").with_attribute("index", 0)).await; let key0 = Blake3::hash(&[&0u64.to_be_bytes()]); let key1 = Blake3::hash(&[&1u64.to_be_bytes()]); let value0 = vec![0, 1, 2]; let value1 = vec![3, 4, 5, 6]; // Commit and sync an initial update so restart recovery has an older watermark. let (db, _) = apply_entries(db, [(key0, Some(value0.clone()))]).await; let db = db.commit().await.unwrap(); let db = db.sync().await.unwrap(); // Persist a later commit without syncing; recovery must replay it after reopen. let (db, _) = apply_entries(db, [(key1, Some(value1.clone()))]).await; let db = db.commit().await.unwrap(); let committed_end = db.bounds().end; let committed_floor = db.inactivity_floor_loc(); drop(db); let db = create_test_store(context.child("store").with_attribute("index", 1)).await; assert_eq!(db.bounds().end, committed_end); assert_eq!(db.inactivity_floor_loc(), committed_floor); assert_eq!(db.get(&key0).await.unwrap(), Some(value0)); assert_eq!(db.get(&key1).await.unwrap(), Some(value1)); db.destroy().await.unwrap(); }); } #[test_traced("DEBUG")] fn test_store_batch() { let executor = deterministic::Runner::default(); executor.start(|mut ctx| async move { let db = create_test_store(ctx.child("store").with_attribute("index", 0)).await; // Ensure the store is empty assert_eq!(db.bounds().end, 1); assert_eq!(db.inactivity_floor_loc, 0); let key = Digest::random(&mut ctx); let value = vec![2, 3, 4, 5]; let batch = db.new_batch(); // Attempt to get a key that does not exist let result = batch.get(&key).await; assert!(result.unwrap().is_none()); // Insert a key-value pair let batch = batch.update(key, value.clone()); assert_eq!(db.bounds().end, 1); // The batch is not applied yet assert_eq!(db.inactivity_floor_loc, 0); // Fetch the value let fetched_value = batch.get(&key).await.unwrap(); assert_eq!(fetched_value.unwrap(), value); let changeset = batch.finalize(None); db.apply_batch(changeset, &mut Proportional).await.unwrap(); // Re-open the store let db = create_test_store(ctx.child("store").with_attribute("index", 1)).await; // Ensure the batch was not applied since we didn't commit. assert_eq!(db.bounds().end, 1); assert_eq!(db.inactivity_floor_loc, 0); assert!(db.get_metadata().await.unwrap().is_none()); // Insert a key-value pair and persist the change. let metadata = vec![99, 100]; let batch = db .new_batch() .update(key, value.clone()) .finalize(Some(metadata.clone())); let applied_start = db.size(); let db = db.apply_batch(batch, &mut Proportional).await.unwrap(); let range = applied_start..db.size(); assert_eq!(range.start, 1); assert_eq!(range.end, 4); let db = db.commit().await.unwrap(); assert_eq!(db.get_metadata().await.unwrap(), Some(metadata.clone())); drop(db); // Re-open the store let db = create_test_store(ctx.child("store").with_attribute("index", 2)).await; // Ensure the re-opened store retained the committed operations assert_eq!(db.bounds().end, 4); assert_eq!(db.inactivity_floor_loc, 2); // Fetch the value, ensuring it is still present let fetched_value = db.get(&key).await.unwrap(); assert_eq!(fetched_value.unwrap(), value); // Destroy the store db.destroy().await.unwrap(); }); } /// A [Db] keyed by variable-length byte keys. type VecKeyStore = Db, Vec, TwoCap>; #[test_traced("DEBUG")] fn test_store_variable_length_keys() { let executor = deterministic::Runner::default(); executor.start(|context| async move { // Configure the operation codec for variable-length keys. let cfg = Config { log: JournalConfig { partition: "journal".into(), write_buffer: NZUsize!(64 * 1024), replay_buffer: NZUsize!(64 * 1024), compression: None, codec_config: (((0..=64).into(), ()), ((0..=10000).into(), ())), items_per_section: NZU64!(7), page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE), }, translator: TwoCap, init_cache: Some(NZUsize!(1024)), init_buffer: NZUsize!(1 << 21), }; // Commit two keys of different lengths that share a translated prefix. let db = VecKeyStore::init( context.child("store").with_attribute("index", 0), cfg.clone(), None, ) .await .unwrap(); let short = b"key".to_vec(); let long = b"key-extended".to_vec(); let batch = db .new_batch() .update(short.clone(), vec![1]) .update(long.clone(), vec![2]) .finalize(None); let db = db.apply_batch(batch, &mut Proportional).await.unwrap(); let db = db.commit().await.unwrap(); assert_eq!(db.get(&short).await.unwrap(), Some(vec![1])); assert_eq!(db.get(&long).await.unwrap(), Some(vec![2])); drop(db); // Reopen the store and verify both committed values. let db = VecKeyStore::init(context.child("store").with_attribute("index", 1), cfg, None) .await .unwrap(); assert_eq!(db.get(&short).await.unwrap(), Some(vec![1])); assert_eq!(db.get(&long).await.unwrap(), Some(vec![2])); db.destroy().await.unwrap(); }); } /// A store keyed by SHA-256 digests whose translator buckets keys by their first byte. type PolicyStore = Db; async fn open>( context: deterministic::Context, partition: &str, ) -> PolicyStore { let cfg = Config { log: JournalConfig { partition: partition.into(), write_buffer: NZUsize!(64 * 1024), replay_buffer: NZUsize!(64 * 1024), compression: None, codec_config: ((), ()), items_per_section: NZU64!(7), page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE), }, translator: OneCap, init_cache: Some(NZUsize!(1024)), init_buffer: NZUsize!(1 << 21), }; PolicyStore::init(context, cfg, None).await.unwrap() } fn digest(i: u64) -> sha256::Digest { Sha256::hash(&[&i.to_be_bytes()]) } /// Return writes of `n` keys in ascending order. The `i`th write has the value /// `digest(100 + i)`. fn seed(n: u64) -> Vec<(sha256::Digest, Option)> { let mut keys: Vec<_> = (0..n).map(digest).collect(); keys.sort(); (100..) .zip(keys) .map(|(i, key)| (key, Some(digest(i)))) .collect() } /// Apply `writes` with `policy` and without metadata. async fn apply( db: PolicyStore, writes: impl IntoIterator)>, policy: &mut P, ) -> (PolicyStore, Range) where V: VariableValue, P: Policy, { let applied_start = db.size(); let db = db .apply_batch(writes.into_iter().collect(), policy) .await .unwrap(); let applied_end = db.size(); (db, applied_start..applied_end) } /// [`Hold`] keeps the inherited floor while the batch's writes apply, and moves it to the new /// commit location when the batch empties the store. #[test_traced("WARN")] fn test_store_policy_hold() { deterministic::Runner::default().start(|context| async move { // Seed four updates with a held floor. let db = open(context.child("store"), "hold").await; let seed = seed(4); let (db, _) = apply(db, seed.clone(), &mut Hold).await; let floor = db.inactivity_floor_loc(); // Hold the floor for a batch that updates, deletes, and creates keys. let created = digest(4); let writes = [ (seed[0].0, Some(digest(200))), (seed[1].0, None), (created, Some(digest(104))), ]; let (db, _) = apply(db, writes, &mut Hold).await; // The writes apply and the untouched keys keep their values. assert_eq!(db.inactivity_floor_loc(), floor); let expected = [ (seed[0].0, Some(digest(200))), (seed[1].0, None), seed[2], seed[3], (created, Some(digest(104))), ]; for (key, value) in expected { assert_eq!(db.get(&key).await.unwrap(), value); } // Deleting every key moves the floor to the new commit location. let deletes = [seed[0].0, seed[2].0, seed[3].0, created].map(|key| (key, None)); let (db, _) = apply(db, deletes, &mut Hold).await; assert!(db.is_empty()); assert_eq!(db.inactivity_floor_loc(), db.size() - 1); db.destroy().await.unwrap(); }); } /// A policy that stops at the first active update leaves the floor at its location and the /// update in place. The next batch's policy decides that update without spending a skip. #[test_traced("WARN")] fn test_store_policy_stop() { deterministic::Runner::default().start(|context| async move { // Seed four updates in key order at 1..5 with a held floor. let db = open(context.child("store"), "stop").await; let seed = seed(4); let (db, _) = apply(db, seed.clone(), &mut Hold).await; let (key, value) = (seed[0].0, seed[0].1.unwrap()); // Pass the initial commit and stop at the first update. let mut policy = Script::new(usize::MAX, u64::MAX, |_: &sha256::Digest| Choice::Stop); let (db, range) = apply(db, [], &mut policy).await; let first = Location::new(1); assert_eq!(policy.visited, [(first, key, value)]); // The batch appends only its commit and moves the floor to the stopped update's // location. assert_eq!(*range.start..*range.end, 6..7); assert_eq!(db.inactivity_floor_loc(), first); // A policy of the next batch decides the stopped update without spending a skip. let mut policy = Script::new(1, 0, keep); let (db, _) = apply(db, [], &mut policy).await; assert_eq!(policy.visited, [(first, key, value)]); assert_eq!(*db.inactivity_floor_loc(), 2); for (key, value) in seed { assert_eq!(db.get(&key).await.unwrap(), value); } assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// A proportional move probes the snapshot index once, when it rewrites the update's slot. #[test_traced("WARN")] fn test_store_proportional_probes_snapshot_once() { // A translator that counts its key transforms, one per snapshot probe. #[derive(Clone)] struct CountingTranslator(Arc); impl Translator for CountingTranslator { type Key = u8; fn transform(&self, key: &[u8]) -> u8 { self.0.fetch_add(1, Ordering::Relaxed); OneCap.transform(key) } } impl BuildHasher for CountingTranslator { type Hasher = ::Hasher; fn build_hasher(&self) -> Self::Hasher { OneCap.build_hasher() } } deterministic::Runner::default().start(|context| async move { type CountedStore = Db, CountingTranslator>; let lookups = Arc::new(AtomicUsize::new(0)); let cfg = test_config(); let db = CountedStore::init( context, Config { log: cfg.log, translator: CountingTranslator(lookups.clone()), init_cache: cfg.init_cache, init_buffer: cfg.init_buffer, }, None, ) .await .unwrap(); // Seed three keys in one translated-key bucket at 1..4 with a held floor. let writes = (0..3).map(|i| { let mut key = [0xAA; 32]; key[31] = i; (Digest::from(key), Some(vec![i])) }); let db = db.apply_batch(writes.collect(), &mut Hold).await.unwrap(); // An empty batch moves one update for its previous commit. lookups.store(0, Ordering::Relaxed); let applied_start = db.size(); let db = db .apply_batch(core::iter::empty().collect(), &mut Proportional) .await .unwrap(); assert_eq!(*db.size() - *applied_start, 2); assert_eq!(lookups.load(Ordering::Relaxed), 1); assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// A policy that replaces an update writes its value for the key at the tip. #[test_traced("WARN")] fn test_store_policy_replace() { deterministic::Runner::default().start(|context| async move { // Seed three updates in key order at 1..4 with a held floor. let db = open(context.child("store"), "replace").await; let seed = seed(3); let (db, _) = apply(db, seed.clone(), &mut Hold).await; let tip = db.size(); // Keep the first and last updates and replace the second. let (replaced, value) = (seed[1].0, digest(201)); let mut policy = Script::new(usize::MAX, u64::MAX, move |key: &sha256::Digest| { if *key == replaced { Choice::Replace(value) } else { Choice::Keep } }); let (db, range) = apply(db, [], &mut policy).await; assert_eq!(policy.locations(), [1, 2, 3].map(Location::new)); assert_eq!(db.inactivity_floor_loc(), tip); // The replacement takes the second update's place among the moves to the tip. assert_eq!(*range.start..*range.end, 5..9); let writes = [seed[0], (replaced, Some(value)), seed[2]]; for (loc, (key, value)) in (5..).zip(writes) { let op = db.get_op(Location::new(loc)).await.unwrap(); assert_eq!(op, Operation::Update(Update(key, value.unwrap()))); assert_eq!(db.get(&key).await.unwrap(), value); } assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// A policy that evicts every update it reaches receives the previously applied, /// colliding-key, and batch-written updates in location order. Values read from the log /// arrive with no clone, and the batch's own value is cloned once. The evictions empty the /// store. #[test_traced("WARN")] fn test_store_policy_evicts_owned() { deterministic::Runner::default().start(|context| async move { let colliding = |i| colliding_digest(0xAA, i); let other = colliding_digest(0xBB, 0); // Seed three keys in one translated-key bucket and one other key at 1..5 with a held // floor. let db = open(context.child("store"), "evict").await; let seed = [colliding(0), colliding(1), colliding(2), other] .into_iter() .zip(0u8..) .map(|(key, tag)| (key, Some(CountedValue::new(tag)))); let (db, _) = apply(db, seed, &mut Hold).await; // The batch rewrites the first colliding key at 6, and the policy evicts every active // update, including that write. let mut policy = Evict::default(); let write = (colliding(0), Some(CountedValue::new(20))); let (db, range) = apply(db, [write], &mut policy).await; // Evictions arrive in location order. Values read from the log move out of the reads, // and the batch's own value is copied once from its unappended operation. let evicted: Vec<_> = policy .evicted .iter() .map(|(loc, key, value, before, after)| (**loc, *key, value.0, *before, *after)) .collect(); assert_eq!( evicted, [ (2, colliding(1), 1, 0, 0), (3, colliding(2), 2, 0, 0), (4, other, 3, 0, 0), (6, colliding(0), 20, 1, 1), ] ); // The batch appends its write, a delete of each evicted key, and its commit. The // empty store commits the commit's own location as the floor. assert_eq!(*range.start..*range.end, 6..12); let Operation::Update(Update(key, value)) = db.get_op(Location::new(6)).await.unwrap() else { panic!("expected the write at 6"); }; assert_eq!((key, value.0), (colliding(0), 20)); for (loc, evicted) in (7..).zip([colliding(1), colliding(2), other, colliding(0)]) { let op = db.get_op(Location::new(loc)).await.unwrap(); assert!(matches!(op, Operation::Delete(key) if key == evicted)); } let op = db.get_op(Location::new(11)).await.unwrap(); assert!(matches!(op, Operation::CommitFloor(None, floor) if *floor == 11)); // Every key reads `None`. for (_, key, ..) in &policy.evicted { assert!(db.get(key).await.unwrap().is_none()); } assert!(db.is_empty()); assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// A walk leaves the value of a batch write it cannot reach uncloned. On a fresh store with /// no skips, the write at 1 lies inside the window but past the inactive commit at 0. #[test_traced("WARN")] fn test_store_unreachable_own_write_is_not_cloned() { deterministic::Runner::default().start(|context| async move { let db = open::(context.child("store"), "unreachable-clone").await; let value = CountedValue::new(7); let witness = value.clone(); assert_eq!(witness.clones(), 1); // The window ends at min(2, 0 + 0 + 2) = 2, so the write at 1 is a candidate, but // reaching it costs the skip the policy lacks: the floor stays at 0 and nothing is // decided. The walk copies only the operations of candidates it reaches, so the count // stays at the witness's own clone. let mut policy = Bounded { entries: 2, skips: 0, }; let (db, range) = apply(db, [(digest(0), Some(value))], &mut policy).await; assert_eq!(*range.start..*range.end, 1..3); assert_eq!(*db.inactivity_floor_loc(), 0); assert_eq!(witness.clones(), 1); // The batch appends only its write and its commit. let Operation::Update(Update(key, value)) = db.get_op(Location::new(1)).await.unwrap() else { panic!("expected the write at 1"); }; assert_eq!((key, value.0), (digest(0), 7)); let op = db.get_op(Location::new(2)).await.unwrap(); assert!(matches!(op, Operation::CommitFloor(None, floor) if *floor == 0)); assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// A policy in a batch with writes passes the locations the writes supersede as inactive, /// keeps or evicts the previously applied updates it reaches, and decides the batch's own /// updates. The result survives commit, reopen, and prune. #[test_traced("WARN")] fn test_store_policy_keep_evict_and_recover() { deterministic::Runner::default().start(|context| async move { // Seed four updates in key order at 1..5 with a held floor. let child = |index| context.child("store").with_attribute("index", index); let db = open(child(0), "recover").await; let seed = seed(4); let (db, _) = apply(db, seed.clone(), &mut Hold).await; assert_eq!(*db.size(), 6); // The batch updates the second key, deletes the third, and creates a fifth. Its // operations follow key order from 6, so its updates lie at 6 and 7 and its delete at 8. let created = digest(4); let writes = [ (seed[1].0, Some(digest(201))), (seed[2].0, None), (created, Some(digest(104))), ]; let mut written = [writes[0], writes[2]]; written.sort(); assert!(written.iter().all(|(key, _)| *key < seed[2].0)); // The policy spends its four skips on the initial commit, the two locations the writes // supersede, and the seed commit. It keeps the first update, evicts the last seeded // update, and keeps both updates the batch wrote. let evicted = seed[3].0; let mut policy = Script::new(usize::MAX, 4, move |key: &sha256::Digest| { if *key == evicted { Choice::Evict } else { Choice::Keep } }); let (db, _) = apply(db, writes, &mut policy).await; let visited: Vec<_> = [(1, seed[0]), (4, seed[3]), (6, written[0]), (7, written[1])] .into_iter() .map(|(loc, (key, value))| (Location::new(loc), key, value.unwrap())) .collect(); assert_eq!(policy.visited, visited); // No skip remains for the batch's delete at 8, so the floor stops there, one short of // the tip the writes reached. assert_eq!(*db.inactivity_floor_loc(), 8); let expected = [ seed[0], (seed[1].0, Some(digest(201))), (seed[2].0, None), (evicted, None), (created, Some(digest(104))), ]; for (key, value) in expected { assert_eq!(db.get(&key).await.unwrap(), value); } // The state survives commit and reopen. let db = db.commit().await.unwrap(); let (size, floor) = (db.size(), db.inactivity_floor_loc()); drop(db); let db: PolicyStore = open(child(1), "recover").await; assert_eq!((db.size(), db.inactivity_floor_loc()), (size, floor)); for (key, value) in expected { assert_eq!(db.get(&key).await.unwrap(), value); } // The state survives prune and reopen. let db = db.prune(floor).await.unwrap(); drop(db); let db: PolicyStore = open(child(2), "recover").await; assert_eq!((db.size(), db.inactivity_floor_loc()), (size, floor)); for (key, value) in expected { assert_eq!(db.get(&key).await.unwrap(), value); } assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); }); } /// The Store's walk reads only the candidates its limits let it reach, a stricter bound than /// the [`Policy`] contract's window. Candidates in the batch's own writes are already in /// memory, so a batch without writes reads at most the updates it reaches, and a policy that /// keeps every update it is offered reads each of them exactly once. /// /// Each case seeds `n` keys and then rewrites every `step`th key in each of `rounds` batches, /// all with a held floor, leaving the active updates at `active` and the size at `tip`. Each /// pair of `entries` and `skips` runs on a fresh store with that layout. #[rstest] // Seed 100 updates at 1..101, then rewrite every fourth key at 102..127. #[case::sparse( 100, 4, 1, (1..101).filter(|loc| loc % 4 != 1).chain(102..127).collect(), 128, &[0, 1, 10, 74, 75, 100, 200], &[0, 1, 2, 5, 30, u64::MAX], )] // Rewrite one key 19 times. The log holds the initial commit at 0, superseded updates and // commits at 1..39, the only active update at 39, and a commit at 40. #[case::long_gap(1, 1, 19, vec![39], 41, &[0, 1, 2], &[0, 1, 38, 39, 40, u64::MAX])] #[test_traced("WARN")] fn test_store_policy_reads_reached_candidates( #[case] n: u64, #[case] step: usize, #[case] rounds: usize, #[case] active: Vec, #[case] tip: u64, #[case] entries: &'static [usize], #[case] skips: &'static [u64], ) { deterministic::Runner::default().start(move |context| async move { let reads = || counter(&context, "log_items_read_total"); let child = |index| context.child("store").with_attribute("index", index); let seed = seed(n); let rewrites: Vec<_> = seed .iter() .step_by(step) .map(|(key, _)| (*key, Some(digest(500)))) .collect(); let mut trial = 0; for &entries in entries { for &skips in skips { let mut db = open(child(trial), &format!("reached-{trial}")).await; trial += 1; (db, _) = apply(db, seed.clone(), &mut Hold).await; for _ in 0..rounds { (db, _) = apply(db, rewrites.clone(), &mut Hold).await; } assert_eq!(*db.size(), tip); // The walk decides and reaches what [`walk_model`] predicts, and reads each // update it decides once. let before = reads(); let mut policy = Script::new(entries, skips, keep); let (db, _) = apply(db, [], &mut policy).await; let read = reads() - before; let (floor, decided) = walk_model(&active, 0, tip, entries, skips, &[]); assert_eq!( read, Widen::widen(decided.len()), "entries={entries} skips={skips}" ); let decided: Vec<_> = decided.into_iter().map(Location::new).collect(); assert_eq!( policy.locations(), decided, "entries={entries} skips={skips}" ); assert_eq!(*db.inactivity_floor_loc(), floor); db.destroy().await.unwrap(); } } }); } /// Under every pair of limits, a policy in a batch without writes, and in one that updates, /// deletes, and creates keys, decides the updates and reaches the floor that [`walk_model`] /// predicts over the active updates below the tip of the batch's writes, including the /// updates the batch wrote. Each decided update moves to the tip with its value. #[test_traced("WARN")] fn test_store_policy_limits() { deterministic::Runner::default().start(|context| async move { // Lay out the initial commit at 0, superseded updates at 1..4, active updates at // 4..7, the seed commit at 7, active updates at 8..11, and the last commit at 11. let seed = seed(6); let updates: Vec<_> = (200..) .zip(&seed[..3]) .map(|(i, (key, _))| (*key, Some(digest(i)))) .collect(); let applied: Vec<(u64, sha256::Digest, sha256::Digest)> = (4..7) .chain(8..11) .zip(seed[3..].iter().chain(&updates)) .map(|(loc, (key, value))| (loc, *key, value.unwrap())) .collect(); // The batch with writes updates the keys at 4 and 8, deletes the key at 5, and // creates two keys. Its five operations follow key order from 12, so the tip its // writes reach is 17. let mut writes = vec![ (seed[3].0, Some(digest(300))), (seed[0].0, Some(digest(301))), (seed[4].0, None), (digest(6), Some(digest(302))), (digest(7), Some(digest(303))), ]; writes.sort(); let mut trial = 0; for writes in [Vec::new(), writes] { // The active updates below the tip are the applied updates of keys the batch does // not write and the batch's own updates. let tip = 12 + Widen::widen(writes.len()); let mut live: BTreeMap = applied .iter() .filter(|(_, key, _)| writes.iter().all(|(written, _)| written != key)) .map(|(loc, key, value)| (*loc, (*key, *value))) .collect(); for (loc, (key, value)) in (12..).zip(&writes) { if let Some(value) = value { live.insert(loc, (*key, *value)); } } let active: Vec = live.keys().copied().collect(); assert_eq!(active.len(), 6 + usize::from(!writes.is_empty())); for entries in 0..=8 { for skips in (0..=12).chain([u64::MAX]) { let child = context.child("store").with_attribute("index", trial); let db = open(child, &format!("limits-{trial}")).await; trial += 1; let (db, _) = apply(db, seed.clone(), &mut Hold).await; let (db, _) = apply(db, updates.clone(), &mut Hold).await; assert_eq!((*db.inactivity_floor_loc(), *db.size()), (0, 12)); // The floor and the decided updates match [`walk_model`]. let (expected, decisions) = walk_model(&active, 0, tip, entries, skips, &[]); let decided: Vec<_> = decisions .iter() .map(|loc| (Location::new(*loc), live[loc].0, live[loc].1)) .collect(); let mut policy = Script::new(entries, skips, keep); let (db, range) = apply(db, writes.clone(), &mut policy).await; let limits = format!("writes={} entries={entries} skips={skips}", writes.len()); assert_eq!(policy.visited, decided, "{limits}"); assert_eq!(*db.inactivity_floor_loc(), expected, "{limits}"); // Each decided update moves to the tip with its value after the writes // and before the commit. assert_eq!( *range.start..*range.end, 12..tip + 1 + Widen::widen(decided.len()) ); for (loc, (_, key, value)) in (tip..).zip(&decided) { let op = db.get_op(Location::new(loc)).await.unwrap(); assert_eq!(op, Operation::Update(Update(*key, *value))); } for (key, value) in live.values() { assert_eq!(db.get(key).await.unwrap(), Some(*value)); } for (key, _) in writes.iter().filter(|(_, value)| value.is_none()) { assert_eq!(db.get(key).await.unwrap(), None); } assert_eq!(db.active_keys, live.len()); assert_bitmap_consistent(&db).await; db.destroy().await.unwrap(); } } } }); } /// A policy decides the batch's own updates like applied ones: keeping one moves it to the /// tip, replacing one writes the policy's value at the tip, evicting one deletes its key even /// when the batch created it, and stopping at one leaves the floor at its location. #[test_traced("WARN")] fn test_store_policy_decides_own_writes() { deterministic::Runner::default().start(|context| async move { let key = |i: u8| sha256::Digest::from([i; 32]); let child = |index| context.child("store").with_attribute("index", index); // Seed four keys at 1..5 with a held floor. The seed commit lies at 5. let db = open(child(0), "own").await; let seed = [1, 2, 4, 5].map(|i| (key(i), Some(digest(u64::from(i))))); let (db, _) = apply(db, seed, &mut Hold).await; assert_eq!(*db.size(), 6); // The batch rewrites every seeded key and creates key 3. In key order its updates lie // at 6..11, and they supersede 1..5. let written = |i: u8| digest(100 + u64::from(i)); let writes = [1, 2, 3, 4, 5].map(|i| (key(i), Some(written(i)))); // The policy spends its six skips on 0..6, then keeps key 1, replaces key 2, evicts // keys 3 and 4, and stops at key 5. let replacement = digest(200); let mut policy = Script::new(usize::MAX, 6, move |k: &sha256::Digest| { if *k == key(1) { Choice::Keep } else if *k == key(2) { Choice::Replace(replacement) } else if *k == key(5) { Choice::Stop } else { Choice::Evict } }); let (db, range) = apply(db, writes, &mut policy).await; let visited: Vec<_> = (6..) .zip(1..=5) .map(|(loc, i)| (Location::new(loc), key(i), written(i))) .collect(); assert_eq!(policy.visited, visited); assert_eq!(*db.inactivity_floor_loc(), 10); // The decisions follow the writes in decision order. The stopped update stays at 10. assert_eq!(*range.start..*range.end, 6..16); let expected = [ Operation::Update(Update(key(1), written(1))), Operation::Update(Update(key(2), replacement)), Operation::Delete(key(3)), Operation::Delete(key(4)), Operation::CommitFloor(None, Location::new(10)), ]; for (loc, expected) in (11..).zip(expected) { assert_eq!(db.get_op(Location::new(loc)).await.unwrap(), expected); } let values = [ (key(1), Some(written(1))), (key(2), Some(replacement)), (key(3), None), (key(4), None), (key(5), Some(written(5))), ]; for (key, value) in values { assert_eq!(db.get(&key).await.unwrap(), value); } assert_eq!(db.active_keys, 3); assert_bitmap_consistent(&db).await; // The state survives commit and reopen, and the next batch's policy decides the // stopped update without spending a skip. let db = db.commit().await.unwrap(); drop(db); let db: PolicyStore = open(child(1), "own").await; assert_eq!((*db.size(), *db.inactivity_floor_loc()), (16, 10)); for (key, value) in values { assert_eq!(db.get(&key).await.unwrap(), value); } assert_bitmap_consistent(&db).await; let mut policy = Script::new(1, 0, keep); let (db, _) = apply(db, [], &mut policy).await; assert_eq!(policy.visited, [(Location::new(10), key(5), written(5))]); assert_eq!(*db.inactivity_floor_loc(), 11); db.destroy().await.unwrap(); }); } /// A batch reads its policy's limits once. #[test_traced("WARN")] fn test_store_policy_reads_limits_once() { deterministic::Runner::default().start(|context| async move { // Seed four updates with a held floor. let db = open(context.child("store"), "limits").await; let (db, _) = apply(db, seed(4), &mut Hold).await; // A policy that allows one entry on its only limits read decides one update. let mut policy = Growing { exact: 1, reads: Cell::new(0), decided: 0, }; let (db, _) = apply(db, [], &mut policy).await; assert_eq!(policy.reads.get(), 1); assert_eq!(policy.decided, 1); db.destroy().await.unwrap(); }); } /// Assert that the activity bitmap, the snapshot, and the active key count of `db` are exact /// against a replay of its retained log. /// /// The bitmap covers the log and has pruned only whole chunks below the log's start. Every /// unpruned bit is set if and only if its location holds a live update or the last commit, /// and every live update lies at or above the inactivity floor. The snapshot maps each live /// key to its update and holds no other entry. async fn assert_bitmap_consistent(db: &Db) where E: Context, K: Key, V: VariableValue, T: Translator, { let bounds = db.log.bounds(); assert_eq!(db.bitmap.len(), bounds.end); let chunk_bits = bitmap::Prunable::::CHUNK_SIZE_BITS; assert_eq!( db.bitmap.pruned_bits(), bounds.start / chunk_bits * chunk_bits ); let positions: Vec = (bounds.start..bounds.end).collect(); let ops = db.log.read_many(&positions).await.unwrap(); let mut live = BTreeMap::new(); replay(&mut live, Location::new(bounds.start), &ops); let floor = db.inactivity_floor_loc(); assert!( live.values().all(|loc| *loc >= floor), "a live update lies below the floor {floor}", ); assert_bits(&db.bitmap, &live); assert_eq!(db.active_keys, live.len()); assert_eq!(db.snapshot.items(), live.len()); for (key, loc) in &live { assert!( db.snapshot.get(key).any(|entry| entry == loc), "snapshot misses the live update at {loc}", ); } } /// Over random batches of updates, deletes, and recreations of colliding keys, each followed by /// an empty batch whose policy keeps, replaces, evicts, or stops, the log replay matches the /// bitmap and the snapshot, and the store serves the modeled values. Pruning to the floor and /// reopening after a commit preserve that. #[test_traced] fn test_store_bitmap_tracks_activity() { struct BitmapPolicy<'a> { action: usize, expected: &'a mut BTreeMap>, decisions: &'a mut [usize; 4], } impl Policy> for BitmapPolicy<'_> { fn evicts(&self) -> bool { true } fn limits(&self, _: usize) -> Limits { Limits { entries: 3, skips: u64::MAX, } } fn decide<'a>( &mut self, entry: Entry<'a, crate::mmr::Family, Digest, Vec>, ) -> crate::qmdb::floor::Decision<'a, Vec> { self.decisions[self.action] += 1; match self.action { 0 => entry.keep(), 1 => { let value = vec![0xFF]; self.expected.insert(*entry.key(), value.clone()); entry.replace(value) } 2 => { self.expected.remove(entry.key()); entry.evict().0 } _ => entry.stop(), } } } deterministic::Runner::default().start(|context| async move { let mut rng = test_rng(); let mut db = create_test_store(context.child("store").with_attribute("index", 0)).await; // A small key universe whose keys share 3 translated prefixes, so batches routinely // update, delete, and recreate colliding keys. let keys: Vec = (0u8..24) .map(|i| { let mut key = Blake3::hash(&[&[i]]); key.0[0..2].copy_from_slice(&[0, i % 3]); key }) .collect(); let mut expected = BTreeMap::new(); let mut decisions = [0; 4]; // After each phase, the log replay matches the bitmap and the snapshot, and the store // serves exactly the modeled values. let assert_state = async |db: &TestStore, expected: &BTreeMap>| { assert_bitmap_consistent(db).await; assert_eq!(db.active_keys, expected.len()); for key in &keys { assert_eq!(db.get(key).await.unwrap(), expected.get(key).cloned()); } }; for round in 0..60u64 { let mut changes = Vec::new(); for _ in 0..rng.random_range(0..12) { let key = keys[rng.random_range(0..keys.len())]; if rng.random_bool(0.3) { changes.push((key, None)); expected.remove(&key); } else { let value = round.to_be_bytes().to_vec(); changes.push((key, Some(value.clone()))); expected.insert(key, value); } } (db, _) = apply_entries(db, changes).await; assert_state(&db, &expected).await; let mut policy = BitmapPolicy { action: round as usize % 4, expected: &mut expected, decisions: &mut decisions, }; db = db .apply_batch(core::iter::empty().collect(), &mut policy) .await .unwrap(); assert_state(&db, &expected).await; if round % 10 == 9 { let floor = db.inactivity_floor_loc(); db = db.prune(floor).await.unwrap(); assert_state(&db, &expected).await; } if round % 20 == 19 { db.commit().await.unwrap().sync().await.unwrap(); db = create_test_store( context.child("store").with_attribute("index", round + 1), ) .await; assert_state(&db, &expected).await; } } assert!(decisions.into_iter().all(|count| count > 0)); db.destroy().await.unwrap(); }); } /// Pruning past a whole bitmap chunk drops it, and reopening rebuilds the same unpruned bits /// and values. #[test_traced("WARN")] fn test_store_bitmap_chunk_prune_recovery() { deterministic::Runner::default().start(|context| async move { // Seed 300 keys at 1..301 and rewrite them at 302..602 with a held floor. An empty // batch with unlimited limits passes 0..302, keeps the rewrites by moving them to // 603..903, and reaches the tip of its writes, 603. Its commit lies at 903. let db = open(context.child("seed"), "bitmap-chunk").await; let writes = seed(300); let (db, _) = apply(db, writes.clone(), &mut Hold).await; let rewrites: Vec<_> = writes .iter() .map(|(key, _)| (*key, Some(digest(500)))) .collect(); let (db, _) = apply(db, rewrites, &mut Hold).await; let mut policy = Bounded { entries: usize::MAX, skips: u64::MAX, }; let (db, _) = apply(db, [], &mut policy).await; assert_eq!((*db.inactivity_floor_loc(), *db.size()), (603, 904)); // Pruning to the floor retains the log section starting at 602 and drops the bitmap's // first 512-bit chunk. let db = db.commit().await.unwrap(); let floor = db.inactivity_floor_loc(); let db = db.prune(floor).await.unwrap(); assert_eq!(db.log.bounds().start, 602); assert_eq!(db.bitmap.pruned_bits(), 512); assert_bitmap_consistent(&db).await; let bits: Vec<_> = (512..904).map(|loc| db.bitmap.get_bit(loc)).collect(); drop(db); // Reopening prunes the same chunk and rebuilds the same bits and values. let db: PolicyStore = open(context.child("reopen"), "bitmap-chunk").await; assert_eq!(db.bitmap.pruned_bits(), 512); assert_eq!(db.bitmap.len(), 904); let reopened: Vec<_> = (512..904).map(|loc| db.bitmap.get_bit(loc)).collect(); assert_eq!(reopened, bits); assert_bitmap_consistent(&db).await; for (key, _) in writes { assert_eq!(db.get(&key).await.unwrap(), Some(digest(500))); } db.destroy().await.unwrap(); }); } fn is_send(_: T) {} #[allow(dead_code)] fn assert_read_futures_are_send(db: TestStore, key: Digest, loc: Location) { is_send(db.get(&key)); is_send(db.get_metadata()); is_send(db.prune(loc)); } #[allow(dead_code)] fn assert_sync_is_send(db: TestStore) { is_send(db.sync()); } #[allow(dead_code)] fn assert_write_futures_are_send( db: Db, TwoCap>, key: Digest, value: Vec, ) { is_send(db.get(&key)); let batch = db.new_batch(); is_send(batch.get(&key)); is_send(db.apply_batch(Changeset::from([(key, Some(value))]), &mut Proportional)); } #[allow(dead_code)] fn assert_commit_is_send(db: Db, TwoCap>) { is_send(db.commit()); } }