From b49cb6c9e3639dfc3d784c7297f2fd144884b7e7 Mon Sep 17 00:00:00 2001 From: Lukasz Klimek <842586+lklimek@users.noreply.github.com> Date: Wed, 14 Oct 2026 08:49:52 +0000 Subject: [PATCH] refactor(platform-wallet): wake SPV readiness wait on events instead of polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `SpvRuntime::wait_until_ready` slept 250 ms between readiness checks. It now parks on a `tokio::sync::Notify` woken by the two transitions that can make a broadcast possible: `start` storing the client, and `PeerTracker` storing a dash-spv `PeersUpdated` snapshot. The waiter enables its `Notified` before each check, so a transition racing the check is never lost; it stays inside `tokio::time::timeout`, so huge FFI budgets still saturate instead of panicking. The "peers listed but client gone" window that `stop` leaves until it clears the tracker no longer costs a re-check every 250 ms: the wait stays parked until something actually changes. Requested in review of #5307 (point 6). 🤖 Co-authored by [Claudius the Magnificent](https://github.com/lklimek/claudius) AI Agent Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PSi8cX229kW2fmLxKW6ir4 --- CHANGELOG.md | 4 + packages/rs-platform-wallet/src/spv/peers.rs | 17 +- .../rs-platform-wallet/src/spv/runtime.rs | 156 ++++++++++++++++-- 3 files changed, 165 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d429ff095f..476a0e194c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,10 @@ (`WalletEvent::ChainTruncated`) updates the in-memory balance but is not yet mirrored to persisted wallet state ([#5380](https://github.com/dashpay/platform/issues/5380)). +- **platform-wallet:** the SPV broadcast-readiness wait (`SpvRuntime::wait_until_ready`) + wakes when the client starts or dash-spv reports a peer update, instead of + re-checking every 250 ms, so a broadcast held for readiness goes out as soon + as the first peer connects. - **Breaking (Rust features):** remove DPP's `core_key_wallet_bip_38` and SDK's `core_key_wallet_bip38` feature forwards because upstream removed BIP38 support. - **Breaking (C API):** the unified FFI no longer exports diff --git a/packages/rs-platform-wallet/src/spv/peers.rs b/packages/rs-platform-wallet/src/spv/peers.rs index 6c179b5d58b..2b8afa4d43c 100644 --- a/packages/rs-platform-wallet/src/spv/peers.rs +++ b/packages/rs-platform-wallet/src/spv/peers.rs @@ -9,12 +9,13 @@ use std::collections::HashMap; use std::net::SocketAddr; -use std::sync::{Mutex, MutexGuard, PoisonError}; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use dash_spv::network::NetworkEvent; use dash_spv::EventHandler; use dashcore::sml::masternode_list::MasternodeList; use dashcore::sml::masternode_list_entry::EntryMasternodeType; +use tokio::sync::Notify; /// Node type of a connected SPV peer, resolved against the masternode list. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -46,9 +47,20 @@ pub struct SpvPeerInfo { #[derive(Debug, Default)] pub(crate) struct PeerTracker { connected: Mutex>, + /// Woken (`notify_waiters`) after every `PeersUpdated` snapshot is + /// stored, so a readiness wait learns about peers without polling. + updated: Arc, } impl PeerTracker { + /// A tracker that wakes `updated`'s waiters after each new snapshot. + pub(crate) fn new(updated: Arc) -> Self { + Self { + connected: Mutex::default(), + updated, + } + } + /// Lock the snapshot, recovering from poisoning: it is only cloned, /// cleared or replaced whole, so a panicking holder cannot tear it. fn connected(&self) -> MutexGuard<'_, Vec> { @@ -73,6 +85,9 @@ impl EventHandler for PeerTracker { fn on_network_event(&self, event: &NetworkEvent) { if let NetworkEvent::PeersUpdated { addresses, .. } = event { *self.connected() = addresses.clone(); + // After the store (guard dropped above): a woken waiter must see + // the new snapshot. + self.updated.notify_waiters(); } } } diff --git a/packages/rs-platform-wallet/src/spv/runtime.rs b/packages/rs-platform-wallet/src/spv/runtime.rs index 73d4b7902cc..678a26a3214 100644 --- a/packages/rs-platform-wallet/src/spv/runtime.rs +++ b/packages/rs-platform-wallet/src/spv/runtime.rs @@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use std::task::{Context, Poll, Waker}; use std::time::Duration; -use tokio::sync::RwLock; +use tokio::sync::{Notify, RwLock}; use tokio::task::{JoinError, JoinHandle}; use dashcore::sml::llmq_entry_verification::LLMQEntryVerificationStatus; @@ -45,9 +45,6 @@ pub struct SpvQuorumPublicKey { /// Maximum wait per stop call; the owned teardown continues after this deadline. const SPV_STOP_TIMEOUT: Duration = Duration::from_secs(15); -/// How often readiness is rechecked for a client with connected peers. -const SPV_READINESS_POLL_INTERVAL: Duration = Duration::from_millis(250); - #[derive(Default)] enum Shutdown { #[default] @@ -83,6 +80,11 @@ pub struct SpvRuntime { /// each stop or storage clear, so none of them interleave. shutdown: tokio::sync::Mutex, peer_tracker: Arc, + /// Woken (`notify_waiters`) on every transition that can make + /// [`Self::is_broadcast_ready`] true: a client stored by `start`, and a + /// `PeersUpdated` snapshot stored by `peer_tracker`. Losing readiness + /// wakes no one; waiters only care about gaining it. + readiness_changed: Arc, } /// Classify a failure from the SPV acceptance-check path @@ -115,6 +117,7 @@ impl SpvRuntime { wallet_manager: Arc>>, event_manager: Arc, ) -> Self { + let readiness_changed = Arc::new(Notify::new()); Self { event_manager, wallet_manager, @@ -122,7 +125,8 @@ impl SpvRuntime { last_config: RwLock::new(None), task: Mutex::new(None), shutdown: tokio::sync::Mutex::new(Shutdown::Idle), - peer_tracker: Arc::new(PeerTracker::default()), + peer_tracker: Arc::new(PeerTracker::new(Arc::clone(&readiness_changed))), + readiness_changed, } } @@ -181,6 +185,10 @@ impl SpvRuntime { let mut client = self.client.write().await; *client = Some(spv_client); *self.last_config.write().await = Some(retained_config); + // Release the client before waking readiness waiters so they can + // read it. + drop(client); + self.readiness_changed.notify_waiters(); Ok(()) } @@ -281,10 +289,13 @@ impl SpvRuntime { /// ("client not started") verdict, and — having no retry — stays /// un-broadcast for the whole session. /// - /// Readiness is polled rather than pushed: "started" is a `client` - /// transition and "has peers" arrives as a dash-spv `PeersUpdated` - /// event, with no combined signal to subscribe to. The poll interval is - /// irrelevant next to the network latency being waited on. + /// Readiness is pushed, not polled: the wait re-checks only when + /// `start` stores a client or a `PeersUpdated` snapshot lands. While + /// neither happens it stays parked — including the "peers listed but + /// client gone" window `stop` leaves until it clears the tracker. It + /// registers for the wake-up before each check, so a transition racing + /// the check is never missed. Cancel-safe: dropping it drops only the + /// registration. /// /// The bound is a plain `Duration` and is applied with /// [`tokio::time::timeout`], which saturates an unrepresentable deadline @@ -294,8 +305,13 @@ impl SpvRuntime { /// FFI frame aborts the host process. pub async fn wait_until_ready(&self, timeout: Duration) -> bool { tokio::time::timeout(timeout, async { - while !self.is_broadcast_ready().await { - tokio::time::sleep(SPV_READINESS_POLL_INTERVAL).await; + loop { + let mut changed = pin!(self.readiness_changed.notified()); + changed.as_mut().enable(); + if self.is_broadcast_ready().await { + return; + } + changed.await; } }) .await @@ -1856,6 +1872,124 @@ mod tests { let _ = std::fs::remove_dir_all(&storage); } + /// The `PeersUpdated` event dash-spv pushes after a connection change. + fn one_peer_connected() -> dash_spv::network::NetworkEvent { + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + + dash_spv::network::NetworkEvent::PeersUpdated { + connected_count: 1, + addresses: vec![SocketAddr::new( + IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), + 19999, + )], + best_height: Some(1_100_000), + } + } + + /// Spawn a readiness wait and prove it parked: it is still pending + /// after the runtime has had every chance to run it. + async fn parked_readiness_wait(runtime: &Arc) -> tokio::task::JoinHandle { + let waiter = { + let runtime = Arc::clone(runtime); + tokio::spawn(async move { runtime.wait_until_ready(Duration::from_secs(30)).await }) + }; + for _ in 0..16 { + tokio::task::yield_now().await; + } + assert!(!waiter.is_finished(), "the wait must park while not ready"); + waiter + } + + /// Readiness is pushed, not polled: the first `PeersUpdated` of a + /// started client resolves a parked wait without any virtual time + /// passing. A sleep-poll would only notice on its next tick. + #[tokio::test(start_paused = true)] + async fn should_resolve_a_parked_wait_on_the_first_peers_updated_event() { + use dash_spv::EventHandler; + + let (runtime, _storage) = offline_runtime().await; + let waiter = parked_readiness_wait(&runtime).await; + let parked_at = tokio::time::Instant::now(); + + runtime.peer_tracker.on_network_event(&one_peer_connected()); + + assert!( + waiter.await.unwrap(), + "the first peer makes the client ready" + ); + assert_eq!( + parked_at.elapsed(), + Duration::ZERO, + "the event itself must wake the wait, not a later poll tick" + ); + runtime.stop().await.unwrap(); + } + + /// The client-start transition is the other half of readiness, so it + /// must wake a parked wait on its own too — here with the peer snapshot + /// already populated, so `start` is the transition that completes it. + #[tokio::test(start_paused = true)] + async fn should_resolve_a_parked_wait_when_the_client_starts() { + use dash_spv::EventHandler; + + let runtime = Arc::new(unstarted_runtime()); + runtime.peer_tracker.on_network_event(&one_peer_connected()); + let waiter = parked_readiness_wait(&runtime).await; + let parked_at = tokio::time::Instant::now(); + + let storage = tempfile::tempdir().unwrap(); + runtime + .start( + ClientConfig::testnet() + .with_storage_path(storage.path()) + .with_restrict_to_configured_peers(true), + ) + .await + .unwrap(); + + assert!( + waiter.await.unwrap(), + "a started client with a peer is ready" + ); + assert_eq!( + parked_at.elapsed(), + Duration::ZERO, + "the start itself must wake the wait, not a later poll tick" + ); + runtime.stop().await.unwrap(); + } + + /// `stop` takes the client before it clears the peer snapshot, so + /// "peers but no client" is a real, possibly long-lived state while + /// teardown runs. A wait in it must sleep until something changes — + /// not re-check on a timer for its whole budget. + #[tokio::test(start_paused = true)] + async fn should_not_recheck_readiness_while_nothing_changes() { + use dash_spv::EventHandler; + use std::future::Future; + use std::sync::atomic::{AtomicUsize, Ordering}; + + let runtime = unstarted_runtime(); + runtime.peer_tracker.on_network_event(&one_peer_connected()); + + let polls = AtomicUsize::new(0); + let mut wait = std::pin::pin!(runtime.wait_until_ready(Duration::from_secs(30))); + let ready = std::future::poll_fn(|cx| { + polls.fetch_add(1, Ordering::Relaxed); + wait.as_mut().poll(cx) + }) + .await; + + assert!(!ready, "peers without a client are not ready"); + let polls = polls.into_inner(); + // One poll to park, one when the budget expires. A 250 ms poll + // would take 120 over the 30 s budget. + assert!( + polls <= 2, + "the wait re-checked {polls} times with no change" + ); + } + /// Every other error on the acceptance path may follow a partial send /// and must stay `MaybeSent`. #[test]