Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
llbartekll merged 3 commits into from
Sep 3, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fd2752e
fix(platform-wallet): never drop a wallet event on the wallets-map lock
jeanpierreroma Aug 31, 2026
cdb8f02
fix(platform-wallet): keep a load rollback to the generation it publi…
jeanpierreroma Aug 31, 2026
6fda0ec
fix(platform-wallet): close the restore path's balance window, and co…
jeanpierreroma Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 24 additions & 21 deletions packages/rs-platform-wallet/src/manager/accessors.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -373,22 +373,31 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
}

/// 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<Arc<PlatformWallet>> {
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<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.
///
/// 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<WalletId> {
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.
Expand Down Expand Up @@ -452,10 +461,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
Expand All @@ -476,9 +484,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
Expand Down Expand Up @@ -641,9 +647,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 {
Expand Down Expand Up @@ -1008,10 +1012,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 {
Expand Down
9 changes: 4 additions & 5 deletions packages/rs-platform-wallet/src/manager/dashpay_sync.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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()
};

Expand Down
13 changes: 6 additions & 7 deletions packages/rs-platform-wallet/src/manager/dpns_sync.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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>,
Expand All @@ -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 {
Expand Down Expand Up @@ -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()
};

Expand Down
Loading
Loading