generalize sync protocol on merkle family

Author: roberto-bayardo

examples/sync/src/bin/client.rs

@@ -8,7 +8,7 @@ use commonware_codec::{EncodeShared, Read};
 use commonware_runtime::{
     tokio as tokio_runtime, BufferPooler, Clock, Metrics, Network, Runner, Spawner, Storage,
 };
-use commonware_storage::qmdb::sync;
+use commonware_storage::{mmr, qmdb::sync};
 use commonware_sync::{
     any, crate_version, current, databases::DatabaseType, immutable, keyless, net::Resolver,
     Digest, Error, Key,
@@ -60,9 +60,9 @@ struct Config {
 async fn target_update_task<E, Op, D>(
     context: E,
     resolver: Resolver<Op, D>,
-    update_tx: mpsc::Sender<sync::Target<D>>,
+    update_tx: mpsc::Sender<sync::Target<mmr::Family, D>>,
     interval_duration: Duration,
-    initial_target: sync::Target<D>,
+    initial_target: sync::Target<mmr::Family, D>,
 ) -> Result<(), Error>
 where
     E: Clock,

examples/sync/src/bin/server.rs

@@ -160,19 +160,19 @@ where
     state.request_counter.inc();
 
     // Get the current database state
-    let (root, inactivity_floor, size) = {
+    let (root, sync_boundary, size) = {
         let database = state.database.read().await;
         (
             database.root(),
-            database.inactivity_floor().await,
+            database.sync_boundary().await,
             database.size().await,
         )
     };
     let response = wire::GetSyncTargetResponse::<Key> {
         request_id: request.request_id,
         target: Target {
             root,
-            range: non_empty_range!(inactivity_floor, size),
+            range: non_empty_range!(sync_boundary, size),
         },
     };
 
@@ -430,7 +430,7 @@ where
         .collect::<String>();
     info!(
         size = ?database.size().await,
-        inactivity_floor = ?database.inactivity_floor().await,
+        sync_boundary = ?database.sync_boundary().await,
         root = %root_hex,
         "{} database ready",
         DB::name()

examples/sync/src/databases/any.rs

@@ -123,8 +123,8 @@ where
         self.bounds().await.end
     }
 
-    async fn inactivity_floor(&self) -> Location {
-        self.inactivity_floor_loc()
+    async fn sync_boundary(&self) -> Location {
+        self.sync_boundary()
     }
 
     fn historical_proof(

examples/sync/src/databases/current.rs

@@ -139,8 +139,9 @@ where
         self.bounds().await.end
     }
 
-    async fn inactivity_floor(&self) -> Location {
-        self.inactivity_floor_loc()
+    async fn sync_boundary(&self) -> Location {
+        self.sync_boundary()
+            .expect("sync_boundary should not overflow")
     }
 
     fn historical_proof(

examples/sync/src/databases/immutable.rs

@@ -129,10 +129,8 @@ where
         self.bounds().await.end
     }
 
-    async fn inactivity_floor(&self) -> Location {
-        // For Immutable databases, all retained operations are active,
-        // so the inactivity floor equals the pruning boundary.
-        self.bounds().await.start
+    async fn sync_boundary(&self) -> Location {
+        self.sync_boundary().await
     }
 
     fn historical_proof(

examples/sync/src/databases/keyless.rs

@@ -126,10 +126,8 @@ where
         self.bounds().await.end
     }
 
-    async fn inactivity_floor(&self) -> Location {
-        // Keyless databases have no inactivity floor concept.
-        // Use the pruning boundary, same as immutable.
-        self.bounds().await.start
+    async fn sync_boundary(&self) -> Location {
+        self.sync_boundary().await
     }
 
     async fn historical_proof(

examples/sync/src/databases/mod.rs

@@ -75,8 +75,11 @@ pub trait Syncable: Sized {
     /// Get the total number of operations in the database (including pruned operations).
     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<Self::Family>> + Send;
+    /// Get the most recent location from which this database can safely be synced.
+    ///
+    /// Callers constructing a sync target should use this value (or any earlier retained
+    /// location) as the `range.start`.
+    fn sync_boundary(&self) -> impl Future<Output = Location<Self::Family>> + Send;
 
     /// Get historical proof and operations.
     fn historical_proof(

examples/sync/src/net/resolver.rs

@@ -3,7 +3,10 @@ use crate::net::request_id;
 use commonware_codec::{EncodeShared, IsUnit, Read};
 use commonware_cryptography::Digest;
 use commonware_runtime::{Network, Spawner};
-use commonware_storage::{mmr::Location, qmdb::sync};
+use commonware_storage::{
+    mmr::{self, Location},
+    qmdb::sync,
+};
 use commonware_utils::channel::{mpsc, oneshot};
 use std::num::NonZeroU64;
 
@@ -42,7 +45,7 @@ where
     }
 
     /// Returns the current sync target from the server.
-    pub async fn get_sync_target(&self) -> Result<sync::Target<D>, crate::Error> {
+    pub async fn get_sync_target(&self) -> Result<sync::Target<mmr::Family, D>, crate::Error> {
         let request_id = self.request_id_generator.next();
         let request =
             wire::Message::GetSyncTargetRequest(wire::GetSyncTargetRequest { request_id });
@@ -75,6 +78,7 @@ where
     Op::Cfg: IsUnit,
     D: Digest,
 {
+    type Family = mmr::Family;
     type Digest = D;
     type Op = Op;
     type Error = crate::Error;
@@ -86,7 +90,8 @@ where
         max_ops: NonZeroU64,
         include_pinned_nodes: bool,
         _cancel_rx: oneshot::Receiver<()>,
-    ) -> Result<sync::resolver::FetchResult<Self::Op, Self::Digest>, Self::Error> {
+    ) -> Result<sync::resolver::FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error>
+    {
         let request_id = self.request_id_generator.next();
         let request = wire::Message::GetOperationsRequest(wire::GetOperationsRequest {
             request_id,

examples/sync/src/net/wire.rs

@@ -5,7 +5,7 @@ use commonware_codec::{
 use commonware_cryptography::Digest;
 use commonware_runtime::{Buf, BufMut};
 use commonware_storage::{
-    mmr::{Location, Proof},
+    mmr::{self, Location, Proof},
     qmdb::sync::Target,
 };
 use std::num::NonZeroU64;
@@ -51,7 +51,7 @@ where
     D: Digest,
 {
     pub request_id: RequestId,
-    pub target: Target<D>,
+    pub target: Target<mmr::Family, D>,
 }
 
 /// Messages that can be sent over the wire.
@@ -334,7 +334,7 @@ where
     type Cfg = ();
     fn read_cfg(buf: &mut impl Buf, _: &()) -> Result<Self, CodecError> {
         let request_id = RequestId::read_cfg(buf, &())?;
-        let target = Target::<D>::read_cfg(buf, &())?;
+        let target = Target::<mmr::Family, D>::read_cfg(buf, &())?;
         Ok(Self { request_id, target })
     }
 }

storage/conformance.toml

@@ -158,7 +158,7 @@ hash = "c692a20fa40180844888e7a26401f99a45ce3127faeca5f1228a41b454424623"
 n_cases = 65536
 hash = "b41d6c6ec560bde9caf2e206526864c618a0721af367585a1719617ca7ce9291"
 
-["commonware_storage::qmdb::sync::target::tests::conformance::CodecConformance<Target<sha256::Digest>>"]
+["commonware_storage::qmdb::sync::target::tests::conformance::CodecConformance<Target<MmrFamily,sha256::Digest>>"]
 n_cases = 65536
 hash = "f742d92a0af0af78a9519bf637bc52ea869965a85a84101a4c53f53eb39325ca"