diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativePersistenceBridge.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativePersistenceBridge.kt index 7d6ba3275c5..bdd3dc5a50f 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativePersistenceBridge.kt +++ b/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, diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt index 4a5d469053d..84bff6e7f15 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt +++ b/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(), diff --git a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt index 978b0c43e09..7e3819d0823 100644 --- a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt +++ b/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(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) diff --git a/packages/rs-platform-wallet-ffi/src/persistence.rs b/packages/rs-platform-wallet-ffi/src/persistence.rs index bbc3861e23a..c4ac6d71745 100644 --- a/packages/rs-platform-wallet-ffi/src/persistence.rs +++ b/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( diff --git a/packages/rs-platform-wallet-ffi/src/shielded_persistence.rs b/packages/rs-platform-wallet-ffi/src/shielded_persistence.rs index db1a704ab31..fa8c850ae77 100644 --- a/packages/rs-platform-wallet-ffi/src/shielded_persistence.rs +++ b/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. diff --git a/packages/rs-platform-wallet-ffi/src/shielded_sync.rs b/packages/rs-platform-wallet-ffi/src/shielded_sync.rs index 1371dfaabd1..6f50b9cbaee 100644 --- a/packages/rs-platform-wallet-ffi/src/shielded_sync.rs +++ b/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. diff --git a/packages/rs-platform-wallet/src/changeset/shielded_changeset.rs b/packages/rs-platform-wallet/src/changeset/shielded_changeset.rs index 3ac11cca4b1..78c6edf3f63 100644 --- a/packages/rs-platform-wallet/src/changeset/shielded_changeset.rs +++ b/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>, /// 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 diff --git a/packages/rs-platform-wallet/src/wallet/platform_wallet.rs b/packages/rs-platform-wallet/src/wallet/platform_wallet.rs index 50a564b04df..7704df6f5f5 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_wallet.rs +++ b/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 diff --git a/packages/rs-platform-wallet/src/wallet/shielded/balance_tests.rs b/packages/rs-platform-wallet/src/wallet/shielded/balance_tests.rs index 03adbe56175..4df93bfa2d1 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/balance_tests.rs +++ b/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( + coordinator: &Arc, + on_batch: F, +) -> Arc +where + F: Fn(Arc, usize) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + 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, 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" + ); +} diff --git a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs index 884f44d8dda..a4ab33aa9d8 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs @@ -78,7 +78,7 @@ pub type ShieldedProgressCallback = Arc; /// /// [`sync_notes_across`]: super::sync::sync_notes_across pub type ShieldedTreeProgressCallback = Arc; -use std::time::{Duration, Instant}; +use std::time::Instant; use tokio::sync::RwLock; @@ -93,6 +93,7 @@ use super::store::{ IdentityDebitRecoveryRecord, IdentityDebitRecoveryStatus, ShieldedStore, StalePendingSpend, SubwalletId, }; +use super::sync::{EpochWatch, NoteSyncOutcome, StoreEpoch}; use super::CAUGHT_UP_COOLDOWN; use crate::error::PlatformWalletError; use crate::manager::shielded_sync::{ShieldedSyncPassSummary, WalletShieldedOutcome}; @@ -267,8 +268,43 @@ pub struct NetworkShieldedCoordinator { /// counter before the load and re-reading it inside the transaction /// makes that snapshot detectably stale. clear_generation: std::sync::atomic::AtomicU64, + + /// Generations a sync pass re-checks before every store write; + /// lifecycle mutations that conflict with a pass's commit bump them + /// under their store write guard, and [`sync`](Self::sync) retries a + /// superseded pass from a fresh registry snapshot. See [`StoreEpoch`]. + store_epoch: StoreEpoch, + + /// Serializes [`sync`](Self::sync) passes on this coordinator. The + /// manager already runs one pass at a time, but a pass no longer + /// holds the store lock for its duration, so this is what keeps the + /// pass the shared tree's only appender (a second concurrent pass + /// would otherwise only be caught, and abandoned, by the leaf-count + /// check). [`clear`](Self::clear) holds it too, so no pass can start + /// between its reset and its registry clear and write into the emptied + /// store. Taken after `lifecycle` (by Clear) and before `fan_out` and + /// any store lock; nothing takes them the other way round. + sync_pass: tokio::sync::Mutex<()>, + + /// Makes a sync pass's changeset fan-out atomic with the lifecycle + /// operations that purge store state (unregister, a re-bind that drops + /// or re-keys accounts, Clear): either the purge, its persister change + /// and its [`StoreEpoch::bump_purged`] land before the fan-out checks + /// them, or after its last delivery. Not the `lifecycle` mutex itself, + /// so store-free lifecycle reads such as + /// [`local_balance_snapshot`](Self::local_balance_snapshot) never wait + /// out a host write. Lock order is `lifecycle` → `sync_pass` → + /// `fan_out` → `store`. + fan_out: tokio::sync::Mutex<()>, } +/// Times [`NetworkShieldedCoordinator::sync`] re-runs a note scan that a +/// concurrent lifecycle change superseded before giving up on the pass. +/// Each retry re-snapshots the registry, so it scans for exactly the +/// wallets bound by then; supersession needs a lifecycle mutation landing +/// mid-download, so more than one retry in a row is already unusual. +const MAX_SUPERSEDED_SYNC_RETRIES: usize = 2; + /// Exclusive handle on one wallet's shielded install transaction, /// returned by /// [`begin_install`](NetworkShieldedCoordinator::begin_install). @@ -405,6 +441,9 @@ impl NetworkShieldedCoordinator { lifecycle: tokio::sync::Mutex::new(()), hydrated: RwLock::new(std::collections::BTreeSet::new()), clear_generation: std::sync::atomic::AtomicU64::new(0), + store_epoch: StoreEpoch::default(), + sync_pass: tokio::sync::Mutex::new(()), + fan_out: tokio::sync::Mutex::new(()), } } @@ -733,6 +772,14 @@ impl NetworkShieldedCoordinator { } } + // A purge must not land between a pass's fan-out check and its + // delivery — see the `fan_out` field doc. + let _fan_out = if stale.is_empty() { + None + } else { + Some(self.fan_out.lock().await) + }; + // Persister FIRST, accounts second. `sync()` snapshots // `accounts` at pass start and looks the persister up only // when it queues the pass's changeset — with the reverse @@ -764,14 +811,18 @@ impl NetworkShieldedCoordinator { // Purge ONLY the dropped / re-keyed subwallets' store state. // Skipped entirely on the identical-registration path (empty - // `stale`), which keeps that path free of the store lock — it - // must never queue behind an in-flight sync pass (that - // blocking, followed by a purge, was the original - // wipe-the-pass bug). A changed-set re-bind may block here, - // but it only ever touches subwallets the caller explicitly + // `stale`), which keeps that path free of the store lock and of + // the sync-supersession bump below (an in-flight pass over this + // wallet keeps its work). A purge mid-pass followed by the pass's + // commit was the original wipe-the-pass bug; now the bump makes + // that pass abandon its commit and retry instead. A changed-set + // re-bind only ever touches subwallets the caller explicitly // dropped; retained accounts are left untouched. if !stale.is_empty() { let mut store = self.store.write().await; + self.store_epoch.bump_wallet(&mut store, wallet_id); + self.store_epoch + .bump_purged(&mut store, stale.iter().map(|(id, _)| *id)); for (id, _) in stale { if let Err(e) = store.purge_subwallet(id) { tracing::warn!( @@ -887,13 +938,23 @@ impl NetworkShieldedCoordinator { // restore could repopulate (and re-mark hydrated) the state // purged below after the purge ran. let _lifecycle = self.lifecycle.lock().await; - self.accounts - .write() - .await - .retain(|id, _| id.wallet_id != wallet_id); + let _fan_out = self.fan_out.lock().await; + let removed: Vec = { + let mut accounts = self.accounts.write().await; + let removed = accounts + .keys() + .filter(|id| id.wallet_id == wallet_id) + .copied() + .collect(); + accounts.retain(|id, _| id.wallet_id != wallet_id); + removed + }; self.persisters.write().await.remove(&wallet_id); self.hydrated.write().await.remove(&wallet_id); - if let Err(e) = self.store.write().await.purge_wallet(wallet_id) { + let mut store = self.store.write().await; + self.store_epoch.bump_wallet(&mut store, wallet_id); + self.store_epoch.bump_purged(&mut store, removed); + if let Err(e) = store.purge_wallet(wallet_id) { tracing::warn!( wallet_id = %hex::encode(wallet_id), error = %e, @@ -932,8 +993,9 @@ impl NetworkShieldedCoordinator { return Ok(ShieldedLocalBalanceState::Unbound); } let Ok(store) = self.store.try_read() else { - // A network scan can hold the store writer for minutes. Do not - // make store-free lifecycle operations wait behind this read. + // The store writer may be busy (a sync pass's append slice or + // final commit, a restore, Clear). Do not make store-free + // lifecycle operations wait behind this read. drop(install); // Wait only for readiness, then drop the temporary store guard // BEFORE reacquiring lifecycle. Keeping it would invert the @@ -1061,6 +1123,9 @@ impl NetworkShieldedCoordinator { continue; } } + // This restore writes state an in-flight pass over the wallet + // may have snapshotted (or be about to overwrite). + self.store_epoch.bump_wallet(&mut store, wallet_id); // Nullifiers the store already tracks for this subwallet. // The in-memory state is always at least as fresh as the // host snapshot (the snapshot's rows were produced from @@ -1123,9 +1188,9 @@ impl NetworkShieldedCoordinator { })?; } // Watermark restore is advance-only. A snapshot loaded - // before an in-flight sync pass (a mid-session re-bind - // queues behind the pass's store write lock) carries the - // PRE-pass watermark; applying it unconditionally rewinds + // before a sync pass committed (a mid-session re-bind that + // loaded the host snapshot while the pass was in flight) + // carries the PRE-pass watermark; applying it unconditionally rewinds // the store below what the pass just scanned and forces a // full rescan on every subsequent pass until restart. // The store's own value only ever comes from a completed @@ -1202,6 +1267,10 @@ impl NetworkShieldedCoordinator { // Taken before the store lock to keep the `lifecycle` → `store` // order. let _lifecycle = self.lifecycle.lock().await; + // See the `sync_pass` field doc. The manager's Clear already holds + // sync admission shut, so this does not wait out a pass there. + let _pass = self.sync_pass.lock().await; + let _fan_out = self.fan_out.lock().await; // Reset the persistent store FIRST and bail before mutating any // coordinator registries if it fails. Clearing `accounts` / `persisters` @@ -1215,6 +1284,10 @@ impl NetworkShieldedCoordinator { let mut first_err: Option = None; { let mut store = self.store.write().await; + // Bumped before either reset so a pass that resumes after this + // guard drops never appends into (or commits notes against) the + // emptied tree, whatever the reset outcome. + self.store_epoch.bump_all(&mut store); if let Err(e) = store.reset_commitment_tree() { tracing::warn!(error = %e, "Failed to reset commitment tree on clear"); first_err.get_or_insert_with(|| { @@ -1289,6 +1362,12 @@ impl NetworkShieldedCoordinator { // told the clear failed, keeps its own state. self.accounts.write().await.clear(); self.persisters.write().await.clear(); + // A pass that watched the epoch after the bump above but read the + // registry before it was cleared would cold-rebuild the purged + // store for wallets that are no longer bound (their changesets then + // dropped for want of a persister). Bump again now that they are + // gone. + self.store_epoch.bump_all(&mut self.store.write().await); if let Ok(mut g) = self.last_caught_up_at.lock() { *g = None; } @@ -1339,38 +1418,33 @@ impl NetworkShieldedCoordinator { tracing::debug!("Coordinator sync cancelled before it started"); return ShieldedSyncPassSummary::default(); } + if self.within_cooldown(force) { + return Self::cooldown_skip_summary(self).await; + } - // Network-wide cooldown gate. Snapshot the remaining - // window into a local before any await — `std::sync::Mutex` - // is `!Send` across await points (clippy's - // `await_holding_lock` lint flags this). - let cooldown_remaining: Option = if force { - None - } else { - self.last_caught_up_at - .lock() - .ok() - .and_then(|guard| *guard) - .map(|when| CAUGHT_UP_COOLDOWN.saturating_sub(when.elapsed())) - .filter(|remaining| !remaining.is_zero()) + // One pass at a time — see the `sync_pass` field doc. The wait + // observes the token: nothing has run yet, so a stop that lands + // while this pass is queued behind another (or behind Clear) just + // returns. Re-check the cooldown: the pass this one waited behind + // may have just caught up. + let Some(_pass) = cancel.run_until_cancelled(self.sync_pass.lock()).await else { + tracing::debug!("Coordinator sync cancelled while waiting for the pass in flight"); + return ShieldedSyncPassSummary::default(); }; - - if let Some(remaining) = cooldown_remaining { - tracing::debug!( - cooldown_remaining_secs = remaining.as_secs(), - cooldown_total_secs = CAUGHT_UP_COOLDOWN.as_secs(), - "Coordinator sync skipped — within caught-up cooldown" - ); + if cancel.is_cancelled() { + return ShieldedSyncPassSummary::default(); + } + if self.within_cooldown(force) { return Self::cooldown_skip_summary(self).await; } // Snapshot the flat subwallet registry. This Vec is both // the IVK fan-out for sync_notes_across and the - // identity-map for per-wallet summary demux below. - let subwallets: Vec<(SubwalletId, AccountViewingKeys)> = { - let accounts = self.accounts.read().await; - accounts.iter().map(|(id, v)| (*id, v.clone())).collect() - }; + // identity-map for per-wallet summary demux below. The store + // epoch is watched FIRST — see `StoreEpoch` for why the order + // matters. + let mut epoch = self.store_epoch.watch(); + let mut subwallets = self.subwallet_snapshot().await; let mut summary = ShieldedSyncPassSummary::default(); if subwallets.is_empty() { @@ -1409,18 +1483,62 @@ impl NetworkShieldedCoordinator { // Second, distinct signal: commitments committed to the local // tree as the interleaved consumer drains the SDK stream. let on_tree_progress = self.tree_progress_handler(); - let notes = match super::sync::sync_notes_across( - &self.sdk, - &self.store, - &subwallets, - on_progress.as_ref(), - on_tree_progress.as_ref(), - cancel, - ) - .await - { - Ok(r) => r, - Err(e) => return self.fail_all_wallets(&subwallets, &e), + let mut superseded_retries = 0; + let notes = loop { + match super::sync::sync_notes_across( + &self.sdk, + &self.store, + &epoch, + &subwallets, + on_progress.as_ref(), + on_tree_progress.as_ref(), + cancel, + ) + .await + { + Ok(NoteSyncOutcome::Completed(r)) => break r, + // A scan cut short publishes nothing, whether the stop + // landed before the store lock or between batches: hand + // back the empty summary so the manager treats it as "no + // pass ran" — `build_per_wallet_summary` would otherwise + // mint an all-zero `Ok` for every registered wallet, and + // the host would see a completion event for no work. A + // stop during the prefetch or a superseded retry lands + // here too: the scan checks the token before it starts. + Ok(NoteSyncOutcome::Cancelled) => return self.cancelled_pass_summary(), + // A stop that lands after a supersession: don't retry, and + // don't report the exhausted-retries failure either. + Ok(NoteSyncOutcome::Superseded) if cancel.is_cancelled() => { + return self.cancelled_pass_summary(); + } + // A lifecycle change (wallet removed, accounts re-keyed, + // snapshot restored, Clear) landed mid-download and + // invalidated the scan's snapshot before it committed + // anything but tree leaves. Re-snapshot the registry so the + // retry scans exactly the wallets bound now — never one the + // change just purged. The pre-scan stranded-release snapshot + // stays valid: the retry's scan still runs after it. + Ok(NoteSyncOutcome::Superseded) + if superseded_retries < MAX_SUPERSEDED_SYNC_RETRIES => + { + superseded_retries += 1; + epoch = self.store_epoch.watch(); + subwallets = self.subwallet_snapshot().await; + if subwallets.is_empty() { + summary.sync_unix_seconds = Self::now_unix(); + return summary; + } + } + Ok(NoteSyncOutcome::Superseded) => { + let e = PlatformWalletError::ShieldedSyncFailed( + "shielded sync was repeatedly superseded by concurrent wallet \ + lifecycle changes; it will run again on the next pass" + .to_string(), + ); + return self.fail_all_wallets(&subwallets, &e); + } + Err(e) => return self.fail_all_wallets(&subwallets, &e), + } }; // Everything below that talks to the network is skipped once the // pass is cancelled; the store-only steps and the host flush of a @@ -1430,23 +1548,6 @@ impl NetworkShieldedCoordinator { // about whether the scan reached the tip. A stop that lands during // the store-only steps after a complete scan does not change that. let cancelled = cancel.is_cancelled(); - // A scan cut short publishes nothing (`sync_notes_across` returns - // the empty result whether the stop landed in the prefetch, before - // the store lock or between batches): hand back the empty summary - // so the manager treats it as "no pass ran" — `build_per_wallet_summary` - // would otherwise mint an all-zero `Ok` for every registered - // wallet, and the host would see a completion event for no work. - if cancelled && notes.total_scanned == 0 && Merge::is_empty(¬es.changeset) { - // The scan may have appended new leaves before the stop, so the - // chain has moved past whatever the last completed pass saw: - // drop the caught-up stamp so the next pass walks Platform at - // once instead of taking the cooldown skip. - if let Ok(mut guard) = self.last_caught_up_at.lock() { - *guard = None; - } - tracing::debug!("Coordinator sync cancelled; the scan published nothing"); - return summary; - } // Scan-based spend detection now happens INSIDE // `sync_notes_across`: every scanned action's nullifier is // replayed against each subwallet's store as part of the note @@ -1559,35 +1660,8 @@ impl NetworkShieldedCoordinator { .await; } - // The note-side changeset already carries saves, synced - // indices, AND the scan-detected spends, so split it per - // WalletId directly — each per-wallet `WalletPersister.store` - // only sees its own wallet's deltas. - let consolidated = notes.changeset.clone(); - if !Merge::is_empty(&consolidated) { - let per_wallet = consolidated.split_by_wallet_id(); - let persisters = self.persisters.read().await; - for (wallet_id, cs) in per_wallet { - let Some(persister) = persisters.get(&wallet_id) else { - tracing::warn!( - wallet_id = %hex::encode(wallet_id), - "Shielded sync changeset dropped: no persister registered (host inconsistency)" - ); - continue; - }; - let full = crate::changeset::PlatformWalletChangeSet { - shielded: Some(cs), - ..Default::default() - }; - if let Err(e) = persister.store(full) { - tracing::warn!( - wallet_id = %hex::encode(wallet_id), - error = %e, - "Failed to queue shielded changeset from coordinator" - ); - } - } - } + self.queue_pass_changeset(&epoch, notes.changeset.clone()) + .await; // Cooldown decision based on aggregate activity across // every subwallet — any new commitment scanned or @@ -1620,6 +1694,168 @@ impl NetworkShieldedCoordinator { summary } + /// Queue a completed pass's changeset to each wallet's persister. The + /// note-side changeset already carries saves, synced indices AND the + /// scan-detected spends, so it is split per WalletId directly. + /// + /// The pass committed before releasing the store guard, so an unregister, + /// re-bind or Clear may since have purged part of that state and + /// installed a replacement persister. Drop every subwallet whose state + /// the pass no longer owns ([`EpochWatch::state_kept`]): the host must + /// not get back what the store forgot. Restores and store-free re-binds + /// purge nothing, so their wallets still get the pass's results. + /// + /// A send may also have spent one of the pass's notes since, and its own + /// round may already have reached the host. That round carries the note + /// itself marked spent, even when the host had no row for it yet, and + /// hosts keep a spent flag once set + /// ([`ShieldedChangeSet::notes_saved`](crate::changeset::ShieldedChangeSet::notes_saved)), + /// so this round's unspent copy can't revert it. As a second line of + /// defence for a persister that doesn't, every note delivered unspent + /// that the store has marked spent by the time delivery ends gets a + /// correction round; a send marking it later queues its own round after + /// this one. No store guard is held across the host write. + async fn queue_pass_changeset( + &self, + epoch: &EpochWatch<'_>, + mut changeset: crate::changeset::ShieldedChangeSet, + ) { + let _fan_out = self.fan_out.lock().await; + changeset.retain_subwallets(|id| epoch.state_kept(id)); + if Merge::is_empty(&changeset) { + return; + } + // A note this round already marks spent (the pass's own scan saw + // its spend) needs no correction. + let delivered_unspent: Vec<(SubwalletId, [u8; 32])> = changeset + .notes_saved + .iter() + .flat_map(|(id, notes)| { + let spent_here: HashSet<&[u8; 32]> = changeset + .nullifiers_spent + .get(id) + .into_iter() + .flatten() + .collect(); + notes + .iter() + .filter(move |note| !note.is_spent && !spent_here.contains(¬e.nullifier)) + .map(|note| (*id, note.nullifier)) + }) + .collect(); + let persisters = self.persisters.read().await; + Self::deliver(&persisters, changeset); + if delivered_unspent.is_empty() { + return; + } + let mut correction = crate::changeset::ShieldedChangeSet::default(); + { + let store = self.store.read().await; + for (id, nullifier) in delivered_unspent { + let spent = store.is_note_spent(id, &nullifier).unwrap_or_else(|e| { + tracing::warn!( + wallet_id = %hex::encode(id.wallet_id), + account = id.account_index, + error = %e, + "Could not re-check a synced note's spent flag after queueing it" + ); + false + }); + if !spent { + continue; + } + // Also save the note itself, the store's copy (the current + // one): the host may not have the row for `nullifiers_spent` + // to flip yet. + match store.get_note(id, &nullifier) { + Ok(Some(note)) => correction.record_note(id, note), + Ok(None) => {} + Err(e) => tracing::warn!( + wallet_id = %hex::encode(id.wallet_id), + account = id.account_index, + error = %e, + "Could not read a synced note to save it spent after queueing it" + ), + } + correction.record_nullifier_spent(id, nullifier); + } + } + if !correction.is_empty() { + Self::deliver(&persisters, correction); + } + } + + /// Split `changeset` per wallet and queue each slice on that wallet's + /// persister. + fn deliver( + persisters: &BTreeMap, + changeset: crate::changeset::ShieldedChangeSet, + ) { + for (wallet_id, cs) in changeset.split_by_wallet_id() { + let Some(persister) = persisters.get(&wallet_id) else { + tracing::warn!( + wallet_id = %hex::encode(wallet_id), + "Shielded sync changeset dropped: no persister registered (host inconsistency)" + ); + continue; + }; + let full = crate::changeset::PlatformWalletChangeSet { + shielded: Some(cs), + ..Default::default() + }; + if let Err(e) = persister.store(full) { + tracing::warn!( + wallet_id = %hex::encode(wallet_id), + error = %e, + "Failed to queue shielded changeset from coordinator" + ); + } + } + } + + /// The empty summary of a pass whose scan was cut short. The scan may + /// have appended new leaves before the stop, so the chain has moved past + /// whatever the last completed pass saw: drop the caught-up stamp so the + /// next pass walks Platform at once instead of taking the cooldown skip. + fn cancelled_pass_summary(&self) -> ShieldedSyncPassSummary { + if let Ok(mut guard) = self.last_caught_up_at.lock() { + *guard = None; + } + tracing::debug!("Coordinator sync cancelled; the scan published nothing"); + ShieldedSyncPassSummary::default() + } + + /// Network-wide cooldown gate: whether a non-forced pass should skip + /// because the last one caught up less than [`CAUGHT_UP_COOLDOWN`] ago. + /// Reads the stamp without holding its `std::sync::Mutex` past the + /// call (it is `!Send` across awaits). + fn within_cooldown(&self, force: bool) -> bool { + if force { + return false; + } + let remaining = self + .last_caught_up_at + .lock() + .ok() + .and_then(|guard| *guard) + .map(|when| CAUGHT_UP_COOLDOWN.saturating_sub(when.elapsed())) + .filter(|remaining| !remaining.is_zero()); + if let Some(remaining) = remaining { + tracing::debug!( + cooldown_remaining_secs = remaining.as_secs(), + cooldown_total_secs = CAUGHT_UP_COOLDOWN.as_secs(), + "Coordinator sync skipped — within caught-up cooldown" + ); + } + remaining.is_some() + } + + /// Snapshot of every registered subwallet and its viewing keys. + async fn subwallet_snapshot(&self) -> Vec<(SubwalletId, AccountViewingKeys)> { + let accounts = self.accounts.read().await; + accounts.iter().map(|(id, v)| (*id, v.clone())).collect() + } + /// Build a `ShieldedSyncPassSummary` where every registered /// wallet's outcome is the supplied error string. Used when /// the network-wide SDK note scan (sync_notes_across) errors @@ -2244,6 +2480,7 @@ mod tests { use super::*; use crate::wallet::persister::NoPlatformPersistence; use crate::wallet::shielded::keys::OrchardKeySet; + use std::time::Duration; use crate::changeset::ShieldedChangeSet; use crate::wallet::shielded::activity::compute_activity_id; @@ -3048,8 +3285,8 @@ mod tests { /// Regression guard for the mid-session re-bind stomp: a /// `restore_for_wallet` call whose snapshot predates a completed - /// sync pass (the exact state a re-bind queued behind the pass's - /// store write lock applies) must not rewind the watermark and + /// sync pass (the exact state a re-bind that loaded its snapshot + /// while the pass was in flight applies) must not rewind the watermark and /// must not overwrite notes the store already tracks. /// /// Observed on iOS (rc.2, testnet): pass discovers a note and @@ -3158,6 +3395,495 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// Every lifecycle mutation that would conflict with an in-flight sync + /// pass's commit invalidates that pass's snapshot (so it is superseded + /// and retried rather than writing over the change). Store-free + /// operations — above all the idempotent re-bind fast path — and changes + /// scoped to a wallet the pass isn't scanning leave it alone. + #[tokio::test] + async fn should_invalidate_in_flight_sync_snapshots_on_conflicting_lifecycle_changes() { + use crate::changeset::{ShieldedSubwalletStartState, ShieldedSyncStartState}; + + let dir = temp_dir("store_epoch"); + let coordinator = coordinator_with_one_wallet(&dir).await; + let wallet_id: WalletId = [0x11; 32]; + let other_wallet: WalletId = [0x22; 32]; + let views = |account| { + OrchardKeySet::from_seed(&[0x42u8; 64], dashcore::Network::Testnet, account) + .expect("derive viewing keys") + .viewing_keys() + }; + let coordinator = &coordinator; + let register = |wallet: WalletId, accounts: &[u32]| { + let account_views: BTreeMap = + accounts.iter().map(|a| (*a, views(*a))).collect(); + coordinator.register_wallet( + wallet, + account_views, + WalletPersister::new(wallet, Arc::new(NoPlatformPersistence)), + ) + }; + let restore = |wallet: WalletId, account: u32| async move { + let snapshot = ShieldedSyncStartState { + per_subwallet: BTreeMap::from([( + SubwalletId::new(wallet, account), + ShieldedSubwalletStartState { + has_sync_state: true, + last_synced_index: 4096, + ..Default::default() + }, + )]), + ..Default::default() + }; + coordinator.restore_for_wallet(wallet, &snapshot).await + }; + // What an in-flight pass over `wallet_id`'s account 0 depends on. + let pass = vec![(SubwalletId::new(wallet_id, 0), views(0))]; + + let watch = coordinator.store_epoch.watch(); + register(wallet_id, &[0]).await.unwrap(); + assert!( + watch.still_current(&pass), + "identical re-bind stays store-free" + ); + register(wallet_id, &[0, 1]).await.unwrap(); + assert!( + watch.still_current(&pass), + "adding an account purges nothing" + ); + restore(wallet_id, 7).await.unwrap(); + assert!( + watch.still_current(&pass), + "a restore that writes nothing (unregistered account) changes nothing" + ); + register(other_wallet, &[0]).await.unwrap(); + restore(other_wallet, 0).await.unwrap(); + coordinator.unregister_wallet(other_wallet).await; + assert!( + watch.still_current(&pass), + "another wallet's restore / removal doesn't touch this pass's state" + ); + + register(wallet_id, &[0]).await.unwrap(); + assert!( + !watch.still_current(&pass), + "dropping an account purges its state" + ); + + let watch = coordinator.store_epoch.watch(); + restore(wallet_id, 0).await.unwrap(); + assert!(!watch.still_current(&pass), "a restore writes watermarks"); + + let watch = coordinator.store_epoch.watch(); + coordinator.unregister_wallet(wallet_id).await; + assert!(!watch.still_current(&pass)); + + let watch = coordinator.store_epoch.watch(); + coordinator.clear().await.unwrap(); + assert!( + !watch.still_current(&[]), + "Clear resets the shared tree under every pass" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// Records the shielded part of every changeset queued to it, running + /// `on_store` (if set) as each one arrives. + #[derive(Default)] + struct RecordingPersistence( + std::sync::Mutex>, + std::sync::Mutex>>, + ); + + impl crate::changeset::PlatformWalletPersistence for RecordingPersistence { + fn store( + &self, + _wallet_id: WalletId, + changeset: crate::changeset::PlatformWalletChangeSet, + ) -> Result<(), crate::changeset::PersistenceError> { + if let Some(on_store) = self.1.lock().unwrap().as_mut() { + on_store(); + } + self.0 + .lock() + .unwrap() + .push(changeset.shielded.unwrap_or_default()); + Ok(()) + } + + fn flush(&self, _wallet_id: WalletId) -> Result<(), crate::changeset::PersistenceError> { + Ok(()) + } + + fn load( + &self, + ) -> Result + { + Ok(Default::default()) + } + } + + /// A pass's changeset reaches the host only for subwallets whose store + /// state no lifecycle change purged after the pass committed it: a re-bind + /// dropping an account, an unregister (even one followed by a same-key + /// re-bind) or a Clear must not see that state come back through a + /// replacement persister. A restore purges nothing, so it keeps the slice. + #[tokio::test] + async fn should_not_queue_a_completed_pass_to_state_purged_after_its_commit() { + use crate::changeset::{ShieldedSubwalletStartState, ShieldedSyncStartState}; + + let dir = temp_dir("fan_out_after_lifecycle"); + let coordinator = coordinator_with_one_wallet(&dir).await; + let wallet_id: WalletId = [0x11; 32]; + let views = |account| { + OrchardKeySet::from_seed(&[0x42u8; 64], dashcore::Network::Testnet, account) + .expect("derive viewing keys") + .viewing_keys() + }; + let account = |index| SubwalletId::new(wallet_id, index); + let rebind = |accounts: &'static [u32]| { + let coordinator = &coordinator; + async move { + let recorder = Arc::new(RecordingPersistence::default()); + let account_views: BTreeMap = + accounts.iter().map(|a| (*a, views(*a))).collect(); + coordinator + .register_wallet( + wallet_id, + account_views, + WalletPersister::new(wallet_id, recorder.clone()), + ) + .await + .unwrap(); + recorder + } + }; + // A pass's changeset covers the accounts it scanned. + let queue_pass = |epoch, scanned: &'static [u32]| { + let coordinator = &coordinator; + async move { + let mut cs = ShieldedChangeSet::default(); + for index in scanned { + cs.record_synced_index(account(*index), 4096); + } + coordinator.queue_pass_changeset(&epoch, cs).await; + } + }; + let delivered = |recorder: &RecordingPersistence| -> Vec> { + let rounds = recorder.0.lock().unwrap(); + rounds.iter().map(|cs| cs.synced_indices.clone()).collect() + }; + + let old = rebind(&[0, 1]).await; + let epoch = coordinator.store_epoch.watch(); + let replacement = rebind(&[0]).await; + queue_pass(epoch, &[0, 1]).await; + assert!(delivered(&old).is_empty()); + assert_eq!( + delivered(&replacement), + vec![BTreeMap::from([(account(0), 4096)])], + "the dropped account's watermark must not reach the replacement persister" + ); + + let epoch = coordinator.store_epoch.watch(); + let snapshot = ShieldedSyncStartState { + per_subwallet: BTreeMap::from([( + account(0), + ShieldedSubwalletStartState { + has_sync_state: true, + last_synced_index: 1024, + ..Default::default() + }, + )]), + ..Default::default() + }; + coordinator + .restore_for_wallet(wallet_id, &snapshot) + .await + .unwrap(); + queue_pass(epoch, &[0]).await; + assert_eq!( + delivered(&replacement).last(), + Some(&BTreeMap::from([(account(0), 4096)])), + "a restore keeps the pass's results, so the host still needs them" + ); + + let epoch = coordinator.store_epoch.watch(); + coordinator.unregister_wallet(wallet_id).await; + let same_keys = rebind(&[0]).await; + queue_pass(epoch, &[0]).await; + assert!( + delivered(&same_keys).is_empty(), + "an unregister purged the state even though the same keys came back" + ); + + let epoch = coordinator.store_epoch.watch(); + coordinator.clear().await.unwrap(); + let after_clear = rebind(&[0]).await; + queue_pass(epoch, &[0]).await; + assert!( + delivered(&after_clear).is_empty(), + "nothing from before a Clear reaches the re-bound wallet" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A send that spends one of a pass's notes after the pass's commit must + /// not have that spend reverted on the host by the pass's stale unspent + /// copy, whether it finalizes before the fan-out or during the host write. + #[tokio::test] + async fn should_not_let_a_pass_revert_a_spend_finalized_after_its_commit() { + let dir = temp_dir("fan_out_after_spend"); + let coordinator = coordinator_with_one_wallet(&dir).await; + let wallet_id: WalletId = [0x11; 32]; + let id = SubwalletId::new(wallet_id, 0); + let recorder = Arc::new(RecordingPersistence::default()); + let views = OrchardKeySet::from_seed(&[0x42u8; 64], dashcore::Network::Testnet, 0) + .expect("derive viewing keys") + .viewing_keys(); + coordinator + .register_wallet( + wallet_id, + BTreeMap::from([(0, views)]), + WalletPersister::new(wallet_id, recorder.clone()), + ) + .await + .unwrap(); + // The pass commits both notes unspent. + let (before, during) = ( + test_note([0xA1; 32], 7, false), + test_note([0xA2; 32], 8, false), + ); + let epoch = coordinator.store_epoch.watch(); + let mut pass = ShieldedChangeSet::default(); + for note in [&before, &during] { + coordinator + .store() + .write() + .await + .save_note(id, note) + .unwrap(); + pass.record_note(id, note.clone()); + } + + // One send finalizes before the fan-out, the other during its host + // write (the store is free then: no guard spans host I/O). + coordinator + .store() + .write() + .await + .mark_spent(id, &before.nullifier) + .unwrap(); + let store = Arc::clone(coordinator.store()); + let nullifier = during.nullifier; + *recorder.1.lock().unwrap() = Some(Box::new(move || { + if let Ok(mut store) = store.try_write() { + store.mark_spent(id, &nullifier).unwrap(); + } + })); + coordinator.queue_pass_changeset(&epoch, pass).await; + + let rounds = recorder.0.lock().unwrap(); + let spent = |nullifier: &[u8; 32]| { + rounds.iter().rev().find_map(|round| { + round + .nullifiers_spent + .get(&id) + .is_some_and(|spent| spent.contains(nullifier)) + .then_some(true) + .or_else(|| { + round.notes_saved.get(&id)?.iter().find_map(|note| { + (note.nullifier == *nullifier).then_some(note.is_spent) + }) + }) + }) + }; + assert_eq!( + spent(&before.nullifier), + Some(true), + "spent before the fan-out" + ); + assert_eq!( + spent(&during.nullifier), + Some(true), + "spent during the host write: a correction round must follow" + ); + drop(rounds); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A send can confirm a note a pass found before the pass's receipt + /// round reaches the host, which then has no row for `nullifiers_spent` + /// to flip. The spend's round carries the note itself, so a host that + /// keeps a spent flag ([`ShieldedChangeSet::notes_saved`]) ends with it + /// spent even when the pass's correction round never lands. + #[tokio::test] + async fn should_keep_a_note_spent_when_its_spend_reaches_the_host_before_its_receipt() { + let dir = temp_dir("spend_before_receipt"); + let coordinator = coordinator_with_one_wallet(&dir).await; + let wallet_id: WalletId = [0x11; 32]; + let id = SubwalletId::new(wallet_id, 0); + let recorder = Arc::new(RecordingPersistence::default()); + let views = OrchardKeySet::from_seed(&[0x42u8; 64], dashcore::Network::Testnet, 0) + .expect("derive viewing keys") + .viewing_keys(); + coordinator + .register_wallet( + wallet_id, + BTreeMap::from([(0, views)]), + WalletPersister::new(wallet_id, recorder.clone()), + ) + .await + .unwrap(); + + // The pass commits a new note; its receipt round is still pending. + let note = test_note([0xA3; 32], 9, false); + let epoch = coordinator.store_epoch.watch(); + coordinator + .store() + .write() + .await + .save_note(id, ¬e) + .unwrap(); + let mut pass = ShieldedChangeSet::default(); + pass.record_note(id, note.clone()); + + // A send confirms it first, then the pass delivers its receipt. + super::super::operations::mark_notes_spent( + coordinator.store(), + Some(&WalletPersister::new(wallet_id, recorder.clone())), + wallet_id, + id, + std::slice::from_ref(¬e), + ) + .await + .unwrap(); + coordinator.queue_pass_changeset(&epoch, pass).await; + + // Replay the spend and receipt rounds onto a host that keeps spent + // flags, dropping the correction round as if it had failed. + let rounds = recorder.0.lock().unwrap(); + assert!(rounds.len() >= 2, "a spend round, then the receipt round"); + let mut host: BTreeMap<[u8; 32], bool> = BTreeMap::new(); + for round in &rounds[..2] { + for saved in round.notes_saved.get(&id).into_iter().flatten() { + *host.entry(saved.nullifier).or_default() |= saved.is_spent; + } + for nullifier in round.nullifiers_spent.get(&id).into_iter().flatten() { + if let Some(spent) = host.get_mut(nullifier) { + *spent = true; + } + } + } + assert_eq!( + host.get(¬e.nullifier), + Some(&true), + "the receipt landing after the spend must not leave the note selectable" + ); + drop(rounds); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A spend's round saves the store's copy of the note, not the one taken + /// when it was selected: a sync pass that re-saved the note with newer + /// metadata in between must not have it rolled back on the host. + #[tokio::test] + async fn should_save_the_current_note_not_the_selection_copy_when_a_spend_confirms() { + let dir = temp_dir("spend_saves_current_note"); + let coordinator = coordinator_with_one_wallet(&dir).await; + let wallet_id: WalletId = [0x11; 32]; + let id = SubwalletId::new(wallet_id, 0); + let recorder = Arc::new(RecordingPersistence::default()); + + let selected = test_note([0xA4; 32], 10, false); + let rescanned = crate::wallet::shielded::ShieldedNote { + block_height: selected.block_height + 50, + ..selected.clone() + }; + { + let mut store = coordinator.store().write().await; + store.save_note(id, &selected).unwrap(); + // A sync pass re-saves it between selection and confirmation. + store.save_note(id, &rescanned).unwrap(); + } + super::super::operations::mark_notes_spent( + coordinator.store(), + Some(&WalletPersister::new(wallet_id, recorder.clone())), + wallet_id, + id, + std::slice::from_ref(&selected), + ) + .await + .unwrap(); + + let rounds = recorder.0.lock().unwrap(); + let saved = rounds[0] + .notes_saved + .get(&id) + .expect("the spend saves the note"); + assert_eq!(saved.len(), 1); + assert!(saved[0].is_spent); + assert_eq!( + saved[0].block_height, rescanned.block_height, + "the spend must not roll back the re-saved metadata" + ); + drop(rounds); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A spend's round reaches the host before any sync pass can commit a + /// newer copy of the note: it is queued while the guard that read the + /// copy is still held, so a pass's later round can't overtake it with + /// metadata the spend then rolls back. + #[tokio::test] + async fn should_queue_a_spend_round_before_a_pass_can_commit_a_newer_copy() { + let dir = temp_dir("spend_round_under_guard"); + let coordinator = coordinator_with_one_wallet(&dir).await; + let wallet_id: WalletId = [0x11; 32]; + let id = SubwalletId::new(wallet_id, 0); + let recorder = Arc::new(RecordingPersistence::default()); + + let note = test_note([0xA5; 32], 11, false); + coordinator + .store() + .write() + .await + .save_note(id, ¬e) + .unwrap(); + // A pass trying to commit a re-scanned copy while the host write is + // in progress. + let store = Arc::clone(coordinator.store()); + let pass_committed_during_delivery = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let committed = Arc::clone(&pass_committed_during_delivery); + *recorder.1.lock().unwrap() = Some(Box::new(move || { + if store.try_write().is_ok() { + committed.store(true, std::sync::atomic::Ordering::SeqCst); + } + })); + super::super::operations::mark_notes_spent( + coordinator.store(), + Some(&WalletPersister::new(wallet_id, recorder.clone())), + wallet_id, + id, + std::slice::from_ref(¬e), + ) + .await + .unwrap(); + + assert!( + !pass_committed_during_delivery.load(std::sync::atomic::Ordering::SeqCst), + "the store guard must be held until the spend round is queued" + ); + assert_eq!(recorder.0.lock().unwrap().len(), 1); + + let _ = std::fs::remove_dir_all(&dir); + } + /// A changed-set re-register purges ONLY the dropped subwallet's /// store state: accounts that remain bound with the same viewing /// key keep their in-memory notes and watermark across the @@ -3683,6 +4409,63 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// Clear waits out an in-flight sync pass, and no pass starts while it + /// runs: one starting between Clear's store reset and its registry clear + /// would snapshot the still-bound accounts and write into the emptied + /// store, where Clear's later epoch bump can no longer undo it. + /// + /// Paused clock, same reasoning as the unregister test above. + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn clear_waits_for_an_in_flight_sync_pass() { + let dir = temp_dir("clear_vs_sync"); + let coordinator = Arc::new(coordinator_with_one_wallet(&dir).await); + + let pass = coordinator.sync_pass.lock().await; + let clearer = { + let coordinator = Arc::clone(&coordinator); + tokio::spawn(async move { coordinator.clear().await }) + }; + + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + !coordinator.registered_subwallets().await.is_empty(), + "clear must wait for the in-flight sync pass" + ); + + drop(pass); + clearer.await.expect("clear task").expect("clear succeeds"); + assert!(coordinator.registered_subwallets().await.is_empty()); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A non-forced pass that waited behind another honors the cooldown + /// that pass just started instead of rescanning. + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn should_recheck_the_cooldown_after_waiting_for_the_pass_in_flight() { + let dir = temp_dir("cooldown_after_pass"); + let coordinator = Arc::new(coordinator_with_one_wallet(&dir).await); + + let pass = coordinator.sync_pass.lock().await; + let waiter = { + let coordinator = Arc::clone(&coordinator); + tokio::spawn(async move { coordinator.sync(false, &CancellationToken::new()).await }) + }; + tokio::time::sleep(Duration::from_millis(200)).await; + // The pass in flight caught up. + *coordinator.last_caught_up_at.lock().unwrap() = Some(Instant::now()); + drop(pass); + + let summary = waiter.await.expect("sync task"); + assert!(summary.wallet_results.values().all(|outcome| matches!( + outcome, + WalletShieldedOutcome::Ok(s) if s.is_cooldown_skip + ))); + assert!(!summary.wallet_results.is_empty()); + + let _ = std::fs::remove_dir_all(&dir); + } + /// `wallet_registration_matches` — the idempotent-re-bind gate: /// true only for the exact same wallet + account → FVK map; false /// for an unknown wallet, a different account set, or a different diff --git a/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs b/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs index 0970e75d48a..7d7b30a79cd 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs @@ -542,6 +542,25 @@ impl ShieldedStore for FileBackedShieldedStore { .unwrap_or_default()) } + fn is_note_spent(&self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { + Ok(self + .subwallets + .get(&id) + .is_some_and(|state| state.is_spent(nullifier))) + } + + fn get_note( + &self, + id: SubwalletId, + nullifier: &[u8; 32], + ) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .and_then(|state| state.note(nullifier)) + .cloned()) + } + fn mark_spent(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { let Some(sw) = self.subwallets.get_mut(&id) else { return Ok(false); @@ -1598,6 +1617,31 @@ mod tests { /// session) — the durable SQLite row must be deleted too, not just /// the in-memory record, or it resurrects the reservation on the /// next open. + /// A re-save of a stale unspent copy must not offer a spent note again. + #[test] + fn should_keep_a_note_spent_when_a_stale_copy_is_re_saved() { + let path = temp_tree_path("stale_resave_keeps_spent"); + let id = SubwalletId::new([0x9; 32], 0); + let nf = [0x3B; 32]; + let stale = ShieldedNote { + position: 0, + cmx: [0x1; 32], + nullifier: nf, + block_height: 10, + is_spent: false, + value: 500, + note_data: vec![0u8; 115], + }; + let mut store = FileBackedShieldedStore::open_path(&path, 100).expect("open"); + store.save_note(id, &stale).expect("save"); + assert!(store.mark_spent(id, &nf).expect("mark spent")); + store.save_note(id, &stale).expect("re-save"); + assert!(store.is_note_spent(id, &nf).expect("read")); + assert!(store.get_unspent_notes(id).expect("read").is_empty()); + drop(store); + let _ = std::fs::remove_file(&path); + } + #[test] fn mark_spent_on_restored_spent_note_clears_durable_redrive() { let path = temp_tree_path("mark_spent_idempotent"); diff --git a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs index 8a8f8a4b39b..e0e9aa17306 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs @@ -2747,7 +2747,18 @@ fn select_recorded_spends( /// Also drops any matching pending reservation so the /// confirmed-spent state and the in-flight-spend state can't /// disagree. -async fn mark_notes_spent( +/// +/// The changeset carries each note itself, marked spent, not only its +/// nullifier. A note a sync pass just found can be spent before the +/// pass's receipt round reaches the host; the host then has no row for +/// `nullifiers_spent` to flip, and the receipt would land unspent. A +/// spent row saved here is kept by the host whatever arrives later +/// ([`ShieldedChangeSet::notes_saved`]). The saved copy is the store's, +/// read under the same guard: a sync pass may have re-saved the note +/// with newer metadata since `notes` were selected. The round is queued +/// before that guard drops, so no pass can commit a newer copy in +/// between and have its round reach the host ahead of this one. +pub(super) async fn mark_notes_spent( store: &Arc>, persister: Option<&WalletPersister>, wallet_id: WalletId, @@ -2762,11 +2773,17 @@ async fn mark_notes_spent( .mark_spent(id, ¬e.nullifier) .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))? { + if let Some(stored) = store + .get_note(id, ¬e.nullifier) + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))? + { + changeset.record_note(id, stored); + } changeset.record_nullifier_spent(id, note.nullifier); } } + queue_shielded_changeset(persister, wallet_id, changeset); } - queue_shielded_changeset(persister, wallet_id, changeset); Ok(()) } diff --git a/packages/rs-platform-wallet/src/wallet/shielded/store.rs b/packages/rs-platform-wallet/src/wallet/shielded/store.rs index 8efd0acfdcb..81312132eb6 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/store.rs @@ -239,6 +239,30 @@ pub trait ShieldedStore: Send + Sync { /// Return all notes (spent and unspent) for `id`. fn get_all_notes(&self, id: SubwalletId) -> Result, Self::Error>; + /// Whether `id` holds a note with `nullifier` that is already marked + /// spent. The default scans [`Self::get_all_notes`]; stores with a + /// nullifier index should answer from it. + fn is_note_spent(&self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { + Ok(self + .get_all_notes(id)? + .iter() + .any(|note| note.nullifier == *nullifier && note.is_spent)) + } + + /// `id`'s note with `nullifier`, if the store holds one. The default + /// scans [`Self::get_all_notes`]; stores with a nullifier index + /// should answer from it. + fn get_note( + &self, + id: SubwalletId, + nullifier: &[u8; 32], + ) -> Result, Self::Error> { + Ok(self + .get_all_notes(id)? + .into_iter() + .find(|note| note.nullifier == *nullifier)) + } + /// Mark `id`'s note with `nullifier` as spent. Returns `true` /// if a matching unspent note was found. fn mark_spent(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result; @@ -411,6 +435,12 @@ pub trait ShieldedStore: Send + Sync { fn append_commitment(&mut self, cmx: &[u8; 32], marked: bool) -> Result<(), Self::Error>; /// Create a tree checkpoint at the given identifier. + /// + /// Identifiers must be strictly increasing: an id that is not above + /// the newest checkpoint's is a successful no-op (shardtree's + /// semantics). The sync path relies on this — it checkpoints at the + /// tree size on every committing pass, which is a no-op whenever the + /// tree hasn't grown since the last checkpoint. fn checkpoint_tree(&mut self, checkpoint_id: u32) -> Result<(), Self::Error>; /// Return the current tree root (Sinsemilla anchor, 32 bytes). @@ -602,10 +632,15 @@ impl SubwalletState { /// Re-saving a note with a known nullifier overwrites the /// existing entry instead of appending a duplicate — Orchard /// nullifiers are globally unique, so a re-scan of the same - /// chunk shouldn't double-count. + /// chunk shouldn't double-count. A re-save never clears + /// `is_spent`: a spend is final, and a stale copy saved after it + /// must not offer the note again. pub(super) fn save_note(&mut self, note: &ShieldedNote) { if let Some(&existing_idx) = self.nullifier_index.get(¬e.nullifier) { - self.notes[existing_idx] = note.clone(); + let existing = &mut self.notes[existing_idx]; + let was_spent = existing.is_spent; + *existing = note.clone(); + existing.is_spent |= was_spent; return; } let idx = self.notes.len(); @@ -635,6 +670,18 @@ impl SubwalletState { self.notes.clone() } + pub(super) fn note(&self, nullifier: &[u8; 32]) -> Option<&ShieldedNote> { + self.nullifier_index + .get(nullifier) + .map(|&idx| &self.notes[idx]) + } + + pub(super) fn is_spent(&self, nullifier: &[u8; 32]) -> bool { + self.nullifier_index + .get(nullifier) + .is_some_and(|&idx| self.notes[idx].is_spent) + } + pub(super) fn mark_spent(&mut self, nullifier: &[u8; 32]) -> MarkSpentOutcome { let Some(&idx) = self.nullifier_index.get(nullifier) else { return MarkSpentOutcome::default(); @@ -896,6 +943,25 @@ impl ShieldedStore for InMemoryShieldedStore { .unwrap_or_default()) } + fn is_note_spent(&self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { + Ok(self + .subwallets + .get(&id) + .is_some_and(|state| state.is_spent(nullifier))) + } + + fn get_note( + &self, + id: SubwalletId, + nullifier: &[u8; 32], + ) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .and_then(|state| state.note(nullifier)) + .cloned()) + } + fn mark_spent(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { // In-memory store: the redrive map lives in the same // `SubwalletState`, so `mark_spent` already dropped any resolved @@ -1072,7 +1138,9 @@ impl ShieldedStore for InMemoryShieldedStore { } fn checkpoint_tree(&mut self, checkpoint_id: u32) -> Result<(), Self::Error> { - self.checkpoints.push(checkpoint_id); + if self.checkpoints.last() < Some(&checkpoint_id) { + self.checkpoints.push(checkpoint_id); + } Ok(()) } @@ -1189,6 +1257,35 @@ mod tests { assert_eq!(state.spendable_balance(), Some(700)); } + #[test] + fn should_report_spent_state_by_nullifier() { + let mut store = InMemoryShieldedStore::new(); + let id = test_id(0); + store.save_note(id, ¬e_with_nullifier([1; 32])).unwrap(); + assert!(!store.is_note_spent(id, &[1; 32]).unwrap()); + store.mark_pending(id, &[1; 32]).unwrap(); + assert!( + !store.is_note_spent(id, &[1; 32]).unwrap(), + "a reservation is not a spend" + ); + store.mark_spent(id, &[1; 32]).unwrap(); + assert!(store.is_note_spent(id, &[1; 32]).unwrap()); + assert!(!store.is_note_spent(id, &[2; 32]).unwrap()); + assert!(!store.is_note_spent(test_id(1), &[1; 32]).unwrap()); + } + + #[test] + fn should_keep_a_note_spent_when_a_stale_copy_is_re_saved() { + let mut store = InMemoryShieldedStore::new(); + let id = test_id(0); + let stale = note_with_nullifier([1; 32]); + store.save_note(id, &stale).unwrap(); + store.mark_spent(id, &[1; 32]).unwrap(); + store.save_note(id, &stale).unwrap(); + assert!(store.is_note_spent(id, &[1; 32]).unwrap()); + assert!(store.get_unspent_notes(id).unwrap().is_empty()); + } + #[test] fn should_reject_overflowing_in_memory_spendable_balance() { let mut store = InMemoryShieldedStore::new(); diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync.rs index 7d4cb149951..572b7ae8596 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/sync.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync.rs @@ -35,7 +35,7 @@ use dash_sdk::platform::shielded::{ }; use futures::{Stream, StreamExt}; use grovedb_commitment_tree::{ExtractedNoteCommitment, Note as OrchardNote, PaymentAddress}; -use tokio::sync::RwLock; +use tokio::sync::{RwLock, RwLockWriteGuard}; use tokio_util::sync::CancellationToken; use tracing::{debug, info}; @@ -43,6 +43,7 @@ use super::keys::AccountViewingKeys; use super::store::{ShieldedStore, SubwalletId}; use crate::changeset::ShieldedChangeSet; use crate::error::PlatformWalletError; +use crate::wallet::platform_wallet::WalletId; /// On-chain MMR chunk size for the shielded notes tree — must stay /// in lock-step with `SHIELDED_NOTES_CHUNK_POWER` in @@ -176,6 +177,168 @@ impl MultiSyncNotesResult { } } +/// Generations a sync pass re-checks under the store write guard before each +/// write, since it no longer holds that lock across its download. +/// +/// Lifecycle operations that conflict with an in-flight pass (wallet removal, +/// a re-bind that drops or re-keys accounts, a host snapshot restore, Clear) +/// bump a generation under the same guard, so either the pass's write lands +/// first or the pass sees the bump and returns +/// [`NoteSyncOutcome::Superseded`] without writing. Wallet changes bump only +/// that wallet, so binding one wallet doesn't abandon a pass over others; +/// Clear bumps the tree generation, which every pass depends on. +/// +/// A pass must [`watch`](Self::watch) before it snapshots the account +/// registry, because lifecycle operations change the registry before they +/// bump. Clear resets the store before clearing the registry, so it bumps +/// both before the reset and after. +#[derive(Debug, Default)] +pub(crate) struct StoreEpoch(std::sync::Mutex); + +#[derive(Debug, Default, Clone)] +struct Generations { + tree: u64, + wallets: BTreeMap, + purges: BTreeMap, +} + +impl Generations { + fn wallet(&self, wallet_id: &WalletId) -> u64 { + self.wallets.get(wallet_id).copied().unwrap_or(0) + } + + fn purges(&self, id: &SubwalletId) -> u64 { + self.purges.get(id).copied().unwrap_or(0) + } +} + +impl StoreEpoch { + fn lock(&self) -> std::sync::MutexGuard<'_, Generations> { + // A plain counter map can't be left half-updated; recover from a + // poisoning panic elsewhere rather than wedging every sync. + self.0 + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Record the current generations for a sync pass about to snapshot + /// the registry and the store. + pub(crate) fn watch(&self) -> EpochWatch<'_> { + EpochWatch { + epoch: self, + observed: self.lock().clone(), + } + } + + /// Invalidate in-flight passes that scan any of `wallet_id`'s + /// subwallets. Call while still holding the write guard of the + /// mutation that invalidates them. + pub(crate) fn bump_wallet( + &self, + _store_write_guard: &mut RwLockWriteGuard<'_, S>, + wallet_id: WalletId, + ) { + *self.lock().wallets.entry(wallet_id).or_default() += 1; + } + + /// Record that `ids`' store state was purged, so a pass that committed + /// before the purge doesn't queue that state to the host afterwards (see + /// [`EpochWatch::state_kept`]). Call while still holding the write guard + /// of the purge. + pub(crate) fn bump_purged( + &self, + _store_write_guard: &mut RwLockWriteGuard<'_, S>, + ids: impl IntoIterator, + ) { + let mut generations = self.lock(); + for id in ids { + *generations.purges.entry(id).or_default() += 1; + } + } + + /// Invalidate every in-flight pass (the shared tree itself changed). + /// Call while still holding the write guard of the mutation that + /// invalidates them. + /// + /// Also drops the per-wallet and per-subwallet generations, so they + /// hold only what changed since the last Clear: every watch taken + /// before this bump fails on the tree generation anyway, so none of + /// them needs the old counters, and a counter restarting at 0 can't + /// make an older watch look current. + pub(crate) fn bump_all( + &self, + _store_write_guard: &mut RwLockWriteGuard<'_, S>, + ) { + let mut generations = self.lock(); + generations.tree += 1; + generations.wallets.clear(); + generations.purges.clear(); + } +} + +/// The [`StoreEpoch`] generations a sync pass observed before snapshotting. +#[derive(Debug)] +pub(crate) struct EpochWatch<'a> { + epoch: &'a StoreEpoch, + observed: Generations, +} + +impl EpochWatch<'_> { + /// Whether nothing the pass over `subwallets` depends on has been + /// bumped since [`StoreEpoch::watch`]. + pub(super) fn still_current(&self, subwallets: &[(SubwalletId, AccountViewingKeys)]) -> bool { + let now = self.epoch.lock(); + now.tree == self.observed.tree + && subwallets + .iter() + .all(|(id, _)| now.wallet(&id.wallet_id) == self.observed.wallet(&id.wallet_id)) + } + + /// Whether `id`'s store state is still the one the pass committed to: + /// neither purged nor touched by a Clear (even a failed one, which may + /// have reset part of the store) since [`StoreEpoch::watch`]. + pub(super) fn state_kept(&self, id: &SubwalletId) -> bool { + let now = self.epoch.lock(); + now.tree == self.observed.tree && now.purges(id) == self.observed.purges(id) + } +} + +/// Outcome of one [`sync_notes_across`] attempt. +#[derive(Debug)] +pub(super) enum NoteSyncOutcome { + /// The pass committed; carries its result. + Completed(MultiSyncNotesResult), + /// A concurrent store mutation (see [`StoreEpoch`]) invalidated the + /// pass's snapshot mid-download. Nothing but already-appended tree + /// leaves was written — no checkpoint, notes, spends or watermarks — + /// which is the same state an interrupted pass leaves behind: the next + /// pass re-fetches from the unchanged watermarks, gate-skips those + /// leaves, and checkpoints them on commit even if it appends nothing + /// new itself. The caller should retry from a fresh registry snapshot. + Superseded, + /// `cancel` fired before the stream ended. Nothing is published but the + /// tree leaves already appended — no notes, spent marks or watermarks — + /// and those are checkpointed only if the store is free and the + /// snapshot still current; otherwise the next completed pass + /// checkpoints them. Takes precedence over [`Self::Superseded`]: both + /// write nothing, and a stopped pass must not be retried. A consumed prefix is not a complete scan — a receipt in + /// it may already be spent by an action in a chunk it never reached, + /// and the ledger can't hold such a note without offering it for + /// selection. The next pass re-walks from the last completed watermark + /// and gate-skips the appended leaves, so a stop costs only the + /// download and decryption of the chunks it had consumed. + Cancelled, +} + +/// Leaves appended per store write-lock acquisition. Each file-store append +/// is its own SQLite transaction (~0.1 ms), and a streamed batch carries up +/// to several MMR chunks (thousands of leaves), so appending a whole batch +/// under one hold would still park a waiting send for hundreds of +/// milliseconds during catch-up. Slicing bounds the hold to a few tens of +/// milliseconds; the per-slice cost (one lock round trip plus one +/// `tree_size` read) is negligible next to the appends. +const APPEND_SLICE: usize = 256; + /// Single-fetch, multi-IVK trial-decrypt across an arbitrary set /// of registered subwallets — the Phase 2b primitive that /// collapses N per-wallet SDK calls into one. @@ -196,126 +359,129 @@ impl MultiSyncNotesResult { /// No `SpendAuthorizingKey` is needed by sync — the spend /// surface re-attaches it at call time. /// -/// Cancellation: `cancel` is observed before the store is locked and -/// between batches. A cancelled pass stops pulling batches (dropping -/// the in-flight network fetches), checkpoints the commitments it -/// appended to the local tree, and publishes NOTHING else: no notes, no -/// spent marks, no watermark, an empty result. A consumed prefix is not -/// a complete scan — a receipt in it may already be spent by an action -/// in a chunk it never reached, and the ledger has no way to hold such a -/// note without offering it for selection. The next pass re-walks from -/// the last completed watermark; the appended leaves make that re-walk -/// skip every append and checkpoint it already made, so what a stop -/// costs is the download and decryption of the chunks it had consumed. -/// It is a clean early stop, not an error. +/// # Locking +/// +/// The store write lock is never held across a network await. Each +/// streamed batch is parsed and trial-decrypted with no lock held; its new +/// commitments are then appended in [`APPEND_SLICE`]-sized write-lock +/// holds, and everything else the pass produces — the checkpoint, notes, +/// outgoing notes, scan-detected spends and watermarks — is committed in a +/// single write-lock hold after the stream ends, so readers still observe +/// those atomically per pass. Every acquisition first checks +/// [`snapshot_still_current`] and returns [`NoteSyncOutcome::Superseded`] +/// if anything moved. `epoch` must have been watched before the registry +/// was snapshotted into `subwallets`. +/// +/// Appends that land between acquisitions are invisible to spends: the +/// spend path only witnesses against checkpoints, and shardtree computes a +/// checkpoint-depth witness as of that checkpoint's position, so leaves +/// appended after the last checkpoint change no witness or anchor until +/// this pass checkpoints them together with the notes they belong to. +/// +/// # Cancellation +/// +/// `cancel` is observed before the scan starts, while waiting for a store +/// lock, and between batches; a stop returns +/// [`NoteSyncOutcome::Cancelled`]. It is a clean early stop, not an error. pub(super) async fn sync_notes_across( sdk: &Arc, store: &Arc>, + epoch: &EpochWatch<'_>, subwallets: &[(SubwalletId, AccountViewingKeys)], on_progress: Option<&super::coordinator::ShieldedProgressCallback>, on_tree_progress: Option<&super::coordinator::ShieldedTreeProgressCallback>, cancel: &CancellationToken, -) -> Result { - if subwallets.is_empty() { - return Ok(MultiSyncNotesResult::default()); - } - - // Drive the FIRST subwallet's IVK as the streaming driver. Its hits - // come back per batch as `batch.decrypted`; every other subwallet's - // are produced by local trial-decryption against `batch.notes`. - let driver_ivk = subwallets[0].1.prepared_ivk.clone(); - - // Builds the SDK stream once the consumer has snapshotted its start - // position and the pre-stream tree size (both feed the config). - let make_stream = |aligned_start: u64, tree_size: u64| { - // Network-only config carrying the caller's "downloaded" progress - // callback; the SDK fires it once per completed network chunk inside - // the stream. The tree-progress ("checked") callback is owned by - // the consumer and fired there — it never travels through the SDK - // config (the SDK doesn't append to a tree). - // Fetch up to 16 chunks concurrently (default is 4) to parallelize - // across the network's ~13 nodes. The per-chunk cost is dominated by - // server-side proof generation, so more in-flight requests is the main - // 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: MAX_CONCURRENT_CHUNK_FETCHES, - ..Default::default() - }; - if let Some(cb) = on_progress { - // Clamp the SDK's "downloaded" value up to the pre-stream tree leaf - // count before forwarding it to the host. The SDK reports downloaded - // as `aligned_start + scanned`, where `aligned_start` is the rewound - // MIN watermark across subwallets and can sit far below `tree_size` - // (a second subwallet binding at watermark 0 collapses it to 0; a - // partial-chunk-tail resume rewinds it below the tail). The "checked" - // signal is the absolute tree size, so over the gate-skipped re-scan - // region (`global_pos < tree_size`, no new appends) the unclamped - // download value reads *below* checked and breaks the advertised - // `Checked ≤ Downloaded ≤ total` invariant. Everything already in the - // tree was necessarily downloaded on a prior pass, so counting it as - // downloaded is accurate; the clamp is a no-op once the stream - // advances past `tree_size`, and the cold-sync case (tree_size == 0) - // is entirely unaffected. - let cb = cb.clone(); - let tree_baseline = tree_size; - sync_config.on_chunk_completed = Some(Arc::new(move |downloaded: u64, height: u64| { - cb(downloaded.max(tree_baseline), height); - })); - } - sync_shielded_notes_stream(sdk, &driver_ivk, aligned_start, Some(sync_config)) - }; - - sync_notes_from_stream(store, subwallets, on_tree_progress, cancel, make_stream).await +) -> Result { + sync_notes_from_stream( + store, + epoch, + subwallets, + on_tree_progress, + cancel, + |aligned_start, tree_size| { + // Network-only config carrying the caller's "downloaded" + // progress callback; the SDK fires it once per completed network + // chunk inside the stream. The tree-progress ("checked") callback + // is owned by the consumer and never travels through the SDK + // config (the SDK doesn't append to a tree). + // Fetch up to 16 chunks concurrently (default is 4) to + // parallelize across the network's ~13 nodes. The per-chunk cost + // is dominated by server-side proof generation, so more in-flight + // requests is the main 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: MAX_CONCURRENT_CHUNK_FETCHES, + ..Default::default() + }; + if let Some(cb) = on_progress { + // Clamp the SDK's "downloaded" value up to the pre-stream + // tree leaf count before forwarding it to the host. The SDK + // reports downloaded as `aligned_start + scanned`, where + // `aligned_start` is the rewound MIN watermark across + // subwallets and can sit far below `tree_size` (a second + // subwallet binding at watermark 0 collapses it to 0; a + // partial-chunk-tail resume rewinds it below the tail). The + // "checked" signal is the absolute tree size, so over the + // gate-skipped re-scan region (`global_pos < tree_size`, no + // new appends) the unclamped download value reads *below* + // checked and breaks the advertised `Checked ≤ Downloaded ≤ + // total` invariant. Everything already in the tree was + // necessarily downloaded on a prior pass, so counting it as + // downloaded is accurate; the clamp is a no-op once the + // stream advances past `tree_size`, and the cold-sync case + // (tree_size == 0) is entirely unaffected. + let cb = cb.clone(); + sync_config.on_chunk_completed = + Some(Arc::new(move |downloaded: u64, height: u64| { + cb(downloaded.max(tree_size), height); + })); + } + // Drive the FIRST subwallet's IVK as the streaming driver. Its + // hits come back per batch as `batch.decrypted`; every other + // subwallet's are produced by local trial-decryption against + // `batch.notes`. + let driver_ivk = &subwallets[0].1.prepared_ivk; + sync_shielded_notes_stream(sdk, driver_ivk, aligned_start, Some(sync_config)) + }, + ) + .await } -/// The consumer half of [`sync_notes_across`]: snapshots the watermarks, -/// takes the store write lock, drains the batch stream `make_stream` -/// builds for `(aligned_start, tree_size)`, and finalizes the store. +/// [`sync_notes_across`]'s consumer, generic over where the note batches +/// come from. /// -/// Split out so a test can feed a synthetic stream and exercise the -/// cancellation and finalize paths without a network. -pub(super) async fn sync_notes_from_stream( +/// `open_stream(aligned_start, tree_size)` is called exactly once, after +/// the pass's watermark / tree-size snapshot, and must return the batches +/// covering `aligned_start..` in ascending tree order — the contract of +/// the SDK's `sync_shielded_notes_stream`, which is what production passes. +/// Tests pass a hand-driven stream so a pass can be held mid-download or +/// cancelled between batches. `subwallets[0]` is the driver whose hits +/// arrive pre-decrypted in `batch.decrypted`. +pub(super) async fn sync_notes_from_stream( store: &Arc>, + epoch: &EpochWatch<'_>, subwallets: &[(SubwalletId, AccountViewingKeys)], on_tree_progress: Option<&super::coordinator::ShieldedTreeProgressCallback>, cancel: &CancellationToken, - make_stream: F, -) -> Result + open_stream: F, +) -> Result where S: ShieldedStore, - St: Stream>, F: FnOnce(u64, u64) -> St, + St: Stream>, { if subwallets.is_empty() { - return Ok(MultiSyncNotesResult::default()); + return Ok(NoteSyncOutcome::Completed(MultiSyncNotesResult::default())); } - // Nothing has been touched yet, so a stop that landed before the scan // costs nothing: skip the lock and the fetch entirely. if cancel.is_cancelled() { debug!("Shielded note sync cancelled before the scan started"); - return Ok(MultiSyncNotesResult::default()); + return Ok(NoteSyncOutcome::Cancelled); } - // Keep the snapshot and streamed consume under one write guard so another - // sync cannot append between them and leave this pass using stale gates. - // 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`. - // - // The wait for the guard observes the token too: a pass queued behind - // another writer (a spend flow, a concurrent scan in tests) has made - // no mutation yet, so a stop that lands here returns at once instead - // of after the unrelated holder is done. - let Some(mut store) = cancel.run_until_cancelled(store.write()).await else { - debug!("Shielded note sync cancelled while waiting for the store lock"); - return Ok(MultiSyncNotesResult::default()); - }; - // Snapshot each subwallet's watermark and the shared tree's // current leaf count. Two distinct quantities drive the rest // of this pass: @@ -332,7 +498,15 @@ where // caught-up subwallet doesn't re-derive nullifiers for notes // it already stored, while a lagging one still saves from its // own start. + // + // The wait for the guard observes the token: a pass queued behind a + // writer (a spend flow holding the store) has made no mutation yet, so + // a stop that lands here returns at once. let (watermarks, already_have, tree_size) = { + let Some(store) = cancel.run_until_cancelled(store.read()).await else { + debug!("Shielded note sync cancelled while waiting for the store lock"); + return Ok(NoteSyncOutcome::Cancelled); + }; let mut watermarks: BTreeMap = BTreeMap::new(); let mut min_idx: Option = None; for (id, _) in subwallets { @@ -375,9 +549,7 @@ where // (total 0) until the first batch lands. let mut total_target: u64 = tree_size; - // The FIRST subwallet drives the stream (see `sync_notes_across`); - // its hits arrive pre-decrypted in `batch.decrypted`. - let (driver_id, _) = &subwallets[0]; + let driver_id = &subwallets[0].0; // Route decryptions to the subwallet that owns the IVK, accumulated // across every batch. @@ -430,9 +602,10 @@ where // for correctness, and the shielded pool is small. A future // optimization can prune auth paths for positions no live // subwallet owns once all wallets have caught past them. - let mut appended = 0u32; // Cumulative leaves committed to the tree this pass — the - // tree-progress ("checked") signal numerator. Includes only + // tree-progress ("checked") signal numerator, and the exact leaf + // count the tree must report whenever this pass re-takes the lock + // (anything else means another writer moved it). Includes only // positions we actually appended (gate-skipped re-fetched // positions are already in the tree, so they don't advance the // commit count beyond `tree_size`). @@ -469,29 +642,26 @@ where let mut scanned_nullifiers: Vec<[u8; 32]> = Vec::new(); // The stream lives in this block so it — and the up-to-`max_concurrent` - // network fetches it holds in flight — is dropped before the finalize - // below runs, whether the loop ran to the end or was cancelled. - let cancelled = { - let stream = make_stream(aligned_start, tree_size); + // network fetches it holds in flight — is dropped before the commit + // below, whether the loop ran to the end or was cancelled. No store + // lock is held while it is polled (see "Locking" on + // `sync_notes_across`). Backpressure is still pull-based — a slow + // consumer polls the stream less often, capping in-flight network + // fetches at `max_concurrent`. + let cancelled = 'scan: { + let stream = open_stream(aligned_start, tree_size); futures::pin_mut!(stream); loop { // Batches arrive in tree order as a contiguous prefix, so - // stopping between two of them leaves the tree one checkpoint - // further along and nothing else to publish. `biased` so a - // pending stop wins over a ready batch. + // stopping between two of them leaves nothing half-consumed. + // `biased` so a pending stop wins over a ready batch. let item = tokio::select! { biased; - _ = cancel.cancelled() => { - debug!( - total_scanned = total_notes_scanned, - "Shielded note sync cancelled; committing tree progress only" - ); - break true; - } + _ = cancel.cancelled() => break 'scan true, next = stream.next() => match next { Some(item) => item, - None => break false, + None => break 'scan false, }, }; let batch = item.map_err(|e| PlatformWalletError::ShieldedSyncFailed(e.to_string()))?; @@ -505,7 +675,7 @@ where last_nonempty = Some((batch.start_index, batch.is_partial)); } - // 1. Append commitments for THIS batch, applying the same + // 1. Collect THIS batch's new commitments, applying the same // idempotency gate against the pre-stream `tree_size` // snapshot. `batch.start_index + i` is the global tree // position. Each action's `nullifier` is the nullifier of @@ -516,6 +686,9 @@ where // still surface a spend whose receipt only persisted on a // later pass), since `mark_spent` is an idempotent no-op for // nullifiers the wallet doesn't own or already marked spent. + // Parsed in full before anything is appended, so a malformed + // note fails the pass without a partially-appended batch. + let mut new_commitments: Vec<[u8; 32]> = Vec::new(); for (i, raw_note) in batch.notes.iter().enumerate() { // A malformed nullifier length means the proven note item is corrupt; // fail fast (consistent with the cmx check below) rather than silently @@ -533,14 +706,27 @@ where let cmx_bytes: [u8; 32] = raw_note.cmx.as_slice().try_into().map_err(|_| { PlatformWalletError::ShieldedSyncFailed("Invalid cmx length".into()) })?; - store - .append_commitment(&cmx_bytes, true) - .map_err(|e| PlatformWalletError::ShieldedTreeUpdateFailed(e.to_string()))?; - appended += 1; - leaves_committed += 1; + new_commitments.push(cmx_bytes); + } + + // 2. Append them under short write-lock holds, re-validating + // the pass's snapshot each time the lock is (re)taken. + for slice in new_commitments.chunks(APPEND_SLICE) { + let Some(mut guard) = cancel.run_until_cancelled(store.write()).await else { + break 'scan true; + }; + if !snapshot_still_current(&*guard, epoch, subwallets, leaves_committed)? { + return Ok(superseded(leaves_committed)); + } + for cmx_bytes in slice { + guard.append_commitment(cmx_bytes, true).map_err(|e| { + PlatformWalletError::ShieldedTreeUpdateFailed(e.to_string()) + })?; + leaves_committed += 1; + } } - // 2. Fire the tree-progress ("checked") callback once per batch + // 3. Fire the tree-progress ("checked") callback once per batch // (already coarse at ~8192-note batches). `total_target` is // sourced from this batch's proof above; it is set by the first // batch and stable thereafter. It is 0 (indeterminate on the @@ -550,7 +736,7 @@ where cb(leaves_committed, total_target); } - // 3. Trial-decrypt THIS batch. Driver hits come pre-decrypted + // 4. Trial-decrypt THIS batch (no lock held). Driver hits come pre-decrypted // in `batch.decrypted`; other subwallets via local // trial-decryption over `batch.notes`. for dn in batch.decrypted { @@ -585,7 +771,7 @@ where } } - // 4. OVK recovery (outgoing/SENT notes). Attempt to open each + // 5. OVK recovery (outgoing/SENT notes). Attempt to open each // scanned note's `out_ciphertext` with EVERY subwallet's OVK // (driver included — unlike IVK trial-decrypt, the driver's // sent notes are NOT pre-recovered by the SDK). A note only @@ -640,50 +826,38 @@ where "SDK stream consumed" ); - if appended > 0 { - // Checkpoint at the tree's true post-append leaf count. The - // tree only ever grows, so this id is strictly monotonic - // and collision-free across consecutive syncs — unlike - // `result.next_start_index` (rewinds to the last partial - // chunk's start) or `aligned_start + total_notes_scanned` - // (can repeat when a lagging subwallet rewinds the fetch). - // shardtree's `checkpoint(id)` silently dedups duplicate - // ids; a non-monotonic id pins depth-0 at the first - // checkpoint while later appends extend the tree past it, - // so the depth-0 witness then reflects a state Platform - // never recorded. - let new_tree_size = store - .tree_size() - .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; - // Hard-fail rather than saturate at u32::MAX: a saturated id - // would reintroduce shardtree's silent-dedup (every checkpoint - // past the ceiling pins to the same id) — the exact corruption - // this monotonic-id scheme exists to avoid. Unreachable today - // (>4.29B notes scanned) but fail loudly before proving. - let checkpoint_id: u32 = new_tree_size.try_into().map_err(|_| { - PlatformWalletError::ShieldedTreeUpdateFailed(format!( - "commitment tree size {new_tree_size} exceeds u32 checkpoint-id range" - )) - })?; - store - .checkpoint_tree(checkpoint_id) - .map_err(|e| PlatformWalletError::ShieldedTreeUpdateFailed(e.to_string()))?; - } - - // A cancelled pass ends here: the appended leaves are checkpointed - // (the next pass skips them), and nothing the prefix decrypted is - // saved, replayed, watermarked or handed to the caller. See the - // cancellation note on `sync_notes_across` for why a consumed prefix - // must not reach the ledger. if cancelled { - drop(store); - info!( - total_scanned = total_notes_scanned, - appended, "Shielded note sync stopped; tree progress kept, nothing published" + return finish_cancelled( + store, + epoch, + subwallets, + tree_size, + leaves_committed, + total_notes_scanned, ); - return Ok(MultiSyncNotesResult::default()); } + // Commit the rest of the pass in ONE write-lock hold, so readers see + // its checkpoint, notes, spends and watermarks together — never notes + // whose positions no checkpoint covers yet (a spend selecting one would + // find no witness), nor a watermark ahead of the notes it vouches for. + // Everything below is local CPU work; no network await happens under + // this guard. + let mut store = store.write().await; + if !snapshot_still_current(&*store, epoch, subwallets, leaves_committed)? { + return Ok(superseded(leaves_committed)); + } + + // Checkpoint whenever the tree is non-empty, not only when this pass + // appended: the notes saved below may sit in leaves an earlier, + // interrupted (superseded, cancelled, errored, killed) pass appended + // without checkpointing, and a note no checkpoint covers is + // unwitnessable at spend time. When the newest checkpoint already sits + // at this size (the usual caught-up pass) the call is a no-op — + // shardtree ignores a checkpoint id that isn't above the current + // maximum. + checkpoint_tree_at(&mut *store, leaves_committed)?; + // Save decrypted notes per subwallet; count new notes per // subwallet; build a single consolidated changeset. let mut per_subwallet_new_notes: BTreeMap = BTreeMap::new(); @@ -699,16 +873,14 @@ where // Gate on THIS subwallet's own watermark (not the network // min): a caught-up subwallet skips re-deriving nullifiers // for notes it already stored, while a lagging one still - // saves everything from its own start. `save_note` is an - // idempotent overwrite-by-nullifier, so a stray re-save is - // harmless — but gating per-subwallet keeps the + // saves everything from its own start, which also keeps the // `per_subwallet_new_notes` count honest. let sub_watermark = watermarks.get(id).copied().unwrap_or(0); for d in discovered { if d.position < sub_watermark { continue; } - let nullifier = d.note.nullifier(&views.full_viewing_key); + let nullifier = d.note.nullifier(&views.full_viewing_key).to_bytes(); let value = d.note.value().inner(); debug!( wallet_id = %hex::encode(id.wallet_id), @@ -718,6 +890,14 @@ where "Note DECRYPTED" ); let note_data = serialize_note(&d.note); + // A re-scanned receipt can already be on file, and a send may + // have confirmed it spent (`mark_notes_spent`) while this pass + // was downloading without the lock. `save_note` replaces the + // stored note wholesale, so carry that spent flag over; scan + // data that predates the spend can't restore it. + let is_spent = store + .is_note_spent(*id, &nullifier) + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; // Stamp the note with ITS chunk's proven height — the same // per-batch height OVK-recovered outgoing notes get below — // never a pass-wide max. The activity deriver clusters @@ -728,9 +908,9 @@ where note_data, position: d.position, cmx: d.cmx, - nullifier: nullifier.to_bytes(), + nullifier, block_height: d.block_height, - is_spent: false, + is_spent, value, }; store @@ -798,18 +978,16 @@ where // streamed batch's note count — identical to the one-shot path's // `result.total_notes_scanned`. // - // Never rewind. `aligned_start` is the MIN watermark across the - // union, so a pass that re-walks a lagging subwallet's range can - // end below a subwallet that was already further along; a completed - // pass reaches the tip, which is at or past every watermark. Only a - // completed pass gets here (a cancelled one returned above), so the - // write is unconditional: the setter also stamps the scan coverage - // and the `ScannedThisSession` balance source, which is exactly what - // reaching the tip means. + // Never rewind. `aligned_start` is the MIN watermark across the union, + // so a pass that re-walks a lagging subwallet's range can end below a + // subwallet that was already further along; a completed pass reaches + // the tip, which is at or past every watermark. Only a completed pass + // gets here, so the write is unconditional: the setter also stamps the + // scan coverage and the `ScannedThisSession` balance source, which is + // exactly what reaching the tip means. let new_index = aligned_start + total_notes_scanned; for (id, _) in subwallets { - let prior = watermarks.get(id).copied().unwrap_or(0); - let advanced = prior.max(new_index); + let advanced = watermarks.get(id).copied().unwrap_or(0).max(new_index); store .set_last_synced_note_index(*id, advanced) .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; @@ -834,12 +1012,110 @@ where // rewind), and the host counter is documented as scan throughput, // not tree growth. let scanned_volume = (aligned_start + total_notes_scanned).saturating_sub(already_have); - Ok(MultiSyncNotesResult { + Ok(NoteSyncOutcome::Completed(MultiSyncNotesResult { per_subwallet_new_notes, per_subwallet_newly_spent: newly_spent_per_subwallet, total_scanned: scanned_volume, changeset, - }) + })) +} + +/// Whether the store still matches what this pass snapshotted and has +/// itself written since: no relevant [`StoreEpoch`] bump, and the tree holds exactly +/// the `expected_tree_size` leaves (snapshot size plus this pass's +/// appends). Called under the write guard the pass is about to write +/// through. +/// +/// The epoch covers every known concurrent mutation; the leaf-count check +/// is the backstop that keeps an unknown writer from ever turning into a +/// duplicated or misplaced leaf — the corruption that makes witnesses +/// resolve against roots Platform never recorded. +fn snapshot_still_current( + store: &S, + epoch: &EpochWatch<'_>, + subwallets: &[(SubwalletId, AccountViewingKeys)], + expected_tree_size: u64, +) -> Result { + if !epoch.still_current(subwallets) { + return Ok(false); + } + let tree_size = store + .tree_size() + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; + Ok(tree_size == expected_tree_size) +} + +/// Log and build the [`NoteSyncOutcome::Superseded`] result. +fn superseded(leaves_committed: u64) -> NoteSyncOutcome { + info!( + leaves_committed, + "Shielded note sync superseded by a concurrent store change; abandoning this pass" + ); + NoteSyncOutcome::Superseded +} + +/// End a cancelled pass: checkpoint the leaves it appended when that costs +/// no wait, and publish nothing else (see [`NoteSyncOutcome::Cancelled`]). +/// +/// Never waits for the store lock: a stop must not queue behind a writer. +/// Leaves left uncheckpointed — the store was busy, or a concurrent change +/// made the snapshot stale — are checkpointed by the next completed pass, +/// exactly as after a superseded one. +fn finish_cancelled( + store: &RwLock, + epoch: &EpochWatch<'_>, + subwallets: &[(SubwalletId, AccountViewingKeys)], + tree_size: u64, + leaves_committed: u64, + total_scanned: u64, +) -> Result { + let mut checkpointed = false; + if leaves_committed > tree_size { + if let Ok(mut guard) = store.try_write() { + if snapshot_still_current(&*guard, epoch, subwallets, leaves_committed)? { + checkpoint_tree_at(&mut *guard, leaves_committed)?; + checkpointed = true; + } + } + } + info!( + total_scanned, + leaves_committed, + checkpointed, + "Shielded note sync stopped; tree progress kept, nothing published" + ); + Ok(NoteSyncOutcome::Cancelled) +} + +/// Checkpoint the tree at its true post-append leaf count, a no-op for an +/// empty tree. The tree only ever grows, so this id is strictly monotonic +/// and collision-free across consecutive syncs — unlike +/// `result.next_start_index` (rewinds to the last partial chunk's start) +/// or `aligned_start + total_notes_scanned` (can repeat when a lagging +/// subwallet rewinds the fetch). shardtree's `checkpoint(id)` silently +/// dedups duplicate ids; a non-monotonic id pins depth-0 at the first +/// checkpoint while later appends extend the tree past it, so the depth-0 +/// witness then reflects a state Platform never recorded. +fn checkpoint_tree_at( + store: &mut S, + leaves_committed: u64, +) -> Result<(), PlatformWalletError> { + if leaves_committed == 0 { + return Ok(()); + } + // Hard-fail rather than saturate at u32::MAX: a saturated id would + // reintroduce shardtree's silent-dedup (every checkpoint past the + // ceiling pins to the same id) — the exact corruption this + // monotonic-id scheme exists to avoid. Unreachable today (>4.29B notes + // scanned) but fail loudly before proving. + let checkpoint_id: u32 = leaves_committed.try_into().map_err(|_| { + PlatformWalletError::ShieldedTreeUpdateFailed(format!( + "commitment tree size {leaves_committed} exceeds u32 checkpoint-id range" + )) + })?; + store + .checkpoint_tree(checkpoint_id) + .map_err(|e| PlatformWalletError::ShieldedTreeUpdateFailed(e.to_string())) } /// Scan-based spend detection: replay the set of nullifiers seen on @@ -1201,3 +1477,8 @@ mod cancel_tests; #[cfg(test)] mod concurrent_sync_tests; + +/// Store-lock behaviour of a sync pass: short wallet operations must not +/// wait on its network I/O, and a superseded pass must not commit. +#[cfg(test)] +mod lock_contention_tests; diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync/cancel_tests.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync/cancel_tests.rs index a683da2d53c..a2a3883816b 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/sync/cancel_tests.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync/cancel_tests.rs @@ -15,7 +15,7 @@ use futures::{Stream, StreamExt}; use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; -use super::{sync_notes_from_stream, MultiSyncNotesResult}; +use super::{sync_notes_from_stream, MultiSyncNotesResult, NoteSyncOutcome, StoreEpoch}; use crate::wallet::shielded::balance::ShieldedBalanceSource; use crate::wallet::shielded::file_store::FileBackedShieldedStore; use crate::wallet::shielded::keys::{AccountViewingKeys, OrchardKeySet}; @@ -89,16 +89,38 @@ async fn run( subwallets: &[(SubwalletId, AccountViewingKeys)], cancel: &CancellationToken, stream: impl Stream, -) -> MultiSyncNotesResult { +) -> NoteSyncOutcome { + let epoch = StoreEpoch::default(); tokio::time::timeout( std::time::Duration::from_secs(5), - sync_notes_from_stream(store, subwallets, None, cancel, move |_, _| stream), + sync_notes_from_stream( + store, + &epoch.watch(), + subwallets, + None, + cancel, + move |_, _| stream, + ), ) .await .expect("a cancelled pass must return promptly") .expect("pass must succeed") } +fn assert_cancelled(outcome: NoteSyncOutcome) { + assert!( + matches!(outcome, NoteSyncOutcome::Cancelled), + "a cut-short pass publishes nothing, got {outcome:?}" + ); +} + +fn expect_completed(outcome: NoteSyncOutcome) -> MultiSyncNotesResult { + match outcome { + NoteSyncOutcome::Completed(result) => result, + other => panic!("pass should have completed, got {other:?}"), + } +} + #[tokio::test] async fn should_keep_tree_progress_and_publish_nothing_when_cancelled_mid_stream() { let path = temp_tree_path("prefix"); @@ -109,7 +131,7 @@ async fn should_keep_tree_progress_and_publish_nothing_when_cancelled_mid_stream let sub = subwallets[0].0; let cancel = CancellationToken::new(); - let result = run( + let outcome = run( &store, &subwallets, &cancel, @@ -118,8 +140,7 @@ async fn should_keep_tree_progress_and_publish_nothing_when_cancelled_mid_stream .await; // The appended leaves stay, checkpointed; nothing else is published. - assert_eq!(result.total_scanned, 0, "a cut-short pass reports no scan"); - assert!(crate::changeset::merge::Merge::is_empty(&result.changeset)); + assert_cancelled(outcome); let store = store.read().await; assert_eq!(store.tree_size().unwrap(), 8, "both batches appended"); assert!( @@ -149,8 +170,9 @@ async fn should_return_empty_without_touching_store_when_cancelled_before_scan() let cancel = CancellationToken::new(); cancel.cancel(); - let result = sync_notes_from_stream( + let outcome = sync_notes_from_stream( &store, + &StoreEpoch::default().watch(), &subwallets, None, &cancel, @@ -161,8 +183,7 @@ async fn should_return_empty_without_touching_store_when_cancelled_before_scan() .await .expect("pass must succeed"); - assert_eq!(result.total_scanned, 0); - assert!(result.per_subwallet_new_notes.is_empty()); + assert_cancelled(outcome); let store = store.read().await; assert_eq!(store.tree_size().unwrap(), 0); assert_eq!(store.last_synced_note_index(sub).unwrap(), 0); @@ -181,13 +202,15 @@ async fn should_rewalk_the_appended_prefix_without_reappending_on_next_pass() { // Pass 1: cancelled after two batches. let cancel = CancellationToken::new(); - run( - &store, - &subwallets, - &cancel, - batches_then_cancel(vec![batch(0, 4, 12), batch(4, 4, 12)], cancel.clone()), - ) - .await; + assert_cancelled( + run( + &store, + &subwallets, + &cancel, + batches_then_cancel(vec![batch(0, 4, 12), batch(4, 4, 12)], cancel.clone()), + ) + .await, + ); assert_eq!(store.read().await.last_synced_note_index(sub).unwrap(), 0); assert_eq!(store.read().await.tree_size().unwrap(), 8); @@ -195,17 +218,19 @@ async fn should_rewalk_the_appended_prefix_without_reappending_on_next_pass() { // re-delivers the first two batches — gate-skipped by the tree-size // snapshot — and then the new one. Runs to completion. let cancel = CancellationToken::new(); - let result = run( - &store, - &subwallets, - &cancel, - futures::stream::iter( - vec![batch(0, 4, 12), batch(4, 4, 12), batch(8, 4, 12)] - .into_iter() - .map(Ok), - ), - ) - .await; + let result = expect_completed( + run( + &store, + &subwallets, + &cancel, + futures::stream::iter( + vec![batch(0, 4, 12), batch(4, 4, 12), batch(8, 4, 12)] + .into_iter() + .map(Ok), + ), + ) + .await, + ); assert_eq!( result.total_scanned, 12, @@ -253,7 +278,7 @@ async fn should_leave_every_watermark_alone_when_cancelled() { .expect("seed the advanced watermark"); let cancel = CancellationToken::new(); - let result = run( + let outcome = run( &store, &subwallets, &cancel, @@ -261,6 +286,7 @@ async fn should_leave_every_watermark_alone_when_cancelled() { ) .await; + assert_cancelled(outcome); let store = store.read().await; assert_eq!(store.tree_size().unwrap(), 8); assert_eq!(store.last_synced_note_index(lagging).unwrap(), 0); @@ -269,10 +295,6 @@ async fn should_leave_every_watermark_alone_when_cancelled() { 12, "a partial prefix below a subwallet's watermark must not rewind it" ); - assert!( - crate::changeset::merge::Merge::is_empty(&result.changeset), - "a cancelled pass sends nothing to the host" - ); let _ = std::fs::remove_file(&path); } @@ -294,7 +316,7 @@ async fn should_leave_store_and_changeset_empty_when_cancelled_before_first_batc .expect("seed the watermark"); let cancel = CancellationToken::new(); - let result = run( + let outcome = run( &store, &subwallets, &cancel, @@ -302,11 +324,7 @@ async fn should_leave_store_and_changeset_empty_when_cancelled_before_first_batc ) .await; - assert_eq!(result.total_scanned, 0); - assert!( - crate::changeset::merge::Merge::is_empty(&result.changeset), - "a pass that consumed nothing records nothing" - ); + assert_cancelled(outcome); let store = store.read().await; assert_eq!(store.tree_size().unwrap(), 0); assert_eq!(store.last_synced_note_index(sub).unwrap(), 4); @@ -327,7 +345,9 @@ async fn should_return_while_waiting_for_the_store_lock_when_cancelled() { let cancel = CancellationToken::new(); let held = store.write().await; - let pass = sync_notes_from_stream(&store, &subwallets, None, &cancel, |_, _| { + let epoch = StoreEpoch::default(); + let watch = epoch.watch(); + let pass = sync_notes_from_stream(&store, &watch, &subwallets, None, &cancel, |_, _| { futures::stream::iter(vec![batch(0, 4, 4)].into_iter().map(Ok)) }); futures::pin_mut!(pass); @@ -337,12 +357,11 @@ async fn should_return_while_waiting_for_the_store_lock_when_cancelled() { ); cancel.cancel(); - let result = tokio::time::timeout(std::time::Duration::from_secs(2), pass.as_mut()) + let outcome = tokio::time::timeout(std::time::Duration::from_secs(2), pass.as_mut()) .await .expect("a cancelled pass returns without the guard") .expect("pass must succeed"); - assert_eq!(result.total_scanned, 0); - assert!(crate::changeset::merge::Merge::is_empty(&result.changeset)); + assert_cancelled(outcome); drop(held); let store = store.read().await; 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 index 1a5486fc4ee..b229335e5a8 100644 --- 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 @@ -8,7 +8,7 @@ use drive_proof_verifier::types::{ use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; -use super::{sync_notes_across, MAX_CONCURRENT_CHUNK_FETCHES}; +use super::{sync_notes_across, NoteSyncOutcome, StoreEpoch, MAX_CONCURRENT_CHUNK_FETCHES}; use crate::wallet::shielded::keys::{AccountViewingKeys, OrchardKeySet}; use crate::wallet::shielded::store::{InMemoryShieldedStore, ShieldedStore, SubwalletId}; @@ -77,28 +77,62 @@ async fn assert_concurrent_sync_matches_sequential(initial_size: u8) { .collect(); let sequential = seeded_store(initial_size, subwallets[0].0); let concurrent = seeded_store(initial_size, subwallets[0].0); + let epoch = StoreEpoch::default(); let never = CancellationToken::new(); let baseline_sdk = pool_sdk(4).await; - sync_notes_across(&baseline_sdk, &sequential, &subwallets, None, None, &never) - .await - .unwrap(); + let baseline = sync_notes_across( + &baseline_sdk, + &sequential, + &epoch.watch(), + &subwallets, + None, + None, + &never, + ) + .await + .unwrap(); + assert!(matches!(baseline, NoteSyncOutcome::Completed(_))); 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, &never); - let second = sync_notes_across(&second_sdk, &concurrent, &subwallets, None, None, &never); + let (first_watch, second_watch) = (epoch.watch(), epoch.watch()); + let first = sync_notes_across( + &first_sdk, + &concurrent, + &first_watch, + &subwallets, + None, + None, + &never, + ); + let second = sync_notes_across( + &second_sdk, + &concurrent, + &second_watch, + &subwallets, + None, + None, + &never, + ); 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. + // Releasing the writer lets both passes snapshot the same tree size before + // either appends. Each append re-checks that size under the write lock, so + // whichever pass appends second finds the tree moved and is superseded + // instead of appending the same positions again. 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 first = first.expect("first concurrent sync"); + let second = second.expect("second concurrent sync"); + assert!( + matches!(first, NoteSyncOutcome::Completed(_)) + || matches!(second, NoteSyncOutcome::Completed(_)), + "one of the overlapping passes must commit" + ); { let expected = sequential.read().await; @@ -120,9 +154,18 @@ async fn assert_concurrent_sync_matches_sequential(initial_size: u8) { // 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, &never) - .await - .unwrap(); + let next = sync_notes_across( + &next_sdk, + &concurrent, + &epoch.watch(), + &subwallets, + None, + None, + &never, + ) + .await + .unwrap(); + assert!(matches!(next, NoteSyncOutcome::Completed(_))); let actual = concurrent.read().await; assert_eq!(actual.tree_size().unwrap(), 5); for (id, _) in &subwallets { diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync/lock_contention_tests.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync/lock_contention_tests.rs new file mode 100644 index 00000000000..0c795cdb3d2 --- /dev/null +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync/lock_contention_tests.rs @@ -0,0 +1,766 @@ +//! Store-lock behaviour of a sync pass: short wallet operations (sends, +//! reservations, activity writes) must not wait on the pass's network I/O, +//! and a pass whose snapshot a concurrent writer invalidated must abandon +//! its commit rather than write over that change. +//! +//! Every test drives `sync_notes_from_stream` with a hand-fed stream, so a +//! pass can be held "mid-download" for as long as the test needs. + +use std::collections::BTreeSet; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use dash_sdk::platform::shielded::notes_sync::types::{DecryptedNote, ShieldedChunkBatch}; +use drive_proof_verifier::types::ShieldedEncryptedNote; +use futures::channel::mpsc; +use futures::StreamExt; +use grovedb_commitment_tree::{ExtractedNoteCommitment, Note, NoteValue, RandomSeed, Rho}; +use rand::rngs::StdRng; +use rand::{RngCore, SeedableRng}; +use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +use super::{sync_notes_from_stream, NoteSyncOutcome, StoreEpoch}; +use crate::error::PlatformWalletError; +use crate::wallet::shielded::coordinator::ShieldedTreeProgressCallback; +use crate::wallet::shielded::file_store::FileBackedShieldedStore; +use crate::wallet::shielded::keys::OrchardKeySet; +use crate::wallet::shielded::store::{ShieldedNote, ShieldedStore, SubwalletId}; + +/// How long the fake network withholds the next batch. +const STREAM_STALL: Duration = Duration::from_millis(1500); +/// Bound on a store-lock wait while the pass is stalled on the network. +/// Before the fix the wait was the whole stall. +const LOCK_WAIT_BOUND: Duration = Duration::from_millis(250); + +/// Bound on taking the store lock while a pass is parked on the network, +/// where the lock should be free. A pass holding it across `stream.next()` +/// would otherwise deadlock the test feeding that stream instead of failing. +const MID_PASS_LOCK_BOUND: Duration = Duration::from_secs(5); + +type Store = Arc>; + +async fn mid_pass_write(store: &Store) -> RwLockWriteGuard<'_, FileBackedShieldedStore> { + tokio::time::timeout(MID_PASS_LOCK_BOUND, store.write()) + .await + .expect("sync held the store lock while waiting for the next batch") +} + +async fn mid_pass_read(store: &Store) -> RwLockReadGuard<'_, FileBackedShieldedStore> { + tokio::time::timeout(MID_PASS_LOCK_BOUND, store.read()) + .await + .expect("sync held the store lock while waiting for the next batch") +} + +type BatchSender = mpsc::UnboundedSender>; +type PassHandle = JoinHandle>; + +const WALLET: [u8; 32] = [0x5A; 32]; + +fn subwallet() -> SubwalletId { + SubwalletId::new(WALLET, 0) +} + +fn temp_tree_path(tag: &str) -> std::path::PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir().join(format!("sync_lock_{tag}_{nanos}.sqlite")) +} + +/// Removes the test's SQLite tree on drop, panics included. +struct TempTree(std::path::PathBuf); + +impl Drop for TempTree { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.0); + } +} + +/// A fresh file-backed store plus the epoch its passes watch. Keep the +/// `TempTree` alive for the whole test. +fn open_store( + tag: &str, +) -> ( + Arc>, + Arc, + TempTree, +) { + let path = temp_tree_path(tag); + let store = FileBackedShieldedStore::open_path(&path, 100).unwrap(); + ( + Arc::new(RwLock::new(store)), + Arc::new(StoreEpoch::default()), + TempTree(path), + ) +} + +/// A canonical (small) Pallas base-field element per tree position, so the +/// commitment is a valid `ExtractedNoteCommitment` and witnesses can be +/// turned into roots. +fn cmx(position: u64) -> [u8; 32] { + let mut c = [0u8; 32]; + c[..8].copy_from_slice(&(position + 1).to_le_bytes()); + c +} + +/// A batch of `len` undecryptable notes at `start_index..`, each carrying a +/// distinct nullifier, as the SDK stream would yield it. +fn batch(start_index: u64, len: u64) -> ShieldedChunkBatch { + ShieldedChunkBatch { + start_index, + notes: (start_index..start_index + len) + .map(|p| ShieldedEncryptedNote { + cmx: cmx(p).to_vec(), + nullifier: { + let mut nf = [0xEE; 32]; + nf[..8].copy_from_slice(&p.to_le_bytes()); + nf.to_vec() + }, + cv_net: Vec::new(), + encrypted_note: Vec::new(), + }) + .collect(), + decrypted: Vec::new(), + block_height: 7, + is_partial: false, + total_count: start_index + len, + } +} + +/// Start a pass over `store` fed by the returned sender. The returned +/// receiver yields the tree-progress numerator once per batch, right after +/// the batch's commitments are appended; on the current-thread test runtime +/// the pass then finishes the batch (trial decryption) and parks on the +/// "network" before the test task runs again. +fn start_pass( + store: &Arc>, + epoch: &Arc, +) -> (BatchSender, mpsc::UnboundedReceiver, PassHandle) { + start_cancellable_pass(store, epoch, CancellationToken::new()) +} + +/// [`start_pass`] under a caller-held cancellation token. +fn start_cancellable_pass( + store: &Arc>, + epoch: &Arc, + cancel: CancellationToken, +) -> (BatchSender, mpsc::UnboundedReceiver, PassHandle) { + let views = OrchardKeySet::from_seed(&[0x42; 64], dashcore::Network::Testnet, 0) + .unwrap() + .viewing_keys(); + let subwallets = vec![(subwallet(), views)]; + let (tx, rx) = mpsc::unbounded(); + let (progress_tx, progress_rx) = mpsc::unbounded(); + let on_tree: ShieldedTreeProgressCallback = Arc::new(move |committed, _| { + let _ = progress_tx.unbounded_send(committed); + }); + let store = Arc::clone(store); + let epoch = Arc::clone(epoch); + let pass = tokio::spawn(async move { + let watch = epoch.watch(); + sync_notes_from_stream( + &store, + &watch, + &subwallets, + Some(&on_tree), + &cancel, + move |_, _| rx, + ) + .await + }); + (tx, progress_rx, pass) +} + +fn expect_completed(outcome: NoteSyncOutcome) -> super::MultiSyncNotesResult { + match outcome { + NoteSyncOutcome::Completed(result) => result, + other => panic!("pass should have completed, got {other:?}"), + } +} + +fn expect_cancelled(outcome: NoteSyncOutcome) { + assert!( + matches!(outcome, NoteSyncOutcome::Cancelled), + "pass should have been cancelled, got {outcome:?}" + ); +} + +fn expect_superseded(outcome: NoteSyncOutcome) { + assert!( + matches!(outcome, NoteSyncOutcome::Superseded), + "pass should have been superseded, got {outcome:?}" + ); +} + +/// The latency regression: a send's store work — reserving notes, probing +/// the anchor/witnesses — must not wait for a sync pass's next network +/// batch. Before the fix the pass held the store write lock across +/// `stream.next().await`, so this wait equalled the whole network stall. +#[tokio::test] +async fn should_not_hold_store_lock_while_sync_waits_for_network() { + let (store, epoch, _tree) = open_store("contention"); + let (tx, mut progress, pass) = start_pass(&store, &epoch); + + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + + // The "network" now stalls before the next batch. + let release = tokio::spawn(async move { + tokio::time::sleep(STREAM_STALL).await; + tx.unbounded_send(Ok(batch(2048, 100))).unwrap(); + }); + + let started = Instant::now(); + { + // A send reserving its notes (operations.rs `reserve_notes`). + let mut guard = store.write().await; + let waited = started.elapsed(); + guard.mark_pending(subwallet(), &[1; 32]).unwrap(); + eprintln!("send: store write-lock wait while sync stalled on network: {waited:?}"); + assert!( + waited < LOCK_WAIT_BOUND, + "store write lock waited {waited:?} behind a sync pass stalled on the network" + ); + } + let started = Instant::now(); + { + // A send's anchor/witness probe (operations.rs + // `extract_spends_and_anchor`). + let guard = store.read().await; + let waited = started.elapsed(); + guard.tree_size().unwrap(); + eprintln!("send: store read-lock wait while sync stalled on network: {waited:?}"); + assert!(waited < LOCK_WAIT_BOUND, "read lock waited {waited:?}"); + } + + release.await.unwrap(); + assert_eq!(progress.next().await, Some(2148)); + drop(progress); + // `release` dropped the sender, ending the stream. + let result = expect_completed(pass.await.unwrap().unwrap()); + assert_eq!(result.total_scanned, 2148); + + let store = store.read().await; + assert_eq!(store.tree_size().unwrap(), 2148); + assert_eq!(store.last_synced_note_index(subwallet()).unwrap(), 2148); + assert!( + store.witness_at_depth(2147, 0).unwrap().is_some(), + "the pass checkpoints the tree at its final size on commit" + ); +} + +/// The worst-case store-lock wait a send sees while a pass ingests a large +/// catch-up batch: bounded by one `APPEND_SLICE` of appends, not by the +/// batch (let alone the download). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn should_bound_store_lock_hold_while_appending_a_large_batch() { + let (store, epoch, _tree) = open_store("append_slices"); + let (tx, mut progress, pass) = start_pass(&store, &epoch); + + // The prober queues for the store lock back to back from its own OS + // thread, so it is always holding the lock or waiting for it, whatever + // the async scheduler does. tokio's RwLock is fair: a prober waiting + // while the pass holds one slice's lock gets it before the pass's next + // slice, and records the partly appended tree it sees there. + let done = StopOnDrop(Arc::new(std::sync::atomic::AtomicBool::new(false))); + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let prober = { + let store = Arc::clone(&store); + let done = Arc::clone(&done.0); + std::thread::spawn(move || { + let mut ready = Some(ready_tx); + let mut probes = Vec::new(); + while !done.load(std::sync::atomic::Ordering::Acquire) { + let started = Instant::now(); + let guard = store.blocking_write(); + probes.push((started.elapsed(), guard.tree_size().unwrap())); + drop(guard); + if let Some(ready) = ready.take() { + let _ = ready.send(()); + } + } + probes + }) + }; + ready_rx + .recv_timeout(MID_PASS_LOCK_BOUND) + .expect("lock prober could not acquire the store before the first batch"); + + let batch_started = Instant::now(); + tx.unbounded_send(Ok(batch(0, 4 * 2048))).unwrap(); + assert_eq!(progress.next().await, Some(4 * 2048)); + let batch_elapsed = batch_started.elapsed(); + drop(done); + drop(tx); + expect_completed(pass.await.unwrap().unwrap()); + + let probes = prober.join().unwrap(); + let mid_ingest: BTreeSet = probes + .iter() + .map(|(_, size)| *size) + .filter(|size| (1..4 * 2048).contains(size)) + .collect(); + // The batch spans 32 append slices. A hold per batch would leave the + // tree empty or full at every probe; a hold per handful of slices would + // show only a few distinct sizes. + assert!( + mid_ingest.len() >= 8, + "probes saw only {mid_ingest:?} mid-ingest: {probes:?}" + ); + let max_wait = probes.iter().map(|(wait, _)| *wait).max().unwrap(); + eprintln!( + "8192-leaf batch ingested in {batch_elapsed:?}; {} lock probes, {} distinct \ + mid-ingest sizes, max wait {max_wait:?}", + probes.len(), + mid_ingest.len() + ); + assert!( + max_wait < batch_elapsed / 2, + "a probe waited {max_wait:?} of a {batch_elapsed:?} batch ingest" + ); +} + +/// Stops the prober thread when dropped, so a failed assertion doesn't +/// leave it contending for the store under later tests. +struct StopOnDrop(Arc); + +impl Drop for StopOnDrop { + fn drop(&mut self) { + self.0.store(true, std::sync::atomic::Ordering::Release); + } +} + +/// Appends a pass makes between lock holds are invisible to a send: the +/// spend path witnesses only at checkpoints, and shardtree computes a +/// checkpoint-depth witness as of that checkpoint's position. So a send +/// that probes mid-pass gets exactly the anchor it would have got before +/// the pass started, and after the pass commits that anchor is still +/// reachable one checkpoint deeper. +#[tokio::test] +async fn should_keep_spend_witnesses_stable_while_a_pass_is_mid_download() { + let (store, epoch, _tree) = open_store("witness_stable"); + let note_cmx = ExtractedNoteCommitment::from_bytes(&cmx(0)).unwrap(); + { + let mut guard = store.write().await; + for p in 0..3 { + guard.append_commitment(&cmx(p), true).unwrap(); + } + guard.checkpoint_tree(3).unwrap(); + } + let root_before = { + let guard = store.read().await; + guard + .witness_at_depth(0, 0) + .unwrap() + .unwrap() + .root(note_cmx) + }; + + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + { + let guard = mid_pass_read(&store).await; + assert_eq!(guard.tree_size().unwrap(), 2048, "batch appended mid-pass"); + let mid_pass = guard + .witness_at_depth(0, 0) + .unwrap() + .unwrap() + .root(note_cmx); + assert_eq!( + mid_pass.to_bytes(), + root_before.to_bytes(), + "uncheckpointed mid-pass appends must not move the depth-0 anchor" + ); + } + + drop(tx); + expect_completed(pass.await.unwrap().unwrap()); + let guard = store.read().await; + let depth0 = guard + .witness_at_depth(0, 0) + .unwrap() + .unwrap() + .root(note_cmx); + let depth1 = guard + .witness_at_depth(0, 1) + .unwrap() + .unwrap() + .root(note_cmx); + assert_ne!(depth0.to_bytes(), root_before.to_bytes()); + assert_eq!(depth1.to_bytes(), root_before.to_bytes()); +} + +/// A lifecycle change landing mid-download (here `unregister_wallet`'s +/// purge, which bumps the epoch under its store guard) supersedes the +/// pass: it appends nothing more and commits no watermark — which would +/// otherwise resurrect sync state for the wallet that was just removed. +#[tokio::test] +async fn should_supersede_pass_when_a_lifecycle_change_lands_mid_download() { + let (store, epoch, _tree) = open_store("superseded_purge"); + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + + { + let mut guard = mid_pass_write(&store).await; + epoch.bump_wallet(&mut guard, WALLET); + guard.purge_wallet(WALLET).unwrap(); + } + tx.unbounded_send(Ok(batch(2048, 100))).unwrap(); + drop(tx); + + expect_superseded(pass.await.unwrap().unwrap()); + let guard = store.read().await; + assert_eq!( + guard.tree_size().unwrap(), + 2048, + "no append after the change" + ); + assert_eq!( + guard.last_synced_note_index(subwallet()).unwrap(), + 0, + "the purged wallet's watermark must not be resurrected" + ); + assert!( + guard.witness_at_depth(0, 0).unwrap().is_none(), + "a superseded pass must not checkpoint" + ); +} + +/// Clear resets the tree mid-download; the pass must not commit (or append) +/// on top of the emptied tree even when the change lands after the last +/// batch, right before the final commit. +#[tokio::test] +async fn should_supersede_pass_when_tree_is_reset_before_commit() { + let (store, epoch, _tree) = open_store("superseded_clear"); + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + + { + let mut guard = mid_pass_write(&store).await; + epoch.bump_all(&mut guard); + guard.reset_commitment_tree().unwrap(); + guard.purge_all_subwallets().unwrap(); + } + drop(tx); + + expect_superseded(pass.await.unwrap().unwrap()); + let guard = store.read().await; + assert_eq!(guard.tree_size().unwrap(), 0); + assert_eq!(guard.last_synced_note_index(subwallet()).unwrap(), 0); +} + +/// Clear compacts the per-wallet and per-subwallet generations without +/// letting a watch from before it look current again, even once a wallet's +/// counter restarts from 0. +#[tokio::test] +async fn should_compact_generations_on_clear_without_reviving_older_watches() { + let (store, epoch, _tree) = open_store("compact_on_clear"); + let views = OrchardKeySet::from_seed(&[0x42; 64], dashcore::Network::Testnet, 0) + .unwrap() + .viewing_keys(); + let subwallets = vec![(subwallet(), views)]; + + let before_clear = epoch.watch(); + { + let mut guard = store.write().await; + epoch.bump_wallet(&mut guard, WALLET); + epoch.bump_purged(&mut guard, [subwallet()]); + epoch.bump_all(&mut guard); + } + { + let generations = epoch.lock(); + assert!(generations.wallets.is_empty() && generations.purges.is_empty()); + } + assert!(!before_clear.still_current(&subwallets)); + assert!(!before_clear.state_kept(&subwallet())); + + let after_clear = epoch.watch(); + assert!(after_clear.still_current(&subwallets)); + epoch.bump_wallet(&mut store.write().await, WALLET); + assert!(!after_clear.still_current(&subwallets)); + assert!( + !before_clear.still_current(&subwallets), + "a restarted counter must not match a watch from before the Clear" + ); +} + +/// Backstop for writers that don't bump the epoch: if the tree's leaf count +/// is not exactly what the pass itself appended, the pass stops before +/// appending at positions that are no longer the ones it fetched (which +/// would duplicate or misplace leaves and corrupt every later witness). +#[tokio::test] +async fn should_supersede_pass_when_another_writer_moves_the_tree() { + let (store, epoch, _tree) = open_store("superseded_foreign"); + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + + mid_pass_write(&store) + .await + .append_commitment(&cmx(9_999), true) + .unwrap(); + tx.unbounded_send(Ok(batch(2048, 100))).unwrap(); + drop(tx); + + expect_superseded(pass.await.unwrap().unwrap()); + let guard = store.read().await; + assert_eq!(guard.tree_size().unwrap(), 2049, "no append after the move"); + assert_eq!(guard.last_synced_note_index(subwallet()).unwrap(), 0); +} + +/// A change scoped to another wallet (its restore, removal, or account +/// purge) does not supersede a pass that isn't scanning that wallet. +#[tokio::test] +async fn should_not_supersede_pass_for_another_wallets_lifecycle_change() { + let (store, epoch, _tree) = open_store("unrelated_wallet"); + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + { + let mut guard = mid_pass_write(&store).await; + epoch.bump_wallet(&mut guard, [0x77; 32]); + guard.purge_wallet([0x77; 32]).unwrap(); + } + tx.unbounded_send(Ok(batch(2048, 100))).unwrap(); + drop(tx); + expect_completed(pass.await.unwrap().unwrap()); + assert_eq!( + store + .read() + .await + .last_synced_note_index(subwallet()) + .unwrap(), + 2148 + ); +} + +/// A superseded pass leaves only uncheckpointed leaves behind — the state an +/// interrupted pass always could. The retry resumes from the unchanged +/// watermark and gate-skips those leaves; even when the chain has not moved +/// (so the retry appends nothing itself) it must checkpoint them, or the +/// notes it saves there would be unwitnessable at spend time. +#[tokio::test] +async fn should_checkpoint_leaves_left_by_a_superseded_pass_on_retry() { + let (store, epoch, _tree) = open_store("retry"); + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + { + // E.g. a host snapshot restore of the wallet being scanned. + let mut guard = mid_pass_write(&store).await; + epoch.bump_wallet(&mut guard, WALLET); + } + drop(tx); + expect_superseded(pass.await.unwrap().unwrap()); + assert!( + store.read().await.witness_at_depth(0, 0).unwrap().is_none(), + "superseded pass left its leaves uncheckpointed" + ); + + // Retry over an unchanged chain: the same 2048 leaves, nothing new. + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + drop(tx); + assert_eq!( + progress.next().await, + Some(2048), + "re-fetched leaves skipped" + ); + expect_completed(pass.await.unwrap().unwrap()); + + let guard = store.read().await; + assert_eq!(guard.tree_size().unwrap(), 2048, "no leaf duplicated"); + assert_eq!(guard.last_synced_note_index(subwallet()).unwrap(), 2048); + let first = guard.witness_at_depth(0, 0).unwrap().expect("witnessable"); + let last = guard + .witness_at_depth(2047, 0) + .unwrap() + .expect("witnessable"); + let cmx_at = |p| ExtractedNoteCommitment::from_bytes(&cmx(p)).unwrap(); + assert_eq!( + first.root(cmx_at(0)).to_bytes(), + last.root(cmx_at(2047)).to_bytes(), + "every position witnesses against the same committed root" + ); + drop(guard); + + // A further caught-up pass re-checkpoints at the same size: a no-op. + let (tx, mut progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + drop(tx); + assert_eq!(progress.next().await, Some(2048)); + expect_completed(pass.await.unwrap().unwrap()); + let guard = store.read().await; + assert!( + guard.witness_at_depth(0, 1).unwrap().is_none(), + "an unchanged tree must not gain a second checkpoint" + ); +} + +/// A real Orchard note to `keys`' default address, plus its nullifier. +/// Deterministic per `value`. +fn owned_note(keys: &OrchardKeySet, value: u64) -> (Note, [u8; 32]) { + let mut rng = StdRng::seed_from_u64(value); + let rho = loop { + let mut b = [0u8; 32]; + rng.fill_bytes(&mut b); + if let Some(rho) = Rho::from_bytes(&b).into_option() { + break rho; + } + }; + let rseed = loop { + let mut b = [0u8; 32]; + rng.fill_bytes(&mut b); + if let Some(rseed) = RandomSeed::from_bytes(b, &rho).into_option() { + break rseed; + } + }; + let note = Note::from_parts(keys.address_at(0), NoteValue::from_raw(value), rho, rseed) + .into_option() + .expect("valid note parts"); + let nullifier = note.nullifier(&keys.full_viewing_key).to_bytes(); + (note, nullifier) +} + +/// A send can confirm a note spent while a pass is downloading. When the +/// pass re-scans that note (here: a restored notes-only snapshot left the +/// watermark at 0) from scan data older than the spend, committing the +/// receipt must keep it spent, not resurrect it as selectable balance. +#[tokio::test] +async fn should_keep_a_spend_confirmed_mid_download_when_rescanning_its_note() { + let (store, epoch, _tree) = open_store("rescan_spent"); + let keys = OrchardKeySet::from_seed(&[0x42; 64], dashcore::Network::Testnet, 0).unwrap(); + let (note, nullifier) = owned_note(&keys, 1_000); + let position = 5; + let note_cmx = ExtractedNoteCommitment::from(note.commitment()).to_bytes(); + store + .write() + .await + .save_note( + subwallet(), + &ShieldedNote { + note_data: super::serialize_note(¬e), + position, + cmx: note_cmx, + nullifier, + block_height: 7, + is_spent: false, + value: 1_000, + }, + ) + .unwrap(); + + let (tx, mut progress, pass) = start_pass(&store, &epoch); + let mut scanned = batch(0, 2048); + scanned.decrypted.push(DecryptedNote { + position, + note, + address: keys.address_at(0), + nullifier, + cmx: note_cmx, + }); + tx.unbounded_send(Ok(scanned)).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + + { + // The send's reservation and its confirmation (`mark_notes_spent`). + let mut guard = mid_pass_write(&store).await; + guard.mark_pending(subwallet(), &nullifier).unwrap(); + assert!(guard.mark_spent(subwallet(), &nullifier).unwrap()); + } + drop(tx); + let result = expect_completed(pass.await.unwrap().unwrap()); + + let guard = store.read().await; + let stored = guard.get_all_notes(subwallet()).unwrap(); + assert_eq!(stored.len(), 1); + assert!(stored[0].is_spent, "the confirmed spend was overwritten"); + assert!(guard.get_unspent_notes(subwallet()).unwrap().is_empty()); + assert_eq!(guard.spendable_balance(subwallet()).unwrap(), 0); + let emitted = &result.changeset.notes_saved[&subwallet()]; + assert!( + emitted + .iter() + .all(|n| n.nullifier != nullifier || n.is_spent), + "the changeset must not persist an unspent replacement" + ); +} + +/// A stop must not queue behind a writer holding the store: the pass returns +/// while the guard is still held, leaving its appended leaves for the next +/// completed pass to checkpoint. +#[tokio::test] +async fn should_return_promptly_when_cancelled_while_a_writer_holds_the_store() { + let (store, epoch, _tree) = open_store("cancel_held"); + let cancel = CancellationToken::new(); + let (tx, mut progress, pass) = start_cancellable_pass(&store, &epoch, cancel.clone()); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + + let held = mid_pass_write(&store).await; + tx.unbounded_send(Ok(batch(2048, 100))).unwrap(); + tokio::task::yield_now().await; + cancel.cancel(); + let outcome = tokio::time::timeout(MID_PASS_LOCK_BOUND, pass) + .await + .expect("a cancelled pass must not wait for the held store lock") + .unwrap() + .unwrap(); + expect_cancelled(outcome); + drop(held); + + { + let guard = store.read().await; + assert_eq!(guard.tree_size().unwrap(), 2048, "no append after the stop"); + assert_eq!(guard.last_synced_note_index(subwallet()).unwrap(), 0); + assert!( + guard.witness_at_depth(0, 0).unwrap().is_none(), + "the busy store left the leaves uncheckpointed" + ); + } + + let (tx, _progress, pass) = start_pass(&store, &epoch); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + tx.unbounded_send(Ok(batch(2048, 100))).unwrap(); + drop(tx); + expect_completed(pass.await.unwrap().unwrap()); + let guard = store.read().await; + assert_eq!(guard.tree_size().unwrap(), 2148, "no leaf duplicated"); + assert!( + guard.witness_at_depth(0, 0).unwrap().is_some(), + "the next completed pass checkpoints the stopped pass's leaves" + ); +} + +/// A stop after a lifecycle change invalidated the snapshot reports +/// `Cancelled` — the coordinator must not retry a stopped pass, nor report +/// it failed once retries run out — and checkpoints nothing on the stale +/// snapshot. +#[tokio::test] +async fn should_report_cancelled_when_stopped_after_a_supersession() { + let (store, epoch, _tree) = open_store("cancel_superseded"); + let cancel = CancellationToken::new(); + let (tx, mut progress, pass) = start_cancellable_pass(&store, &epoch, cancel.clone()); + tx.unbounded_send(Ok(batch(0, 2048))).unwrap(); + assert_eq!(progress.next().await, Some(2048)); + { + let mut guard = mid_pass_write(&store).await; + epoch.bump_wallet(&mut guard, WALLET); + } + cancel.cancel(); + + expect_cancelled(pass.await.unwrap().unwrap()); + drop(tx); + let guard = store.read().await; + assert_eq!(guard.last_synced_note_index(subwallet()).unwrap(), 0); + assert!( + guard.witness_at_depth(0, 0).unwrap().is_none(), + "a stale snapshot must not checkpoint" + ); +} diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentShieldedNote.swift b/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentShieldedNote.swift index 4016e7dfd92..180c89fff51 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentShieldedNote.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentShieldedNote.swift @@ -9,7 +9,8 @@ import SwiftData /// `(walletId, accountIndex, position)` and re-saves with the same /// nullifier overwrite the existing row in place — Orchard /// nullifiers are globally unique, so repeated discovery of the -/// same note (e.g. after a re-sync) shouldn't double-count. +/// same note (e.g. after a re-sync) shouldn't double-count. A +/// re-save never clears `isSpent` (see `persistShieldedNotes`). /// /// On cold start the matching `loadShieldedNotes` callback streams /// every row back to Rust so `ShieldedWallet::restore_from_snapshot` diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedLocalBalance.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedLocalBalance.swift index fdf8d615bdb..6d2fa547400 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedLocalBalance.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedLocalBalance.swift @@ -114,11 +114,11 @@ extension PlatformWalletManager { /// Reads one coherent local balance snapshot after shielded binding. /// This performs no network requests. Call it before starting network /// sync during launch. Native lock waits have a 100 ms deadline. A scan - /// holds the store write lock throughout its streamed download, which - /// may take minutes; reads can return `PlatformWalletError.walletOperation` - /// (busy) for that entire period. Repeated immediate reads do not bypass - /// the scan. Retain the last usable balance, allow network sync to continue, - /// and consume its completion event or request a fresh read after it ends. + /// does not hold the store lock across its network I/O, so reads normally + /// succeed while it runs; they can still return + /// `PlatformWalletError.walletOperation` (busy) if they contend with a + /// scan's tree appends or final commit, a restore, or Clear. On busy, + /// retain the last usable balance and request a fresh read later. /// /// Each invocation performs one native read. A successful bind overlapping /// delivery throws `ShieldedLocalBalanceReadError.bindingChanged`; the host diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift index f1d499682d7..dcb2b5985b7 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift @@ -5673,21 +5673,27 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable { /// Upsert a batch of decrypted shielded notes by `nullifier`. /// Re-saves with the same nullifier overwrite the existing - /// row in place — Orchard nullifiers are globally unique. - func persistShieldedNotes(walletId: Data, snapshots: [ShieldedNoteSnapshot]) { - onQueue { + /// row in place — Orchard nullifiers are globally unique — + /// except that a re-save never clears `isSpent` (see + /// `ShieldedChangeSet::notes_saved`): a sync pass can deliver a + /// note unspent after a concurrent send's round marked it spent. + /// The existing-row read is what keeps that flag, so a read that + /// fails throws and fails the round rather than inserting the + /// incoming copy over a row it couldn't see. + func persistShieldedNotes(walletId: Data, snapshots: [ShieldedNoteSnapshot]) throws { + try onQueue { for snap in snapshots { let nf = snap.nullifier let predicate = #Predicate { $0.nullifier == nf } var descriptor = FetchDescriptor(predicate: predicate) descriptor.fetchLimit = 1 - if let existing = try? backgroundContext.fetch(descriptor).first { + if let existing = try modelFetcher.fetch(descriptor, in: backgroundContext).first { existing.walletId = snap.walletId existing.accountIndex = snap.accountIndex existing.position = snap.position existing.cmx = snap.cmx existing.blockHeight = snap.blockHeight - existing.isSpent = snap.isSpent + existing.isSpent = existing.isSpent || snap.isSpent existing.value = snap.value existing.noteData = snap.noteData existing.lastUpdated = Date() @@ -10861,8 +10867,12 @@ private func persistShieldedNotesCallback( )) } } - handler.persistShieldedNotes(walletId: walletId, snapshots: snapshots) - return 0 + do { + try handler.persistShieldedNotes(walletId: walletId, snapshots: snapshots) + return 0 + } catch { + return -1 + } } private func persistShieldedNullifiersSpentCallback( diff --git a/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/ShieldedNoteSpentPersistenceTests.swift b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/ShieldedNoteSpentPersistenceTests.swift new file mode 100644 index 00000000000..316fd2d2678 --- /dev/null +++ b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/ShieldedNoteSpentPersistenceTests.swift @@ -0,0 +1,122 @@ +import SwiftData +import XCTest + +@testable import SwiftDashSDK + +/// A shielded sync pass can deliver a note unspent after a concurrent send's +/// round already marked it spent. The persister must keep the spent flag: +/// the Rust store never un-spends a note, so a durable row that did would +/// offer the spent note again after a restart. +@MainActor +final class ShieldedNoteSpentPersistenceTests: XCTestCase { + + private let walletId = Data(repeating: 0xD1, count: 32) + private let nullifier = Data(repeating: 0xD2, count: 32) + + private func saveNote(_ handler: PlatformWalletPersistenceHandler, isSpent: Bool) throws { + handler.beginChangeset(walletId: walletId) + try handler.persistShieldedNotes( + walletId: walletId, + snapshots: [ + .init( + walletId: walletId, + accountIndex: 0, + position: 8, + cmx: Data(repeating: 0xD3, count: 32), + nullifier: nullifier, + blockHeight: 51, + isSpent: isSpent, + value: 200_000, + noteData: Data(repeating: 0xD4, count: 115) + ) + ] + ) + _ = handler.endChangeset(walletId: walletId, success: true) + } + + private func storedNote(_ container: ModelContainer) throws -> PersistentShieldedNote { + let context = ModelContext(container) + let rows = try context.fetch(FetchDescriptor()) + XCTAssertEqual(rows.count, 1, "one row per nullifier") + return try XCTUnwrap(rows.first) + } + + func testResaveNeverClearsSpent() throws { + let container = try DashModelContainer.createInMemory() + let handler = PlatformWalletPersistenceHandler( + modelContainer: container, network: .testnet) + + try saveNote(handler, isSpent: false) + // A send's round marks the note spent. + handler.beginChangeset(walletId: walletId) + handler.persistShieldedNullifiersSpent( + walletId: walletId, + entries: [(walletId: walletId, accountIndex: 0, nullifier: nullifier)] + ) + _ = handler.endChangeset(walletId: walletId, success: true) + XCTAssertTrue(try storedNote(container).isSpent) + + // A sync pass's later round re-saves it unspent. + try saveNote(handler, isSpent: false) + XCTAssertTrue(try storedNote(container).isSpent, "a re-save must not un-spend a note") + } + + /// A send can confirm a note before the sync pass that found it has + /// stored it. The spend's round then inserts the note spent, and the + /// pass's later receipt must not clear it. + func testReceiptAfterTheSpendRoundKeepsTheNoteSpent() throws { + let container = try DashModelContainer.createInMemory() + let handler = PlatformWalletPersistenceHandler( + modelContainer: container, network: .testnet) + + try saveNote(handler, isSpent: true) + try saveNote(handler, isSpent: false) + XCTAssertTrue(try storedNote(container).isSpent) + } + + /// The existing-row read is what keeps a spent flag. If it fails, the + /// save must fail the round, not insert the stale copy over the row. + func testUnreadableRowFailsTheSaveInsteadOfInsertingItUnspent() throws { + let container = try DashModelContainer.createInMemory() + let handler = PlatformWalletPersistenceHandler( + modelContainer: container, network: .testnet) + try saveNote(handler, isSpent: true) + + let faulting = PlatformWalletPersistenceHandler( + modelContainer: container, + network: .testnet, + modelFetcher: FetchFaultInjector(faulting: PersistentShieldedNote.self) + ) + faulting.beginChangeset(walletId: walletId) + XCTAssertThrowsError(try faulting.persistShieldedNotes( + walletId: walletId, + snapshots: [ + .init( + walletId: walletId, + accountIndex: 0, + position: 8, + cmx: Data(repeating: 0xD3, count: 32), + nullifier: nullifier, + blockHeight: 51, + isSpent: false, + value: 200_000, + noteData: Data(repeating: 0xD4, count: 115) + ) + ] + )) + _ = faulting.endChangeset(walletId: walletId, success: false) + + XCTAssertTrue(try storedNote(container).isSpent) + } + + func testResaveStillMarksAnUnspentNoteSpent() throws { + let container = try DashModelContainer.createInMemory() + let handler = PlatformWalletPersistenceHandler( + modelContainer: container, network: .testnet) + + try saveNote(handler, isSpent: false) + XCTAssertFalse(try storedNote(container).isSpent) + try saveNote(handler, isSpent: true) + XCTAssertTrue(try storedNote(container).isSpent) + } +}