perf(platform-wallet): don't hold the shielded store lock across sync network I/O

packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativePersistenceBridge.kt

@@ -786,7 +786,14 @@ abstract class NativePersistenceBridge {
 
     // ── Shielded persist ──────────────────────────────────────────────
 
-    /** One `ShieldedNoteFFI`. Descriptor `([B[BIJ[B[BJBJ[B)I`. */
+    /**
+     * One `ShieldedNoteFFI`. Descriptor `([B[BIJ[B[BJBJ[B)I`.
+     *
+     * An upsert by nullifier must keep an existing row's spent flag, whatever
+     * `isSpent` says: a sync pass can deliver a note unspent after a send's
+     * round already saved it spent. Fail the round if the existing row can't
+     * be read (see `ShieldedNoteFFI`).
+     */
     @Suppress("LongParameterList")
     open fun onPersistShieldedNote(
         walletId: ByteArray,

packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt

@@ -2463,6 +2463,10 @@ class PlatformWalletPersistenceHandler(
         noteData: ByteArray,
     ): Int = guarded {
         stage(walletId) { db ->
+            // A re-save never clears a spent flag (see
+            // `ShieldedChangeSet::notes_saved`): a sync pass can deliver a
+            // note unspent after a concurrent send's round marked it spent.
+            val alreadySpent = db.shieldedDao().getNoteByNullifier(nullifier)?.isSpent == true
             db.shieldedDao().upsertNote(
                 ShieldedNoteEntity(
                     nullifier = nullifier,
@@ -2471,7 +2475,7 @@ class PlatformWalletPersistenceHandler(
                     position = position,
                     cmx = cmx,
                     blockHeight = blockHeight,
-                    isSpent = isSpent.toInt() != 0,
+                    isSpent = alreadySpent || isSpent.toInt() != 0,
                     value = value,
                     noteData = noteData,
                     lastUpdated = now(),

packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt

@@ -2040,6 +2040,89 @@ class PlatformWalletPersistenceHandlerTest {
         assertTrue(noteData.contentEquals(note.noteData))
     }
 
+    @Test
+    fun shieldedNoteResaveNeverClearsSpent() = runTest {
+        val nullifier = ByteArray(32) { 23 }
+        suspend fun saveNote(isSpent: Byte) {
+            handler.onChangesetBegin(walletId)
+            handler.onPersistShieldedNote(
+                walletId = walletId,
+                noteWalletId = walletId,
+                accountIndex = 0,
+                position = 8,
+                cmx = ByteArray(32) { 24 },
+                nullifier = nullifier,
+                blockHeight = 51,
+                isSpent = isSpent,
+                value = 200_000,
+                noteData = ByteArray(115) { 25 },
+            )
+            handler.onChangesetEnd(walletId, success = true)
+        }
+
+        saveNote(isSpent = 0)
+        // A send's round marks the note spent.
+        handler.onChangesetBegin(walletId)
+        handler.onPersistShieldedNullifierSpent(walletId, walletId, 0, nullifier)
+        handler.onChangesetEnd(walletId, success = true)
+        // A sync pass's later round re-saves it unspent.
+        saveNote(isSpent = 0)
+
+        assertTrue(db.shieldedDao().getNoteByNullifier(nullifier)!!.isSpent)
+    }
+
+    @Test
+    fun shieldedNoteReceiptAfterTheSpendRoundStaysSpent() = runTest {
+        // A send can confirm a note before the sync pass that found it has
+        // stored it: the spend's round inserts it spent, then the pass's
+        // receipt arrives unspent.
+        val nullifier = ByteArray(32) { 29 }
+        for (isSpent in listOf<Byte>(1, 0)) {
+            handler.onChangesetBegin(walletId)
+            handler.onPersistShieldedNote(
+                walletId = walletId,
+                noteWalletId = walletId,
+                accountIndex = 0,
+                position = 10,
+                cmx = ByteArray(32) { 30 },
+                nullifier = nullifier,
+                blockHeight = 53,
+                isSpent = isSpent,
+                value = 400_000,
+                noteData = ByteArray(115) { 31 },
+            )
+            handler.onChangesetEnd(walletId, success = true)
+        }
+
+        assertTrue(db.shieldedDao().getNoteByNullifier(nullifier)!!.isSpent)
+    }
+
+    @Test
+    fun shieldedNoteResaveStillMarksAnUnspentNoteSpent() = runTest {
+        val nullifier = ByteArray(32) { 26 }
+        suspend fun saveNote(isSpent: Byte) {
+            handler.onChangesetBegin(walletId)
+            handler.onPersistShieldedNote(
+                walletId = walletId,
+                noteWalletId = walletId,
+                accountIndex = 0,
+                position = 9,
+                cmx = ByteArray(32) { 27 },
+                nullifier = nullifier,
+                blockHeight = 52,
+                isSpent = isSpent,
+                value = 300_000,
+                noteData = ByteArray(115) { 28 },
+            )
+            handler.onChangesetEnd(walletId, success = true)
+        }
+
+        saveNote(isSpent = 0)
+        assertFalse(db.shieldedDao().getNoteByNullifier(nullifier)!!.isSpent)
+        saveNote(isSpent = 1)
+        assertTrue(db.shieldedDao().getNoteByNullifier(nullifier)!!.isSpent)
+    }
+
     @Test
     fun shieldedSyncStateAdvancesMonotonically() = runTest {
         handler.onChangesetBegin(walletId)

packages/rs-platform-wallet-ffi/src/persistence.rs

@@ -948,7 +948,9 @@ pub struct PersistenceCallbacks {
     // wallet; the outer `wallet_id` argument on the `store`
     // callback identifies the wallet the changeset originated from
     // (always identical to every entry's nested `wallet_id`).
-    /// Per-subwallet decrypted notes upserts.
+    /// Per-subwallet decrypted notes upserts. An upsert must keep an
+    /// existing row's spent flag (see
+    /// [`ShieldedNoteFFI`](crate::shielded_persistence::ShieldedNoteFFI)).
     #[cfg(feature = "shielded")]
     pub on_persist_shielded_notes_fn: Option<
         unsafe extern "C" fn(

packages/rs-platform-wallet-ffi/src/shielded_persistence.rs

@@ -20,6 +20,13 @@ use std::ffi::c_void;
 /// `(wallet_id, account_index, nullifier)` overwrite the existing
 /// row in place — Orchard nullifiers are globally unique, so a
 /// rescan after a restart shouldn't produce duplicates.
+///
+/// Except `is_spent`: once a row has it set, a re-save must keep it,
+/// whatever the incoming entry says. A sync pass can deliver a note
+/// unspent after a send's round already saved it spent, possibly
+/// before the host had any row for it. A host that can't read the
+/// existing row must fail the round rather than write the entry over
+/// it (`ShieldedChangeSet::notes_saved`).
 #[repr(C)]
 pub struct ShieldedNoteFFI {
     /// 32-byte wallet identifier.

packages/rs-platform-wallet-ffi/src/shielded_sync.rs

@@ -146,7 +146,8 @@ unsafe fn local_shielded_balance_snapshot_with_budget(
     });
     let (wallet, coordinator) = unwrap_option_or_return!(lookup);
     // Only the owned Arcs cross this wait. Holding the global registry's read
-    // guard while a scan owns the store lock can block an unrelated writer,
+    // guard while the store lock is held elsewhere (a sync pass's append
+    // slice or commit, a restore, Clear) can block an unrelated writer,
     // which then blocks the stop lookup behind it on the fair registry lock.
     let result = runtime().block_on(async {
         // A missing wallet is an API error, not a legitimate unbound state.

packages/rs-platform-wallet/src/changeset/shielded_changeset.rs

@@ -30,6 +30,15 @@ pub struct ShieldedChangeSet {
     /// subwallet. Keyed by `(wallet_id, account_index)`. Order
     /// inside the `Vec` is insertion order — the persister can
     /// upsert by `(SubwalletId, position)`.
+    ///
+    /// A saved row never clears a spent flag: the persister keeps
+    /// `is_spent` once a row has it, whatever a later row says. A sync
+    /// pass can deliver a note unspent after a concurrent send's round
+    /// already marked it, and the store itself never un-spends a note,
+    /// so overwriting the flag would leave the durable row behind the
+    /// store. A confirmed spend's round saves the note itself marked
+    /// spent, so this holds even when it reaches the host before the
+    /// pass's receipt does.
     pub notes_saved: BTreeMap<SubwalletId, Vec<ShieldedNote>>,
     /// Nullifiers freshly observed as spent on-chain, keyed by
     /// the subwallet that owns the corresponding note. The
@@ -108,6 +117,24 @@ impl ShieldedChangeSet {
         self.viewing_keys.insert(id, fvk_bytes.to_vec());
     }
 
+    /// Drop every delta for a subwallet `keep` rejects.
+    pub fn retain_subwallets(&mut self, keep: impl Fn(&SubwalletId) -> bool) {
+        let ShieldedChangeSet {
+            notes_saved,
+            nullifiers_spent,
+            outgoing_notes,
+            synced_indices,
+            activity_entries,
+            viewing_keys,
+        } = self;
+        notes_saved.retain(|id, _| keep(id));
+        nullifiers_spent.retain(|id, _| keep(id));
+        outgoing_notes.retain(|id, _| keep(id));
+        synced_indices.retain(|id, _| keep(id));
+        activity_entries.retain(|id, _| keep(id));
+        viewing_keys.retain(|id, _| keep(id));
+    }
+
     /// Split a consolidated shielded changeset into one
     /// `ShieldedChangeSet` per `WalletId`. Used by the
     /// network-scoped coordinator's sync path: the free

packages/rs-platform-wallet/src/wallet/platform_wallet.rs

@@ -1040,7 +1040,7 @@ impl PlatformWallet {
         // and watermark. A re-bind racing an in-flight sync pass
         // therefore cannot wipe the pass's results (an
         // unregister-then-register cycle here would purge the whole
-        // wallet behind the pass's store lock and then restore a
+        // wallet — superseding the pass — and then restore a
         // pre-pass snapshot — the "note discovered by sync is
         // unspendable until app restart" / "every pass rescans from
         // 0" failure). Registration also runs BEFORE the restore so

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

@@ -576,3 +576,184 @@ async fn should_reject_restored_local_balance_overflow() {
         u64::MAX
     );
 }
+
+// ── Sync passes superseded by a concurrent lifecycle change ─────────────
+//
+// A pass no longer holds the store lock across its download, so a
+// lifecycle change can land mid-pass. These drive `coordinator.sync` over
+// the mock SDK and inject the change from the tree-progress hook, which
+// fires inside the pass between store-lock holds.
+
+/// Install a tree-progress hook that runs `on_batch(call_index)` inside the
+/// pass (once per streamed batch), blocking on the lifecycle future.
+/// Nothing in the pass holds a coordinator or store lock while the hook
+/// runs, so the future completes without contention. It is driven on the
+/// runtime via `block_in_place` (the tests use the multi-thread flavor),
+/// `unconstrained` (the pass's task may have spent its coop budget) and
+/// under a timeout, so a regression that fires the hook while holding a
+/// lock the future needs fails the test instead of hanging it. Returns the
+/// call counter.
+fn hook_tree_progress<F, Fut>(
+    coordinator: &Arc<NetworkShieldedCoordinator>,
+    on_batch: F,
+) -> Arc<std::sync::atomic::AtomicUsize>
+where
+    F: Fn(Arc<NetworkShieldedCoordinator>, usize) -> Fut + Send + Sync + 'static,
+    Fut: std::future::Future<Output = ()> + Send,
+{
+    let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let weak = Arc::downgrade(coordinator);
+    let counter = Arc::clone(&calls);
+    coordinator.install_tree_progress_handler(Some(Arc::new(move |_, _| {
+        let call = counter.fetch_add(1, Ordering::SeqCst);
+        if let Some(coordinator) = weak.upgrade() {
+            let lifecycle_change = tokio::task::unconstrained(tokio::time::timeout(
+                std::time::Duration::from_secs(10),
+                on_batch(coordinator, call),
+            ));
+            tokio::task::block_in_place(|| {
+                tokio::runtime::Handle::current()
+                    .block_on(lifecycle_change)
+                    .expect("lifecycle change blocked inside the sync pass")
+            });
+        }
+    })));
+    calls
+}
+
+/// A host snapshot carrying sync state for `wallet`'s account 0.
+fn synced_start_state(wallet: [u8; 32]) -> ShieldedSyncStartState {
+    ShieldedSyncStartState {
+        per_subwallet: BTreeMap::from([(
+            SubwalletId::new(wallet, 0),
+            ShieldedSubwalletStartState {
+                has_sync_state: true,
+                ..Default::default()
+            },
+        )]),
+        ..Default::default()
+    }
+}
+
+/// Coordinator over the empty mock pool, plus how many batches (hook calls)
+/// one uninterrupted pass over it streams — the unit the tests below count
+/// scan attempts in.
+async fn empty_pool_coordinator() -> (Arc<NetworkShieldedCoordinator>, usize) {
+    let path = tree_path();
+    let coordinator = Arc::new(NetworkShieldedCoordinator::new(
+        Arc::new(empty_pool_sdk().await),
+        dashcore::Network::Testnet,
+        path.clone(),
+        FileBackedShieldedStore::open_path(&path, 100).unwrap(),
+    ));
+    register(&coordinator, [0x30; 32], &[0]).await;
+    let calls = hook_tree_progress(&coordinator, |_, _| async {});
+    assert_eq!(
+        coordinator
+            .sync(true, &CancellationToken::new())
+            .await
+            .success_count(),
+        1
+    );
+    let batches_per_pass = calls.load(Ordering::SeqCst);
+    assert!(batches_per_pass > 0);
+    coordinator.unregister_wallet([0x30; 32]).await;
+    (coordinator, batches_per_pass)
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn should_retry_sync_superseded_by_wallet_removal_without_resurrecting_it() {
+    let (coordinator, batches_per_pass) = empty_pool_coordinator().await;
+    let kept = [0x31; 32];
+    let removed = [0x32; 32];
+    register(&coordinator, kept, &[0]).await;
+    register(&coordinator, removed, &[0]).await;
+    let calls = hook_tree_progress(&coordinator, move |coordinator, call| async move {
+        if call == 0 {
+            coordinator.unregister_wallet(removed).await;
+        }
+    });
+
+    let outcome = coordinator.sync(true, &CancellationToken::new()).await;
+    assert_eq!(outcome.success_count(), 1, "{outcome:?}");
+    assert!(outcome.wallet_results.contains_key(&kept));
+    assert!(!outcome.wallet_results.contains_key(&removed));
+    assert_eq!(
+        calls.load(Ordering::SeqCst),
+        2 * batches_per_pass,
+        "the superseded scan was retried once"
+    );
+    let store = coordinator.store().read().await;
+    assert_eq!(
+        store
+            .local_account_balance(SubwalletId::new(kept, 0))
+            .unwrap()
+            .source,
+        ShieldedBalanceSource::ScannedThisSession
+    );
+    assert_eq!(
+        store
+            .local_account_balance(SubwalletId::new(removed, 0))
+            .unwrap()
+            .source,
+        ShieldedBalanceSource::NoHistory,
+        "the superseded pass must not write sync state for the removed wallet"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn should_fail_sync_after_repeated_supersession() {
+    let (coordinator, batches_per_pass) = empty_pool_coordinator().await;
+    let wallet = [0x33; 32];
+    register(&coordinator, wallet, &[0]).await;
+    // Every attempt is superseded by a restore of the wallet it scans.
+    let calls = hook_tree_progress(&coordinator, move |coordinator, _| async move {
+        let snapshot = synced_start_state(wallet);
+        coordinator
+            .restore_for_wallet(wallet, &snapshot)
+            .await
+            .unwrap();
+    });
+
+    let outcome = coordinator.sync(true, &CancellationToken::new()).await;
+    assert_eq!(outcome.success_count(), 0, "{outcome:?}");
+    assert!(
+        matches!(&outcome.wallet_results[&wallet],
+            crate::manager::shielded_sync::WalletShieldedOutcome::Err(message)
+                if message.contains("superseded")),
+        "{outcome:?}"
+    );
+    assert_eq!(
+        calls.load(Ordering::SeqCst),
+        3 * batches_per_pass,
+        "one attempt plus two retries"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn should_not_supersede_sync_for_a_wallet_bound_mid_pass() {
+    let (coordinator, batches_per_pass) = empty_pool_coordinator().await;
+    let scanning = [0x34; 32];
+    let late = [0x35; 32];
+    register(&coordinator, scanning, &[0]).await;
+    // A second wallet binds (register + restore) while the pass runs.
+    let calls = hook_tree_progress(&coordinator, move |coordinator, call| async move {
+        if call == 0 {
+            register(&coordinator, late, &[0]).await;
+            let snapshot = synced_start_state(late);
+            coordinator
+                .restore_for_wallet(late, &snapshot)
+                .await
+                .unwrap();
+        }
+    });
+
+    let outcome = coordinator.sync(true, &CancellationToken::new()).await;
+    assert_eq!(outcome.success_count(), 1, "{outcome:?}");
+    assert!(outcome.wallet_results.contains_key(&scanning));
+    assert_eq!(
+        calls.load(Ordering::SeqCst),
+        batches_per_pass,
+        "binding another wallet must not discard the in-flight scan"
+    );
+}