fix(platform-wallet): never drop a wallet event on the wallets-map lock

packages/rs-platform-wallet/src/manager/accessors.rs

@@ -374,20 +374,24 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
 
     /// Get a clone of a wallet by its ID.
     pub async fn get_wallet(&self, wallet_id: &WalletId) -> Option<Arc<PlatformWallet>> {
-        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<PlatformWallet>` 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<PlatformWallet>` 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<Arc<PlatformWallet>> {
-        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<WalletId> {
-        let wallets = self.wallets.read().await;
+        let wallets = self.wallets.load();
         wallets.keys().copied().collect()
     }
 
@@ -452,10 +456,9 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     // -----------------------------------------------------------------
 
     /// Atomic snapshot of every wallet id currently registered on the
-    /// manager. Cheap (`Arc<RwLock>` read + `BTreeMap` key clone).
+    /// manager. Cheap (wait-free `ArcSwap` load + `BTreeMap` key clone).
     pub fn list_wallet_ids_blocking(&self) -> Vec<WalletId> {
-        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<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     /// 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<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
         &self,
         wallet_id: &WalletId,
     ) -> Option<PlatformAddressProviderStateSnapshot> {
-        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<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
         // 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<PlatformWallet>` 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<PlatformWallet>` 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 {

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<RwLock<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
+    wallets: Arc<ArcSwap<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
     /// 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<RwLock<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
+        wallets: Arc<ArcSwap<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
         registry: Arc<ThreadRegistry<WalletWorker>>,
     ) -> Self {
         Self {
@@ -364,7 +363,7 @@ impl DashPaySyncManager {
         }
 
         let snapshot: Vec<(WalletId, Arc<PlatformWallet>)> = {
-            let wallets = self.wallets.read().await;
+            let wallets = self.wallets.load();
             wallets.iter().map(|(id, w)| (*id, Arc::clone(w))).collect()
         };
 

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<RwLock<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
+    wallets: Arc<ArcSwap<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
     registry: Arc<ThreadRegistry<WalletWorker>>,
     /// Dispatches `on_dpns_marketplace_sync_completed` after each pass.
     events: Arc<PlatformEventManager>,
@@ -144,7 +143,7 @@ pub struct DpnsSyncManager {
 
 impl DpnsSyncManager {
     pub fn new(
-        wallets: Arc<RwLock<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
+        wallets: Arc<ArcSwap<BTreeMap<WalletId, Arc<PlatformWallet>>>>,
         registry: Arc<ThreadRegistry<WalletWorker>>,
         events: Arc<PlatformEventManager>,
     ) -> Self {
@@ -289,7 +288,7 @@ impl DpnsSyncManager {
         }
 
         let snapshot: Vec<(WalletId, Arc<PlatformWallet>)> = {
-            let wallets = self.wallets.read().await;
+            let wallets = self.wallets.load();
             wallets.iter().map(|(id, w)| (*id, Arc::clone(w))).collect()
         };
 

packages/rs-platform-wallet/src/manager/load.rs

@@ -208,9 +208,11 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
             }
 
             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<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
             // 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;

packages/rs-platform-wallet/src/manager/mod.rs

@@ -332,7 +332,17 @@ pub struct PlatformWalletManager<P: PlatformWalletPersistence + 'static> {
     /// update their lock-free balance atomics from event-handler
     /// context, without touching the SPV-contended `wallet_manager`
     /// lock.
-    pub(super) wallets: Arc<RwLock<std::collections::BTreeMap<WalletId, Arc<PlatformWallet>>>>,
+    ///
+    /// 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<arc_swap::ArcSwap<std::collections::BTreeMap<WalletId, Arc<PlatformWallet>>>>,
     /// Notified on InstantLock / ChainLock events for `AssetLockManager` waiters.
     pub(super) lock_notify: Arc<Notify>,
     pub(super) spv_manager: Arc<SpvRuntime>,
@@ -471,7 +481,9 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
             .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<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
 
         // 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<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
             ));
         };
 
-        // 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<Arc<PlatformWallet>> = {
-            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<Arc<PlatformWallet>> = 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<dyn PlatformWalletPersistence>,
             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");
+    }
 }