diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync.rs index 4f01ea876ef..cd0e19553d1 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/sync.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync.rs @@ -201,6 +201,11 @@ pub(super) async fn sync_notes_across( 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 +223,6 @@ pub(super) async fn sync_notes_across( // 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 = BTreeMap::new(); let mut min_idx: Option = None; for (id, _) in subwallets { @@ -302,15 +306,6 @@ pub(super) async fn sync_notes_across( })); } - // 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 +1069,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; diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs new file mode 100644 index 00000000000..f8e896fa403 --- /dev/null +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs @@ -0,0 +1,141 @@ +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; +use crate::wallet::shielded::keys::{AccountViewingKeys, OrchardKeySet}; +use crate::wallet::shielded::store::{InMemoryShieldedStore, ShieldedStore, SubwalletId}; + +async fn pool_sdk(note_count: u8) -> Arc { + 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..16 { + 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> { + 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. + // Releasing the writer grants both queued readers their permits: each can + // snapshot before either acquires the append lock. An atomic write-locked + // snapshot instead serializes these same two queued syncs. + 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; +}