[storage/qmdb/current] Generalized grafting core to support current QMDB (#3514)

Author: roberto-bayardo

examples/sync/src/bin/server.rs

@@ -8,7 +8,7 @@ use commonware_runtime::{
     tokio as tokio_runtime, BufferPooler, Clock, Listener, Metrics, Network, Runner, SinkOf,
     Spawner, Storage, StreamOf,
 };
-use commonware_storage::qmdb::sync::Target;
+use commonware_storage::{mmr, qmdb::sync::Target};
 use commonware_stream::utils::codec::{recv_frame, send_frame};
 use commonware_sync::{
     any, crate_version, current,
@@ -106,7 +106,7 @@ async fn maybe_add_operations<DB, E>(
     config: &Config,
 ) -> Result<(), Box<dyn std::error::Error>>
 where
-    DB: Syncable,
+    DB: Syncable<Family = mmr::Family>,
     E: Storage + Clock + Metrics + RngCore,
 {
     let now = context.current();
@@ -155,7 +155,7 @@ async fn handle_get_sync_target<DB>(
     request: wire::GetSyncTargetRequest,
 ) -> Result<wire::GetSyncTargetResponse<Key>, Error>
 where
-    DB: Syncable,
+    DB: Syncable<Family = mmr::Family>,
 {
     state.request_counter.inc();
 
@@ -186,7 +186,7 @@ async fn handle_get_operations<DB>(
     request: wire::GetOperationsRequest,
 ) -> Result<wire::GetOperationsResponse<DB::Operation, Key>, Error>
 where
-    DB: Syncable,
+    DB: Syncable<Family = mmr::Family>,
 {
     state.request_counter.inc();
     request.validate()?;
@@ -263,7 +263,7 @@ async fn handle_message<DB>(
     message: wire::Message<DB::Operation, Key>,
 ) -> wire::Message<DB::Operation, Key>
 where
-    DB: Syncable,
+    DB: Syncable<Family = mmr::Family>,
 {
     let request_id = message.request_id();
     match message {
@@ -317,7 +317,7 @@ async fn recv_loop<DB, E>(
     response_sender: mpsc::Sender<wire::Message<DB::Operation, Key>>,
     client_addr: SocketAddr,
 ) where
-    DB: Syncable + Send + Sync + 'static,
+    DB: Syncable<Family = mmr::Family> + Send + Sync + 'static,
     DB::Operation: Read + Send,
     <DB::Operation as Read>::Cfg: commonware_codec::IsUnit,
     E: Metrics + Network + Spawner,
@@ -367,7 +367,7 @@ async fn handle_client<DB, E>(
     client_addr: SocketAddr,
 ) -> Result<(), Box<dyn std::error::Error>>
 where
-    DB: Syncable + Send + Sync + 'static,
+    DB: Syncable<Family = mmr::Family> + Send + Sync + 'static,
     DB::Operation: Read + Send,
     <DB::Operation as Read>::Cfg: commonware_codec::IsUnit,
     E: Storage + Clock + Metrics + Network + Spawner,
@@ -408,7 +408,7 @@ async fn initialize_database<DB, E>(
     context: &mut E,
 ) -> Result<DB, Box<dyn std::error::Error>>
 where
-    DB: Syncable,
+    DB: Syncable<Family = mmr::Family>,
     E: RngCore,
 {
     info!("starting {} database", DB::name());
@@ -446,7 +446,7 @@ async fn run_helper<DB, E>(
     database: DB,
 ) -> Result<(), Box<dyn std::error::Error>>
 where
-    DB: Syncable + Send + Sync + 'static,
+    DB: Syncable<Family = mmr::Family> + Send + Sync + 'static,
     DB::Operation: Read + Send,
     <DB::Operation as Read>::Cfg: commonware_codec::IsUnit,
     E: Storage + Clock + Metrics + Network + Spawner + RngCore + Clone,

examples/sync/src/databases/any.rs

@@ -54,6 +54,7 @@ impl<E> crate::databases::Syncable for Database<E>
 where
     E: Storage + Clock + Metrics,
 {
+    type Family = mmr::Family;
     type Operation = Operation;
 
     fn create_test_operations(count: usize, seed: u64) -> Vec<Self::Operation> {

examples/sync/src/databases/current.rs

@@ -1,15 +1,15 @@
 //! Current database types and helpers for the sync example.
 //!
 //! A `current` database extends an `any` database with an activity bitmap that tracks which
-//! operations are active (i.e. represent the current state of their key) vs inactive
-//! (superseded or deleted). Its canonical root folds the ops root, a grafted MMR root
-//! (combining bitmap chunks with ops subtree roots), and an optional partial-chunk digest.
-//! See [current] module documentation for more details.
+//! operations are active (i.e. represent the current state of their key) vs inactive (superseded or
+//! deleted). Its canonical root folds the ops root, a grafted merkle root (combining bitmap chunks
+//! with ops subtree roots), and an optional partial-chunk digest. See [current] module
+//! documentation for more details.
 //!
-//! For sync, the engine targets the **ops root** (not the canonical root). The operations and
-//! proof format are identical to `any` -- the bitmap is reconstructed deterministically from
-//! the operations after sync completes. See the
-//! [Root structure](commonware_storage::qmdb::current) module documentation for details.
+//! For sync, the engine targets the **ops root** (not the canonical root). The operations and proof
+//! format are identical to `any` -- the bitmap is reconstructed deterministically from the
+//! operations after sync completes. See the [Root structure](commonware_storage::qmdb::current)
+//! module documentation for details.
 //!
 //! This module re-uses the same [`Operation`] type as [`super::any`] since the underlying
 //! operations log is the same.
@@ -36,7 +36,8 @@ use tracing::error;
 const CHUNK_SIZE: usize = sha256::Digest::SIZE;
 
 /// Database type alias.
-pub type Database<E> = current::unordered::fixed::Db<E, Key, Value, Hasher, Translator, CHUNK_SIZE>;
+pub type Database<E> =
+    current::unordered::fixed::Db<mmr::Family, E, Key, Value, Hasher, Translator, CHUNK_SIZE>;
 
 /// Operation type alias. Same as the `any` operation type.
 pub type Operation = FixedOperation<mmr::Family, Key, Value>;
@@ -45,7 +46,7 @@ pub type Operation = FixedOperation<mmr::Family, Key, Value>;
 pub fn create_config(context: &impl BufferPooler) -> Config<Translator> {
     let page_cache = buffer::paged::CacheRef::from_pooler(context, NZU16!(2048), NZUsize!(10));
     Config {
-        mmr_config: MmrConfig {
+        merkle_config: MmrConfig {
             journal_partition: "mmr-journal".into(),
             metadata_partition: "mmr-metadata".into(),
             items_per_blob: NZU64!(4096),
@@ -59,7 +60,7 @@ pub fn create_config(context: &impl BufferPooler) -> Config<Translator> {
             write_buffer: NZUsize!(4096),
             page_cache,
         },
-        grafted_mmr_metadata_partition: "grafted-mmr-metadata".into(),
+        grafted_metadata_partition: "grafted-mmr-metadata".into(),
         translator: Translator::default(),
     }
 }
@@ -68,6 +69,7 @@ impl<E> super::Syncable for Database<E>
 where
     E: Storage + Clock + Metrics,
 {
+    type Family = mmr::Family;
     type Operation = Operation;
 
     fn create_test_operations(count: usize, seed: u64) -> Vec<Self::Operation> {

examples/sync/src/databases/immutable.rs

@@ -87,6 +87,7 @@ impl<E> super::Syncable for Database<E>
 where
     E: Storage + Clock + Metrics,
 {
+    type Family = mmr::Family;
     type Operation = Operation;
 
     fn create_test_operations(count: usize, seed: u64) -> Vec<Self::Operation> {

examples/sync/src/databases/mod.rs

@@ -3,7 +3,7 @@
 use crate::Key;
 use commonware_codec::Encode;
 use commonware_storage::{
-    mmr::{self, Location, Proof},
+    merkle::{self, Location, Proof},
     qmdb::{self, operation::Operation},
 };
 use std::{future::Future, num::NonZeroU64};
@@ -48,8 +48,11 @@ impl DatabaseType {
 /// Helper trait for databases that can be synced.
 #[allow(clippy::type_complexity)]
 pub trait Syncable: Sized {
+    /// The merkle family used by this database.
+    type Family: merkle::Family;
+
     /// The type of operations in the database.
-    type Operation: Operation<mmr::Family> + Encode + Sync + 'static;
+    type Operation: Operation<Self::Family> + Encode + Sync + 'static;
 
     /// Create test operations with the given count and seed.
     /// The returned operations must end with a commit operation.
@@ -60,30 +63,35 @@ pub trait Syncable: Sized {
     fn add_operations(
         &mut self,
         operations: Vec<Self::Operation>,
-    ) -> impl Future<Output = Result<(), qmdb::Error<mmr::Family>>>;
+    ) -> impl Future<Output = Result<(), qmdb::Error<Self::Family>>>;
 
     /// Get the database's root digest.
     fn root(&self) -> Key;
 
     /// Get the total number of operations in the database (including pruned operations).
-    fn size(&self) -> impl Future<Output = Location> + Send;
+    fn size(&self) -> impl Future<Output = Location<Self::Family>> + Send;
 
     /// Get the inactivity floor, the location below which all operations are inactive.
-    fn inactivity_floor(&self) -> impl Future<Output = Location> + Send;
+    fn inactivity_floor(&self) -> impl Future<Output = Location<Self::Family>> + Send;
 
     /// Get historical proof and operations.
     fn historical_proof(
         &self,
-        op_count: Location,
-        start_loc: Location,
+        op_count: Location<Self::Family>,
+        start_loc: Location<Self::Family>,
         max_ops: NonZeroU64,
-    ) -> impl Future<Output = Result<(Proof<Key>, Vec<Self::Operation>), qmdb::Error<mmr::Family>>> + Send;
+    ) -> impl Future<
+        Output = Result<
+            (Proof<Self::Family, Key>, Vec<Self::Operation>),
+            qmdb::Error<Self::Family>,
+        >,
+    > + Send;
 
-    /// Get the pinned MMR nodes for a lower operation boundary of `loc`.
+    /// Get the pinned nodes for a lower operation boundary of `loc`.
     fn pinned_nodes_at(
         &self,
-        loc: Location,
-    ) -> impl Future<Output = Result<Vec<Key>, qmdb::Error<mmr::Family>>> + Send;
+        loc: Location<Self::Family>,
+    ) -> impl Future<Output = Result<Vec<Key>, qmdb::Error<Self::Family>>> + Send;
 
     /// Get the database type name for logging.
     fn name() -> &'static str;

storage/fuzz/fuzz_targets/current_crash_recovery.rs

@@ -15,7 +15,7 @@ use commonware_runtime::{
 };
 use commonware_storage::{
     journal::contiguous::variable::Config as VConfig,
-    mmr::{journaled::Config as MmrConfig, Location},
+    mmr::{self, journaled::Config as MmrConfig, Location},
     qmdb::current::{unordered::variable::Db as Current, VariableConfig},
     translator::TwoCap,
 };
@@ -34,7 +34,7 @@ type RawValue = [u8; 32];
 /// Maximum write buffer size.
 const MAX_WRITE_BUF: usize = 2048;
 
-type Db = Current<deterministic::Context, Key, Value, Sha256, TwoCap, 32>;
+type Db = Current<mmr::Family, deterministic::Context, Key, Value, Sha256, TwoCap, 32>;
 
 fn bounded_page_size(u: &mut Unstructured<'_>) -> Result<u16> {
     u.int_in_range(1..=256)
@@ -98,7 +98,7 @@ fn make_config(
 ) -> VariableConfig<TwoCap, ((), ())> {
     let page_cache = CacheRef::from_pooler(ctx, page_size, page_cache_size);
     VariableConfig {
-        mmr_config: MmrConfig {
+        merkle_config: MmrConfig {
             journal_partition: format!("crash-mmr-journal-{suffix}"),
             metadata_partition: format!("crash-mmr-metadata-{suffix}"),
             items_per_blob: NZU64!(mmr_items_per_blob),
@@ -114,7 +114,7 @@ fn make_config(
             codec_config: ((), ()),
             page_cache,
         },
-        grafted_mmr_metadata_partition: format!("crash-grafted-mmr-metadata-{suffix}"),
+        grafted_metadata_partition: format!("crash-grafted-mmr-metadata-{suffix}"),
         translator: TwoCap,
     }
 }

storage/fuzz/fuzz_targets/current_ordered_operations.rs

@@ -5,7 +5,7 @@ use commonware_cryptography::{sha256::Digest, Hasher, Sha256};
 use commonware_runtime::{buffer::paged::CacheRef, deterministic, Runner};
 use commonware_storage::{
     journal::contiguous::fixed::Config as FConfig,
-    mmr::{journaled::Config as MmrConfig, Location},
+    mmr::{self, journaled::Config as MmrConfig, Location},
     qmdb::current::{ordered::fixed::Db as CurrentDb, FixedConfig as Config},
     translator::TwoCap,
 };
@@ -20,7 +20,7 @@ type Key = FixedBytes<32>;
 type Value = FixedBytes<32>;
 type RawKey = [u8; 32];
 type RawValue = [u8; 32];
-type Db = CurrentDb<deterministic::Context, Key, Value, Sha256, TwoCap, 32>;
+type Db = CurrentDb<mmr::Family, deterministic::Context, Key, Value, Sha256, TwoCap, 32>;
 
 #[derive(Arbitrary, Debug, Clone)]
 enum CurrentOperation {
@@ -115,7 +115,7 @@ fn fuzz(data: FuzzInput) {
             NZUsize!(PAGE_CACHE_SIZE),
         );
         let cfg = Config {
-            mmr_config: MmrConfig {
+            merkle_config: MmrConfig {
                 journal_partition: "fuzz-current-mmr-journal".into(),
                 metadata_partition: "fuzz-current-mmr-metadata".into(),
                 items_per_blob: NZU64!(MMR_ITEMS_PER_BLOB),
@@ -129,7 +129,7 @@ fn fuzz(data: FuzzInput) {
                 write_buffer: NZUsize!(WRITE_BUFFER_SIZE),
                 page_cache,
             },
-            grafted_mmr_metadata_partition: "fuzz-current-grafted-mmr-metadata".into(),
+            grafted_metadata_partition: "fuzz-current-grafted-mmr-metadata".into(),
             translator: TwoCap,
         };
 

storage/fuzz/fuzz_targets/current_unordered_batch_root.rs

@@ -5,7 +5,7 @@ use commonware_cryptography::Sha256;
 use commonware_runtime::{buffer::paged::CacheRef, deterministic, BufferPooler, Runner};
 use commonware_storage::{
     journal::contiguous::fixed::Config as FConfig,
-    mmr::journaled::Config as MmrConfig,
+    mmr::{self, journaled::Config as MmrConfig},
     qmdb::current::{unordered::fixed::Db as CurrentDb, FixedConfig as Config},
     translator::OneCap,
 };
@@ -15,7 +15,7 @@ use std::num::NonZeroU16;
 
 type Key = FixedBytes<32>;
 type Value = FixedBytes<32>;
-type Db = CurrentDb<deterministic::Context, Key, Value, Sha256, OneCap, 32>;
+type Db = CurrentDb<mmr::Family, deterministic::Context, Key, Value, Sha256, OneCap, 32>;
 
 const PAGE_SIZE: NonZeroU16 = NZU16!(137);
 const COLLISION_GROUPS: u8 = 4;
@@ -76,7 +76,7 @@ impl<'a> Arbitrary<'a> for FuzzInput {
 fn test_config(name: &str, pooler: &impl BufferPooler) -> Config<OneCap> {
     let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, NZUsize!(2));
     Config {
-        mmr_config: MmrConfig {
+        merkle_config: MmrConfig {
             journal_partition: format!("{name}-mmr"),
             metadata_partition: format!("{name}-meta"),
             items_per_blob: NZU64!(17),
@@ -90,7 +90,7 @@ fn test_config(name: &str, pooler: &impl BufferPooler) -> Config<OneCap> {
             write_buffer: NZUsize!(1024),
             page_cache,
         },
-        grafted_mmr_metadata_partition: format!("{name}-grafted"),
+        grafted_metadata_partition: format!("{name}-grafted"),
         translator: OneCap,
     }
 }

storage/fuzz/fuzz_targets/current_unordered_operations.rs

@@ -5,7 +5,7 @@ use commonware_cryptography::{sha256::Digest, Hasher, Sha256};
 use commonware_runtime::{buffer::paged::CacheRef, deterministic, Runner};
 use commonware_storage::{
     journal::contiguous::fixed::Config as FConfig,
-    mmr::{journaled::Config as MmrConfig, Location},
+    mmr::{self, journaled::Config as MmrConfig, Location},
     qmdb::current::{unordered::fixed::Db as CurrentDb, FixedConfig as Config},
     translator::TwoCap,
 };
@@ -20,7 +20,7 @@ type Key = FixedBytes<32>;
 type Value = FixedBytes<32>;
 type RawKey = [u8; 32];
 type RawValue = [u8; 32];
-type Db = CurrentDb<deterministic::Context, Key, Value, Sha256, TwoCap, 32>;
+type Db = CurrentDb<mmr::Family, deterministic::Context, Key, Value, Sha256, TwoCap, 32>;
 
 #[derive(Arbitrary, Debug, Clone)]
 enum CurrentOperation {
@@ -105,7 +105,7 @@ fn fuzz(data: FuzzInput) {
             NZUsize!(PAGE_CACHE_SIZE),
         );
         let cfg = Config {
-            mmr_config: MmrConfig {
+            merkle_config: MmrConfig {
                 journal_partition: "fuzz-current-mmr-journal".into(),
                 metadata_partition: "fuzz-current-mmr-metadata".into(),
                 items_per_blob: NZU64!(MMR_ITEMS_PER_BLOB),
@@ -119,7 +119,7 @@ fn fuzz(data: FuzzInput) {
                 write_buffer: NZUsize!(WRITE_BUFFER_SIZE),
                 page_cache,
             },
-            grafted_mmr_metadata_partition: "fuzz-current-grafted-mmr-metadata".into(),
+            grafted_metadata_partition: "fuzz-current-grafted-mmr-metadata".into(),
             translator: TwoCap,
         };