Merge remote-tracking branch 'origin/main' into danlaine/keyless-inactivity-floor

Author: danlaine

Cargo.lock

@@ -117,7 +117,7 @@ curl http://localhost:9090/metrics
 4. **Sync Iteration Loop**: For each sync iteration:
    - **Database Initialization**: Client opens a new database (or reopens existing one)
    - **Connection**: Client establishes connection to server
-   - **Initial Sync Target**: Client requests server metadata to determine sync target (inactivity floor, size, and root digest)
+   - **Initial Sync Target**: Client requests server metadata to determine sync target (sync boundary, size, and root digest)
    - **Dynamic Target Updates**: Client periodically requests target updates during sync to handle new operations added by the server
    - **Sync Completion**: Client continues until all operations are applied and state matches server's target
    - **Database Closure**: Client closes the database to prepare for next iteration

examples/sync/README.md

@@ -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/client.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/bin/server.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/any.rs

@@ -139,8 +139,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

@@ -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/immutable.rs

@@ -144,8 +144,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()
     }
 
     async fn historical_proof(

examples/sync/src/databases/keyless.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/databases/mod.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,