From c2b7185cd6a5664387bfff36f8a5809a46581163 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Mon, 10 Aug 2026 11:20:55 +0700 Subject: [PATCH] fix(platform-wallet): wait for SPV transport before resuming asset locks that need broadcast MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The app-launch asset-lock catch-up races SPV client startup and loses permanently. Hosts drive the catch-up from wallet load, which runs before `startSpv` — in dashwallet-ios roughly fifty lines and two setup steps before it. `resume_asset_lock`'s `Built` arm broadcasts into a client that has not started, and that failure is classified `BroadcastError::Rejected`, the "provably never sent" verdict. Nothing reschedules the catch-up, so the lock never leaves `Built` and every later session repeats the identical race. One observed lock held 2 DASH for days until a manual retry. Gate the broadcast on transport readiness rather than reordering the hosts. `TransactionBroadcaster` gains `wait_until_ready`, defaulted to "always ready" so `DapiBroadcaster` and the test doubles are unchanged; `SpvBroadcaster` delegates to a new `SpvRuntime::wait_until_ready` that waits for a started client with at least one connected peer — both halves matter, since zero-peers is the other pre-send `Rejected` shape at launch. Only the two arms that actually broadcast wait, and the wait is deducted from the caller's timeout so the total stays inside the requested budget. The `InstantSendLocked` / `ChainLocked` / `RecoveredFromChain` arms already hold a proof and broadcast nothing, so they never wait — that exclusion is load-bearing, not an optimization: four callers pass `timeout: None` on exactly that branch and gating them would turn a cheap path re-derivation into an indefinite hang. A readiness timeout is not fatal; the broadcast is attempted anyway and reports the same error it would have reported without the wait. Regression tests cover all three behaviours: a `Built` resume under the catch-up's unbounded timeout records no broadcast attempt until the transport comes up and then sends the original transaction; a bounded caller still fails fast instead of parking; and a chain-locked lock resumes without consulting readiness at all. Co-Authored-By: Claude Opus 5 --- .../src/asset_lock/sync.rs | 8 + .../rs-platform-wallet/src/broadcaster.rs | 34 ++ .../rs-platform-wallet/src/spv/runtime.rs | 44 +++ .../src/wallet/asset_lock/sync/recovery.rs | 303 ++++++++++++++++++ .../PlatformWalletManager.swift | 8 + 5 files changed, 397 insertions(+) diff --git a/packages/rs-platform-wallet-ffi/src/asset_lock/sync.rs b/packages/rs-platform-wallet-ffi/src/asset_lock/sync.rs index 5b840f9d039..55ecf2ec9d0 100644 --- a/packages/rs-platform-wallet-ffi/src/asset_lock/sync.rs +++ b/packages/rs-platform-wallet-ffi/src/asset_lock/sync.rs @@ -96,6 +96,14 @@ pub unsafe extern "C" fn asset_lock_manager_resume( /// this on a background queue — `runtime().block_on(...)` parks the /// calling thread for up to `timeout_secs` (or **indefinitely** when /// `timeout_secs == 0`, since a ChainLock is guaranteed finality). +/// +/// Safe to call before SPV is up. Hosts drive this at wallet load, +/// which typically precedes `platform_wallet_manager_start_spv`, so a +/// lock that still needs broadcasting would otherwise take a +/// definitive never-sent rejection and — nothing retries the catch-up +/// — stay stranded for the session. `resume_asset_lock` therefore +/// waits for the SPV transport (within `timeout_secs`) before +/// broadcasting; callers need no readiness gate of their own. #[no_mangle] pub unsafe extern "C" fn asset_lock_manager_catch_up_blocking( handle: Handle, diff --git a/packages/rs-platform-wallet/src/broadcaster.rs b/packages/rs-platform-wallet/src/broadcaster.rs index 633af12ae25..075be0592b7 100644 --- a/packages/rs-platform-wallet/src/broadcaster.rs +++ b/packages/rs-platform-wallet/src/broadcaster.rs @@ -70,6 +70,23 @@ pub trait TransactionBroadcaster: Send + Sync { /// transport ambiguity, or unverifiable response must be /// [`BroadcastError::MaybeSent`]. async fn broadcast(&self, transaction: &Transaction) -> Result; + + /// Resolve once this broadcaster's transport can actually reach the + /// network, or when `timeout` elapses (`None` waits indefinitely). + /// Returns whether readiness was reached. + /// + /// Callers that resume queued work at app start use this to avoid + /// racing a transport that is still coming up: a transport-not-ready + /// failure is definitive ([`BroadcastError::Rejected`]) and the resume + /// paths do not retry, so losing that race strands the transaction for + /// the session. + /// + /// The default is "always ready" — correct for any broadcaster with no + /// startup phase of its own, such as [`DapiBroadcaster`], whose gRPC + /// requests carry their own connection handling. + async fn wait_until_ready(&self, _timeout: Option) -> bool { + true + } } /// Broadcasts transactions via Platform's DAPI gRPC endpoint. @@ -148,6 +165,11 @@ trait SpvChannel: Send + Sync { transaction: &Transaction, timeout: Option, ) -> Result; + + /// Resolve once the SPV client is started and has at least one + /// connected peer — the two conditions whose absence makes + /// `broadcast_and_wait` fail before any send. + async fn wait_until_ready(&self, timeout: Option) -> bool; } #[async_trait] @@ -160,6 +182,10 @@ impl SpvChannel for SpvRuntime { self.broadcast_transaction_and_wait(transaction, timeout) .await } + + async fn wait_until_ready(&self, timeout: Option) -> bool { + SpvRuntime::wait_until_ready(self, timeout).await + } } /// Broadcasts purely over the SPV P2P network — no DAPI involvement. @@ -219,6 +245,10 @@ impl TransactionBroadcaster for SpvBroadcaster { Err(other) => Err(other), } } + + async fn wait_until_ready(&self, timeout: Option) -> bool { + self.spv.wait_until_ready(timeout).await + } } #[cfg(test)] @@ -256,6 +286,10 @@ mod tests { .take() .expect("one acceptance check") } + + async fn wait_until_ready(&self, _timeout: Option) -> bool { + true + } } fn transaction() -> Transaction { diff --git a/packages/rs-platform-wallet/src/spv/runtime.rs b/packages/rs-platform-wallet/src/spv/runtime.rs index 38a65512744..fd4b82ac2f0 100644 --- a/packages/rs-platform-wallet/src/spv/runtime.rs +++ b/packages/rs-platform-wallet/src/spv/runtime.rs @@ -49,6 +49,10 @@ const SPV_CLIENT_STOP_BUDGET: Duration = Duration::from_secs(15); /// graceful timeout above was meant to escape. const SPV_ABORT_GRACE: Duration = Duration::from_secs(2); +/// How often [`SpvRuntime::wait_until_ready`] re-checks for a started client +/// with connected peers. +const SPV_READINESS_POLL_INTERVAL: Duration = Duration::from_millis(250); + /// Join a stopped SPV runner, escalating to cancellation after `timeout`. /// /// Returns `None` once Tokio has confirmed the task terminated. Returns @@ -196,6 +200,46 @@ impl SpvRuntime { self.client.try_read().map(|c| c.is_some()).unwrap_or(false) } + /// Whether a broadcast issued right now could reach the network: the + /// client is started *and* at least one peer is connected. + /// + /// Both halves are required because both are pre-send rejections in + /// [`broadcast_transaction_and_wait`](Self::broadcast_transaction_and_wait) + /// — an unstarted client and dash-spv's zero-connected-peers check. + async fn is_broadcast_ready(&self) -> bool { + self.client.read().await.is_some() && !self.peer_tracker.snapshot().is_empty() + } + + /// Resolve once a broadcast could actually reach the network, or when + /// `timeout` elapses. `None` waits indefinitely. + /// + /// Returns whether readiness was reached. This closes the launch race + /// where work resumed at app start (asset-lock catch-up in particular) + /// broadcasts into a client that has not finished starting, takes the + /// definitive `Rejected { "client not started" }` verdict, and — having + /// no retry — strands the transaction 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. + pub async fn wait_until_ready(&self, timeout: Option) -> bool { + let deadline = timeout.map(|t| tokio::time::Instant::now() + t); + loop { + if self.is_broadcast_ready().await { + return true; + } + let now = tokio::time::Instant::now(); + match deadline { + None => tokio::time::sleep(SPV_READINESS_POLL_INTERVAL).await, + Some(deadline) if now < deadline => { + tokio::time::sleep_until(deadline.min(now + SPV_READINESS_POLL_INTERVAL)).await + } + Some(_) => return false, + } + } + } + /// Broadcast a transaction through SPV peers and wait for dash-spv's /// network-acceptance verdict. /// diff --git a/packages/rs-platform-wallet/src/wallet/asset_lock/sync/recovery.rs b/packages/rs-platform-wallet/src/wallet/asset_lock/sync/recovery.rs index bce7e73c9e8..a2329671885 100644 --- a/packages/rs-platform-wallet/src/wallet/asset_lock/sync/recovery.rs +++ b/packages/rs-platform-wallet/src/wallet/asset_lock/sync/recovery.rs @@ -189,6 +189,49 @@ impl AssetLockManager { // --------------------------------------------------------------------------- impl AssetLockManager { + /// Wait for the broadcaster's transport to come up before a resume + /// broadcasts, and return the finality budget left over. + /// + /// A resume that still needs a broadcast (`Built` / `Broadcast`) is + /// typically driven by the app-launch catch-up, which runs while the + /// SPV client is still starting. Broadcasting into an unstarted client + /// fails as `Rejected` — definitively never sent — and no resume path + /// retries, so a `Built` lock that loses this race stays un-broadcast + /// across every subsequent session and its funds stay stranded until + /// someone retries by hand. + /// + /// The wait is bounded by the caller's own `timeout` and the time it + /// consumes is deducted from it, so the total stays within the budget + /// the caller asked for. `None` (the catch-up's unbounded wait for + /// finality) waits for the transport indefinitely — no weaker bound + /// would help, since the proof that wait is for arrives over the same + /// transport. + /// + /// Timing out is not fatal here: the broadcast is attempted anyway and + /// reports the same failure it would have reported without the wait. + async fn await_broadcast_ready( + &self, + out_point: &OutPoint, + timeout: Option, + ) -> Option { + let started = tokio::time::Instant::now(); + if !self.broadcaster.wait_until_ready(timeout).await { + tracing::warn!( + outpoint = %out_point, + ?timeout, + "resume_asset_lock: broadcast transport still not ready; \ + attempting the broadcast anyway" + ); + } + let waited = started.elapsed(); + tracing::debug!( + outpoint = %out_point, + ?waited, + "resume_asset_lock: broadcast transport wait finished" + ); + timeout.map(|t| t.saturating_sub(waited)) + } + /// Resume a tracked asset lock from whatever stage it's at. /// /// Looks up the tracked lock by `txid`, then: @@ -207,6 +250,13 @@ impl AssetLockManager { /// caller passes `derivation_path` to the same signer used for the /// build phase when the credit output is later consumed on Platform. /// + /// The two arms that broadcast first wait for the broadcaster's + /// transport to come up (see + /// [`await_broadcast_ready`](Self::await_broadcast_ready)), so a resume + /// driven by the app-launch catch-up doesn't race the SPV client's + /// startup and strand the lock. `InstantSendLocked` / `ChainLocked` + /// already hold a proof, broadcast nothing, and so never wait. + /// /// `timeout` is `Option` and is only consulted when the lock /// still needs a proof (`Built` / `Broadcast`): `None` waits /// **indefinitely** for finality. For `InstantSendLocked` / `ChainLocked` @@ -252,6 +302,7 @@ impl AssetLockManager { let proof = match status { AssetLockStatus::Built => { // Re-broadcast and wait for proof. + let timeout = self.await_broadcast_ready(out_point, timeout).await; self.broadcaster.broadcast(&tx).await?; let cs = self .advance_asset_lock_status(out_point, AssetLockStatus::Broadcast, None) @@ -282,6 +333,7 @@ impl AssetLockManager { // rather than failing the resume on a tx that is actually // fine. If the tx really was mined, `wait_for_proof` // resolves immediately from the SPV/persisted record. + let timeout = self.await_broadcast_ready(out_point, timeout).await; if let Err(e) = self.broadcaster.broadcast(&tx).await { tracing::debug!( outpoint = %out_point, @@ -488,6 +540,7 @@ impl AssetLockManager { #[cfg(test)] mod tests { use std::collections::BTreeMap; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -833,4 +886,254 @@ mod tests { "re-derived credit-output path must match the build-time path" ); } + + /// Broadcaster whose transport comes up partway through the test, the + /// way the SPV client does during app launch. Before `make_ready`, a + /// `broadcast` reproduces `SpvRuntime`'s pre-send verdict for an + /// unstarted client: `Rejected` — definitively never sent. + #[derive(Default)] + struct StartingUpBroadcaster { + ready: AtomicBool, + readiness_waits: AtomicUsize, + /// Every broadcast attempt, paired with the readiness state at the + /// moment it was attempted. + attempts: Mutex>, + } + + impl StartingUpBroadcaster { + fn make_ready(&self) { + self.ready.store(true, Ordering::SeqCst); + } + + fn attempts(&self) -> Vec<(Transaction, bool)> { + self.attempts.lock().expect("attempts mutex").clone() + } + } + + #[async_trait] + impl TransactionBroadcaster for StartingUpBroadcaster { + async fn broadcast(&self, transaction: &Transaction) -> Result { + let ready = self.ready.load(Ordering::SeqCst); + self.attempts + .lock() + .expect("attempts mutex") + .push((transaction.clone(), ready)); + if !ready { + return Err(BroadcastError::Rejected { + reason: "SPV broadcast not sent: client not started".to_string(), + }); + } + Ok(transaction.txid()) + } + + async fn wait_until_ready(&self, timeout: Option) -> bool { + self.readiness_waits.fetch_add(1, Ordering::SeqCst); + let deadline = timeout.map(|t| tokio::time::Instant::now() + t); + loop { + if self.ready.load(Ordering::SeqCst) { + return true; + } + if deadline.is_some_and(|d| tokio::time::Instant::now() >= d) { + return false; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + } + } + + /// Build a wallet holding one tracked asset lock at `status`, ready to + /// be resumed. + async fn tracked_lock_fixture( + broadcaster: Arc, + status: AssetLockStatus, + proof: Option, + ) -> ( + AssetLockManager, + OutPoint, + Transaction, + ) { + let (wallet_manager, wallet_id, _balance, signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let sdk = Arc::new( + dash_sdk::SdkBuilder::new_mock() + .with_network(Network::Testnet) + .build() + .expect("mock sdk"), + ); + let manager = AssetLockManager::new( + sdk, + Arc::clone(&wallet_manager), + wallet_id, + Arc::new(Notify::new()), + broadcaster, + WalletPersister::new(wallet_id, Arc::new(RecordingPersistence::default())), + ); + let (transaction, _path) = manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 4, + &signer, + ) + .await + .expect("build asset lock"); + let out_point = OutPoint::new(transaction.txid(), 0); + { + let mut wm = wallet_manager.write().await; + wm.get_wallet_info_mut(&wallet_id) + .expect("wallet must remain registered") + .tracked_asset_locks + .insert( + out_point, + TrackedAssetLock { + out_point, + transaction: transaction.clone(), + account_index: 0, + funding_type: AssetLockFundingType::IdentityRegistration, + identity_index: 4, + amount: 1_000_000, + status, + proof, + }, + ); + } + (manager, out_point, transaction) + } + + /// The app-launch catch-up (`timeout: None`) resumes a `Built` lock + /// while the SPV client is still starting. It must hold the broadcast + /// until the transport is up rather than spending its one attempt on a + /// client that cannot send: the failure is `Rejected` ("never sent"), + /// nothing reschedules the catch-up, and a lock left un-broadcast stays + /// stranded — with its funds — across every later session. + #[tokio::test] + async fn built_resume_waits_for_the_broadcast_transport_to_come_up() { + let broadcaster = Arc::new(StartingUpBroadcaster::default()); + let (manager, out_point, transaction) = + tracked_lock_fixture(Arc::clone(&broadcaster), AssetLockStatus::Built, None).await; + + // The unbounded wait the catch-up uses. `wait_for_proof` never + // resolves here (no IS/CL event is ever fired), so the resume is + // observed through the broadcaster rather than by joining it. + let resume = tokio::spawn(async move { manager.resume_asset_lock(&out_point, None).await }); + + // While the transport is down, the broadcast must not be spent. + tokio::time::sleep(Duration::from_millis(50)).await; + assert!( + broadcaster.attempts().is_empty(), + "a Built resume must not broadcast before the transport is ready" + ); + assert_eq!( + broadcaster.readiness_waits.load(Ordering::SeqCst), + 1, + "the Built arm must wait on transport readiness exactly once" + ); + + broadcaster.make_ready(); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + let attempts = loop { + let attempts = broadcaster.attempts(); + if !attempts.is_empty() { + break attempts; + } + assert!( + tokio::time::Instant::now() < deadline, + "the resume must broadcast once the transport comes up" + ); + tokio::time::sleep(Duration::from_millis(5)).await; + }; + + assert_eq!(attempts.len(), 1, "exactly one broadcast"); + let (broadcast_tx, was_ready) = &attempts[0]; + assert!( + was_ready, + "the broadcast must land after the transport is ready, not before" + ); + assert_eq!( + broadcast_tx, &transaction, + "recovery must re-broadcast the original transaction, not a rebuild" + ); + + resume.abort(); + } + + /// A bounded caller must not inherit an unbounded park: when the + /// transport never comes up, the wait is capped by the caller's own + /// timeout and the resume then reports the same never-sent failure it + /// would have reported without the wait. + #[tokio::test] + async fn built_resume_bounds_the_transport_wait_by_the_caller_timeout() { + let broadcaster = Arc::new(StartingUpBroadcaster::default()); + let (manager, out_point, _transaction) = + tracked_lock_fixture(Arc::clone(&broadcaster), AssetLockStatus::Built, None).await; + + let started = tokio::time::Instant::now(); + let error = manager + .resume_asset_lock(&out_point, Some(Duration::from_millis(100))) + .await + .expect_err("a transport that never comes up must fail, not hang"); + let elapsed = started.elapsed(); + + assert!( + matches!(error, PlatformWalletError::TransactionBroadcast(_)), + "the never-sent verdict must still surface: {error:?}" + ); + assert!( + elapsed < Duration::from_secs(2), + "the wait must be bounded by the caller's timeout, took {elapsed:?}" + ); + let attempts = broadcaster.attempts(); + assert_eq!( + attempts.len(), + 1, + "the broadcast is still attempted once the wait times out" + ); + } + + /// A lock already at ChainLocked has its proof and needs no broadcast, + /// so it must resume without waiting on the transport at all — an + /// SPV client that never starts cannot block it. + #[tokio::test] + async fn chain_locked_resume_does_not_wait_for_the_broadcast_transport() { + use dpp::identity::state_transition::asset_lock_proof::chain::ChainAssetLockProof; + + let broadcaster = Arc::new(StartingUpBroadcaster::default()); + // The proof has to name the real outpoint, which only exists once + // the transaction is built — so stage the lock, then re-stamp it. + let (manager, out_point, _transaction) = + tracked_lock_fixture(Arc::clone(&broadcaster), AssetLockStatus::Built, None).await; + { + let proof = dpp::prelude::AssetLockProof::Chain(ChainAssetLockProof { + core_chain_locked_height: 1_234, + out_point, + }); + let mut wm = manager.wallet_manager.write().await; + let lock = wm + .get_wallet_info_mut(&manager.wallet_id) + .expect("wallet") + .tracked_asset_locks + .get_mut(&out_point) + .expect("tracked lock"); + lock.status = AssetLockStatus::ChainLocked; + lock.proof = Some(proof); + } + + let (proof, _path) = manager + .resume_asset_lock(&out_point, None) + .await + .expect("a chain-locked lock resumes from its own proof"); + + assert!(matches!(proof, dpp::prelude::AssetLockProof::Chain(_))); + assert_eq!( + broadcaster.readiness_waits.load(Ordering::SeqCst), + 0, + "an IS/CL-staged lock needs no broadcast and must not wait on SPV" + ); + assert!( + broadcaster.attempts().is_empty(), + "an IS/CL-staged lock must not broadcast" + ); + } } diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift index fa9b787c1a3..cfe6d9ad9ed 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift @@ -923,6 +923,14 @@ public class PlatformWalletManager: ObservableObject { /// Called from `loadFromPersistor` after every wallet is /// inserted. App-foreground / network-reconnect callers can /// invoke this directly to retry whatever was still pending. + /// + /// No SPV readiness gate is needed at this call site, and hosts + /// must not add one: load runs before `startSpv`, and a lock at + /// `Built` still needs a broadcast, so Rust's `resume_asset_lock` + /// waits for the SPV transport itself (within the per-lock + /// timeout) rather than spending its one attempt on a client that + /// has not started. Gating here would only delay the locks that + /// need no broadcast at all. public func catchUpStuckAssetLocks(wallets: [ManagedPlatformWallet]) { guard let persistenceHandler = persistenceHandler else { return } for wallet in wallets {