From fd2752e603aefa127eae0a701efbcd1384766b2c Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Mon, 31 Aug 2026 17:24:58 +0300 Subject: [PATCH 1/3] fix(platform-wallet): never drop a wallet event on the wallets-map lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The manager's `wallets` map was a `tokio::sync::RwLock`, and the two synchronous event handlers that resolve a wallet through it cannot await: `BalanceUpdateHandler` probed with `try_read()` and dropped the event's balance snapshot whenever a manager lifecycle write (create / remove / load) was in flight, and `SpendObservationHandler` carried a whole deferral queue to survive the same probe failing. The bus neither retries nor coalesces, so a dropped snapshot leaves superseded totals on screen until some later balance-bearing event happens to arrive, and nothing guarantees one does. Convert the map to `arc_swap::ArcSwap` (already this crate's idiom for rare-write / hot-read state): readers take a wait-free snapshot that can never fail or block, so the drop window no longer exists rather than being papered over. The rare lifecycle writers publish via `rcu`, preserving the generation-checked removal's check-and-remove atomicity, and the sync-context accessors that used `blocking_read()` become wait-free loads, removing their panic-inside-runtime hazard. With the read infallible, `SpendObservationHandler`'s pending queue loses its premise: there is no contention outcome left to defer, so the queue, its 4096-outpoint cap and the shedding warning go, and the handler applies every observation at delivery. Its regression test keeps `dashpay/platform#4309` pinned against the closest window the new type admits — a lifecycle writer parked mid-`rcu` across the delivery — as does the balance handler's own test, which asserts the snapshot lands before that writer commits. --- .../src/manager/accessors.rs | 36 ++-- .../src/manager/dashpay_sync.rs | 9 +- .../src/manager/dpns_sync.rs | 13 +- .../rs-platform-wallet/src/manager/load.rs | 19 +- .../rs-platform-wallet/src/manager/mod.rs | 121 ++++++++++-- .../src/manager/platform_address_sync.rs | 13 +- .../src/manager/wallet_lifecycle.rs | 54 ++++-- .../src/wallet/core/balance_handler.rs | 43 +++-- .../src/wallet/core/broadcast.rs | 92 +++++---- .../src/wallet/core/spend_observer.rs | 181 ++++-------------- 10 files changed, 297 insertions(+), 284 deletions(-) diff --git a/packages/rs-platform-wallet/src/manager/accessors.rs b/packages/rs-platform-wallet/src/manager/accessors.rs index af4c76dfc5f..dbb55e25513 100644 --- a/packages/rs-platform-wallet/src/manager/accessors.rs +++ b/packages/rs-platform-wallet/src/manager/accessors.rs @@ -374,20 +374,24 @@ impl PlatformWalletManager

{ /// Get a clone of a wallet by its ID. pub async fn get_wallet(&self, wallet_id: &WalletId) -> Option> { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); wallets.get(wallet_id).cloned() } - /// Blocking twin of [`Self::get_wallet`] for synchronous FFI entry - /// points that need to clone the `Arc` out before doing - /// network work outside the handle-storage guard. + /// Synchronous twin of [`Self::get_wallet`] for FFI entry points that + /// need to clone the `Arc` out before doing network work + /// outside the handle-storage guard. + /// + /// Named `_blocking` for the callers it serves, not for what it does: the + /// wallets map is an `ArcSwap`, so this load is wait-free and cannot block + /// or panic inside a runtime the way the previous `blocking_read` could. pub fn get_wallet_blocking(&self, wallet_id: &WalletId) -> Option> { - self.wallets.blocking_read().get(wallet_id).cloned() + self.wallets.load().get(wallet_id).cloned() } /// List all wallet IDs. pub async fn wallet_ids(&self) -> Vec { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); wallets.keys().copied().collect() } @@ -452,10 +456,9 @@ impl PlatformWalletManager

{ // ----------------------------------------------------------------- /// Atomic snapshot of every wallet id currently registered on the - /// manager. Cheap (`Arc` read + `BTreeMap` key clone). + /// manager. Cheap (wait-free `ArcSwap` load + `BTreeMap` key clone). pub fn list_wallet_ids_blocking(&self) -> Vec { - let wallets = self.wallets.blocking_read(); - wallets.keys().copied().collect() + self.wallets.load().keys().copied().collect() } /// Network a registered wallet belongs to, or `None` when the id is @@ -476,9 +479,7 @@ impl PlatformWalletManager

{ /// registered wallet participates in each pass since the sync /// manager doesn't keep a separate watch list. pub fn platform_address_sync_config_blocking(&self) -> PlatformAddressSyncConfigSnapshot { - let wallets = self.wallets.blocking_read(); - let count = wallets.len(); - drop(wallets); + let count = self.wallets.load().len(); let interval = self.platform_address_sync_manager.interval(); let last = self .platform_address_sync_manager @@ -641,9 +642,7 @@ impl PlatformWalletManager

{ &self, wallet_id: &WalletId, ) -> Option { - let wallets = self.wallets.blocking_read(); - let wallet = wallets.get(wallet_id)?.clone(); - drop(wallets); + let wallet = self.wallets.load().get(wallet_id)?.clone(); let provider_lock = wallet.platform().provider_for_diagnostics(); let guard = provider_lock.blocking_read(); let Some(provider) = guard.as_ref() else { @@ -1008,10 +1007,9 @@ impl PlatformWalletManager

{ // byte strings for the same G1 point — no collision). let mut operator_index: std::collections::HashMap<[u8; 48], u32> = std::collections::HashMap::new(); - // Clone the `Arc` out and drop the `wallets` read - // guard before deriving (the derive calls take the wallet's own - // state lock — don't hold `wallets` across them). - let platform_wallet = self.wallets.blocking_read().get(wallet_id).cloned(); + // Clone the `Arc` out of the map snapshot before + // deriving (the derive calls take the wallet's own state lock). + let platform_wallet = self.wallets.load().get(wallet_id).cloned(); if let Some(platform_wallet) = platform_wallet { use crate::wallet::provider_key_at_index::ProviderKeyKind; for index in 0..operator_scan_max { diff --git a/packages/rs-platform-wallet/src/manager/dashpay_sync.rs b/packages/rs-platform-wallet/src/manager/dashpay_sync.rs index 6f1f4340996..7c1b45e1d7e 100644 --- a/packages/rs-platform-wallet/src/manager/dashpay_sync.rs +++ b/packages/rs-platform-wallet/src/manager/dashpay_sync.rs @@ -52,8 +52,7 @@ use std::sync::{ }; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use tokio::sync::RwLock; - +use arc_swap::ArcSwap; use dash_async::{ThreadRegistry, WorkerConfig}; use crate::error::PlatformWalletError; @@ -132,7 +131,7 @@ impl DashPaySyncSummary { /// without any re-registration — and crucially without consulting the /// token registry, so DashPay-only identities are never skipped. pub struct DashPaySyncManager { - wallets: Arc>>>, + wallets: Arc>>>, /// Shared registry that owns this loop's lifecycle: it spawns the /// OS thread (with the deep-stack config below), owns its cancellation /// token, and joins it at shutdown. A generation-guarded slot handles a @@ -154,7 +153,7 @@ pub struct DashPaySyncManager { impl DashPaySyncManager { pub fn new( - wallets: Arc>>>, + wallets: Arc>>>, registry: Arc>, ) -> Self { Self { @@ -364,7 +363,7 @@ impl DashPaySyncManager { } let snapshot: Vec<(WalletId, Arc)> = { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); wallets.iter().map(|(id, w)| (*id, Arc::clone(w))).collect() }; diff --git a/packages/rs-platform-wallet/src/manager/dpns_sync.rs b/packages/rs-platform-wallet/src/manager/dpns_sync.rs index ef376031921..6344c425c53 100644 --- a/packages/rs-platform-wallet/src/manager/dpns_sync.rs +++ b/packages/rs-platform-wallet/src/manager/dpns_sync.rs @@ -9,8 +9,8 @@ //! //! **Wallet-driven, not registry-driven — by design.** A sibling of //! [`DashPaySyncManager`](super::dashpay_sync::DashPaySyncManager): it -//! holds the same `wallets` map, snapshots the wallet `Arc`s under a -//! read guard each sweep, and refreshes **every** wallet. It is a +//! holds the same `wallets` map, snapshots the wallet `Arc`s from its +//! wait-free map each sweep, and refreshes **every** wallet. It is a //! separate coordinator (not a seventh DashPay step) because the DashPay //! pass is contact/profile-scoped and runs at a 15s cadence, while //! marketplace state changes are rare — this loop defaults to 60s. @@ -43,8 +43,7 @@ use std::sync::{ }; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use tokio::sync::RwLock; - +use arc_swap::ArcSwap; use dash_async::{ThreadRegistry, WorkerConfig}; use crate::events::PlatformEventManager; @@ -129,7 +128,7 @@ impl DpnsSyncPassSummary { /// [`DashPaySyncManager`](super::dashpay_sync::DashPaySyncManager) /// verbatim. pub struct DpnsSyncManager { - wallets: Arc>>>, + wallets: Arc>>>, registry: Arc>, /// Dispatches `on_dpns_marketplace_sync_completed` after each pass. events: Arc, @@ -144,7 +143,7 @@ pub struct DpnsSyncManager { impl DpnsSyncManager { pub fn new( - wallets: Arc>>>, + wallets: Arc>>>, registry: Arc>, events: Arc, ) -> Self { @@ -289,7 +288,7 @@ impl DpnsSyncManager { } let snapshot: Vec<(WalletId, Arc)> = { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); wallets.iter().map(|(id, w)| (*id, Arc::clone(w))).collect() }; diff --git a/packages/rs-platform-wallet/src/manager/load.rs b/packages/rs-platform-wallet/src/manager/load.rs index 3588aef2b54..7c8f445f05c 100644 --- a/packages/rs-platform-wallet/src/manager/load.rs +++ b/packages/rs-platform-wallet/src/manager/load.rs @@ -208,9 +208,11 @@ impl PlatformWalletManager

{ } let platform_wallet = Arc::new(platform_wallet); - let mut wallets_guard = self.wallets.write().await; - wallets_guard.insert(wallet_id, platform_wallet); - drop(wallets_guard); + self.wallets.rcu(|wallets| { + let mut wallets = std::collections::BTreeMap::clone(wallets); + wallets.insert(wallet_id, Arc::clone(&platform_wallet)); + wallets + }); inserted_in_wallets.push(wallet_id); } @@ -220,10 +222,13 @@ impl PlatformWalletManager

{ // remove from `self.wallets` first (UI surface), then // from the inner `wallet_manager`. if !inserted_in_wallets.is_empty() { - let mut wallets_guard = self.wallets.write().await; - for id in &inserted_in_wallets { - wallets_guard.remove(id); - } + self.wallets.rcu(|wallets| { + let mut wallets = std::collections::BTreeMap::clone(wallets); + for id in &inserted_in_wallets { + wallets.remove(id); + } + wallets + }); } if !inserted_in_manager.is_empty() { let mut wm = self.wallet_manager.write().await; diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index 9192dfae148..a444035cc74 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/packages/rs-platform-wallet/src/manager/mod.rs @@ -332,7 +332,17 @@ pub struct PlatformWalletManager { /// update their lock-free balance atomics from event-handler /// context, without touching the SPV-contended `wallet_manager` /// lock. - pub(super) wallets: Arc>>>, + /// + /// An [`arc_swap::ArcSwap`] rather than a lock: readers take a + /// wait-free snapshot that can never fail or block, which the + /// balance handler depends on — the event bus neither retries nor + /// coalesces, so a snapshot dropped during a lifecycle write would + /// be lost for good (see `BalanceUpdateHandler`). Writers are the + /// rare manager lifecycle paths (create/remove/load) and publish + /// via `rcu`, whose closure must stay pure map manipulation — it + /// can run more than once under a concurrent-writer retry. + pub(super) wallets: + Arc>>>, /// Notified on InstantLock / ChainLock events for `AssetLockManager` waiters. pub(super) lock_notify: Arc, pub(super) spv_manager: Arc, @@ -471,7 +481,9 @@ impl PlatformWalletManager

{ .take_persistence_receiver() .expect("persistence receiver is available exactly once on a fresh WalletManager"); let wallet_manager = Arc::new(RwLock::new(wallet_manager_inner)); - let wallets = Arc::new(RwLock::new(std::collections::BTreeMap::new())); + let wallets = Arc::new(arc_swap::ArcSwap::from_pointee( + std::collections::BTreeMap::new(), + )); let lock_notify = Arc::new(Notify::new()); // Shared registry that owns the coordinators' loop-thread join // handles for a clean, panic-aware shutdown join. @@ -495,10 +507,11 @@ impl PlatformWalletManager

{ // Build handler list: app handler + internal handlers. // BalanceUpdateHandler holds a clone of the wallets map (a - // separate lock from wallet_manager) so it can look up - // PlatformWallets and write to their lock-free balance - // atomics from broadcast-handler context without contending - // with SPV's write lock. + // wait-free `ArcSwap`, separate from the wallet_manager lock) + // so it can look up PlatformWallets and write to their + // lock-free balance atomics from broadcast-handler context + // without contending with SPV's write lock — and without any + // window in which a lifecycle write could make the lookup fail. let lock_handler = Arc::new(LockNotifyHandler::new(Arc::clone(&lock_notify))); let balance_handler = Arc::new(BalanceUpdateHandler::new(Arc::clone(&wallets))); // SpendObservationHandler releases in-broadcast input fences when the @@ -843,14 +856,10 @@ impl PlatformWalletManager

{ )); }; - // Snapshot Arc clones under a short read lock; never hold the - // `wallets` read guard across the per-wallet `.await`s below — - // that would block registration and invite lock-ordering - // issues against each wallet's `wallet_manager` lock. - let wallets: Vec> = { - let guard = self.wallets.read().await; - guard.values().cloned().collect() - }; + // Snapshot Arc clones from the wait-free map; clone out rather + // than holding the `ArcSwap` guard across the per-wallet + // `.await`s below. + let wallets: Vec> = self.wallets.load().values().cloned().collect(); for wallet in wallets { wallet.platform().reset_sync_state().await; @@ -1112,7 +1121,11 @@ mod tests { Arc::new(NoopPersister) as Arc, Arc::new(crate::broadcaster::SpvBroadcaster::new(spv)), )); - mgr.wallets.write().await.insert(wallet_id, wallet); + mgr.wallets.rcu(|wallets| { + let mut next = std::collections::BTreeMap::clone(wallets); + next.insert(wallet_id, Arc::clone(&wallet)); + next + }); // Fence an outpoint the way a dispatch does: pin, then settle into the // pending-spend phase that only an observed spend may end. @@ -1355,4 +1368,82 @@ mod tests { "guard must clear the slot during unwind" ); } + + /// A balance snapshot delivered while a lifecycle write to the + /// `wallets` map is in flight must still land in the wallet's + /// lock-free balance atomics. The event bus neither retries nor + /// coalesces, so a snapshot dropped here is gone for good: the + /// wallet keeps displaying the superseded totals until some later + /// event happens to carry a fresh balance, and nothing guarantees + /// one arrives. + /// + /// When the map was a `tokio::sync::RwLock` and the handler used + /// `try_read()`, this exact delivery-under-contention scenario + /// dropped the snapshot (the pre-fix form of this test held + /// `wallets.write()` across the delivery and failed). With the map + /// an `ArcSwap`, the closest reachable window is a lifecycle writer + /// parked mid-`rcu`; the handler's `load()` must observe a committed + /// map and apply the balance immediately, before that writer + /// completes. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn balance_snapshot_survives_wallets_map_write_contention() { + use std::collections::BTreeMap; + + use crate::test_support::test_platform_wallet_manager; + use crate::wallet::core::BalanceUpdateHandler; + use key_wallet::wallet::balance::WalletCoreBalance; + + let (manager, wallet_id) = test_platform_wallet_manager().await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("fixture wallet is registered"); + + // The production unit under test, holding the same map the + // manager registers at construction. + let handler = BalanceUpdateHandler::new(Arc::clone(&manager.wallets)); + + // Park a lifecycle writer mid-publication: its `rcu` closure has + // read the current map but not yet committed the replacement. + // This pins open the window in which the old lock-based map + // made `try_read()` fail and lose the event. + let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let wallets_for_writer = Arc::clone(&manager.wallets); + let writer = std::thread::spawn(move || { + wallets_for_writer.rcu(|current| { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + Arc::clone(current) + }); + }); + entered_rx + .recv() + .expect("the writer must reach its rcu closure"); + + // Deliver the balance-bearing event while the write is in flight. + let corrected = WalletCoreBalance::new(1_234, 0, 0, 0); + handler.on_wallet_event(&crate::events::WalletEvent::BlockProcessed { + wallet_id, + height: 1_000, + chain_lock: None, + inserted: vec![], + updated: vec![], + matured: vec![], + balance: corrected, + account_balances: BTreeMap::new(), + addresses_derived: vec![], + }); + + // Observable immediately — before the lifecycle writer commits. + assert_eq!( + wallet.balance().confirmed(), + corrected.confirmed(), + "the balance snapshot was dropped: a lifecycle write to the wallets map \ + was in flight during delivery, and the bus will not re-deliver it" + ); + + release_tx.send(()).expect("writer still parked"); + writer.join().expect("writer thread completes"); + } } diff --git a/packages/rs-platform-wallet/src/manager/platform_address_sync.rs b/packages/rs-platform-wallet/src/manager/platform_address_sync.rs index f6e971d891e..f8e7cefb127 100644 --- a/packages/rs-platform-wallet/src/manager/platform_address_sync.rs +++ b/packages/rs-platform-wallet/src/manager/platform_address_sync.rs @@ -15,12 +15,11 @@ use std::sync::{ }; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use arc_swap::ArcSwapOption; +use arc_swap::{ArcSwap, ArcSwapOption}; use dash_sdk::platform::address_sync::{AddressSyncConfig, AddressSyncResult}; use key_wallet::PlatformP2PKHAddress; use crate::wallet::PlatformAddressTag; -use tokio::sync::RwLock; use dash_async::ThreadRegistry; @@ -98,7 +97,7 @@ impl PlatformAddressSyncSummary { /// `sync_now` again returns an empty summary immediately (the caller can /// check `is_syncing()` to distinguish). pub struct PlatformAddressSyncManager { - wallets: Arc>>>, + wallets: Arc>>>, event_manager: Arc, /// Shared registry that owns this loop's lifecycle: it spawns the /// OS thread, owns its cancellation token, and joins it at shutdown. @@ -127,7 +126,7 @@ pub struct PlatformAddressSyncManager { impl PlatformAddressSyncManager { pub fn new( - wallets: Arc>>>, + wallets: Arc>>>, event_manager: Arc, registry: Arc>, ) -> Self { @@ -357,7 +356,7 @@ impl PlatformAddressSyncManager { } let snapshot: Vec<(WalletId, Arc)> = { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); wallets.iter().map(|(id, w)| (*id, Arc::clone(w))).collect() }; @@ -454,7 +453,7 @@ impl PlatformAddressSyncManager { } let wallet = { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); wallets.get(wallet_id).cloned() }; let wallet = @@ -514,7 +513,7 @@ mod tests { /// but still drives the full flag → gate → completion-event protocol /// we're testing here. fn make_manager() -> (Arc, Arc) { - let wallets = Arc::new(RwLock::new(BTreeMap::new())); + let wallets = Arc::new(ArcSwap::from_pointee(BTreeMap::new())); let counter = Arc::new(CompletionCounter::new()); let event_manager = Arc::new(PlatformEventManager::new(vec![ Arc::clone(&counter) as Arc diff --git a/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs b/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs index 71bef57d723..b11c1fefdf6 100644 --- a/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs +++ b/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs @@ -58,11 +58,10 @@ fn parse_mnemonic_any_language(phrase: &str) -> Result { /// publish a NEW generation into both maps — the id is free in the inner /// manager from the moment the removal above completes, and nothing gates /// registration. Reproducing it deterministically from outside is not possible: -/// the window is bounded by two *different* locks, and the only lock a test -/// could hold to park the remover inside it (`self.wallets`) is the same lock -/// the registration must acquire to publish, so parking the remover would also -/// block the registration — and `tokio`'s `RwLock` hands the writer queue out -/// in FIFO order, which puts the remover first. A rendezvous is therefore the +/// the window is bounded by two *different* synchronization domains — the +/// inner manager's lock and the public map's `ArcSwap` publication — and a +/// test holds no lock that could park the remover between them without also +/// stalling the registration's own publish. A rendezvous is therefore the /// only way to pin this ordering without a sleep or a completion-order race. /// /// Compiled under `cfg(test)` only: neither this static nor its call site @@ -579,11 +578,14 @@ impl PlatformWalletManager

{ let platform_wallet = Arc::new(platform_wallet); - // Register the PlatformWallet handle. - { - let mut wallets = self.wallets.write().await; + // Register the PlatformWallet handle. `rcu` publishes a new map + // snapshot; the closure can rerun under a concurrent-writer + // retry, so it must stay pure map manipulation. + self.wallets.rcu(|wallets| { + let mut wallets = std::collections::BTreeMap::clone(wallets); wallets.insert(wallet_id, Arc::clone(&platform_wallet)); - } + wallets + }); // Re-seed the lock-free balance atomic from the wallet's inner // balance now that the wallet is in `self.wallets`. @@ -695,7 +697,7 @@ impl PlatformWalletManager

{ /// So once this method has removed generation G1 from the inner /// `wallet_manager`, the id is free and a concurrent registration can publish /// a *different* generation G2 into both maps before this method reaches its - /// own `self.wallets` removal — the two removals are separately locked, with + /// own `self.wallets` removal — the two removals are separately synchronized, with /// no happens-before edge between them and the registration. Removing by key /// there would take G2 out of the public map (leaving it registered in the /// inner manager, invisible and unremovable) and hand G2 to `tear_down`, @@ -736,7 +738,7 @@ impl PlatformWalletManager

{ // matched against. let (removed, _teardown) = loop { let candidate = { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); match wallets.get(wallet_id) { None => { return Err(PlatformWalletError::WalletNotFound(hex::encode(wallet_id))) @@ -746,7 +748,7 @@ impl PlatformWalletManager

{ }; let guard = candidate.generation().teardown_guard().await; let still_current = { - let wallets = self.wallets.read().await; + let wallets = self.wallets.load(); wallets .get(wallet_id) .is_some_and(|wallet| Arc::ptr_eq(wallet.generation(), candidate.generation())) @@ -877,13 +879,27 @@ impl PlatformWalletManager

{ // this method); removing by key would evict that live wallet and hand it // to `tear_down` under the wrong gate. { - let mut wallets = self.wallets.write().await; - let entry_is_ours = wallets - .get(wallet_id) - .is_some_and(|wallet| Arc::ptr_eq(wallet.generation(), &generation)); - if entry_is_ours { - wallets.remove(wallet_id); - } else { + // `rcu` keeps the generation check and the removal atomic: the + // closure sees the map the CAS will replace, and a concurrent + // publication retries the whole closure against the new map. + // The `Cell` therefore ends up holding the verdict of the + // attempt that actually committed. + let entry_is_ours = std::cell::Cell::new(false); + self.wallets.rcu(|wallets| { + let ours = wallets + .get(wallet_id) + .is_some_and(|wallet| Arc::ptr_eq(wallet.generation(), &generation)); + entry_is_ours.set(ours); + if ours { + let mut next = std::collections::BTreeMap::clone(wallets); + next.remove(wallet_id); + Arc::new(next) + } else { + // Not ours: publish the map unchanged. + Arc::clone(wallets) + } + }); + if !entry_is_ours.get() { tracing::warn!( wallet_id = %hex::encode(wallet_id), "remove_wallet: a new generation was registered under this id while the \ diff --git a/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs b/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs index 27797ec92e0..db655bda2d2 100644 --- a/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs +++ b/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs @@ -4,8 +4,8 @@ use std::collections::BTreeMap; use std::sync::Arc; +use arc_swap::ArcSwap; use dash_spv::EventHandler; -use tokio::sync::RwLock; use crate::events::{PlatformEventHandler, WalletEvent}; use crate::wallet::platform_wallet::WalletId; @@ -23,18 +23,27 @@ use crate::wallet::PlatformWallet; /// /// Registered in `PlatformWalletManager`'s handler list. The handler /// holds an `Arc` clone of the manager's `wallets` map (a *separate* -/// lock from the heavily-contended `wallet_manager` SPV write lock). -/// SPV holds the wallet-manager write lock for the entire duration of -/// block processing — looking the balance up through *that* lock would -/// silently lose every event during initial sync. The wallets map is -/// only written by manager lifecycle methods (`create_wallet_from_*`, -/// `remove_wallet`), so a `try_read()` here essentially never contends. +/// structure from the heavily-contended `wallet_manager` SPV write +/// lock). SPV holds the wallet-manager write lock for the entire +/// duration of block processing — looking the balance up through *that* +/// lock would silently lose every event during initial sync. +/// +/// The map is an [`ArcSwap`] so this lookup is wait-free and can never +/// fail: `load()` always returns the latest published map, even while a +/// manager lifecycle write (wallet insert / remove / load) is publishing +/// a new one. That infallibility is load-bearing, not a convenience. +/// `on_wallet_event` is synchronous and the bus neither retries nor +/// coalesces, so a snapshot missed here is gone for good: nothing +/// guarantees a later event carries the same correction, and until one +/// does the wallet displays superseded totals. A fallible lookup (the +/// previous `RwLock::try_read`) dropped exactly that snapshot whenever +/// it raced a lifecycle write. pub struct BalanceUpdateHandler { - wallets: Arc>>>, + wallets: Arc>>>, } impl BalanceUpdateHandler { - pub fn new(wallets: Arc>>>) -> Self { + pub fn new(wallets: Arc>>>) -> Self { Self { wallets } } } @@ -60,16 +69,12 @@ impl EventHandler for BalanceUpdateHandler { WalletEvent::ChainLockProcessed { .. } => return, }; - // try_read on the wallets map (NOT the wallet_manager - // SPV-contended lock). The map is only written by manager - // lifecycle methods, so this almost never contends. - let Ok(wallets) = self.wallets.try_read() else { - tracing::debug!( - wallet = %hex::encode(wallet_id), - "Wallet balance update dropped: wallets-map lock contended" - ); - return; - }; + // Wait-free snapshot of the wallets map; cannot fail or block, + // so no balance-bearing event is ever dropped here. A wallet + // not in the snapshot is one registered concurrently with this + // event — its creation path re-seeds the balance atomics from + // the inner wallet after publishing it, covering that window. + let wallets = self.wallets.load(); if let Some(pw) = wallets.get(wallet_id) { pw.balance().set( balance.confirmed(), diff --git a/packages/rs-platform-wallet/src/wallet/core/broadcast.rs b/packages/rs-platform-wallet/src/wallet/core/broadcast.rs index 50a7d2f5fe8..93512c50faa 100644 --- a/packages/rs-platform-wallet/src/wallet/core/broadcast.rs +++ b/packages/rs-platform-wallet/src/wallet/core/broadcast.rs @@ -824,15 +824,15 @@ mod tests { } /// A wallets map — the production `BTreeMap>` - /// behind its own `RwLock` — holding one entry per fixture wallet. + /// behind its own `ArcSwap` — holding one entry per fixture wallet. fn wallets_map( cores: &[&CoreWallet], ) -> Arc< - tokio::sync::RwLock< + arc_swap::ArcSwap< std::collections::BTreeMap>, >, > { - Arc::new(tokio::sync::RwLock::new( + Arc::new(arc_swap::ArcSwap::from_pointee( cores .iter() .map(|core| (core.wallet_id(), platform_wallet_sharing(core))) @@ -1141,26 +1141,27 @@ mod tests { core.abandon_transaction(&after).await; } - /// `dashpay/platform#4309` — A CONTENDED WALLETS MAP MUST DEFER A SPEND - /// OBSERVATION, NEVER DISCARD IT. + /// `dashpay/platform#4309` — A WALLETS-MAP WRITE IN FLIGHT MUST NOT COST + /// A SPEND OBSERVATION. /// - /// `SpendObservationHandler::on_wallet_event` is synchronous, so it probes - /// the wallets map with `try_read`. That probe fails while wallet - /// registration/removal holds the map's write lock — and the handler used - /// to DROP the observation on that failure. For a DAPI-path dispatch the - /// dropped `TransactionDetected` can be the only spend-bearing event the - /// wallet ever gets: InstantLock promotions carry no record here by - /// design, and an evicted or never-confirmed transaction produces no - /// inserted `BlockProcessed` record. With no deadline behind the - /// pending-spend fence, one moment of lock contention then left the input - /// fenced for the manager's lifetime even though the wallet HAD observed - /// it spent. + /// `SpendObservationHandler::on_wallet_event` is synchronous, so while the + /// wallets map was a `tokio::sync::RwLock` it could only probe with + /// `try_read` — a probe that fails while wallet registration/removal holds + /// the write lock. For a DAPI-path dispatch the observation lost that way + /// can be the only spend-bearing event the wallet ever gets: InstantLock + /// promotions carry no record here by design, and an evicted or + /// never-confirmed transaction produces no inserted `BlockProcessed` + /// record. With no deadline behind the pending-spend fence, one moment of + /// lock contention left the input fenced for the manager's lifetime even + /// though the wallet HAD observed it spent. /// - /// The handler now queues the observation and applies it on the next - /// delivered event — ANY event, including one carrying no spend at all — - /// so the evidence survives the contention instead of vanishing with it. + /// The map is now an `ArcSwap`, so the read cannot fail and the handler + /// applies every observation at delivery — no deferral queue, no window + /// to lose it in. The closest reachable analogue of the old contention is + /// a lifecycle writer parked mid-`rcu`, which this test pins open across + /// the delivery: the fence must clear anyway, before that writer commits. #[tokio::test] - async fn a_contended_wallets_map_defers_but_never_drops_a_spend_observation() { + async fn a_wallets_map_write_in_flight_does_not_cost_a_spend_observation() { let (core, signer, outputs) = funded_core_wallet( StandardAccountType::BIP44Account, Arc::new(AlwaysOkBroadcaster), @@ -1189,40 +1190,37 @@ mod tests { let map = wallets_map(&[&core]); let handler = SpendObservationHandler::new(Arc::clone(&map)); - // Wallet registration/removal holds the wallets-map WRITE lock at the - // instant the wallet's own spend event is delivered: `try_read` fails. - let contended = map.write().await; - dash_spv::EventHandler::on_wallet_event(&handler, &spend_event(&core, &sent_tx)); - drop(contended); - - // The observation was deferred, not applied: the fence still stands. - // (Pre-fix this holds too — by discarding rather than deferring — so - // the discriminating half is what the NEXT event does.) - expect_mid_broadcast( - try_finalize_tx(&core, AccountTypePreference::BIP44, &outputs, &signer).await, - "a deferred observation must not have been applied under contention", - ); + // Park a lifecycle writer mid-publication: its `rcu` closure has read + // the current map but not yet committed the replacement. This is the + // window the lock-based map failed its `try_read` in. + let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let map_for_writer = Arc::clone(&map); + let writer = std::thread::spawn(move || { + map_for_writer.rcu(|current| { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + Arc::clone(current) + }); + }); + entered_rx + .recv() + .expect("the writer must reach its rcu closure"); - // The next delivered event drains the queue — even a bare watermark - // advance that resolves no wallet and carries no spend of its own. - // This does NOT retire the fence by chain progress: it applies - // evidence that already arrived and was queued. - dash_spv::EventHandler::on_wallet_event( - &handler, - &key_wallet_manager::WalletEvent::SyncHeightAdvanced { - wallet_id: core.wallet_id(), - height: stamped + 17_001, - }, - ); + dash_spv::EventHandler::on_wallet_event(&handler, &spend_event(&core, &sent_tx)); + // Applied at delivery — before the parked writer commits. let after = try_finalize_tx(&core, AccountTypePreference::BIP44, &outputs, &signer).await; let after = after.unwrap_or_else(|error| { panic!( - "an observation deferred by wallets-map contention must be \ - applied on the next event, not dropped — the fence never \ - cleared: {error:?}" + "an observation delivered while a wallets-map write was in \ + flight must still release the fence: {error:?}" ) }); + + release_tx.send(()).expect("writer still parked"); + writer.join().expect("writer thread completes"); + core.abandon_transaction(&after).await; } diff --git a/packages/rs-platform-wallet/src/wallet/core/spend_observer.rs b/packages/rs-platform-wallet/src/wallet/core/spend_observer.rs index 47fa6420d15..ba9151db56a 100644 --- a/packages/rs-platform-wallet/src/wallet/core/spend_observer.rs +++ b/packages/rs-platform-wallet/src/wallet/core/spend_observer.rs @@ -7,12 +7,12 @@ //! network; this side takes it down when the wallet can actually see that the //! outpoints are spent. -use std::collections::{BTreeMap, BTreeSet}; -use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use std::collections::BTreeMap; +use std::sync::Arc; +use arc_swap::ArcSwap; use dash_spv::EventHandler; use dashcore::OutPoint; -use tokio::sync::RwLock; use crate::changeset::core_bridge::spent_outpoints; use crate::events::{PlatformEventHandler, WalletEvent}; @@ -70,158 +70,61 @@ use crate::wallet::PlatformWallet; /// processing, which holds the wallet-manager WRITE lock for the whole batch. /// Resolving the generation through *that* lock would deadlock or silently drop /// every event during initial sync, so this handler holds an `Arc` clone of the -/// manager's `wallets` map instead — a separate lock, written only by manager -/// lifecycle methods, so `try_read` essentially never contends. Releasing the -/// fence then takes only the generation's `in_broadcast` `std::sync::Mutex` for -/// a few hash operations and never awaits. +/// manager's `wallets` map instead. That map is an [`ArcSwap`], so the lookup +/// is wait-free and INFALLIBLE: a manager lifecycle write (wallet insert / +/// remove / load) publishes a new map without ever making a reader fail or +/// block. Releasing the fence then takes only the generation's `in_broadcast` +/// `std::sync::Mutex` for a few hash operations and never awaits. /// -/// # A contended map DEFERS the observation; it never discards it +/// The infallibility is what retires the deferral this handler used to need. +/// While the map was a `tokio::sync::RwLock`, a `try_read` losing to a +/// lifecycle writer had to queue the observation for the next delivered +/// event — dropping it was unacceptable, since `TransactionDetected` can be +/// the ONLY spend-bearing event a dispatch ever produces (InstantLock +/// promotions carry no record here by design, and an evicted or +/// never-confirmed transaction inserts no `BlockProcessed` record) and the +/// pending-spend fence has no deadline behind it, so one lost observation +/// fenced an input for the manager's lifetime (`dashpay/platform#4309`). With +/// a read that cannot fail there is no such window and nothing to queue: every +/// observation is applied at delivery. /// -/// `try_read` can still fail — wallet registration/removal holds the map's -/// write lock for a moment — and this handler used to drop the observation on -/// that failure and call the drop fail-safe. The *direction* was fail-safe (a -/// dropped observation can only delay a release, never cause one), but the -/// cost was not a wait: `TransactionDetected` can be the ONLY spend-bearing -/// event a dispatch ever produces (InstantLock promotions carry no record -/// here by design, and an evicted or never-confirmed transaction inserts no -/// `BlockProcessed` record), and the pending-spend fence has no deadline -/// behind it. One moment of lock contention could therefore fence an input -/// for the manager's lifetime even though the wallet HAD observed it spent -/// (`dashpay/platform#4309`). -/// -/// So a failed `try_read` now queues the observation in [`pending`] -/// (per-wallet outpoint sets under a `std::sync::Mutex`), and EVERY delivered -/// event — spend-bearing or not — first retries the queue. The evidence -/// survives the contention and is applied at the next delivery, which during -/// any sync activity is moments away. Draining on a spend-free event (a bare -/// `SyncHeightAdvanced`, say) does not retire a fence by chain progress: it -/// applies an observation that already arrived and was queued. -/// -/// Two outcomes remain terminal, deliberately: -/// -/// * a wallet id that resolves to NO entry in a successfully read map — the -/// wallet is unregistered; that is a resolution, not contention, and it -/// matches the pre-queue behaviour; -/// * a queue already holding [`MAX_QUEUED_SPEND_OBSERVATIONS`] outpoints — a -/// pathology no real contention window reaches (entries are deduplicated -/// per wallet and drain on the next event), shed with a warning rather -/// than allowed to grow without bound. -/// -/// # Why the queue cannot deadlock -/// -/// The handler's lock order is strictly nested and acyclic: the private -/// `pending` mutex, then `wallets.try_read()`, then each resolved -/// generation's `in_broadcast` mutex (inside -/// [`observe_spent`](super::WalletGeneration::observe_spent)). Nothing else -/// in the crate takes the `pending` mutex at all; the `in_broadcast` critical -/// sections are pure hash operations that acquire no further lock; and the -/// one lock another thread can hold long — the `wallets` map, on the -/// registration/removal write path — is only ever PROBED here (`try_read`), -/// never awaited, so this handler can never be the blocked edge in a cycle -/// with those writers. +/// One outcome stays terminal, deliberately: a wallet id that resolves to no +/// entry in the map is unregistered — a resolution, not contention, with +/// nothing to retry against. /// -/// [`pending`]: Self::pending pub struct SpendObservationHandler { - wallets: Arc>>>, - /// Observations whose wallets-map probe was contended, awaiting the next - /// delivered event — see the type docs. Keyed per wallet with the - /// outpoints deduplicated: observations are idempotent set-unions - /// ([`observe_spent`](super::WalletGeneration::observe_spent)), so - /// neither ordering nor multiplicity needs preserving. - pending: Mutex>>, + wallets: Arc>>>, } -/// Upper bound on the total outpoints [`SpendObservationHandler`] will hold -/// queued across all wallets. Queued entries drain on the very next delivered -/// event and are deduplicated, so reaching this bound takes a wallets-map -/// writer stalled across thousands of distinct own-spend observations — a -/// pathology, and one worth a warning rather than unbounded growth. -const MAX_QUEUED_SPEND_OBSERVATIONS: usize = 4096; - impl SpendObservationHandler { - pub fn new(wallets: Arc>>>) -> Self { - Self { - wallets, - pending: Mutex::new(BTreeMap::new()), - } + pub fn new(wallets: Arc>>>) -> Self { + Self { wallets } } - /// Recovers from a poisoned mutex rather than panicking — the guarded - /// data is a plain outpoint-set map with no invariant a partial write - /// could break, and panicking here would take down SPV's event fan-out. - /// (Same policy as the fence map itself.) - fn pending_lock(&self) -> MutexGuard<'_, BTreeMap>> { - self.pending.lock().unwrap_or_else(PoisonError::into_inner) - } - - /// Queue `observation` (if any), then apply everything queued if the - /// wallets map can be read right now. + /// Apply `observation` to its wallet's generation. /// - /// The pending mutex is held across the whole attempt; see the type docs - /// for why that nesting cannot deadlock. On a contended map everything — - /// this observation included — simply stays queued for the next event. - fn observe(&self, observation: Option<(WalletId, Vec)>) { - let mut pending = self.pending_lock(); - if let Some((wallet_id, outpoints)) = observation { - Self::enqueue(&mut pending, wallet_id, outpoints); - } - if pending.is_empty() { - return; - } - // try_read on the wallets map, NOT the SPV-contended wallet_manager - // lock — see the type docs. - let Ok(wallets) = self.wallets.try_read() else { - tracing::debug!( - wallets = pending.len(), - "in-broadcast fence release deferred: wallets-map lock \ - contended; observation queued for the next event" - ); - return; - }; - for (wallet_id, outpoints) in std::mem::take(&mut *pending) { - if let Some(wallet) = wallets.get(&wallet_id) { - wallet.generation().observe_spent(outpoints); - } - // A wallet absent from a successfully read map is unregistered: - // resolved, not deferred — nothing to retry against. - } - } - - /// Add one observation to the queue, shedding (with a warning) only past - /// [`MAX_QUEUED_SPEND_OBSERVATIONS`] total queued outpoints. - fn enqueue( - pending: &mut BTreeMap>, - wallet_id: WalletId, - outpoints: Vec, - ) { - let mut total: usize = pending.values().map(BTreeSet::len).sum(); - let entry = pending.entry(wallet_id).or_default(); - for outpoint in outpoints { - if total >= MAX_QUEUED_SPEND_OBSERVATIONS { - tracing::warn!( - wallet = %hex::encode(wallet_id), - %outpoint, - "spend-observation queue full: shedding an observation; \ - any fence on this outpoint waits for its next spend event" - ); - continue; - } - if entry.insert(outpoint) { - total += 1; - } + /// The wallets-map read is a wait-free `ArcSwap` load — it cannot fail or + /// block, so the observation is never deferred and never dropped. A wallet + /// absent from the loaded map is unregistered: resolved, not contended. + fn observe(&self, wallet_id: WalletId, outpoints: Vec) { + // The wallets map, NOT the SPV-contended wallet_manager lock — see + // the type docs. + if let Some(wallet) = self.wallets.load().get(&wallet_id) { + wallet.generation().observe_spent(outpoints); } } } impl EventHandler for SpendObservationHandler { fn on_wallet_event(&self, event: &WalletEvent) { - // Every delivery retries the queue, so an observation deferred by a - // contended wallets map is applied at the next event of ANY variant — - // including the spend-free ones this handler otherwise ignores. - let observation = observing_wallet(event) - .map(|wallet_id| (*wallet_id, observed_spends(event))) - .filter(|(_, outpoints)| !outpoints.is_empty()); - self.observe(observation); + let Some(wallet_id) = observing_wallet(event) else { + return; + }; + let outpoints = observed_spends(event); + if outpoints.is_empty() { + return; + } + self.observe(*wallet_id, outpoints); } } From cdb8f029fcd0bf44cffda25d3e2b428c1c64fd0c Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Mon, 31 Aug 2026 19:41:53 +0300 Subject: [PATCH 2/3] fix(platform-wallet): keep a load rollback to the generation it published MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-ups on the review of this PR. `cargo fmt` on the `wallets_map` test helper, whose return type the ArcSwap change left wrapped. And the rollback in `load_from_persistor` tracked only `WalletId`, so it removed by id alone. That is safe while nothing else touches the map, and this is the interleaving where something does: this load publishes a generation under an id, a concurrent `remove_wallet` frees that id, a registration publishes a NEW generation under it, and only then does a later iteration fail and reach the rollback. Removing by id would delete that new registration — a live wallet this call never created and whose owner is still using it — and the inner-manager unwind that follows would strip its backing too. The rollback is now generation-checked, the same rule `remove_wallet` applies to its own removal: an entry is reclaimed only while it still holds the `Arc` this load inserted, and the inner-manager unwind keys off that same answer. An id that never reached `self.wallets` — this call failed between the two inserts — has no such owner and unwinds as before. The decision is a pure `rollback_targets`, so the invariant is pinned without racing a real load against a real re-registration: `rollback_only_reclaims_the_generation_this_load_published` asserts both halves — reclaimed while ours, refused once superseded. Pre-existing: the id-only removal predates the ArcSwap change, which altered how the map is written, not what the rollback matched on. --- .../rs-platform-wallet/src/manager/load.rs | 122 +++++++++++++++++- .../src/wallet/core/broadcast.rs | 4 +- 2 files changed, 117 insertions(+), 9 deletions(-) diff --git a/packages/rs-platform-wallet/src/manager/load.rs b/packages/rs-platform-wallet/src/manager/load.rs index 7c8f445f05c..cb1575d60f7 100644 --- a/packages/rs-platform-wallet/src/manager/load.rs +++ b/packages/rs-platform-wallet/src/manager/load.rs @@ -63,7 +63,10 @@ impl PlatformWalletManager

{ // boundary with no Swift-side reset path, so transactional // semantics matter for this hydration API. let mut inserted_in_manager: Vec = Vec::new(); - let mut inserted_in_wallets: Vec = Vec::new(); + // The generation travels with the id: a rollback may only remove the + // registration THIS call published (see the rollback block below). + let mut inserted_in_wallets: Vec<(WalletId, Arc)> = + Vec::new(); let mut load_error: Option = None; 'load: for (expected_wallet_id, wallet_state) in wallets { @@ -213,7 +216,7 @@ impl PlatformWalletManager

{ wallets.insert(wallet_id, Arc::clone(&platform_wallet)); wallets }); - inserted_in_wallets.push(wallet_id); + inserted_in_wallets.push((wallet_id, Arc::clone(platform_wallet.generation()))); } if let Some(err) = load_error { @@ -221,18 +224,44 @@ impl PlatformWalletManager

{ // manager state matches what it was before. Order: // remove from `self.wallets` first (UI surface), then // from the inner `wallet_manager`. + // Generation-checked, exactly like `remove_wallet`'s own removal: + // a concurrent removal frees an id and a registration can publish + // a DIFFERENT generation under it before this rollback runs. + // Removing by id alone would delete that live wallet — one this + // call never created and whose owner is still using it. + let rolled_back = std::cell::RefCell::new(Vec::::new()); if !inserted_in_wallets.is_empty() { self.wallets.rcu(|wallets| { - let mut wallets = std::collections::BTreeMap::clone(wallets); - for id in &inserted_in_wallets { - wallets.remove(id); + // `rcu` may retry, so this is rebuilt per attempt rather + // than accumulated across them. + let ours = rollback_targets(&inserted_in_wallets, wallets); + let mut next = std::collections::BTreeMap::clone(wallets); + for id in &ours { + next.remove(id); } - wallets + *rolled_back.borrow_mut() = ours; + next }); } + let rolled_back = rolled_back.into_inner(); if !inserted_in_manager.is_empty() { let mut wm = self.wallet_manager.write().await; for id in &inserted_in_manager { + // A published id whose generation is no longer ours belongs + // to a newer registration; taking it out of the inner + // manager would strip a live wallet of its backing. An id + // that never reached `self.wallets` (this call failed + // between the two inserts) has no such owner and is unwound + // as before. + let published = inserted_in_wallets.iter().any(|(w, _)| w == id); + if published && !rolled_back.contains(id) { + tracing::warn!( + wallet_id = %hex::encode(id), + "rollback after load failure: a new generation was registered under \ + this id, leaving the new registration in place" + ); + continue; + } if let Err(e) = wm.remove_wallet(id) { tracing::warn!( wallet_id = %hex::encode(id), @@ -249,11 +278,41 @@ impl PlatformWalletManager

{ } } +/// Of the registrations this load published, the ones a rollback may still +/// take back: those whose map entry is *still the same generation* this call +/// inserted. +/// +/// A concurrent `remove_wallet` frees an id, and a registration can publish a +/// different generation under it before a later iteration's failure reaches +/// the rollback. Removing by id alone would delete that live wallet — one this +/// call never created and whose owner is still using it. Same rule +/// `remove_wallet` applies to its own removal. +/// +/// Pure so the invariant is unit-testable without racing a real load against a +/// real re-registration. +fn rollback_targets( + published: &[(WalletId, Arc)], + current: &BTreeMap>, +) -> Vec { + published + .iter() + .filter(|(id, generation)| { + current + .get(id) + .is_some_and(|wallet| Arc::ptr_eq(wallet.generation(), generation)) + }) + .map(|(id, _)| *id) + .collect() +} + #[cfg(test)] mod idempotent_load_tests { use std::collections::BTreeMap; use std::sync::Arc; + use super::rollback_targets; + use crate::wallet::core::WalletGeneration; + use key_wallet::test_utils::TestWalletContext; use key_wallet::wallet::ManagedWalletInfo; use key_wallet::Wallet; @@ -367,4 +426,55 @@ mod idempotent_load_tests { "idempotent reloads must not duplicate or drop the wallet" ); } + + /// `dashpay/platform#4309`-adjacent lifecycle hazard: a rollback must not + /// remove a registration it did not make. + /// + /// The interleaving: this load publishes generation G1 under an id, a + /// concurrent `remove_wallet` frees that id, a registration publishes G2 + /// under it, and only then does a later iteration of this load fail and + /// reach the rollback. Removing by id alone deletes G2 — a live wallet + /// whose owner is still using it, and one this call never created. + /// + /// Both halves are pinned: the entry is reclaimed while it is still ours, + /// and refused once it is not. The inner-manager rollback keys off this + /// same answer, so a wallet left in `self.wallets` is never stripped of + /// its backing either. + #[tokio::test] + async fn rollback_only_reclaims_the_generation_this_load_published() { + let ctx = TestWalletContext::new_random(); + let expected_id = ctx.wallet.compute_wallet_id(); + let manager = make_manager(SingleWalletPersister { + wallet: ctx.wallet, + managed: ctx.managed_wallet, + }); + manager + .load_from_persistor() + .await + .expect("first load succeeds"); + + let published = manager.wallets.load(); + let wallet = published + .get(&expected_id) + .expect("the load registered the wallet"); + let ours = Arc::clone(wallet.generation()); + + assert_eq!( + rollback_targets(&[(expected_id, Arc::clone(&ours))], &published), + vec![expected_id], + "a registration still holding this load's generation is ours to roll back" + ); + + // The same id, a different generation — what a removal plus a + // re-registration leaves behind. + let superseding = Arc::new(WalletGeneration::new()); + assert!( + !Arc::ptr_eq(&ours, &superseding), + "the fixture must model two distinct generations" + ); + assert!( + rollback_targets(&[(expected_id, superseding)], &published).is_empty(), + "a generation this load never published must survive its rollback" + ); + } } diff --git a/packages/rs-platform-wallet/src/wallet/core/broadcast.rs b/packages/rs-platform-wallet/src/wallet/core/broadcast.rs index 93512c50faa..be29c1fe8e2 100644 --- a/packages/rs-platform-wallet/src/wallet/core/broadcast.rs +++ b/packages/rs-platform-wallet/src/wallet/core/broadcast.rs @@ -828,9 +828,7 @@ mod tests { fn wallets_map( cores: &[&CoreWallet], ) -> Arc< - arc_swap::ArcSwap< - std::collections::BTreeMap>, - >, + arc_swap::ArcSwap>>, > { Arc::new(arc_swap::ArcSwap::from_pointee( cores From 6fda0ecf77f277eea726cfdeaffa1c53af538545 Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Thu, 3 Sep 2026 13:28:09 +0300 Subject: [PATCH 3/3] fix(platform-wallet): close the restore path's balance window, and correct what the docs claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on this PR. One real defect, the rest are the code telling a future reader something that is not true. `load_from_persistor` seeded the balance atomic BEFORE `insert_wallet`, and the wallet becomes SPV-visible the moment that insert lands — several `.await`s before the `rcu` publishes it. A `BlockProcessed` arriving in that window finds the wallet absent from the map and its snapshot is dropped, leaving a restored wallet showing the persisted total while the inner balance has moved on, with no later event guaranteed to correct it. `register_wallet` already re-seeds after its publish and its comment claimed the restore path did too; now it does. Both sites note the ordering they accept: the seed can briefly lose a race with the handler, and the next event corrects that — worth more than the window it closes. The field doc said an `rcu` closure "must stay pure map manipulation", while two of the three call sites deliberately write captured state and are correct in doing so. It now states the rule they actually rely on: only the invocation whose compare-and-swap succeeds is published, so captured state must be overwritten per attempt, never accumulated. The rollback's warning claimed a new generation had been registered under the id whenever the entry was not ours — including when something else had simply removed it and nothing replaced it, which sends anyone reading it after a wallet-disappeared report after a generation that does not exist. The two states are now distinguished. `get_wallet` and `wallet_ids` became character-for-character copies of their `_blocking` twins, so they delegate rather than drift; their `async` signatures stay for source compatibility, with the doc saying they no longer suspend. Tests: `a_failed_load_rolls_back_the_wallet_it_had_already_published` fails a load after a wallet is published, so the rollback's `rcu` closure, its per-attempt verdict hand-off and the branch deciding whether the inner-manager entry is removed all execute — none of which the pure-function test reaches. And both rendezvous closures now park once: `rcu` may re-run its closure, and a second `recv()` on a send-once channel would hang the suite rather than fail it. --- .../src/manager/accessors.rs | 13 +- .../rs-platform-wallet/src/manager/load.rs | 147 +++++++++++++++++- .../rs-platform-wallet/src/manager/mod.rs | 24 ++- .../src/manager/wallet_lifecycle.rs | 6 + .../src/wallet/core/balance_handler.rs | 14 +- .../src/wallet/core/broadcast.rs | 14 +- 6 files changed, 198 insertions(+), 20 deletions(-) diff --git a/packages/rs-platform-wallet/src/manager/accessors.rs b/packages/rs-platform-wallet/src/manager/accessors.rs index dbb55e25513..e3dcde5c310 100644 --- a/packages/rs-platform-wallet/src/manager/accessors.rs +++ b/packages/rs-platform-wallet/src/manager/accessors.rs @@ -373,9 +373,12 @@ impl PlatformWalletManager

{ } /// Get a clone of a wallet by its ID. + /// + /// The lookup is wait-free since the map became an `ArcSwap`, so this + /// suspends at no point; it delegates to the synchronous twin and keeps + /// its `async` signature for source compatibility with existing callers. pub async fn get_wallet(&self, wallet_id: &WalletId) -> Option> { - let wallets = self.wallets.load(); - wallets.get(wallet_id).cloned() + self.get_wallet_blocking(wallet_id) } /// Synchronous twin of [`Self::get_wallet`] for FFI entry points that @@ -390,9 +393,11 @@ impl PlatformWalletManager

{ } /// List all wallet IDs. + /// + /// Wait-free like [`Self::get_wallet`]; delegates to the synchronous + /// twin and keeps its `async` signature for source compatibility. pub async fn wallet_ids(&self) -> Vec { - let wallets = self.wallets.load(); - wallets.keys().copied().collect() + self.list_wallet_ids_blocking() } /// Read per-account balance + key-usage snapshots for a wallet. diff --git a/packages/rs-platform-wallet/src/manager/load.rs b/packages/rs-platform-wallet/src/manager/load.rs index cb1575d60f7..a8669115513 100644 --- a/packages/rs-platform-wallet/src/manager/load.rs +++ b/packages/rs-platform-wallet/src/manager/load.rs @@ -217,6 +217,38 @@ impl PlatformWalletManager

{ wallets }); inserted_in_wallets.push((wallet_id, Arc::clone(platform_wallet.generation()))); + + // Re-seed the balance atomic now that the wallet is published. + // + // The seed above ran before `insert_wallet`, and the wallet + // becomes SPV-visible the moment that insert lands — several + // `.await`s before the `rcu` above. Any `BlockProcessed` for it + // in that window finds the wallet absent from the map and its + // snapshot is dropped, leaving the atomic at the persisted total + // while the inner `ManagedWalletInfo` balance has moved on. + // `register_wallet` closes the same window this way; without it + // here, a restored wallet whose catch-up completes inside the + // window keeps a stale total on screen with no later event + // guaranteed to correct it. + // + // Last writer wins between this seed and the handler: if SPV + // processes another block between the read below and the `set`, + // the atomic briefly goes back to the older totals. The next + // balance-bearing event corrects it, and during catch-up those + // arrive continuously — which is why the seed is worth more than + // the window it can briefly re-open. + { + let wm = self.wallet_manager.read().await; + if let Some(info) = wm.get_wallet_info(&wallet_id) { + let b = &info.core_wallet.balance; + platform_wallet.balance().set( + b.confirmed(), + b.unconfirmed(), + b.immature(), + b.locked(), + ); + } + } } if let Some(err) = load_error { @@ -244,6 +276,9 @@ impl PlatformWalletManager

{ }); } let rolled_back = rolled_back.into_inner(); + // Wait-free, purely for the diagnostics below: an id still in the + // map after the rollback is one a same-id re-registration owns. + let still_mapped = self.wallets.load(); if !inserted_in_manager.is_empty() { let mut wm = self.wallet_manager.write().await; for id in &inserted_in_manager { @@ -255,11 +290,27 @@ impl PlatformWalletManager

{ // as before. let published = inserted_in_wallets.iter().any(|(w, _)| w == id); if published && !rolled_back.contains(id) { - tracing::warn!( - wallet_id = %hex::encode(id), - "rollback after load failure: a new generation was registered under \ - this id, leaving the new registration in place" - ); + // Two distinct states reach here, and saying the wrong + // one sends whoever reads this after a + // wallet-disappeared report chasing the wrong + // generation: either something else already removed + // the entry (a completed concurrent `remove_wallet`), + // or a same-id re-registration published a generation + // that is not ours. Only the second leaves anything in + // place. + if still_mapped.contains_key(id) { + tracing::warn!( + wallet_id = %hex::encode(id), + "rollback after load failure: a new generation was registered \ + under this id, leaving the new registration in place" + ); + } else { + tracing::warn!( + wallet_id = %hex::encode(id), + "rollback after load failure: this id was already removed by \ + something else; nothing left to roll back" + ); + } continue; } if let Err(e) = wm.remove_wallet(id) { @@ -367,6 +418,47 @@ mod idempotent_load_tests { } } + /// Two entries: the real wallet under its true id, and the same wallet + /// under a key that cannot be the id it recomputes to. The second entry + /// sorts last, so the loader publishes the first and then fails the + /// id-match check — the only way to drive the rollback without racing a + /// real failure. + struct MismatchedSecondWalletPersister { + wallet: Wallet, + managed: ManagedWalletInfo, + } + + impl PlatformWalletPersistence for MismatchedSecondWalletPersister { + fn store( + &self, + _wallet_id: WalletId, + _changeset: PlatformWalletChangeSet, + ) -> Result<(), PersistenceError> { + Ok(()) + } + + fn flush(&self, _wallet_id: WalletId) -> Result<(), PersistenceError> { + Ok(()) + } + + fn load(&self) -> Result { + let entry = || ClientWalletStartState { + wallet: self.wallet.clone(), + wallet_info: self.managed.clone(), + identity_manager: IdentityManagerStartState::default(), + unused_asset_locks: BTreeMap::new(), + }; + let mut wallets = BTreeMap::new(); + wallets.insert(self.wallet.compute_wallet_id(), entry()); + // Sorts after any real id, so it is processed second. + wallets.insert([0xFF; 32], entry()); + Ok(ClientStartState { + wallets, + ..Default::default() + }) + } + } + struct NoopEventHandler; impl EventHandler for NoopEventHandler {} impl PlatformEventHandler for NoopEventHandler {} @@ -477,4 +569,49 @@ mod idempotent_load_tests { "a generation this load never published must survive its rollback" ); } + + /// Drives the rollback itself, not just the predicate it consults. + /// + /// `rollback_only_reclaims_the_generation_this_load_published` covers + /// `rollback_targets` in isolation; this one fails a load AFTER a wallet + /// has been published, so the `rcu` closure, the per-attempt verdict + /// hand-off, and the branch that decides whether the inner manager entry + /// is removed all execute. Inverting that decision leaves the sibling + /// test green while stripping a live wallet of its backing, so the two + /// are not redundant. + #[tokio::test] + async fn a_failed_load_rolls_back_the_wallet_it_had_already_published() { + let ctx = TestWalletContext::new_random(); + let expected_id = ctx.wallet.compute_wallet_id(); + let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk")); + let event_handler: Arc = Arc::new(NoopEventHandler); + let manager = Arc::new(PlatformWalletManager::new( + sdk, + Arc::new(MismatchedSecondWalletPersister { + wallet: ctx.wallet, + managed: ctx.managed_wallet, + }), + event_handler, + )); + + let result = manager.load_from_persistor().await; + assert!( + result.is_err(), + "the id-mismatched second entry must fail the load" + ); + + assert!( + manager.get_wallet(&expected_id).await.is_none(), + "the wallet published before the failure must be rolled back out of the map" + ); + assert!( + manager + .wallet_manager + .read() + .await + .get_wallet(&expected_id) + .is_none(), + "and out of the inner manager, so a retry can re-insert it" + ); + } } diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index a444035cc74..d1ea58404d3 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/packages/rs-platform-wallet/src/manager/mod.rs @@ -339,8 +339,14 @@ pub struct PlatformWalletManager { /// coalesces, so a snapshot dropped during a lifecycle write would /// be lost for good (see `BalanceUpdateHandler`). Writers are the /// rare manager lifecycle paths (create/remove/load) and publish - /// via `rcu`, whose closure must stay pure map manipulation — it - /// can run more than once under a concurrent-writer retry. + /// via `rcu`. That closure can run more than once under a + /// concurrent-writer retry, and only the invocation whose + /// compare-and-swap succeeds is published — so captured state it + /// writes must be OVERWRITTEN per attempt, never accumulated across + /// them. `remove_wallet`'s generation verdict and the load rollback's + /// reclaimed-id list both rely on exactly that: each attempt recomputes + /// its answer from the map the closure was handed, which is the same + /// map the CAS compares against. pub(super) wallets: Arc>>>, /// Notified on InstantLock / ChainLock events for `AssetLockManager` waiters. @@ -1411,9 +1417,19 @@ mod tests { let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); let wallets_for_writer = Arc::clone(&manager.wallets); let writer = std::thread::spawn(move || { + // `rcu` re-runs its closure if the compare-and-swap loses, so the + // release must be waited on ONCE: a second `recv()` would block + // forever on a channel the test only sends to once, and + // `writer.join()` below would hang the suite instead of failing + // it. Nothing else writes this map today, so the retry is latent + // — which is exactly why it must not be able to wedge the test. + let mut parked = false; wallets_for_writer.rcu(|current| { - let _ = entered_tx.send(()); - let _ = release_rx.recv(); + if !parked { + parked = true; + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + } Arc::clone(current) }); }); diff --git a/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs b/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs index b11c1fefdf6..270d9ff6aa2 100644 --- a/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs +++ b/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs @@ -602,6 +602,12 @@ impl PlatformWalletManager

{ // into the atomic here (as `manager::load` does for restored // wallets); any later block events are applied normally now that // the wallet is mapped. + // + // Last writer wins between this seed and the handler: if SPV + // processes another block between the read below and the `set`, the + // atomic briefly goes back to the older totals. The next + // balance-bearing event corrects it, and during the rescan this + // exists for those arrive continuously. { let wm = self.wallet_manager.read().await; if let Some(info) = wm.get_wallet_info(&wallet_id) { diff --git a/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs b/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs index db655bda2d2..abc626d55b5 100644 --- a/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs +++ b/packages/rs-platform-wallet/src/wallet/core/balance_handler.rs @@ -69,11 +69,15 @@ impl EventHandler for BalanceUpdateHandler { WalletEvent::ChainLockProcessed { .. } => return, }; - // Wait-free snapshot of the wallets map; cannot fail or block, - // so no balance-bearing event is ever dropped here. A wallet - // not in the snapshot is one registered concurrently with this - // event — its creation path re-seeds the balance atomics from - // the inner wallet after publishing it, covering that window. + // Wait-free snapshot of the wallets map; cannot fail or block, so + // a lifecycle write can no longer cost a snapshot. A wallet absent + // from the snapshot is one still inside its creation window: it is + // registered in the inner manager (and therefore SPV-visible) + // several `.await`s before it is published here. Both creation + // paths close that window by re-seeding the atomics from the inner + // wallet immediately after publishing — `register_wallet` and + // `load_from_persistor` alike — so the window costs a snapshot, not + // the balance. let wallets = self.wallets.load(); if let Some(pw) = wallets.get(wallet_id) { pw.balance().set( diff --git a/packages/rs-platform-wallet/src/wallet/core/broadcast.rs b/packages/rs-platform-wallet/src/wallet/core/broadcast.rs index be29c1fe8e2..8db36c69c75 100644 --- a/packages/rs-platform-wallet/src/wallet/core/broadcast.rs +++ b/packages/rs-platform-wallet/src/wallet/core/broadcast.rs @@ -1195,9 +1195,19 @@ mod tests { let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); let map_for_writer = Arc::clone(&map); let writer = std::thread::spawn(move || { + // `rcu` re-runs its closure if the compare-and-swap loses, so the + // release must be waited on ONCE: a second `recv()` would block + // forever on a channel the test only sends to once, and + // `writer.join()` below would hang the suite instead of failing + // it. Nothing else writes this map today, so the retry is latent + // — which is exactly why it must not be able to wedge the test. + let mut parked = false; map_for_writer.rcu(|current| { - let _ = entered_tx.send(()); - let _ = release_rx.recv(); + if !parked { + parked = true; + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + } Arc::clone(current) }); });