[storage/qmdb/sync] Avoid requiring a trusted ops root in current qmdb sync (#3778)

Co-authored-by: Patrick O'Grady <me@patrickogrady.xyz>

Author: roberto-bayardo

examples/sync/src/bin/client.rs

@@ -11,7 +11,11 @@ use commonware_runtime::{
 };
 use commonware_storage::{
     mmr,
-    qmdb::sync::{self, compact},
+    qmdb::{
+        any::sync::Target,
+        current as current_qmdb,
+        sync::{self, compact},
+    },
 };
 use commonware_sync::{
     any, crate_version, current,
@@ -70,9 +74,9 @@ struct Config {
 async fn target_update_task<E, Op, D>(
     context: E,
     resolver: Resolver<Op, D>,
-    update_tx: mpsc::Sender<sync::Target<mmr::Family, D>>,
+    update_tx: mpsc::Sender<Target<mmr::Family, D>>,
     interval_duration: Duration,
-    initial_target: sync::Target<mmr::Family, D>,
+    initial_target: Target<mmr::Family, D>,
 ) -> Result<(), Error>
 where
     E: Clock,
@@ -132,8 +136,8 @@ where
         E,
         Config,
         Resolver<Op, Key>,
-        sync::Target<mmr::Family, Key>,
-        mpsc::Receiver<sync::Target<mmr::Family, Key>>,
+        Target<mmr::Family, Key>,
+        mpsc::Receiver<Target<mmr::Family, Key>>,
         u32,
     ) -> SyncFut,
     SyncFut: Future<Output = Result<DB, Box<dyn std::error::Error>>>,
@@ -188,7 +192,7 @@ where
         |context, config, resolver, initial_target, update_receiver, iteration| async move {
             let db_config = any::create_config(&context);
             let sync_config =
-                sync::engine::Config::<any::Database<_>, Resolver<any::Operation, Key>> {
+                sync::engine::Config::<any::Database<_>, Resolver<any::Operation, Key>, _> {
                     context,
                     db_config,
                     fetch_batch_size: config.batch_size,
@@ -217,44 +221,94 @@ where
 
 /// Repeatedly sync a Current database to the server's state.
 ///
-/// The sync engine targets the ops root (not the canonical root). After sync completes,
-/// the bitmap and grafted MMR are reconstructed from the synced operations.
+/// Uses the `current::sync::sync` wrapper. The wrapper verifies each target's `OpsRootWitness`
+/// before forwarding its ops root to the shared sync engine, then checks the database root for the
+/// target the engine finishes on.
 async fn run_current<E>(context: E, config: Config) -> Result<(), Box<dyn std::error::Error>>
 where
     E: BufferPooler + Storage + Clock + Metrics + Network + Spawner,
 {
-    run_full_sync::<current::Database<_>, current::Operation, _, _, _>(
-        context,
-        config,
-        |context, config, resolver, initial_target, update_receiver, iteration| async move {
-            let db_config = current::create_config(&context);
-            let sync_config =
-                sync::engine::Config::<current::Database<_>, Resolver<current::Operation, Key>> {
-                    context,
-                    db_config,
-                    fetch_batch_size: config.batch_size,
-                    target: initial_target,
-                    resolver,
-                    apply_batch_size: 1024,
-                    max_outstanding_requests: config.max_outstanding_requests,
-                    update_rx: Some(update_receiver),
-                    finish_rx: None,
-                    reached_target_tx: None,
-                    max_retained_roots: 8,
-                };
-            let database: current::Database<_> = sync::sync(sync_config).await?;
-            info!(
-                sync_iteration = iteration,
-                canonical_root = %database.root(),
-                ops_root = %database.ops_root(),
-                sync_interval = ?config.sync_interval,
-                "Current sync completed successfully"
-            );
-            Ok(database)
-        },
-        "Current database",
-    )
-    .await
+    info!("starting Current database sync process");
+    let mut iteration = 0u32;
+    loop {
+        let resolver =
+            Resolver::<current::Operation, Key>::connect(context.child("resolver"), config.server)
+                .await?;
+
+        let initial_target = resolver.get_current_sync_target().await?;
+        info!(
+            root = %initial_target.root,
+            ops_root = %initial_target.ops_root,
+            range = ?initial_target.range,
+            "received current sync target"
+        );
+
+        let (update_sender, update_receiver) = mpsc::channel(UPDATE_CHANNEL_SIZE);
+
+        let target_update_handle = {
+            let resolver = resolver.clone();
+            let mut current_target_root = initial_target.root;
+            let target_update_interval = config.target_update_interval;
+            context
+                .child("target_update")
+                .spawn(move |context| async move {
+                    loop {
+                        context.sleep(target_update_interval).await;
+                        match resolver.get_current_sync_target().await {
+                            Ok(new_target) => {
+                                if current_target_root != new_target.root {
+                                    let new_root = new_target.root;
+                                    match update_sender.clone().try_send(new_target) {
+                                        Ok(()) => {
+                                            info!("target updated");
+                                            current_target_root = new_root;
+                                        }
+                                        Err(mpsc::error::TrySendError::Closed(_)) => return Ok(()),
+                                        Err(err) => {
+                                            warn!(?err, "failed to send target update");
+                                            return Err(Error::TargetUpdateChannel {
+                                                reason: err.to_string(),
+                                            });
+                                        }
+                                    }
+                                }
+                            }
+                            Err(err) => {
+                                warn!(?err, "failed to get sync target from server");
+                            }
+                        }
+                    }
+                })
+        };
+
+        let db_config = current::create_config(&context);
+        let database: current::Database<_> = current_qmdb::sync::sync(current_qmdb::sync::Config {
+            context: context.child("sync"),
+            resolver,
+            target: initial_target,
+            max_outstanding_requests: config.max_outstanding_requests,
+            fetch_batch_size: config.batch_size,
+            apply_batch_size: 1024,
+            db_config,
+            update_rx: Some(update_receiver),
+            finish_rx: None,
+            reached_target_tx: None,
+            max_retained_roots: 8,
+        })
+        .await?;
+
+        target_update_handle.abort();
+        info!(
+            sync_iteration = iteration,
+            root = %database.root(),
+            ops_root = %database.ops_root(),
+            sync_interval = ?config.sync_interval,
+            "Current sync completed successfully"
+        );
+        database.destroy().await?;
+        context.sleep(config.sync_interval).await;
+        iteration += 1;
+    }
 }
 
 /// Repeatedly sync an Immutable database to the server's state.
@@ -270,6 +324,7 @@ where
             let sync_config = sync::engine::Config::<
                 immutable::Database<_>,
                 Resolver<immutable::Operation, Key>,
+                _,
             > {
                 context,
                 db_config,
@@ -307,20 +362,23 @@ where
         config,
         |context, config, resolver, initial_target, update_receiver, iteration| async move {
             let db_config = keyless::create_config(&context);
-            let sync_config =
-                sync::engine::Config::<keyless::Database<_>, Resolver<keyless::Operation, Key>> {
-                    context,
-                    db_config,
-                    fetch_batch_size: config.batch_size,
-                    target: initial_target,
-                    resolver,
-                    apply_batch_size: 1024,
-                    max_outstanding_requests: config.max_outstanding_requests,
-                    update_rx: Some(update_receiver),
-                    finish_rx: None,
-                    reached_target_tx: None,
-                    max_retained_roots: 8,
-                };
+            let sync_config = sync::engine::Config::<
+                keyless::Database<_>,
+                Resolver<keyless::Operation, Key>,
+                _,
+            > {
+                context,
+                db_config,
+                fetch_batch_size: config.batch_size,
+                target: initial_target,
+                resolver,
+                apply_batch_size: 1024,
+                max_outstanding_requests: config.max_outstanding_requests,
+                update_rx: Some(update_receiver),
+                finish_rx: None,
+                reached_target_tx: None,
+                max_retained_roots: 8,
+            };
             let database: keyless::Database<_> = sync::sync(sync_config).await?;
             info!(
                 sync_iteration = iteration,

examples/sync/src/bin/server.rs

@@ -14,7 +14,7 @@ use commonware_runtime::{
 };
 use commonware_storage::{
     mmr,
-    qmdb::sync::{compact, Target},
+    qmdb::{any::sync::Target, sync::compact},
 };
 use commonware_stream::utils::codec::{recv_frame, send_frame};
 use commonware_sync::{
@@ -220,7 +220,7 @@ where
     state.request_counter.inc();
 
     // Get the current database state
-    let (root, sync_boundary, size) = {
+    let (ops_root, sync_boundary, size) = {
         let database = state.database.read().await;
         (
             database.root(),
@@ -230,10 +230,7 @@ where
     };
     let response = wire::GetSyncTargetResponse::<Key> {
         request_id: request.request_id,
-        target: Target {
-            root,
-            range: non_empty_range!(sync_boundary, size),
-        },
+        target: Target::new(ops_root, non_empty_range!(sync_boundary, size)),
     };
 
     debug!(?response, "serving target update");
@@ -407,6 +404,49 @@ where
     }
 }
 
+struct CurrentFullMode;
+
+impl<E> ServeMode<current::Database<E>> for CurrentFullMode
+where
+    E: Storage + Clock + Metrics + Send + Sync + 'static,
+{
+    const LISTENING_MESSAGE: &'static str =
+        "current server listening and continuously adding operations";
+    const SHUTDOWN_MESSAGE: &'static str = "context shutdown, stopping current server";
+
+    async fn handle_message(
+        state: &State<current::Database<E>>,
+        message: wire::Message<current::Operation, Key>,
+    ) -> wire::Message<current::Operation, Key> {
+        let request_id = message.request_id();
+        match message {
+            wire::Message::GetOperationsRequest(request) => dispatch_message!(
+                state,
+                request_id,
+                wire::Message::GetOperationsResponse,
+                handle_get_operations::<current::Database<E>>(state, request)
+            ),
+            wire::Message::GetSyncTargetRequest(request) => {
+                state.request_counter.inc();
+                let database = state.database.read().await;
+                match current::current_sync_target(&*database).await {
+                    Ok(target) => {
+                        debug!(?target, "serving current sync target");
+                        wire::Message::GetCurrentSyncTargetResponse(
+                            wire::GetCurrentSyncTargetResponse {
+                                request_id: request.request_id,
+                                target,
+                            },
+                        )
+                    }
+                    Err(err) => error_message(state, request_id, err.into()),
+                }
+            }
+            _ => unexpected_message(state, request_id),
+        }
+    }
+}
+
 impl<DB> ServeMode<DB> for CompactMode
 where
     DB: CompactSyncable<Family = mmr::Family> + Send + Sync + 'static,
@@ -710,14 +750,18 @@ where
 }
 
 /// Run the Current database server.
-async fn run_current<E>(context: E, config: Config) -> Result<(), Box<dyn std::error::Error>>
+///
+/// Uses `CurrentFullMode` to serve database-root-aware sync targets with an
+/// [OpsRootWitness](commonware_storage::qmdb::current::proof::OpsRootWitness).
+async fn run_current<E>(mut context: E, config: Config) -> Result<(), Box<dyn std::error::Error>>
 where
     E: BufferPooler + Storage + Clock + Metrics + Network + Spawner + RngCore + Send,
 {
     let db_config = current::create_config(&context);
     let database = current::Database::init(context.child("database"), db_config).await?;
+    let database = initialize_database(database, &config, &mut context).await?;
 
-    run_helper(context, config, database).await
+    run_server::<current::Database<_>, E, CurrentFullMode>(context, config, database).await
 }
 
 /// Run the Immutable database server.

examples/sync/src/databases/current.rs

@@ -2,14 +2,16 @@
 //!
 //! 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 merkle root (combining bitmap chunks
+//! deleted). Its database 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`; direct proof verifiers should use `qmdb::hasher`. 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 internally targets the **ops root** (not the database root). The
+//! operations and proof format are identical to `any`; direct proof verifiers should use
+//! `qmdb::hasher`. The bitmap is reconstructed deterministically from the operations after sync
+//! completes. The `current::sync` wrapper verifies each target's `OpsRootWitness` against its
+//! trusted database root, then checks the reconstructed database root for the target the
+//! engine finishes on.
 //!
 //! This module re-uses the same [`Operation`] type as [`super::any`] since the underlying
 //! operations log is the same.
@@ -25,11 +27,11 @@ use commonware_storage::{
     qmdb::{
         self,
         any::unordered::{fixed::Operation as FixedOperation, Update},
-        current::{self, FixedConfig as Config},
+        current::{self, sync::Target as CurrentTarget, FixedConfig as Config},
         operation::Committable,
     },
 };
-use commonware_utils::{NZUsize, NZU16, NZU64};
+use commonware_utils::{non_empty_range, NZUsize, NZU16, NZU64};
 use std::{future::Future, num::NonZeroU64};
 use tracing::error;
 
@@ -145,9 +147,7 @@ where
     }
 
     fn root(&self) -> Key {
-        // Return the ops root (not the canonical root) because this is what the
-        // sync engine verifies against.
-        self.ops_root()
+        self.root()
     }
 
     fn name() -> &'static str {
@@ -186,6 +186,21 @@ where
     }
 }
 
+pub async fn current_sync_target<E: Storage + Clock + Metrics>(
+    db: &Database<E>,
+) -> Result<CurrentTarget<mmr::Family, Key>, qmdb::Error<mmr::Family>> {
+    let hasher = qmdb::hasher::<Hasher>();
+    let witness = db.ops_root_witness(&hasher).await?;
+    let sync_boundary = db.sync_boundary();
+    let size = db.bounds().await.end;
+    Ok(CurrentTarget::new(
+        db.root(),
+        db.ops_root(),
+        witness,
+        non_empty_range!(sync_boundary, size),
+    ))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;