fix(platform-wallet): stop concurrent shielded syncs from duplicating commitments

packages/rs-platform-wallet/src/wallet/shielded/sync.rs

@@ -53,6 +53,9 @@ use crate::error::PlatformWalletError;
 /// here and there together if the chunk power ever changes.
 const CHUNK_SIZE: u64 = 2048;
 
+/// In-flight chunk fetches per note-sync stream; tests mock this many chunks.
+const MAX_CONCURRENT_CHUNK_FETCHES: usize = 16;
+
 /// Result of one note-sync pass.
 #[derive(Debug, Clone, Default)]
 pub struct SyncNotesResult {
@@ -201,6 +204,11 @@ pub(super) async fn sync_notes_across<S: ShieldedStore>(
         return Ok(MultiSyncNotesResult::default());
     }
 
+    // Keep the snapshot and streamed consume under one write guard so another
+    // sync cannot append between them and leave this pass using stale gates.
+    // SDK fetches never access the store, so the stream can run under this lock.
+    let mut store = store.write().await;
+
     // Snapshot each subwallet's watermark and the shared tree's
     // current leaf count. Two distinct quantities drive the rest
     // of this pass:
@@ -218,7 +226,6 @@ pub(super) async fn sync_notes_across<S: ShieldedStore>(
     // it already stored, while a lagging one still saves from its
     // own start.
     let (watermarks, already_have, tree_size) = {
-        let store = store.read().await;
         let mut watermarks: BTreeMap<SubwalletId, u64> = BTreeMap::new();
         let mut min_idx: Option<u64> = None;
         for (id, _) in subwallets {
@@ -277,7 +284,7 @@ pub(super) async fn sync_notes_across<S: ShieldedStore>(
     // client-side lever; the pull-based stream still caps in-flight fetches
     // at this bound and keeps memory bounded.
     let mut sync_config = dash_sdk::platform::shielded::notes_sync::types::ShieldedSyncConfig {
-        max_concurrent: 16,
+        max_concurrent: MAX_CONCURRENT_CHUNK_FETCHES,
         ..Default::default()
     };
     if let Some(cb) = on_progress {
@@ -302,15 +309,6 @@ pub(super) async fn sync_notes_across<S: ShieldedStore>(
         }));
     }
 
-    // Acquire the store write lock for the whole interleaved consume.
-    // This is the only writer during a pass, so holding it across
-    // `stream.next().await` is safe: the stream's producer side is the
-    // SDK's network fetch loop, which never touches the store, so there
-    // is no lock-ordering deadlock. Backpressure is pull-based — a slow
-    // append simply polls the stream less often, capping in-flight
-    // network fetches at `max_concurrent`.
-    let mut store = store.write().await;
-
     let stream = sync_shielded_notes_stream(sdk, &driver_ivk, aligned_start, Some(sync_config));
     futures::pin_mut!(stream);
 
@@ -1074,3 +1072,6 @@ mod ovk_builder_roundtrip_tests;
 /// full-decryption and the OVK send-history recovery primitives.
 #[cfg(test)]
 mod memo_roundtrip_tests;
+
+#[cfg(test)]
+mod concurrent_sync_tests;

packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs

@@ -0,0 +1,140 @@
+use std::sync::Arc;
+
+use dash_sdk::Sdk;
+use dashcore::Network;
+use drive_proof_verifier::types::{
+    ShieldedEncryptedNote, ShieldedEncryptedNotes, ShieldedEncryptedNotesQuery,
+};
+use tokio::sync::RwLock;
+
+use super::{sync_notes_across, MAX_CONCURRENT_CHUNK_FETCHES};
+use crate::wallet::shielded::keys::{AccountViewingKeys, OrchardKeySet};
+use crate::wallet::shielded::store::{InMemoryShieldedStore, ShieldedStore, SubwalletId};
+
+async fn pool_sdk(note_count: u8) -> Arc<Sdk> {
+    let mut sdk = Sdk::new_mock();
+    let count = 2048
+        * u32::from(
+            sdk.version()
+                .drive_abci
+                .query
+                .shielded_queries
+                .max_query_chunks,
+        );
+    // Cover the stream's speculative window; only the first chunk has notes.
+    for chunk in 0..MAX_CONCURRENT_CHUNK_FETCHES as u64 {
+        let notes = if chunk == 0 {
+            (1..=note_count)
+                .map(|i| ShieldedEncryptedNote {
+                    cmx: vec![i; 32],
+                    nullifier: vec![i; 32],
+                    cv_net: vec![0; 32],
+                    encrypted_note: Vec::new(), // No owned receipts in this pool.
+                })
+                .collect()
+        } else {
+            Vec::new()
+        };
+        sdk.mock()
+            .expect_fetch(
+                ShieldedEncryptedNotesQuery {
+                    start_index: u64::from(count) * chunk,
+                    count,
+                },
+                Some(ShieldedEncryptedNotes {
+                    notes,
+                    total_count: u64::from(note_count),
+                }),
+            )
+            .await
+            .unwrap();
+    }
+    Arc::new(sdk)
+}
+
+fn seeded_store(initial_size: u8, caught_up: SubwalletId) -> Arc<RwLock<InMemoryShieldedStore>> {
+    let mut store = InMemoryShieldedStore::new();
+    for i in 1..=initial_size {
+        store.append_commitment(&[i; 32], true).unwrap();
+    }
+    store
+        .set_last_synced_note_index(caught_up, u64::from(initial_size))
+        .unwrap();
+    Arc::new(RwLock::new(store))
+}
+
+async fn assert_concurrent_sync_matches_sequential(initial_size: u8) {
+    let subwallets: Vec<_> = [1, 2]
+        .into_iter()
+        .map(|seed| {
+            let keys = OrchardKeySet::from_seed(&[seed; 32], Network::Testnet, 0).unwrap();
+            (
+                SubwalletId::new([seed; 32], 0),
+                AccountViewingKeys::from_full_viewing_key(keys.full_viewing_key),
+            )
+        })
+        .collect();
+    let sequential = seeded_store(initial_size, subwallets[0].0);
+    let concurrent = seeded_store(initial_size, subwallets[0].0);
+    let baseline_sdk = pool_sdk(4).await;
+    sync_notes_across(&baseline_sdk, &sequential, &subwallets, None, None)
+        .await
+        .unwrap();
+
+    let first_sdk = pool_sdk(4).await;
+    let second_sdk = pool_sdk(4).await;
+    let first = sync_notes_across(&first_sdk, &concurrent, &subwallets, None, None);
+    let second = sync_notes_across(&second_sdk, &concurrent, &subwallets, None, None);
+    futures::pin_mut!(first, second);
+
+    // Queue both at their first store lock, without sleeps or scheduler races.
+    // Were that lock a shared read, releasing the writer would let both passes
+    // snapshot before either appends; the write-locked snapshot serializes them.
+    let guard = concurrent.write().await;
+    assert!(futures::poll!(first.as_mut()).is_pending());
+    assert!(futures::poll!(second.as_mut()).is_pending());
+    drop(guard);
+    let (first, second) = tokio::join!(first, second);
+    first.expect("first concurrent sync");
+    second.expect("second concurrent sync");
+
+    {
+        let expected = sequential.read().await;
+        let actual = concurrent.read().await;
+        assert_eq!(expected.tree_size().unwrap(), 4);
+        assert_eq!(
+            actual.tree_size().unwrap(),
+            expected.tree_size().unwrap(),
+            "concurrent syncs must append each commitment position only once"
+        );
+        for (id, _) in &subwallets {
+            assert_eq!(expected.last_synced_note_index(*id).unwrap(), 4);
+            assert_eq!(
+                actual.last_synced_note_index(*id).unwrap(),
+                expected.last_synced_note_index(*id).unwrap()
+            );
+        }
+    }
+
+    // A later real position must remain appendable after the overlapping scans.
+    let next_sdk = pool_sdk(5).await;
+    sync_notes_across(&next_sdk, &concurrent, &subwallets, None, None)
+        .await
+        .unwrap();
+    let actual = concurrent.read().await;
+    assert_eq!(actual.tree_size().unwrap(), 5);
+    for (id, _) in &subwallets {
+        assert_eq!(actual.last_synced_note_index(*id).unwrap(), 5);
+    }
+}
+
+#[tokio::test]
+async fn should_append_commitments_once_across_concurrent_cold_syncs() {
+    assert_concurrent_sync_matches_sequential(0).await;
+}
+
+#[tokio::test]
+async fn should_append_commitments_once_across_concurrent_resumed_syncs() {
+    // One subwallet is caught up and the other still needs the existing leaves.
+    assert_concurrent_sync_matches_sequential(2).await;
+}