diff --git a/packages/rs-platform-wallet/src/manager/accessors.rs b/packages/rs-platform-wallet/src/manager/accessors.rs index af4c76dfc5f..e3dcde5c310 100644 --- a/packages/rs-platform-wallet/src/manager/accessors.rs +++ b/packages/rs-platform-wallet/src/manager/accessors.rs @@ -373,22 +373,31 @@ 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.read().await; - wallets.get(wallet_id).cloned() + self.get_wallet_blocking(wallet_id) } - /// 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. + /// + /// 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.read().await; - wallets.keys().copied().collect() + self.list_wallet_ids_blocking() } /// Read per-account balance + key-usage snapshots for a wallet. @@ -452,10 +461,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 +484,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 +647,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 +1012,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..a8669115513 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 { @@ -208,10 +211,44 @@ 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); - 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()))); + + // 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 { @@ -219,15 +256,63 @@ 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() { - 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(); + // 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 { + // 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) { + // 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) { tracing::warn!( wallet_id = %hex::encode(id), @@ -244,11 +329,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; @@ -303,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 {} @@ -362,4 +518,100 @@ 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" + ); + } + + /// 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 9192dfae148..d1ea58404d3 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/packages/rs-platform-wallet/src/manager/mod.rs @@ -332,7 +332,23 @@ 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`. 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. pub(super) lock_notify: Arc, pub(super) spv_manager: Arc, @@ -471,7 +487,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 +513,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 +862,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 +1127,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 +1374,92 @@ 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 || { + // `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| { + if !parked { + parked = true; + 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..270d9ff6aa2 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`. @@ -600,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) { @@ -695,7 +703,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 +744,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 +754,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 +885,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..abc626d55b5 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,16 @@ 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 + // 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( 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..8db36c69c75 100644 --- a/packages/rs-platform-wallet/src/wallet/core/broadcast.rs +++ b/packages/rs-platform-wallet/src/wallet/core/broadcast.rs @@ -824,15 +824,13 @@ 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< - std::collections::BTreeMap>, - >, + arc_swap::ArcSwap>>, > { - 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 +1139,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 +1188,47 @@ 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 || { + // `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| { + if !parked { + parked = true; + 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); } }