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

@@ -63,7 +63,10 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
         // boundary with no Swift-side reset path, so transactional
         // semantics matter for this hydration API.
         let mut inserted_in_manager: Vec<WalletId> = Vec::new();
-        let mut inserted_in_wallets: Vec<WalletId> = 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<crate::wallet::core::WalletGeneration>)> =
+            Vec::new();
         let mut load_error: Option<PlatformWalletError> = None;
 
         'load: for (expected_wallet_id, wallet_state) in wallets {
@@ -208,26 +211,57 @@ 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);
-            inserted_in_wallets.push(wallet_id);
+            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, Arc::clone(platform_wallet.generation())));
         }
 
         if let Some(err) = load_error {
             // Walk back every wallet committed in this call so the
             // 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::<WalletId>::new());
             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| {
+                    // `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);
+                    }
+                    *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),
@@ -244,11 +278,41 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     }
 }
 
+/// 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<WalletGeneration>)],
+    current: &BTreeMap<WalletId, Arc<PlatformWallet>>,
+) -> Vec<WalletId> {
+    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;
@@ -362,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"
+        );
+    }
 }