[storage/qmdb] Coordinate state sync finalization

Author: clabby

examples/sync/src/bin/client.rs

@@ -150,6 +150,8 @@ where
                 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,
             };
 
@@ -214,6 +216,8 @@ where
                 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,
             };
 
@@ -275,6 +279,8 @@ where
                 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,
             };
 

storage/fuzz/fuzz_targets/qmdb_any_fixed_sync.rs

@@ -122,6 +122,8 @@ async fn test_sync<
     let sync_config: sync::engine::Config<FixedDb, R> = sync::engine::Config {
         context: context.with_label("sync").with_attribute("id", sync_id),
         update_rx: None,
+        finish_rx: None,
+        reached_target_tx: None,
         db_config,
         fetch_batch_size: NZU64!((fetch_batch_size % 100) + 1),
         target,

storage/src/qmdb/any/sync/tests.rs

@@ -22,13 +22,15 @@ use crate::{
 };
 use commonware_codec::Encode;
 use commonware_cryptography::sha256::Digest;
+use commonware_macros::select;
 use commonware_runtime::{deterministic, BufferPooler, Metrics, Runner as _};
 use commonware_utils::{
     channel::{mpsc, oneshot},
     non_empty_range,
     sync::AsyncRwLock,
     NZU64,
 };
+use futures::{pin_mut, FutureExt};
 use rand::RngCore as _;
 use std::{num::NonZeroU64, sync::Arc};
 
@@ -132,6 +134,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
 
@@ -174,6 +178,8 @@ where
             fetch_batch_size: NZU64!(2),
             db_config,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
 
@@ -223,6 +229,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
 
@@ -304,6 +312,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
 
@@ -378,6 +388,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
         let synced_db: H::Db = sync::sync(config).await.unwrap();
@@ -479,6 +491,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
         let synced_db: H::Db = sync::sync(config).await.unwrap();
@@ -541,6 +555,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 10,
             update_rx: Some(update_receiver),
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 1,
         };
         let client: Engine<H::Db, _> = Engine::new(config).await.unwrap();
@@ -608,6 +624,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 10,
             update_rx: Some(update_receiver),
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 1,
         };
         let client: Engine<H::Db, _> = Engine::new(config).await.unwrap();
@@ -689,6 +707,8 @@ where
                 apply_batch_size: 1024,
                 max_outstanding_requests: 1,
                 update_rx: Some(update_receiver),
+                finish_rx: None,
+                reached_target_tx: None,
                 max_retained_roots: 1,
             };
 
@@ -759,6 +779,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 10,
             update_rx: Some(update_receiver),
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 1,
         };
 
@@ -790,6 +812,109 @@ where
     });
 }
 
+/// Test that explicit finish control waits for a finish signal even after reaching target.
+pub(crate) fn test_sync_waits_for_explicit_finish<H: SyncTestHarness>()
+where
+    Arc<DbOf<H>>: Resolver<Op = OpOf<H>, Digest = Digest>,
+    OpOf<H>: Encode,
+    JournalOf<H>: Contiguous,
+{
+    let executor = deterministic::Runner::default();
+    executor.start(|mut context| async move {
+        let mut target_db = H::init_db(context.with_label("target")).await;
+        target_db = H::apply_ops(target_db, H::create_ops(10)).await;
+        let initial_target = Target {
+            root: H::sync_target_root(&target_db),
+            range: non_empty_range!(
+                target_db.inactivity_floor_loc().await,
+                target_db.bounds().await.end
+            ),
+        };
+
+        target_db = H::apply_ops(target_db, H::create_ops_seeded(5, 1)).await;
+        let updated_lower_bound = target_db.inactivity_floor_loc().await;
+        let updated_upper_bound = target_db.bounds().await.end;
+        let updated_target = Target {
+            root: H::sync_target_root(&target_db),
+            range: non_empty_range!(updated_lower_bound, updated_upper_bound),
+        };
+        let updated_verification_root = target_db.root();
+
+        let (update_sender, update_receiver) = mpsc::channel(1);
+        let (finish_sender, finish_receiver) = mpsc::channel(1);
+        let (reached_sender, mut reached_receiver) = mpsc::channel(1);
+        let target_db = Arc::new(target_db);
+        let config = Config {
+            context: context.with_label("client"),
+            db_config: H::config(&context.next_u64().to_string(), &context),
+            fetch_batch_size: NZU64!(10),
+            target: initial_target.clone(),
+            resolver: target_db.clone(),
+            apply_batch_size: 1024,
+            max_outstanding_requests: 1,
+            update_rx: Some(update_receiver),
+            finish_rx: Some(finish_receiver),
+            reached_target_tx: Some(reached_sender),
+            max_retained_roots: 0,
+        };
+
+        let sync_handle = sync::sync(config);
+        pin_mut!(sync_handle);
+
+        select! {
+            _ = sync_handle.as_mut() => {
+                panic!("sync completed before explicit finish signal");
+            },
+            reached = reached_receiver.recv() => {
+                let reached = reached.expect("engine should report reached-target before finish");
+                assert_eq!(reached, initial_target);
+            }
+        }
+        assert!(
+            sync_handle.as_mut().now_or_never().is_none(),
+            "sync must wait for explicit finish signal after reaching target"
+        );
+
+        update_sender
+            .send(updated_target.clone())
+            .await
+            .expect("target update channel should be open");
+
+        select! {
+            _ = sync_handle.as_mut() => {
+                panic!("sync completed before explicit finish signal for updated target");
+            },
+            reached = reached_receiver.recv() => {
+                let reached = reached.expect("engine should report updated target before finish");
+                assert_eq!(reached, updated_target);
+            }
+        }
+        assert!(
+            sync_handle.as_mut().now_or_never().is_none(),
+            "sync must still wait for explicit finish signal after updated target is reached"
+        );
+
+        finish_sender
+            .send(())
+            .await
+            .expect("finish signal channel should be open");
+
+        let synced_db: H::Db = sync_handle
+            .await
+            .expect("sync should succeed after finish signal");
+        assert_eq!(synced_db.root(), updated_verification_root);
+        assert_eq!(synced_db.bounds().await.end, updated_upper_bound);
+        assert_eq!(synced_db.inactivity_floor_loc().await, updated_lower_bound);
+
+        synced_db.destroy().await.unwrap();
+        Arc::try_unwrap(target_db)
+            .unwrap_or_else(|_| panic!("failed to unwrap Arc"))
+            .destroy()
+            .await
+            .unwrap();
+    });
+}
+
 /// Test that the client can handle target updates during sync execution.
 pub(crate) fn test_target_update_during_sync<H: SyncTestHarness>(
     initial_ops: usize,
@@ -831,6 +956,8 @@ pub(crate) fn test_target_update_during_sync<H: SyncTestHarness>(
                 max_outstanding_requests: 10,
                 apply_batch_size: 1024,
                 update_rx: Some(update_receiver),
+                finish_rx: None,
+                reached_target_tx: None,
                 max_retained_roots: 1,
             };
             let mut client: Engine<H::Db, _> = Engine::new(config).await.unwrap();
@@ -938,6 +1065,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
         let synced_db: H::Db = sync::sync(config).await.unwrap();
@@ -1004,6 +1133,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
         let synced_db: H::Db = sync::sync(config).await.unwrap();
@@ -1357,6 +1488,8 @@ where
             apply_batch_size: 1024,
             max_outstanding_requests: 1,
             update_rx: None,
+            finish_rx: None,
+            reached_target_tx: None,
             max_retained_roots: 0,
         };
 
@@ -1678,6 +1811,11 @@ macro_rules! sync_tests_for_harness {
                 super::test_target_update_on_done_client::<$harness>();
             }
 
+            #[test_traced]
+            fn test_sync_waits_for_explicit_finish() {
+                super::test_sync_waits_for_explicit_finish::<$harness>();
+            }
+
             #[rstest]
             #[case(1, 1)]
             #[case(1, 2)]

storage/src/qmdb/current/sync/tests.rs

@@ -426,6 +426,11 @@ macro_rules! current_sync_tests_for_harness {
                 crate::qmdb::any::sync::tests::test_target_update_on_done_client::<$harness>();
             }
 
+            #[test_traced]
+            fn test_sync_waits_for_explicit_finish() {
+                crate::qmdb::any::sync::tests::test_sync_waits_for_explicit_finish::<$harness>();
+            }
+
             #[rstest]
             #[case(1, 1)]
             #[case(1, 2)]