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 9bc819585cd..585d7c7858b 100644 --- a/packages/rs-platform-wallet-ffi/src/asset_lock/sync.rs +++ b/packages/rs-platform-wallet-ffi/src/asset_lock/sync.rs @@ -9,6 +9,85 @@ use std::ffi::CString; use std::os::raw::c_char; use std::time::Duration; +/// Largest bound a caller may request, in seconds (one year). +/// +/// The wait paths downstream build their deadline as +/// `Instant::now() + timeout`, which **panics** as soon as the resulting +/// instant is not representable — and a panic raised inside an +/// `extern "C"` frame aborts the host process rather than returning a +/// [`PlatformWalletFFIResult`]. `timeout_secs` arrives as an unrestricted +/// `u64`, so a caller passing `UInt64.max` (or any large sentinel) would +/// take the host down. Clamping is preferable to rejecting: every value +/// past this point already means "effectively forever" to a mobile host +/// that will not survive the wait anyway. +const MAX_TIMEOUT_SECS: u64 = 365 * 24 * 60 * 60; + +/// Convert an FFI `timeout_secs` into the `Option` the resume +/// path takes. +/// +/// `0` declines to specify a bound — `resume_asset_lock` reads the +/// resulting `None` as "apply the recovery policy's own default" (see +/// the `# Timeouts` section on [`asset_lock_manager_resume`]). Anything +/// larger than [`MAX_TIMEOUT_SECS`] is clamped so the deadline +/// arithmetic downstream stays representable. +fn resume_timeout(timeout_secs: u64) -> Option { + (timeout_secs != 0).then(|| Duration::from_secs(timeout_secs.min(MAX_TIMEOUT_SECS))) +} + +/// Records the bounds handed to `resume_asset_lock` on this thread. +/// +/// The bound is otherwise unobservable from outside: it is consumed deep +/// inside the async resume, behind a manager handle, on a path that by +/// definition does not return until it expires. Recording it is what lets +/// the clamp be pinned on the **exported** paths rather than on the private +/// conversion helper — an entry point that converts `timeout_secs` some +/// other way, or drops the converted bound on its way to the manager, +/// records something different and fails the pins below instead of quietly +/// restoring the host abort an unrepresentable deadline causes in an +/// `extern "C"` frame. +/// +/// Thread-local, so each test owns the record of the calls it made. +#[cfg(test)] +mod timeout_probe { + use std::cell::RefCell; + use std::time::Duration; + + thread_local! { + static RECORDED: RefCell>> = const { RefCell::new(Vec::new()) }; + } + + pub(super) fn record(timeout: Option) { + RECORDED.with(|recorded| recorded.borrow_mut().push(timeout)); + } + + /// Everything recorded on this thread since the last take, in call order. + pub(super) fn take() -> Vec> { + RECORDED.with(|recorded| recorded.borrow_mut().drain(..).collect()) + } +} + +/// Pass-through that records the bound under `cfg(test)`. +/// +/// It stands **in the argument position** of the `resume_asset_lock` calls +/// below rather than beside them, and that placement is the whole point: a +/// recording statement next to the call still records the right value when +/// the call itself is handed a different one, so the pin passes while the +/// clamp is gone. Being the argument, what is recorded is what is consumed +/// — there is no value in between to diverge. +#[cfg(test)] +fn forwarded(timeout: Option) -> Option { + timeout_probe::record(timeout); + timeout +} + +/// Production twin of the `cfg(test)` recorder above: the identity, inlined +/// away. No ABI, no state, no branch. +#[cfg(not(test))] +#[inline(always)] +fn forwarded(timeout: Option) -> Option { + timeout +} + /// Build an `OutPoint` from a 32-byte raw txid pointer and a vout. /// /// **FFI invariant:** the `txid` parameter is typed `*const [u8; 32]`, @@ -47,9 +126,12 @@ fn parse_outpoint(txid: *const [u8; 32], vout: u32) -> dashcore::OutPoint { /// # Timeouts /// /// `timeout_secs` bounds only the stages that still have to WAIT for a -/// proof (`Built` / `Broadcast`, plus the defensive proof-less -/// `RecoveredFromChain` fallback). `InstantSendLocked` / `ChainLocked` -/// already carry a proof and return without ever consulting it. +/// proof: a `Built` / `Broadcast` row whose local record does not +/// already hold finality, plus the defensive proof-less +/// `RecoveredFromChain` fallback. Every other resume — a row carrying +/// an `InstantSendLocked` / `ChainLocked` proof, and a `Built` / +/// `Broadcast` row the local finality probe settles before any +/// transport work — returns without ever consulting it. /// /// `timeout_secs == 0` does **not** request an unbounded wait — it /// declines to specify one, and `resume_asset_lock` then applies the @@ -76,9 +158,15 @@ fn parse_outpoint(txid: *const [u8; 32], vout: u32) -> dashcore::OutPoint { /// `Broadcast` arms the expiry surfaces as /// `TransactionBroadcastUnconfirmed`. /// -/// A non-zero `timeout_secs` keeps its exact semantics, -/// `FinalityTimeout` included — the substitution above is gated on the -/// caller having declined to choose. +/// A `timeout_secs` in `1..=31_536_000` (one year) keeps its exact +/// semantics, `FinalityTimeout` included — the substitution above is +/// gated on the caller having declined to choose. Anything larger is +/// silently CLAMPED to one year rather than honoured or rejected: the +/// wait paths downstream build their deadline as `Instant::now() + +/// timeout` and panic on an unrepresentable instant, which in an +/// `extern "C"` frame aborts the host process. A caller passing +/// `UInt64.max` as an "effectively forever" sentinel therefore gets one +/// year, not forever. #[no_mangle] pub unsafe extern "C" fn asset_lock_manager_resume( handle: Handle, @@ -95,14 +183,10 @@ pub unsafe extern "C" fn asset_lock_manager_resume( check_ptr!(out_derivation_path); let out_point = parse_outpoint(txid, vout); - // `timeout_secs == 0` declines to specify a bound. `resume_asset_lock` - // reads the resulting `None` as "apply the recovery policy's default": - // the 180s `UNCONFIRMED_BROADCAST_PROOF_TIMEOUT` on every proof-waiting - // arm. See this function's `# Timeouts` section. - let timeout = (timeout_secs != 0).then(|| Duration::from_secs(timeout_secs)); + let timeout = resume_timeout(timeout_secs); let option = ASSET_LOCK_MANAGER_STORAGE.with_item(handle, |manager| { - runtime().block_on(manager.resume_asset_lock(&out_point, timeout)) + runtime().block_on(manager.resume_asset_lock(&out_point, forwarded(timeout))) }); let result = unwrap_option_or_return!(option); let (proof, path) = unwrap_result_or_return!(result); @@ -143,8 +227,11 @@ pub unsafe extern "C" fn asset_lock_manager_resume( /// `UNCONFIRMED_BROADCAST_PROOF_TIMEOUT`, and it applies to every arm /// that waits for a proof — a `Built` re-broadcast whatever the /// broadcaster answered, a `Broadcast` row, the defensive proof-less -/// `RecoveredFromChain` fallback. Pass a non-zero `timeout_secs` for -/// a different upper bound. +/// `RecoveredFromChain` fallback. Pass a `timeout_secs` in +/// `1..=31_536_000` (one year) for a different upper bound; larger +/// values are clamped to one year, because the deadline arithmetic +/// downstream panics on an unrepresentable instant and that panic +/// aborts the host process from an `extern "C"` frame. /// /// That policy is what makes this entry point safe to fan out at /// launch. The catch-up sweep starts one call per stuck lock; when @@ -165,11 +252,7 @@ pub unsafe extern "C" fn asset_lock_manager_catch_up_blocking( check_ptr!(txid); let out_point = parse_outpoint(txid, vout); - // `timeout_secs == 0` declines to specify a bound. `resume_asset_lock` - // reads the resulting `None` as "apply the recovery policy's default": - // the 180s `UNCONFIRMED_BROADCAST_PROOF_TIMEOUT` on every proof-waiting - // arm. See this function's `# Timeouts` section. - let timeout = (timeout_secs != 0).then(|| Duration::from_secs(timeout_secs)); + let timeout = resume_timeout(timeout_secs); tracing::info!( outpoint = %out_point, @@ -178,7 +261,7 @@ pub unsafe extern "C" fn asset_lock_manager_catch_up_blocking( ); let option = ASSET_LOCK_MANAGER_STORAGE.with_item(handle, |manager| { - runtime().block_on(manager.resume_asset_lock(&out_point, timeout)) + runtime().block_on(manager.resume_asset_lock(&out_point, forwarded(timeout))) }); let result = match option { Some(r) => r, @@ -288,3 +371,212 @@ pub unsafe extern "C" fn asset_lock_manager_recover( unwrap_option_or_return!(option); PlatformWalletFFIResult::ok() } + +#[cfg(test)] +mod tests { + use super::{ + asset_lock_manager_catch_up_blocking, asset_lock_manager_resume, resume_timeout, + timeout_probe, MAX_TIMEOUT_SECS, + }; + use crate::error::PlatformWalletFFIResultCode; + use crate::handle::{Handle, ASSET_LOCK_MANAGER_STORAGE}; + use crate::runtime::runtime; + use platform_wallet::test_support::{test_platform_wallet_manager, NoopTestPersister}; + use platform_wallet::PlatformWalletManager; + use std::os::raw::c_char; + use std::sync::Arc; + use std::time::Duration; + + /// The clamp both exported paths must apply. + fn one_year() -> Option { + Some(Duration::from_secs(MAX_TIMEOUT_SECS)) + } + + /// A handle over a real, wallet-backed `AssetLockManager`. + /// + /// A live manager is what makes these pins bite. `HandleStorage::with_item` + /// runs its closure only for a handle it finds, so an absent handle returns + /// before the entry point ever reaches `resume_asset_lock` — and a pin that + /// observes the bound short of that call cannot tell a forwarded bound from + /// a dropped one. The wallet tracks no asset locks, so the resume looks the + /// outpoint up, fails `AssetLockNotTracked` and returns without waiting on + /// anything. + /// + /// The returned manager owns the registered wallet and must be kept alive + /// for the duration of the call. + fn live_manager_handle() -> (Arc>, Handle) { + runtime().block_on(async { + let (manager, wallet_id) = test_platform_wallet_manager().await; + let wallet = manager + .get_wallet(&wallet_id) + .await + .expect("the manager just registered this wallet"); + let handle = ASSET_LOCK_MANAGER_STORAGE.insert(Arc::clone(wallet.asset_locks())); + (manager, handle) + }) + } + + /// Drive [`asset_lock_manager_resume`] against a live manager and report + /// the bound it forwarded alongside its result code. + /// + /// Called from the plain test thread: the entry point does its own + /// `runtime().block_on(...)`, exactly as the host thread does, and nesting + /// that inside an outer `block_on` would abort. + fn resume_extern_with( + handle: Handle, + timeout_secs: u64, + ) -> (PlatformWalletFFIResultCode, Vec>) { + let _ = timeout_probe::take(); + let txid = [7u8; 32]; + let mut proof_bytes: *mut u8 = std::ptr::null_mut(); + let mut proof_len: usize = 0; + let mut derivation_path: *mut c_char = std::ptr::null_mut(); + + let result = unsafe { + asset_lock_manager_resume( + handle, + &txid, + 0, + timeout_secs, + &mut proof_bytes, + &mut proof_len, + &mut derivation_path, + ) + }; + + (result.code, timeout_probe::take()) + } + + /// Drive [`asset_lock_manager_catch_up_blocking`], as above. + fn catch_up_extern_with( + handle: Handle, + timeout_secs: u64, + ) -> (PlatformWalletFFIResultCode, Vec>) { + let _ = timeout_probe::take(); + let txid = [7u8; 32]; + + let result = + unsafe { asset_lock_manager_catch_up_blocking(handle, &txid, 0, timeout_secs) }; + + (result.code, timeout_probe::take()) + } + + /// The clamp has to hold on the ABI itself, not just in the helper. + /// + /// `asset_lock_manager_resume` is one of the two symbols a host can + /// actually reach, and the conversion helper is private: an entry point + /// that stops routing `timeout_secs` through it — or that grows a third + /// conversion of its own, or hands the manager something else entirely — + /// restores the host abort while every helper test stays green. So + /// exercise the exported symbol with the sentinel hosts really pass + /// (`UInt64.max`), against a manager that actually consumes the bound, and + /// pin what arrives there. + #[test] + fn the_exported_resume_clamps_an_extreme_timeout() { + let (_manager, handle) = live_manager_handle(); + + let (code, recorded) = resume_extern_with(handle, u64::MAX); + + assert_eq!( + code, + PlatformWalletFFIResultCode::ErrorAssetLockNotTracked, + "the resume must have run — this code comes from inside \ + `resume_asset_lock`, past the handle lookup, so the bound below \ + is one the manager was really called with" + ); + assert_eq!( + recorded, + vec![one_year()], + "the exported resume must hand the manager `UInt64.max` clamped to \ + one year" + ); + let bound = recorded[0].expect("a non-zero request is bounded"); + // The arithmetic downstream. Aborts the host, from an `extern "C"` + // frame, on an unrepresentable instant. + let _deadline = std::time::Instant::now() + bound; + + ASSET_LOCK_MANAGER_STORAGE.remove(handle); + } + + /// Same contract on the launch catch-up symbol, which is the one the + /// hosts fan out at app start — and therefore the one that takes the + /// process down if its bound is unrepresentable. + #[test] + fn the_exported_catch_up_clamps_an_extreme_timeout() { + let (_manager, handle) = live_manager_handle(); + + let (code, recorded) = catch_up_extern_with(handle, u64::MAX); + + assert_eq!( + code, + PlatformWalletFFIResultCode::ErrorWalletOperation, + "the catch-up maps every resume failure but a double-spend verdict \ + to this one code, so this proves the resume ran rather than the \ + handle lookup missing" + ); + assert_eq!( + recorded, + vec![one_year()], + "the exported catch-up must hand the manager `UInt64.max` clamped \ + to one year" + ); + let bound = recorded[0].expect("a non-zero request is bounded"); + let _deadline = std::time::Instant::now() + bound; + + ASSET_LOCK_MANAGER_STORAGE.remove(handle); + } + + /// Neither exported path may reshape the values a host legitimately + /// passes: `0` still declines to specify a bound (and draws the recovery + /// policy's own default downstream), and the 300s the launch catch-up + /// asks for arrives intact. + #[test] + fn the_exported_paths_leave_ordinary_timeouts_alone() { + let (_manager, handle) = live_manager_handle(); + + assert_eq!(resume_extern_with(handle, 0).1, vec![None]); + assert_eq!(catch_up_extern_with(handle, 0).1, vec![None]); + assert_eq!( + resume_extern_with(handle, 300).1, + vec![Some(Duration::from_secs(300))] + ); + assert_eq!( + catch_up_extern_with(handle, 300).1, + vec![Some(Duration::from_secs(300))] + ); + + ASSET_LOCK_MANAGER_STORAGE.remove(handle); + } + + /// `timeout_secs` is an unrestricted `u64` on both public resume entry + /// points. The wait paths it feeds build their deadline as + /// `Instant::now() + timeout`, which panics once that instant is not + /// representable — and a panic in an `extern "C"` frame aborts the host + /// process instead of returning a result code. A caller passing + /// `UInt64.max` must therefore get a long wait, not a crashed app. + #[test] + fn an_extreme_timeout_stays_a_representable_deadline() { + let timeout = resume_timeout(u64::MAX).expect("a non-zero request is bounded"); + + // The arithmetic the resume path performs on this value. Panics — + // and so aborts the host — on an unrepresentable instant. + let _deadline = std::time::Instant::now() + timeout; + assert_eq!(timeout, Duration::from_secs(MAX_TIMEOUT_SECS)); + } + + /// Zero keeps its meaning: it declines to specify a bound and lets the + /// recovery policy pick its own default. It is NOT a + /// request for a zero-length wait, which would expire every proof wait + /// instantly. + #[test] + fn zero_declines_to_specify_a_bound() { + assert_eq!(resume_timeout(0), None); + } + + /// An ordinary request passes through untouched — the clamp must not + /// quietly reshape the 300s ceiling the launch catch-up asks for. + #[test] + fn an_ordinary_timeout_passes_through_unchanged() { + assert_eq!(resume_timeout(300), Some(Duration::from_secs(300))); + } +} diff --git a/packages/rs-platform-wallet-ffi/src/error.rs b/packages/rs-platform-wallet-ffi/src/error.rs index 63cb50a152c..451b4e72350 100644 --- a/packages/rs-platform-wallet-ffi/src/error.rs +++ b/packages/rs-platform-wallet-ffi/src/error.rs @@ -1520,18 +1520,20 @@ mod tests { /// Code 26 is a promise about cleanup, not about the broadcaster's /// verdict: the row was untracked and the funding reservation released, /// so a rebuild is safe. An asset-lock build whose rejection raced a - /// concurrent resume keeps both — the guard retains the advanced row and - /// the release is skipped — and reports the unknown outcome instead. The - /// two must never collapse to one code across the boundary: a host that - /// read 26 there would rebuild from other UTXOs and create a second asset - /// lock beside a transaction the advance says reached the network. + /// concurrent resume keeps both — a guard retains the row, either + /// because the resume already advanced it or because the resume holds + /// its dispatch window, and the release is skipped — and reports the + /// unknown outcome instead. The two must never collapse to one code + /// across the boundary: a host that read 26 there would rebuild from + /// other UTXOs and create a second asset lock beside a transaction that + /// has either reached the network already or is about to. #[test] fn a_retained_asset_lock_row_reports_the_unknown_outcome_not_the_rejection() { let retained: PlatformWalletFFIResult = PlatformWalletError::TransactionBroadcastUnconfirmed( "asset lock 0000..:0 stays tracked and reserved: the broadcast was \ - rejected, but a concurrent resume had already advanced the row past \ - Built, so the transaction may be on the network" + rejected, but a concurrent resume is driving the same row, so the \ + transaction may be on the network or about to reach it" .to_string(), ) .into(); diff --git a/packages/rs-platform-wallet/src/broadcaster.rs b/packages/rs-platform-wallet/src/broadcaster.rs index 633af12ae25..04d2aca2bd8 100644 --- a/packages/rs-platform-wallet/src/broadcaster.rs +++ b/packages/rs-platform-wallet/src/broadcaster.rs @@ -63,13 +63,42 @@ impl From for PlatformWalletError { /// peer echo / InstantSend lock / confirmation, or by an accepting Core /// endpoint. A successful P2P socket write alone must never satisfy this /// contract. +/// `'static` because the asset-lock manager hands a shared handle onto +/// itself — broadcaster included — to background tasks (the +/// readiness-deferred resume retry), and a spawned task cannot borrow. +/// Every broadcaster is an owned struct anyway; the bound only makes that +/// requirement explicit. #[async_trait] -pub trait TransactionBroadcaster: Send + Sync { +pub trait TransactionBroadcaster: Send + Sync + 'static { /// Contract: [`BroadcastError::Rejected`] is allowed only when the /// transaction definitively did not enter the network. Any timeout, /// 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. Returns whether readiness was + /// reached. + /// + /// Callers that resume queued work at app start use this so they do not + /// race a transport that is still coming up. Losing that race is not a + /// retryable stumble: a transport that never dispatched reports + /// [`BroadcastError::Rejected`], the resume paths treat that verdict as + /// definitive, and nothing reschedules them — so the transaction stays + /// un-broadcast for the whole session. + /// + /// The bound is a plain `Duration` rather than an `Option`, so an + /// unbounded readiness wait is unrepresentable. These waits run under + /// the FFI's `runtime().block_on(...)`, on host threads the host also + /// needs in order to *start* the very transport being waited for; a wait + /// with no ceiling there is a deadlock, not a delay. + /// + /// 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: Duration) -> bool { + true + } } /// Broadcasts transactions via Platform's DAPI gRPC endpoint. @@ -148,6 +177,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: Duration) -> bool; } #[async_trait] @@ -160,6 +194,10 @@ impl SpvChannel for SpvRuntime { self.broadcast_transaction_and_wait(transaction, timeout) .await } + + async fn wait_until_ready(&self, timeout: Duration) -> bool { + SpvRuntime::wait_until_ready(self, timeout).await + } } /// Broadcasts purely over the SPV P2P network — no DAPI involvement. @@ -219,6 +257,10 @@ impl TransactionBroadcaster for SpvBroadcaster { Err(other) => Err(other), } } + + async fn wait_until_ready(&self, timeout: Duration) -> bool { + self.spv.wait_until_ready(timeout).await + } } #[cfg(test)] @@ -231,6 +273,8 @@ mod tests { struct AcceptanceSpy { calls: AtomicUsize, verdict: Mutex>>, + /// Every readiness budget the channel was handed, in call order. + readiness_budgets: Mutex>, } impl AcceptanceSpy { @@ -238,6 +282,7 @@ mod tests { Self { calls: AtomicUsize::new(0), verdict: Mutex::new(Some(verdict)), + readiness_budgets: Mutex::new(Vec::new()), } } } @@ -256,6 +301,14 @@ mod tests { .take() .expect("one acceptance check") } + + async fn wait_until_ready(&self, timeout: Duration) -> bool { + self.readiness_budgets + .lock() + .expect("readiness budget mutex") + .push(timeout); + true + } } fn transaction() -> Transaction { @@ -310,4 +363,29 @@ mod tests { } } } + + /// The readiness gate the resume paths depend on has to reach the SPV + /// channel, budget intact. Callers can only observe readiness through + /// `TransactionBroadcaster`, so a `SpvBroadcaster` that silently kept + /// the trait's "always ready" default would report a transport that has + /// not started as ready and hand the resume straight back into the + /// never-sent rejection this gate exists to avoid — with every + /// recovery-level test still green. + #[tokio::test] + async fn spv_broadcaster_delegates_readiness_to_the_spv_channel() { + let spv = Arc::new(AcceptanceSpy::with(Ok(BroadcastResult::Accepted { + relayed_by: 1, + }))); + let broadcaster = SpvBroadcaster::from_channel(spv.clone()); + + assert!(broadcaster.wait_until_ready(Duration::from_secs(7)).await); + + assert_eq!( + *spv.readiness_budgets + .lock() + .expect("readiness budget mutex"), + vec![Duration::from_secs(7)], + "readiness must reach the SPV channel with the caller's budget" + ); + } } diff --git a/packages/rs-platform-wallet/src/spv/runtime.rs b/packages/rs-platform-wallet/src/spv/runtime.rs index 4839b891e7b..c98ebeabbb7 100644 --- a/packages/rs-platform-wallet/src/spv/runtime.rs +++ b/packages/rs-platform-wallet/src/spv/runtime.rs @@ -50,6 +50,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 @@ -197,6 +201,47 @@ 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 + /// classified by [`classify_spv_send_error`]. + 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. Returns whether readiness was reached. + /// + /// This closes the launch race where work resumed at app start (the + /// 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 — 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. + /// + /// The bound is a plain `Duration` and is applied with + /// [`tokio::time::timeout`], which saturates an unrepresentable deadline + /// instead of panicking the way `Instant::now() + timeout` does. That + /// matters because callers reach here through `extern "C"` entry points + /// whose timeout arrives as an unrestricted `u64`, and a panic in an + /// 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; + } + }) + .await + .is_ok() + } + /// Broadcast a transaction through SPV peers and wait for dash-spv's /// network-acceptance verdict. /// @@ -798,6 +843,7 @@ impl std::fmt::Debug for SpvRuntime { #[cfg(test)] mod tests { use std::sync::Arc; + use std::time::Duration; use dash_spv::error::{NetworkError, SpvError}; use dashcore::Network; @@ -856,6 +902,121 @@ mod tests { ); } + /// The readiness predicate must fail closed on an unstarted client: + /// the acceptance path rejects that state before any send, so reporting + /// it ready hands the caller straight back into the never-sent verdict + /// the gate exists to avoid. + #[tokio::test(start_paused = true)] + async fn readiness_is_not_reached_while_the_client_is_unstarted() { + let wallet_manager = Arc::new(RwLock::new(WalletManager::::new( + Network::Testnet, + ))); + let runtime = SpvRuntime::new(wallet_manager, Arc::new(PlatformEventManager::new(vec![]))); + + assert!( + !runtime.wait_until_ready(Duration::from_secs(30)).await, + "an unstarted client must never report broadcast-ready" + ); + } + + /// An `extern "C"` caller supplies the readiness budget as an + /// unrestricted `u64` of seconds. Building the deadline with + /// `Instant::now() + timeout` panics once that instant is not + /// representable, and a panic inside an FFI frame aborts the host + /// process instead of returning a result code — so the wait has to + /// survive an extreme budget rather than take the host down with it. + #[tokio::test(start_paused = true)] + async fn an_extreme_readiness_budget_does_not_panic() { + let wallet_manager = Arc::new(RwLock::new(WalletManager::::new( + Network::Testnet, + ))); + let runtime = SpvRuntime::new(wallet_manager, Arc::new(PlatformEventManager::new(vec![]))); + + // Never resolves (the client is unstarted), so cut it short: the + // assertion here is that constructing the wait survives, not that + // it finishes. + let outcome = tokio::time::timeout( + Duration::from_secs(1), + runtime.wait_until_ready(Duration::MAX), + ) + .await; + + assert!(outcome.is_err(), "an extreme budget must park, not resolve"); + } + + /// A started client with no peers is the OTHER pre-send rejection, and + /// readiness has to observe both halves and then actually resolve. + /// + /// The launch race this gate exists for ends the moment dash-spv reports + /// its first connection, so the predicate must go from false to true on + /// that event alone — with no restart, and without the caller polling + /// anything itself. A predicate that only ever reported false would keep + /// every recovery test green (they all assert around an expired wait) + /// while turning the gate into a fixed 15s delay before the same + /// never-sent broadcast, which is strictly worse than not waiting. + /// + /// This starts a real client — offline, restricted to a configured peer + /// list that is empty, so it opens its storage and connects to nothing — + /// because "started" is exactly the half a double cannot stand in for. + #[tokio::test(start_paused = true)] + async fn readiness_arrives_when_a_started_client_reports_its_first_peer() { + use dash_spv::network::NetworkEvent; + use dash_spv::{ClientConfig, EventHandler}; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + + // `DiskStorageManager` locks the directory it opens, so the client + // gets one of its own and the stop below releases it. + let storage = std::env::temp_dir().join(format!( + "platform-wallet-spv-readiness-{}", + std::process::id() + )); + std::fs::create_dir_all(&storage).expect("private storage dir"); + let wallet_manager = Arc::new(RwLock::new(WalletManager::::new( + Network::Testnet, + ))); + let runtime = SpvRuntime::new(wallet_manager, Arc::new(PlatformEventManager::new(vec![]))); + runtime + .start( + ClientConfig::testnet() + .with_storage_path(&storage) + .with_restrict_to_configured_peers(true), + ) + .await + .expect("an offline client with no configured peers still starts"); + assert!(runtime.is_started(), "the client must be started"); + + assert!( + !runtime.wait_until_ready(Duration::from_secs(30)).await, + "a started client with no connected peers must not report ready — \ + dash-spv's zero-peer check rejects the send before it dispatches, \ + exactly like an unstarted client" + ); + + // The event dash-spv pushes to its handlers on the first connection. + runtime + .peer_tracker + .on_network_event(&NetworkEvent::PeersUpdated { + connected_count: 1, + addresses: vec![SocketAddr::new( + IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), + 19999, + )], + best_height: Some(1_100_000), + }); + + assert!( + runtime.wait_until_ready(Duration::from_secs(30)).await, + "a started client that has just reported its first peer must \ + report ready — this transition is the whole point of the wait" + ); + + runtime + .stop() + .await + .expect("clean stop releases the data dir"); + let _ = std::fs::remove_dir_all(&storage); + } + /// Every other error on the acceptance path may follow a partial send /// and must stay `MaybeSent`. #[test] diff --git a/packages/rs-platform-wallet/src/wallet/asset_lock/build.rs b/packages/rs-platform-wallet/src/wallet/asset_lock/build.rs index 3b24cec9b40..a70d5805ff0 100644 --- a/packages/rs-platform-wallet/src/wallet/asset_lock/build.rs +++ b/packages/rs-platform-wallet/src/wallet/asset_lock/build.rs @@ -1125,6 +1125,11 @@ impl AssetLockManager { // inputs are re-spendable. A `MaybeSent` failure keeps both the // reservation and the resumable row. // + // A resume that has snapshotted the row but not yet recorded its + // send is still `Built`, so `claim_resume_dispatch` excludes the + // removal. The exclusion survives an ambiguous cancellation after + // a possible send or observed proof, until the row advances. + // // The reported error type and the in-broadcast fence both follow the // cleanup, never the broadcaster's verdict alone — they are decided // by the one predicate. The definite-rejection contract is reported @@ -1150,12 +1155,12 @@ impl AssetLockManager { // spend can still arrive. let cs_untrack = self.untrack_asset_lock(&out_point).await; // Release only when the Built row was actually removed. If - // the untrack guard fired instead — a concurrent - // `resume_asset_lock` advanced the row past `Built`, positive - // evidence the transaction reached the network after all — - // the inputs must stay reserved exactly like a `MaybeSent` - // outcome, or the still-tracked row would be resumable while - // its inputs are re-spendable. + // an untrack guard fired instead — the row advanced past + // `Built`, or active/sticky resume state says this transaction + // may already be live or committed to dispatch — the inputs + // must stay reserved exactly like a `MaybeSent` outcome, or the + // still-tracked row would be resumable while its inputs are + // re-spendable. let removed_built_row = cs_untrack.removed.contains(&out_point); self.queue_asset_lock_changeset(cs_untrack); if removed_built_row { @@ -1184,36 +1189,46 @@ impl AssetLockManager { .await; in_broadcast_pin.settle_released(); } else { - // The untrack guard fired: a concurrent `resume_asset_lock` - // advanced the row past `Built`, which is positive evidence - // the transaction reached the network after all. The - // reservation stays held, and so must the fence. + // An untrack guard fired: the row advanced past `Built`, or + // resume state says the transaction may already be live or + // committed to dispatch. The reservation stays held, and + // so must the fence. in_broadcast_pin.settle_pending_spend(); // The cleanup did not run, so the definite-rejection // contract does not hold either. `TransactionBroadcast` // promises the caller that the row is gone, the inputs are - // free, and a rebuild is safe; here the advanced row is - // still tracked and resumable and its inputs are still - // reserved and fenced, so a caller honouring that promise - // would rebuild from other UTXOs and create a SECOND asset - // lock beside a transaction the advance says reached the - // network. The contract that matches what is actually true - // is the unknown outcome: do not retry, the row and its - // reservation are intact, resume the existing lock. + // free, and a rebuild is safe; here the row is still + // tracked and resumable and its inputs are still reserved + // and fenced, so a caller honouring that promise would + // rebuild from other UTXOs and create a SECOND asset lock + // beside a transaction that has either reached the network + // already or is about to. The contract that matches what is + // actually true is the unknown outcome: do not retry, the + // row and its reservation are intact, resume the existing + // lock. + // + // The price is that the reservation and the fence outlive + // this call: the fence ends on an observed spend, and no + // second cleanup pass exists to reconsider once the + // concurrent resume settles. That is the same price every + // ambiguous outcome already pays, and it is the only side + // that is safe to be wrong on — the row stays tracked and + // resumable, so the value behind it is recovered by a + // later resume rather than lost. tracing::warn!( %txid, error = %e, - "asset lock broadcast was rejected, but a concurrent resume had \ - already advanced the row past Built; keeping the row and its \ - funding reservation and reporting an unknown outcome rather than \ - a definite rejection" + "asset lock broadcast was rejected, but resume state excludes \ + cleanup of the same row; keeping the row and its funding \ + reservation and reporting an unknown outcome rather than a \ + definite rejection" ); return Err(PlatformWalletError::TransactionBroadcastUnconfirmed( format!( "asset lock {out_point} stays tracked and reserved: the \ - broadcast was rejected, but a concurrent resume had already \ - advanced the row past Built, so the transaction may be on \ - the network: {e}" + broadcast was rejected, but resume state excludes cleanup \ + of the same row, so the transaction may be on the network \ + or committed to dispatch: {e}" ), )); } @@ -1334,9 +1349,16 @@ fn map_builder_error(e: AssetLockError, requested: u64) -> PlatformWalletError { #[cfg(test)] mod tests { use std::sync::{Arc, Mutex}; + use std::time::Duration; use dashcore::OutPoint; use key_wallet::account::account_type::StandardAccountType; + use key_wallet::account::AccountType; + use key_wallet::managed_account::managed_account_trait::ManagedAccountTrait; + use key_wallet::managed_account::transaction_record::{ + TransactionDirection, TransactionRecord, + }; + use key_wallet::transaction_checking::{TransactionContext, TransactionType}; use key_wallet::wallet::managed_wallet_info::transaction_building::AccountTypePreference; use tokio::sync::Notify; @@ -1850,9 +1872,12 @@ mod tests { /// Like [`funded_asset_lock_manager`] but over a caller-built persistence /// stub (e.g. one with `fail_flush` set). - async fn funded_asset_lock_manager_with_persistence( + async fn funded_asset_lock_manager_with_persistence< + B: TransactionBroadcaster, + P: PlatformWalletPersistence + 'static, + >( broadcaster: Arc, - persistence: Arc, + persistence: Arc

, ) -> (Arc>, WalletSigner) { let (wallet_manager, wallet_id, _balance, signer) = funded_wallet_manager(StandardAccountType::BIP44Account).await; @@ -2758,4 +2783,796 @@ mod tests { rebuilt.input.len() ); } + + /// Broadcaster double that sequences the initial build's definite + /// rejection against a resume parked in the transport-readiness wait. + /// + /// The first `broadcast` call is the build's: it announces that the + /// `Built` row is tracked and the send is in flight, then waits for the + /// test before returning the rejection that triggers the cleanup. Every + /// later call is a resume's and receives the verdict selected by the + /// test. `wait_until_ready` is the resume's park: it models a transport + /// that comes up exactly when the test says so, which is the only way to + /// hold a resume inside its dispatch window for the whole of the cleanup. + struct RejectTheBuildAndParkTheResume { + calls: std::sync::atomic::AtomicUsize, + /// Every transaction handed to the broadcaster with the verdict it + /// drew, in dispatch order — so a test can tell an attempt that was + /// refused before dispatch from one that actually went out. + dispatched: Mutex>, + at_broadcast: Arc, + reject_gate: Arc, + resume_parked: Arc, + transport_gate: Arc, + resume_sent: Option>, + accept_resume: bool, + } + + impl RejectTheBuildAndParkTheResume { + fn dispatched(&self) -> Vec<(Txid, bool)> { + self.dispatched.lock().expect("dispatch log mutex").clone() + } + } + + #[async_trait] + impl TransactionBroadcaster for RejectTheBuildAndParkTheResume { + async fn broadcast(&self, transaction: &Transaction) -> Result { + let call = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let accepted = call > 0 && self.accept_resume; + self.dispatched + .lock() + .expect("dispatch log mutex") + .push((transaction.txid(), accepted)); + if call == 0 { + self.at_broadcast.wait().await; + self.reject_gate.wait().await; + return Err(BroadcastError::Rejected { + reason: "bad-txns-inputs-missingorspent".to_string(), + }); + } + if !accepted { + return Err(BroadcastError::Rejected { + reason: "simulated pre-dispatch rejection".to_string(), + }); + } + if let Some(resume_sent) = &self.resume_sent { + resume_sent.wait().await; + } + Ok(transaction.txid()) + } + + async fn wait_until_ready(&self, _timeout: Duration) -> bool { + self.resume_parked.wait().await; + self.transport_gate.wait().await; + true + } + } + + /// Persistence double that supplies an InstantSend record from durable + /// storage and pauses on the validation lookup made only after the resume + /// has turned that record into a proof. + struct ObservableLocalProofPersistence { + stored: Mutex>, + record: Mutex>, + lookups: std::sync::atomic::AtomicUsize, + proof_observed: Notify, + proof_gate: std::sync::Barrier, + } + + impl ObservableLocalProofPersistence { + fn new() -> Self { + Self { + stored: Mutex::new(Vec::new()), + record: Mutex::new(None), + lookups: std::sync::atomic::AtomicUsize::new(0), + proof_observed: Notify::new(), + proof_gate: std::sync::Barrier::new(2), + } + } + + fn removed_outpoints(&self) -> Vec { + self.stored + .lock() + .expect("observable persistence mutex") + .iter() + .filter_map(|cs| cs.asset_locks.as_ref()) + .flat_map(|locks| locks.removed.iter().copied()) + .collect() + } + } + + impl PlatformWalletPersistence for ObservableLocalProofPersistence { + fn store( + &self, + _wallet_id: WalletId, + changeset: PlatformWalletChangeSet, + ) -> Result<(), PersistenceError> { + self.stored + .lock() + .expect("observable persistence mutex") + .push(changeset); + Ok(()) + } + + fn flush(&self, _wallet_id: WalletId) -> Result<(), PersistenceError> { + Ok(()) + } + + fn load(&self) -> Result { + Ok(ClientStartState::default()) + } + + fn get_core_tx_record( + &self, + _wallet_id: WalletId, + _txid: &Txid, + ) -> Result, PersistenceError> { + let lookup = self + .lookups + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if lookup == 1 { + self.proof_observed.notify_one(); + tokio::task::block_in_place(|| self.proof_gate.wait()); + } + Ok(self.record.lock().expect("observable record mutex").clone()) + } + } + + fn instant_send_record(transaction: Transaction) -> TransactionRecord { + TransactionRecord::new( + transaction, + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InstantSend( + dashcore::ephemerealdata::instant_lock::InstantLock::default(), + ), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + 0, + ) + } + + /// The single tracked asset-lock outpoint on the fixture wallet. + async fn the_only_tracked_outpoint( + wallet_manager: &Arc>>, + wallet_id: WalletId, + ) -> OutPoint { + let wm = wallet_manager.read().await; + let (_, info) = wm.get_wallet_and_info(&wallet_id).expect("wallet present"); + assert_eq!( + info.tracked_asset_locks.len(), + 1, + "the build must have tracked its Built row before broadcasting" + ); + *info + .tracked_asset_locks + .keys() + .next() + .expect("one tracked lock") + } + + /// THE FENCE VIOLATION: a definite rejection must not release an asset + /// lock's inputs while a resume is still on its way to broadcasting that + /// very transaction. + /// + /// A resume reads the tracked row, its transaction and its status out of + /// the map under one read guard, and only then waits for the broadcast + /// transport, sends, and records the send by advancing the row. The + /// initial build's cleanup runs against a snapshot that is already gone: + /// its removal guard asks whether the row is still `Built`, which a + /// resume that has not reached its status advance yet still is. So the + /// cleanup used to remove the row and release both the funding + /// reservation and the in-broadcast fence in the middle of the resume's + /// dispatch window — and the resume then put the original transaction on + /// the wire from inputs the definite-rejection contract had just told the + /// host were free to rebuild from. The result is a live asset lock + /// beside a replacement spending the same UTXOs. + /// + /// The interleaving is driven, not hoped for: the build suspends inside + /// `broadcast` with its row tracked, the resume is held inside the + /// transport-readiness wait, and only then is the rejection released. The + /// resume's send is deliberately allowed to succeed afterwards — the fix + /// is not to suppress it but to make sure nothing released its inputs + /// first, which is what the rebuild refusal below asserts. + /// + /// The resume names a one-millisecond budget purely so its downstream + /// proof wait cannot outlive the test; the transport wait is over long + /// before that, since it ends on the test's own signal. + #[tokio::test] + async fn a_rejection_cleanup_cannot_release_inputs_under_a_parked_resume() { + let (wallet_manager, wallet_id, _generation, signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let broadcaster = Arc::new(RejectTheBuildAndParkTheResume { + calls: std::sync::atomic::AtomicUsize::new(0), + dispatched: Mutex::new(Vec::new()), + at_broadcast: Arc::new(tokio::sync::Barrier::new(2)), + reject_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_parked: Arc::new(tokio::sync::Barrier::new(2)), + transport_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_sent: None, + accept_resume: true, + }); + let (manager, persistence) = asset_lock_manager_over( + Arc::clone(&wallet_manager), + wallet_id, + Arc::clone(&broadcaster), + ); + + let build = async { + manager + .create_funded_asset_lock_proof( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await + }; + + // Every rendezvous is bounded so that a regression which stops the + // build or the resume from reaching its half of the interleaving + // fails the test with the step it never got to, instead of hanging + // the suite on a barrier nobody will arrive at. + let reach = |step: &'static str, wait| async move { + tokio::time::timeout(Duration::from_secs(30), wait) + .await + .unwrap_or_else(|_| panic!("the interleaving never reached: {step}")); + }; + + let coordinator = async { + // The build is inside `broadcast`: signed, tracked at `Built`, + // nothing decided yet. + reach("the build's broadcast", broadcaster.at_broadcast.wait()).await; + let out_point = the_only_tracked_outpoint(&wallet_manager, wallet_id).await; + + // A resume of that same row snapshots it and parks in the + // transport wait, exactly as the launch catch-up does while the + // SPV client is still starting. + let resume = tokio::spawn({ + let manager = Arc::clone(&manager); + async move { + manager + .resume_asset_lock(&out_point, Some(Duration::from_millis(1))) + .await + } + }); + reach( + "the resume's transport wait", + broadcaster.resume_parked.wait(), + ) + .await; + + // Only now does the build's rejection — and its whole cleanup — + // run. + reach("the build's rejection", broadcaster.reject_gate.wait()).await; + (out_point, resume) + }; + + let (build_result, (out_point, resume)) = tokio::join!(build, coordinator); + + let Err(PlatformWalletError::TransactionBroadcastUnconfirmed(reason)) = &build_result + else { + panic!( + "with a resume holding the dispatch window the cleanup cannot run, so \ + the definite-rejection contract — row gone, inputs released, rebuild \ + safe — does not hold and must not be reported: {build_result:?}" + ); + }; + assert!( + reason.contains("resume"), + "the unknown outcome must name the concurrent resume that kept the row, \ + so the verdict is not read as an ordinary ambiguous broadcast: {reason}" + ); + { + let wm = wallet_manager.read().await; + let (_, info) = wm.get_wallet_and_info(&wallet_id).expect("wallet present"); + assert_eq!( + info.tracked_asset_locks + .get(&out_point) + .map(|lock| lock.status.clone()), + Some(AssetLockStatus::Built), + "the row the parked resume is about to broadcast must survive the cleanup" + ); + } + assert!( + persistence.removed_outpoints().is_empty(), + "and no persisted-row deletion may be queued for it either, got {:?}", + persistence.removed_outpoints() + ); + + // The assertion the whole test exists for: at this point the cleanup + // has finished and the resume has not sent yet. If the inputs were + // reusable here, the send that follows would land beside whatever the + // host rebuilt from them. + let rebuild = manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await; + match rebuild { + Err(PlatformWalletError::AssetLockInsufficientFunds { available: 0, .. }) => {} + other => panic!( + "the rejection cleanup released the inputs of a transaction a parked \ + resume is still about to broadcast — a rebuild from them creates a \ + second asset lock beside a live one: {other:?}" + ), + } + + // Bring the transport up and let the parked resume do exactly what it + // was always going to do. + tokio::time::timeout(Duration::from_secs(30), broadcaster.transport_gate.wait()) + .await + .expect("the resume must still be parked in the transport wait"); + let _ = resume.await.expect("resume task"); + let dispatched = broadcaster.dispatched(); + assert_eq!( + dispatched.len(), + 2, + "the test proves nothing unless the parked resume really did reach \ + the broadcaster after the cleanup ran: {dispatched:?}" + ); + assert!( + dispatched[1].1, + "and that send has to be one the transport accepted — a second \ + refusal would leave nothing on the wire and the interleaving \ + would be harmless for the wrong reason: {dispatched:?}" + ); + assert_eq!( + dispatched[0].0, dispatched[1].0, + "what went out is the very transaction the cleanup judged rejected" + ); + } + + /// A resume that proves its own attempt never left the device must + /// release its cleanup exclusion. If the initial build's rejection is + /// still waiting, that cleanup can then remove the dead row and release + /// its inputs under the ordinary definite-rejection contract. + #[tokio::test] + async fn should_release_cleanup_exclusion_after_a_predispatch_resume_rejection() { + let (wallet_manager, wallet_id, _generation, signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let broadcaster = Arc::new(RejectTheBuildAndParkTheResume { + calls: std::sync::atomic::AtomicUsize::new(0), + dispatched: Mutex::new(Vec::new()), + at_broadcast: Arc::new(tokio::sync::Barrier::new(2)), + reject_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_parked: Arc::new(tokio::sync::Barrier::new(2)), + transport_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_sent: None, + accept_resume: false, + }); + let (manager, persistence) = asset_lock_manager_over( + Arc::clone(&wallet_manager), + wallet_id, + Arc::clone(&broadcaster), + ); + + let build = async { + manager + .create_funded_asset_lock_proof( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await + }; + let coordinator = async { + tokio::time::timeout(Duration::from_secs(30), broadcaster.at_broadcast.wait()) + .await + .expect("the build must reach its broadcast"); + let out_point = the_only_tracked_outpoint(&wallet_manager, wallet_id).await; + let resume = tokio::spawn({ + let manager = Arc::clone(&manager); + async move { + manager + .resume_asset_lock(&out_point, Some(Duration::from_millis(1))) + .await + } + }); + tokio::time::timeout(Duration::from_secs(30), broadcaster.resume_parked.wait()) + .await + .expect("the resume must reach the transport wait"); + tokio::time::timeout(Duration::from_secs(30), broadcaster.transport_gate.wait()) + .await + .expect("the resume transport must be released"); + let resume_result = tokio::time::timeout(Duration::from_secs(30), resume) + .await + .expect("the rejected resume must finish") + .expect("the resume task must not panic"); + assert!( + matches!( + resume_result, + Err(PlatformWalletError::TransactionBroadcastUnconfirmed(_)) + ), + "the resume must report its pre-dispatch rejection as an unknown original \ + outcome: {resume_result:?}" + ); + + tokio::time::timeout(Duration::from_secs(30), broadcaster.reject_gate.wait()) + .await + .expect("the initial rejection must be released after the resume exits"); + out_point + }; + + let (build_result, out_point) = tokio::join!(build, coordinator); + assert!( + matches!( + build_result, + Err(PlatformWalletError::TransactionBroadcast(_)) + ), + "once the proven-predispatch resume releases its claim, the initial rejection can \ + restore the definite-rejection contract: {build_result:?}" + ); + assert!( + persistence.removed_outpoints().contains(&out_point), + "the rejected Built row must be queued for deletion" + ); + assert!( + !wallet_manager + .read() + .await + .get_wallet_and_info(&wallet_id) + .expect("wallet present") + .1 + .tracked_asset_locks + .contains_key(&out_point), + "the rejected Built row must be removed in memory" + ); + assert!( + manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await + .is_ok(), + "cleanup must make the definitely rejected transaction's inputs reusable" + ); + assert_eq!( + broadcaster + .dispatched() + .iter() + .map(|(_, sent)| *sent) + .collect::>(), + vec![false, false], + "both attempts must be proven pre-dispatch for cleanup to be safe" + ); + } + + /// Once a resumed send is observable, cancelling the future cannot make + /// the original build's cleanup treat that transaction as never sent. + #[tokio::test] + async fn cancelling_after_an_observable_resume_send_keeps_cleanup_excluded() { + let (wallet_manager, wallet_id, _generation, signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let broadcaster = Arc::new(RejectTheBuildAndParkTheResume { + calls: std::sync::atomic::AtomicUsize::new(0), + dispatched: Mutex::new(Vec::new()), + at_broadcast: Arc::new(tokio::sync::Barrier::new(2)), + reject_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_parked: Arc::new(tokio::sync::Barrier::new(2)), + transport_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_sent: Some(Arc::new(tokio::sync::Barrier::new(2))), + accept_resume: true, + }); + let (manager, persistence) = asset_lock_manager_over( + Arc::clone(&wallet_manager), + wallet_id, + Arc::clone(&broadcaster), + ); + + let build = async { + manager + .create_funded_asset_lock_proof( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await + }; + let coordinator = async { + tokio::time::timeout(Duration::from_secs(30), broadcaster.at_broadcast.wait()) + .await + .expect("the build must reach its broadcast"); + let out_point = the_only_tracked_outpoint(&wallet_manager, wallet_id).await; + let resume = tokio::spawn({ + let manager = Arc::clone(&manager); + async move { + manager + .resume_asset_lock(&out_point, Some(Duration::from_millis(1))) + .await + } + }); + tokio::time::timeout(Duration::from_secs(30), broadcaster.resume_parked.wait()) + .await + .expect("the resume must reach the transport wait"); + + let wallet_write = wallet_manager.write().await; + tokio::time::timeout(Duration::from_secs(30), broadcaster.transport_gate.wait()) + .await + .expect("the resume transport must be released"); + tokio::time::timeout( + Duration::from_secs(30), + broadcaster + .resume_sent + .as_ref() + .expect("resume send gate") + .wait(), + ) + .await + .expect("the resumed send must become observable"); + resume.abort(); + assert!( + resume + .await + .expect_err("the resume must be cancelled") + .is_cancelled(), + "the resume must be aborted before it records Broadcast" + ); + drop(wallet_write); + tokio::time::timeout(Duration::from_secs(30), broadcaster.reject_gate.wait()) + .await + .expect("the build rejection must be released"); + out_point + }; + + let (build_result, out_point) = tokio::join!(build, coordinator); + assert!( + matches!( + build_result, + Err(PlatformWalletError::TransactionBroadcastUnconfirmed(_)) + ), + "a cancellation after an observable send is an unknown outcome; cleanup must not \ + report that rebuilding is safe: {build_result:?}" + ); + assert!( + persistence.removed_outpoints().is_empty(), + "the possibly-live transaction must not be queued for deletion" + ); + assert_eq!( + wallet_manager + .read() + .await + .get_wallet_and_info(&wallet_id) + .expect("wallet present") + .1 + .tracked_asset_locks + .get(&out_point) + .map(|lock| lock.status.clone()), + Some(AssetLockStatus::Built), + "the cancelled resume did not record Broadcast, so the sticky exclusion must keep \ + the Built row" + ); + assert!( + matches!( + manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await, + Err(PlatformWalletError::AssetLockInsufficientFunds { available: 0, .. }) + ), + "the cleanup must not release inputs of a transaction whose send was observable" + ); + } + + /// A proof read from durable local state is equally conclusive: cancelling + /// before the row can record it must keep cleanup from releasing its spend. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelling_after_observing_local_proof_keeps_cleanup_excluded() { + let broadcaster = Arc::new(RejectTheBuildAndParkTheResume { + calls: std::sync::atomic::AtomicUsize::new(0), + dispatched: Mutex::new(Vec::new()), + at_broadcast: Arc::new(tokio::sync::Barrier::new(2)), + reject_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_parked: Arc::new(tokio::sync::Barrier::new(2)), + transport_gate: Arc::new(tokio::sync::Barrier::new(2)), + resume_sent: None, + accept_resume: true, + }); + let persistence = Arc::new(ObservableLocalProofPersistence::new()); + let (manager, signer) = funded_asset_lock_manager_with_persistence( + Arc::clone(&broadcaster), + Arc::clone(&persistence), + ) + .await; + let wallet_manager = Arc::clone(&manager.wallet_manager); + let wallet_id = manager.wallet_id; + + let build = async { + manager + .create_funded_asset_lock_proof( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await + }; + let coordinator = async { + tokio::time::timeout(Duration::from_secs(30), broadcaster.at_broadcast.wait()) + .await + .expect("the build must reach its broadcast"); + let (out_point, transaction) = { + let mut wm = wallet_manager.write().await; + let (_, info) = wm + .get_wallet_and_info_mut(&wallet_id) + .expect("wallet present"); + let lock = info + .tracked_asset_locks + .values() + .next() + .expect("one tracked lock"); + let out_point = lock.out_point; + let transaction = lock.transaction.clone(); + info.core_wallet + .accounts + .standard_bip44_accounts + .get_mut(&0) + .expect("BIP44 account 0") + .transactions_mut() + .remove(&transaction.txid()); + (out_point, transaction) + }; + *persistence.record.lock().expect("observable record mutex") = + Some(instant_send_record(transaction)); + + let resume = tokio::spawn({ + let manager = Arc::clone(&manager); + async move { manager.resume_asset_lock(&out_point, None).await } + }); + tokio::time::timeout( + Duration::from_secs(30), + persistence.proof_observed.notified(), + ) + .await + .expect("the resume must validate the local proof"); + let wallet_write = wallet_manager.write().await; + tokio::task::block_in_place(|| persistence.proof_gate.wait()); + resume.abort(); + assert!( + resume + .await + .expect_err("the resume must be cancelled") + .is_cancelled(), + "the resume must be aborted before it records the proof" + ); + drop(wallet_write); + tokio::time::timeout(Duration::from_secs(30), broadcaster.reject_gate.wait()) + .await + .expect("the build rejection must be released"); + out_point + }; + + let (build_result, out_point) = tokio::join!(build, coordinator); + assert!( + matches!( + build_result, + Err(PlatformWalletError::TransactionBroadcastUnconfirmed(_)) + ), + "a cancellation after observing local finality is an unknown outcome; cleanup must \ + not report that rebuilding is safe: {build_result:?}" + ); + assert!( + persistence.removed_outpoints().is_empty(), + "the locally-final transaction must not be queued for deletion" + ); + assert_eq!( + wallet_manager + .read() + .await + .get_wallet_and_info(&wallet_id) + .expect("wallet present") + .1 + .tracked_asset_locks + .get(&out_point) + .map(|lock| lock.status.clone()), + Some(AssetLockStatus::Built), + "the cancelled resume did not record its proof, so the sticky exclusion must keep \ + the Built row" + ); + assert!( + matches!( + manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await, + Err(PlatformWalletError::AssetLockInsufficientFunds { available: 0, .. }) + ), + "the cleanup must not release inputs spent by a locally-final transaction" + ); + } + + /// A claim that has not crossed the side-effect boundary is an RAII hold. + /// + /// `untrack_asset_lock` must refuse only while a resume is actually + /// inside its pre-dispatch window, and must remove the row the moment the + /// last releasable claim goes. Two claims are taken so the count itself is + /// pinned: releasing one of them must not be enough. + #[tokio::test] + async fn a_released_dispatch_claim_lets_the_cleanup_remove_the_row() { + let (manager, signer, _persistence) = + funded_asset_lock_manager(Arc::new(AlwaysMaybeSentBroadcaster)).await; + + // An ambiguous broadcast leaves the row tracked at `Built` — the + // state every rejection cleanup and every resume starts from. + let _ = manager + .create_funded_asset_lock_proof( + 1_000_000, + 0, + AssetLockFundingType::IdentityRegistration, + 0, + &signer, + ) + .await; + let out_point = the_only_tracked_outpoint(&manager.wallet_manager, manager.wallet_id).await; + + let first = manager.claim_resume_dispatch(out_point); + let second = manager.claim_resume_dispatch(out_point); + assert!( + manager + .untrack_asset_lock(&out_point) + .await + .removed + .is_empty(), + "a claimed row must survive the cleanup" + ); + + drop(first); + assert!( + manager + .untrack_asset_lock(&out_point) + .await + .removed + .is_empty(), + "one resume releasing its claim says nothing about the other: the \ + row must survive while any dispatch window is still open" + ); + + drop(second); + assert!( + manager + .untrack_asset_lock(&out_point) + .await + .removed + .contains(&out_point), + "with the last claim gone the cleanup must be free again — a claim \ + that outlived its resume would fence these inputs for the rest of \ + the session" + ); + let wm = manager.wallet_manager.read().await; + let (_, info) = wm + .get_wallet_and_info(&manager.wallet_id) + .expect("wallet present"); + assert!( + !info.tracked_asset_locks.contains_key(&out_point), + "and the row itself is gone, not just reported as removed" + ); + } } diff --git a/packages/rs-platform-wallet/src/wallet/asset_lock/manager.rs b/packages/rs-platform-wallet/src/wallet/asset_lock/manager.rs index b9c810b6d26..82197c9cac6 100644 --- a/packages/rs-platform-wallet/src/wallet/asset_lock/manager.rs +++ b/packages/rs-platform-wallet/src/wallet/asset_lock/manager.rs @@ -4,8 +4,10 @@ //! waiting for proofs, and tracking lifecycle status. Shared across sub-wallets //! via `Arc`. +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; +use dashcore::OutPoint; use tokio::sync::{Notify, RwLock}; use crate::broadcaster::TransactionBroadcaster; @@ -72,7 +74,62 @@ pub struct AssetLockManager { /// `wallet_manager.write()` guard would cause across the build→persist /// span. Deliberately NOT held across the broadcast/proof-wait — only the /// snapshot ordering needs serialization. - pub(super) build_persist_serial: tokio::sync::Mutex<()>, + /// + /// Shared (`Arc`) so a [`shared_handle`](Self::shared_handle) sees the + /// same gate. + pub(super) build_persist_serial: Arc>, + /// Outpoints whose `Built` row is excluded from rejected-build cleanup, + /// counted so concurrent resumes of the same lock each hold their own + /// claim. A live resume releases its claim on pre-dispatch exit. Once a + /// send may have had side effects or local finality has been observed, + /// cancellation leaves the claim sticky until the row advances beyond + /// `Built`. + /// + /// The claim is what keeps a parked resume from broadcasting a + /// transaction whose inputs the rejection cleanup has already released. + /// A resume snapshots the row, then waits for the broadcast transport, + /// sends, and only then records the send by advancing the row — + /// suspension points the whole way. The claim remains after an ambiguous + /// cancellation in that interval. The initial build's definite + /// pre-send rejection removes the still-`Built` row and releases both + /// the funding reservation and the in-broadcast fence, and its + /// removal-guard (row still `Built`) cannot see a resume that has not + /// reached its status advance yet. So without a claim the release can + /// land inside the resume's dispatch window, and the resume then puts + /// the original transaction on the wire from inputs a rebuild is free + /// to reselect. [`untrack_asset_lock`](Self::untrack_asset_lock) + /// therefore refuses the removal while a claim stands, which leaves the + /// reservation and fence held and downgrades the build's verdict to the + /// unknown outcome — exactly what it already reports when its guard + /// fires on an advanced row. + /// + /// Atomicity comes from the wallet lock, not from this mutex: the claim + /// is taken while the resume still holds the read guard it snapshotted + /// the row under, and read while the cleanup holds the write guard it + /// removes the row under. The two guards exclude each other, so either + /// the cleanup sees the claim and keeps the row, or it removed the row + /// before the resume could snapshot it and the resume finds nothing to + /// resume. This mutex is only ever held for map arithmetic — never + /// across an await. + /// + /// Per-manager rather than per-wallet state because a registered wallet + /// has exactly one `AssetLockManager`, shared as `Arc` + /// across every sub-wallet, so a build and a resume of the same lock + /// always meet here — the same reasoning that puts + /// `build_persist_serial` above on the manager. A manager handle that + /// outlives its registration keeps its own map, which claims nothing + /// about outpoints: a re-registration allocates a fresh funding index + /// and therefore a different funding transaction, so no build and resume + /// of ONE outpoint can end up on two maps. + pub(super) resume_dispatch_claims: Arc>>, + /// Outpoints whose resume gave up on the broadcast transport and now has + /// a readiness-deferred retry in flight — see + /// [`resume_when_transport_ready`](Self::resume_when_transport_ready). + /// Membership is what keeps a lock the host resumes repeatedly (launch, + /// foreground, reconnect) from stacking up one retry per attempt: a + /// second deferral while the first is still waiting is a no-op. Held + /// only for set arithmetic, never across an await. + pub(super) deferred_resumes: Arc>>, /// Test-only gauge of builds currently at or past the /// `build_persist_serial` gate within `broadcast_funded_asset_lock` /// (incremented before the `lock().await`, RAII-decremented on every @@ -82,7 +139,7 @@ pub struct AssetLockManager { /// build holds the lock, a gauge of 2 proves the second build cannot /// yet have collected its pool snapshot. #[cfg(test)] - pub(super) build_serial_gate: std::sync::atomic::AtomicUsize, + pub(super) build_serial_gate: Arc, } impl AssetLockManager { @@ -102,9 +159,36 @@ impl AssetLockManager { lock_notify, broadcaster, persister, - build_persist_serial: tokio::sync::Mutex::new(()), + build_persist_serial: Arc::new(tokio::sync::Mutex::new(())), + resume_dispatch_claims: Arc::new(std::sync::Mutex::new(BTreeMap::new())), + deferred_resumes: Arc::new(std::sync::Mutex::new(BTreeSet::new())), + #[cfg(test)] + build_serial_gate: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + } + } + + /// A second handle onto the SAME manager: every field is shared, so a + /// build, a resume or a claim made through either is visible through + /// the other. This is what a background task spawned from a `&self` + /// method holds, since such a method has no `Arc` to clone. + /// + /// Deliberately not `impl Clone`: a `Clone` bound invites treating the + /// value as copyable state, and the whole point of the manager's + /// per-wallet mutexes and claim maps is that there is exactly one of + /// each per registered wallet. + pub(super) fn shared_handle(&self) -> Self { + Self { + sdk: Arc::clone(&self.sdk), + wallet_manager: Arc::clone(&self.wallet_manager), + wallet_id: self.wallet_id, + lock_notify: Arc::clone(&self.lock_notify), + broadcaster: Arc::clone(&self.broadcaster), + persister: self.persister.clone(), + build_persist_serial: Arc::clone(&self.build_persist_serial), + resume_dispatch_claims: Arc::clone(&self.resume_dispatch_claims), + deferred_resumes: Arc::clone(&self.deferred_resumes), #[cfg(test)] - build_serial_gate: std::sync::atomic::AtomicUsize::new(0), + build_serial_gate: Arc::clone(&self.build_serial_gate), } } 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 e0a88a2e6f0..e3b58e07998 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 @@ -6,6 +6,8 @@ use crate::broadcaster::{BroadcastError, TransactionBroadcaster}; use std::collections::BTreeSet; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; use std::time::Duration; use dashcore::Address as DashAddress; @@ -193,6 +195,51 @@ impl AssetLockManager { // Resumable asset lock // --------------------------------------------------------------------------- +/// Longest a resume holds a broadcast waiting for the broadcaster's transport +/// to come up. +/// +/// Sized for the gap it exists to cover — an SPV client that is mid-startup +/// while the launch catch-up runs — and no larger. The ceiling is what keeps +/// the wait a delay rather than a deadlock: the hosts drive the catch-up +/// through the FFI's blocking entry point, several locks at a time, on the +/// same fixed-width thread pool the host itself needs in order to reach +/// `start_spv` and bring that transport up. Parking those threads without a +/// ceiling parks the transport's own start. +/// +/// Expiry costs nothing beyond the delay: the broadcast is attempted anyway +/// and reports exactly what it would have reported with no wait at all. +const BROADCAST_TRANSPORT_READY_WAIT: Duration = Duration::from_secs(15); + +/// Longest a readiness-deferred retry waits for the transport before giving +/// the lock back to the next catch-up. +/// +/// Where [`BROADCAST_TRANSPORT_READY_WAIT`] is sized against the host +/// thread it occupies, this one runs on the manager's own runtime and +/// occupies nothing the host needs, so it can afford to cover a slow first +/// peer connection or a network that comes back minutes after launch. It +/// is still a ceiling and not a park: a transport that never comes up in a +/// session must not keep one task per stuck lock alive for the session's +/// whole life. On expiry the row is exactly as it was — tracked, resumable +/// — and the next catch-up (launch, foreground, reconnect) starts over. +const DEFERRED_RESUME_TRANSPORT_WAIT: Duration = Duration::from_secs(10 * 60); + +/// RAII membership in the manager's deferred-resume set: dropped on every +/// exit of the retry task — completion, expiry, or the runtime tearing the +/// task down — so an outpoint can be deferred again afterwards. +struct DeferredResumeMembership { + deferred: Arc>>, + out_point: OutPoint, +} + +impl Drop for DeferredResumeMembership { + fn drop(&mut self) { + self.deferred + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(&self.out_point); + } +} + /// Find the first outpoint of `lock`'s transaction that some **other, /// confirmed** transaction of this wallet already spent, returning /// `(conflicting_input, spending_txid, spender_height)`. @@ -446,6 +493,181 @@ pub(crate) fn seed_observed_input_conflicts(info: &PlatformWalletInfo) { } 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 — or + /// one with no connected peers yet — fails as + /// [`BroadcastError::Rejected`], the never-sent verdict, which on both + /// broadcasting arms ends the resume in an unconfirmed outcome. Nothing + /// reschedules the catch-up, so a lock that loses this race is left + /// exactly as it was and the next session repeats it, with the funds + /// sitting behind it the whole time. + /// + /// The wait is bounded twice over: by + /// [`BROADCAST_TRANSPORT_READY_WAIT`], and by the caller's own `timeout` + /// when it asked for one. Whatever it consumes is deducted from that + /// timeout, so a bounded caller's total stays inside the budget it asked + /// for, and `None` stays `None` — the recovery policy's own + /// [`UNCONFIRMED_BROADCAST_PROOF_TIMEOUT`] default still applies to the + /// proof wait downstream. + /// + /// A wait that expires without readiness is recorded in `missed`, so + /// the resume can hand the lock to a readiness-deferred retry once its + /// own attempt has run its course — see + /// [`resume_asset_lock`](Self::resume_asset_lock). + async fn await_broadcast_ready( + &self, + out_point: &OutPoint, + timeout: Option, + missed: &AtomicBool, + ) -> Option { + let budget = timeout.map_or(BROADCAST_TRANSPORT_READY_WAIT, |t| { + t.min(BROADCAST_TRANSPORT_READY_WAIT) + }); + let started = tokio::time::Instant::now(); + if !self.broadcaster.wait_until_ready(budget).await { + missed.store(true, Ordering::Relaxed); + tracing::warn!( + outpoint = %out_point, + ?budget, + "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)) + } + + /// The status `out_point` would be deferred from: its current status + /// when that is one of the two that still need a send, `None` for a + /// row that is settled, consumed or gone. + async fn status_needing_a_send(&self, out_point: &OutPoint) -> Option { + let wm = self.wallet_manager.read().await; + wm.get_wallet_info(&self.wallet_id) + .and_then(|info| info.tracked_asset_locks.get(out_point)) + .map(|lock| lock.status.clone()) + .filter(|status| matches!(status, AssetLockStatus::Built | AssetLockStatus::Broadcast)) + } + + /// Retry a resume whose bounded transport wait expired, once the + /// transport actually comes up — off the caller's thread. + /// + /// The bounded wait in [`await_broadcast_ready`](Self::await_broadcast_ready) + /// is the most the synchronous FFI entry point can afford: the host + /// drives the catch-up on the same fixed-width pool it needs to reach + /// `start_spv`, so parking there for longer would delay the very + /// transport being waited for. But an expiry followed by the pre-dispatch + /// rejection ends the resume with nothing rescheduling it, and the + /// in-tree hosts start the catch-up only while loading a wallet. A cold + /// start or a first peer connection that takes longer than the ceiling + /// therefore recreated the original stranding: SPV usable seconds later, + /// the transaction unsent until the next launch. + /// + /// This closes that gap without lengthening the host's wait. The retry + /// is a task on the manager's runtime — the one the FFI's + /// `runtime().block_on(...)` already runs on — holding a + /// [`shared_handle`](Self::shared_handle) onto the same manager. It waits + /// for readiness under [`DEFERRED_RESUME_TRANSPORT_WAIT`], re-checks that + /// the row still reads exactly `deferred_from` — any change means another + /// resume, a funding flow or a proof got there first, and the retry + /// stands down rather than add a send of its own — and then runs an + /// ordinary resume with the caller's own `timeout`. That inner resume + /// never defers again: readiness was just observed, so a second miss is + /// a transport that went away, which is the next catch-up's problem. + /// + /// One retry per outpoint at a time: `deferred_resumes` membership makes + /// a second deferral while the first is waiting a no-op, so a host that + /// re-runs its catch-up on foreground and reconnect cannot stack retries. + /// Without a runtime to spawn on — a purely blocking caller — nothing is + /// scheduled and the row is simply left for the next catch-up, exactly + /// as before. + fn resume_when_transport_ready( + &self, + out_point: OutPoint, + deferred_from: AssetLockStatus, + timeout: Option, + ) { + let Ok(runtime) = tokio::runtime::Handle::try_current() else { + tracing::warn!( + outpoint = %out_point, + "resume_asset_lock: transport not ready and no runtime to \ + defer the retry onto; leaving the lock for the next catch-up" + ); + return; + }; + { + let mut deferred = self + .deferred_resumes + .lock() + .unwrap_or_else(|e| e.into_inner()); + if !deferred.insert(out_point) { + tracing::debug!( + outpoint = %out_point, + "resume_asset_lock: a readiness-deferred retry is already \ + waiting for this lock" + ); + return; + } + } + tracing::info!( + outpoint = %out_point, + ceiling = ?DEFERRED_RESUME_TRANSPORT_WAIT, + "resume_asset_lock: transport not ready within the bounded wait; \ + deferring a retry until it comes up" + ); + let manager = self.shared_handle(); + runtime.spawn(async move { + let _membership = DeferredResumeMembership { + deferred: Arc::clone(&manager.deferred_resumes), + out_point, + }; + if !manager + .broadcaster + .wait_until_ready(DEFERRED_RESUME_TRANSPORT_WAIT) + .await + { + tracing::warn!( + outpoint = %out_point, + ceiling = ?DEFERRED_RESUME_TRANSPORT_WAIT, + "deferred resume: transport never came up; leaving the \ + lock tracked for the next catch-up" + ); + return; + } + let current = manager.status_needing_a_send(&out_point).await; + if current.as_ref() != Some(&deferred_from) { + tracing::info!( + outpoint = %out_point, + ?deferred_from, + ?current, + "deferred resume: the row moved on since the deferral; \ + standing down" + ); + return; + } + match manager.resume_asset_lock_attempt(&out_point, timeout).await { + Ok(_) => tracing::info!( + outpoint = %out_point, + "deferred resume: completed once the transport came up" + ), + Err(error) => tracing::warn!( + outpoint = %out_point, + %error, + "deferred resume: the retry did not settle the lock; it \ + stays tracked and resumable" + ), + } + }); + } + /// Re-run the double-spend screen once a proof wait has expired, and /// render a conflict that still stands as the verdict explaining that /// expiry. @@ -617,11 +839,25 @@ impl AssetLockManager { /// caller passes `derivation_path` to the same signer used for the /// build phase when the credit output is later consumed on Platform. /// + /// Before either broadcasting arm touches the transport, the local + /// record is probed once under a zero bound: a row that already holds + /// finality — the ordinary state after a send whose proof arrived with + /// no waiter active — completes offline, sending nothing and waiting for + /// nothing. Only a row that still needs a send gives the broadcaster's + /// transport a bounded chance to come up (see + /// [`await_broadcast_ready`](Self::await_broadcast_ready)), so a resume + /// driven by the app-launch catch-up doesn't spend its one send on an + /// SPV client that has not started yet. `InstantSendLocked` / + /// `ChainLocked` / `RecoveredFromChain` broadcast nothing at all and + /// never wait for the transport. + /// /// `timeout` is `Option` and is only consulted when the lock - /// still needs a proof (`Built` / `Broadcast`, or the defensive - /// proof-less `RecoveredFromChain` fallback). For `InstantSendLocked` / - /// `ChainLocked` the proof already exists and no wait happens, so the - /// value is moot. + /// still needs a proof: `Built` / `Broadcast` rows that the local probe + /// did not settle, and the defensive fallback for a `RecoveredFromChain` + /// row whose persisted proof was lost. Everywhere else — including a + /// `RecoveredFromChain` row that still carries its proof, which is what + /// reconstruction assigns that status with — the proof already exists + /// and no wait happens, so the value is moot. /// /// `None` does not request an unbounded wait — it declines to name a /// bound, and every proof-waiting path then substitutes @@ -666,15 +902,52 @@ impl AssetLockManager { /// lock) is host and user policy; the SDK does not license it /// unilaterally on this evidence. The screen is one-sided — read its /// docs before treating a clean pass as evidence the lock is alive. + /// + /// A `Built` / `Broadcast` resume whose bounded transport wait expires + /// still makes its one attempt and reports what it drew — but if the + /// row still needs a send afterwards, a retry is deferred until the + /// transport comes up, off this thread. See + /// [`resume_when_transport_ready`](Self::resume_when_transport_ready). pub async fn resume_asset_lock( &self, out_point: &OutPoint, timeout: Option, + ) -> Result<(dpp::prelude::AssetLockProof, DerivationPath), PlatformWalletError> { + let transport_missed = AtomicBool::new(false); + let result = self + .resume_asset_lock_with(out_point, timeout, &transport_missed) + .await; + if transport_missed.load(Ordering::Relaxed) { + if let Some(deferred_from) = self.status_needing_a_send(out_point).await { + self.resume_when_transport_ready(*out_point, deferred_from, timeout); + } + } + result + } + + /// One resume attempt, with no readiness-deferred retry: what the + /// deferred retry itself runs, having just observed readiness. + async fn resume_asset_lock_attempt( + &self, + out_point: &OutPoint, + timeout: Option, + ) -> Result<(dpp::prelude::AssetLockProof, DerivationPath), PlatformWalletError> { + self.resume_asset_lock_with(out_point, timeout, &AtomicBool::new(false)) + .await + } + + /// The resume proper. `transport_missed` is set when a broadcasting + /// arm's bounded transport wait expired without readiness. + async fn resume_asset_lock_with( + &self, + out_point: &OutPoint, + timeout: Option, + transport_missed: &AtomicBool, ) -> Result<(dpp::prelude::AssetLockProof, DerivationPath), PlatformWalletError> { tracing::info!(outpoint = %out_point, ?timeout, "resume_asset_lock: entered"); // 1. Look up the tracked lock — snapshot the fields we need. - let (tx, status, existing_proof, account_index, input_conflict) = { + let (tx, status, existing_proof, account_index, input_conflict, mut dispatch_claim) = { let wm = self.wallet_manager.read().await; let info = wm .get_wallet_info(&self.wallet_id) @@ -711,12 +984,26 @@ impl AssetLockManager { | AssetLockStatus::RecoveredFromChain | AssetLockStatus::Consumed => None, }; + // Claim the dispatch window for the two statuses that can still + // send, while the read guard that produced this snapshot is + // still held. From here until the send is made and recorded, the + // initial build's rejection cleanup cannot remove the row and + // release its inputs — see `untrack_asset_lock`. The claim is + // taken HERE rather than at the broadcast because the whole + // point is to cover the suspension points in between; taking it + // later would leave the same window under a shorter name. + let dispatch_claim = matches!( + lock.status, + AssetLockStatus::Built | AssetLockStatus::Broadcast + ) + .then(|| self.claim_resume_dispatch(*out_point)); ( lock.transaction.clone(), lock.status.clone(), lock.proof.clone(), lock.account_index, input_conflict, + dispatch_claim, ) }; @@ -753,23 +1040,82 @@ impl AssetLockManager { ); } + // Local finality is probed BEFORE anything touching the transport. + // A row can already carry its proof in the record while sitting at + // `Built` or `Broadcast` — finality that lands with no waiter active + // enriches the record without advancing the tracked status — and + // that is exactly the state the launch catch-up resumes from, before + // the SPV client has started. Such a resume needs no transport at + // all: waiting for one first made an already-final lock pay the + // readiness ceiling on every offline launch, and then completed it + // from the same record anyway. The probe is a single local + // record/persister check under a zero bound — it never touches the + // network and never waits. A proof outranks a sighted input conflict + // here for the same reason it outranks one at every other point in + // this function: the transaction is final, so no sibling can take it. + let local_finality = match status { + AssetLockStatus::Built | AssetLockStatus::Broadcast => self + .wait_for_proof(out_point, Some(Duration::ZERO)) + .await + .ok(), + AssetLockStatus::InstantSendLocked + | AssetLockStatus::ChainLocked + | AssetLockStatus::RecoveredFromChain + | AssetLockStatus::Consumed => None, + }; + if status == AssetLockStatus::Built && local_finality.is_some() { + dispatch_claim + .as_mut() + .expect("Built resumes hold a cleanup-exclusion claim") + .preserve_on_drop(); + } + // 2. Resume from the current status. - let proof = match status { - AssetLockStatus::Built => { - // Re-broadcast and wait for proof. - // + let proof = match (&status, local_finality) { + // Already final locally: no send, no transport wait, no proof + // wait. The row is advanced and the proof attached below exactly + // as a waited-for proof would be. + (AssetLockStatus::Built | AssetLockStatus::Broadcast, Some(proof)) => { + // The claim is deliberately NOT released here. Nothing will + // be dispatched, but a `Built` row stays `Built` until the + // status advance at the end of this function, and a + // rejection cleanup that removed it in between would release + // the reservation and the fence of a transaction the record + // says is already final — reporting the definite rejection, + // whose contract licenses a rebuild, for an outpoint that is + // settled on chain. Everything from here to that advance is + // record reads and one lock acquisition, so holding the + // claim across them costs a racing cleanup nothing it would + // have been right to do. + tracing::info!( + outpoint = %out_point, + status = ?status, + "resume_asset_lock: the local record already holds finality — \ + completing the resume without waiting for the broadcast \ + transport or sending anything" + ); + self.validate_or_upgrade_proof(proof, account_index, out_point) + .await? + } + (AssetLockStatus::Built, None) => { + // Nothing final locally, so this row still needs a send: + // re-broadcast and wait for proof — but not into a transport + // that cannot carry the send. See `await_broadcast_ready`. + let timeout = self + .await_broadcast_ready(out_point, timeout, transport_missed) + .await; + // No verdict this broadcaster can return ends the resume by - // itself. `MaybeSent` - // means the outcome is unknown — and for a lock stuck at - // `Built` that is the expected answer when the app died - // between a successful broadcast and this status advance: - // the tx is in a mempool (or mined) and every re-broadcast - // reports the same ambiguity. Failing on it left the lock at - // `Built` forever, so each recovery pass repeated the same - // broadcast and the same abort, and the top-up never - // completed. Advancing to `Broadcast` and waiting matches - // what the `Broadcast` arm below already does with the - // identical signal. + // itself. `MaybeSent` means the outcome is unknown — and for + // a lock stuck at `Built` that is the expected answer when + // the app died between a successful broadcast and this + // status advance: the tx is in a mempool (or mined) and + // every re-broadcast reports the same ambiguity. Failing on + // it left the lock at `Built` forever, so each recovery pass + // repeated the same broadcast and the same abort, and the + // top-up never completed. Advancing to `Broadcast` and + // waiting matches what the `Broadcast` arm below already + // does with the identical signal. // // `MaybeSent` is however ALSO what the broadcaster reports // for a genuinely rejected transaction: `DapiBroadcaster` @@ -810,6 +1156,14 @@ impl AssetLockManager { let mut local_proof = None; let mut maybe_sent_reason = None; let mut undispatched = None; + // The broadcaster cannot report whether it had side effects + // until the await completes. Make the claim sticky before + // entering it, so cancellation during or immediately after a + // possible send cannot license cleanup of the inputs. + dispatch_claim + .as_mut() + .expect("Built resumes hold a cleanup-exclusion claim") + .preserve_on_drop(); match self.broadcaster.broadcast(&tx).await { Ok(_) => {} Err(BroadcastError::MaybeSent { reason }) => { @@ -824,8 +1178,19 @@ impl AssetLockManager { maybe_sent_reason = Some(reason); } Err(rejected @ BroadcastError::Rejected { .. }) => { + // This verdict proves the current attempt never left + // the device. Unless the local record supplies proof, + // cancellation from here is a clean pre-dispatch exit. + dispatch_claim + .as_mut() + .expect("Built resumes hold a cleanup-exclusion claim") + .release_on_drop(); match self.wait_for_proof(out_point, Some(Duration::ZERO)).await { Ok(proof) => { + dispatch_claim + .as_mut() + .expect("Built resumes hold a cleanup-exclusion claim") + .preserve_on_drop(); tracing::info!( outpoint = %out_point, error = %rejected, @@ -880,6 +1245,11 @@ impl AssetLockManager { } } let proof = if let Some(proof) = local_proof { + // Held to the status advance for the same reason as the + // already-final arm above: this send never left the + // device and none will follow, but the row reads `Built` + // until then and its transaction is final, so a cleanup + // that removed it would free inputs that are spent. proof } else { // The status advance belongs to a send that actually @@ -887,12 +1257,35 @@ impl AssetLockManager { // the row exactly where it was, so the next resume // re-sends the transaction instead of dropping into the // `Broadcast` arm's wait for a send that never happened. + // + // The advance is conditional on the row still reading + // `Built`. It records "this send dispatched", nothing + // more; a concurrent resume that already carried the + // row to a proof-bearing status must not be regressed, + // and a `Consumed` tombstone laid down by an explicit + // funding flow while this resume was suspended must + // stay one — that case ends the resume here, as the + // consumed lock has nothing left to wait for. if undispatched.is_none() { - let cs = self - .advance_asset_lock_status(out_point, AssetLockStatus::Broadcast, None) - .await?; - self.queue_asset_lock_changeset(cs); + if let Some(cs) = self + .advance_asset_lock_status_if( + out_point, + |current| *current == AssetLockStatus::Built, + AssetLockStatus::Broadcast, + None, + ) + .await? + { + self.queue_asset_lock_changeset(cs); + } + drop(dispatch_claim.take()); } + // A dispatched attempt no longer needs the claim once the + // row is `Broadcast`: that status excludes the cleanup on + // its own. A proven-undispatched attempt keeps its ordinary + // RAII claim during the proof wait below. It releases on a + // timeout or other pre-dispatch exit, but becomes sticky if + // the wait observes local finality. // Every resumed lock waits under a deadline, and a // sighting only shortens it. An accepted re-broadcast is // evidence the transaction reached the network, never @@ -913,7 +1306,15 @@ impl AssetLockManager { timeout.or(Some(UNCONFIRMED_BROADCAST_PROOF_TIMEOUT)) }; match self.wait_for_proof(out_point, bounded).await { - Ok(proof) => proof, + Ok(proof) => { + if undispatched.is_some() { + dispatch_claim + .as_mut() + .expect("Built resumes hold a cleanup-exclusion claim") + .preserve_on_drop(); + } + proof + } Err(expiry @ PlatformWalletError::FinalityTimeout(_)) => { // The wait has now given live synchronization its // window, so the screen is re-read and what it says @@ -986,12 +1387,18 @@ impl AssetLockManager { self.validate_or_upgrade_proof(proof, account_index, out_point) .await? } - AssetLockStatus::Broadcast => { - // Defensive re-broadcast, then wait for proof. A lock can - // sit at `Broadcast` across app restarts long enough for - // its funding tx to be evicted from every mempool (Core's - // default `-mempoolexpiry` is two weeks), or the original - // broadcast may have reached no peers at all (SPV + (AssetLockStatus::Broadcast, None) => { + // Defensive re-broadcast, then wait for proof — and, as on + // the `Built` arm, not into a transport that cannot carry + // the send. See `await_broadcast_ready`. + let timeout = self + .await_broadcast_ready(out_point, timeout, transport_missed) + .await; + + // A lock can sit at `Broadcast` across app restarts long + // enough for its funding tx to be evicted from every mempool + // (Core's default `-mempoolexpiry` is two weeks), or the + // original broadcast may have reached no peers at all (SPV // connectivity gap). Once no node holds the tx, no IS/CL // proof can ever arrive and the wait below can only run out // its bound. A re-broadcast revives an evicted/undelivered @@ -1064,7 +1471,7 @@ impl AssetLockManager { // original send IS provable: a row can sit at `Built` after a // successful broadcast too (app killed between the send and // the status advance), which is precisely why the `Built` arm - // above also only surfaces the error and leaves its row alone. + // above reaches the same verdict from the same position. let mut local_proof = None; if let Err(e) = self.broadcaster.broadcast(&tx).await { if matches!(e, BroadcastError::Rejected { .. }) { @@ -1114,6 +1521,11 @@ impl AssetLockManager { ); } } + // The send is done and this arm records nothing about it (the + // row is already at `Broadcast`), so the dispatch window ends + // here; everything below only waits. + drop(dispatch_claim.take()); + // Bounded like the `Built` arm, and for the same reason. This // arm is only entered on a RESUME, i.e. for a transaction // whose earlier broadcast window already failed to produce a @@ -1173,7 +1585,7 @@ impl AssetLockManager { self.validate_or_upgrade_proof(proof, account_index, out_point) .await? } - AssetLockStatus::InstantSendLocked | AssetLockStatus::ChainLocked => { + (AssetLockStatus::InstantSendLocked | AssetLockStatus::ChainLocked, _) => { // Already have a proof — validate / upgrade if stale. let proof = existing_proof.ok_or_else(|| { PlatformWalletError::AssetLockProofWait(format!( @@ -1184,7 +1596,7 @@ impl AssetLockManager { self.validate_or_upgrade_proof(proof, account_index, out_point) .await? } - AssetLockStatus::RecoveredFromChain => { + (AssetLockStatus::RecoveredFromChain, _) => { // Reconstructed from a chain-locked record after a // restore — Platform-side consumption is unknown. An // explicit resume is allowed to try consuming it: @@ -1232,7 +1644,7 @@ impl AssetLockManager { } } } - AssetLockStatus::Consumed => { + (AssetLockStatus::Consumed, _) => { // Terminal tombstone — the asset lock was already // burned by a successful identity registration / top-up. // Retaining this state makes the typed distinction from @@ -1250,7 +1662,11 @@ impl AssetLockManager { // pending window and resurrect the false-"Pending" rendering // on every restored lock whose resume didn't end in a spend. // Consumption is recorded separately (`consume_asset_lock`) - // when the credit output actually lands on Platform. + // when the credit output actually lands on Platform — and a + // tombstone that landed while this resume was suspended wins: + // `advance_asset_lock_status` refuses to overwrite `Consumed`, + // so the proof gathered here is never handed out for a lock the + // wallet has already spent. let new_status = if status == AssetLockStatus::RecoveredFromChain { AssetLockStatus::RecoveredFromChain } else { @@ -1263,6 +1679,7 @@ impl AssetLockManager { .advance_asset_lock_status(out_point, new_status, Some(proof.clone())) .await?; self.queue_asset_lock_changeset(cs); + drop(dispatch_claim.take()); // 4. Re-derive the one-time credit-output derivation path. let path = { @@ -1383,6 +1800,7 @@ impl AssetLockManager { #[cfg(test)] mod tests { use std::collections::BTreeMap; + use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -1406,13 +1824,14 @@ mod tests { use key_wallet_manager::WalletManager; use tokio::sync::{Notify, RwLock}; + use super::{BROADCAST_TRANSPORT_READY_WAIT, DEFERRED_RESUME_TRANSPORT_WAIT}; use crate::broadcaster::{BroadcastError, TransactionBroadcaster}; use crate::changeset::{ ClientStartState, PersistenceError, PlatformWalletChangeSet, PlatformWalletPersistence, }; use crate::error::PlatformWalletError; use crate::test_support::{ - funded_wallet_manager, AlwaysMaybeSentBroadcaster, AlwaysRejectedBroadcaster, + funded_wallet_manager, AlwaysMaybeSentBroadcaster, AlwaysRejectedBroadcaster, WalletSigner, }; use crate::wallet::asset_lock::manager::AssetLockManager; use crate::wallet::asset_lock::tracked::{AssetLockStatus, TrackedAssetLock}; @@ -1458,6 +1877,31 @@ mod tests { } } + /// `Rejected` on every send — the production `SpvBroadcaster`'s verdict + /// for an unstarted client — while counting the sends it was asked for. + /// The count is what separates "completed from the local record" from + /// "retried the broadcast until something worked". + #[derive(Default)] + struct CountingRejectedBroadcaster { + sends: AtomicU32, + } + + impl CountingRejectedBroadcaster { + fn sends(&self) -> u32 { + self.sends.load(Ordering::SeqCst) + } + } + + #[async_trait] + impl TransactionBroadcaster for CountingRejectedBroadcaster { + async fn broadcast(&self, _transaction: &Transaction) -> Result { + self.sends.fetch_add(1, Ordering::SeqCst); + Err(BroadcastError::Rejected { + reason: "SPV broadcast not sent: client not started".to_string(), + }) + } + } + impl PlatformWalletPersistence for RecordingPersistence { fn store( &self, @@ -1712,9 +2156,18 @@ mod tests { /// Builds a tracked `Built`-status lock on a funded wallet and resumes it /// through `broadcaster`, returning the resume error and the lock's status /// afterwards. Shared by the two ambiguity/rejection cases below. + /// + /// The manager and signer come back too, so a caller can attempt a second + /// build on the SAME wallet — the only way to observe from outside that + /// the funding inputs are still reserved. async fn resume_built_lock_with( broadcaster: Arc, - ) -> (PlatformWalletError, AssetLockStatus) { + ) -> ( + PlatformWalletError, + AssetLockStatus, + AssetLockManager, + WalletSigner, + ) { let (wallet_manager, wallet_id, _balance, signer) = funded_wallet_manager(StandardAccountType::BIP44Account).await; let sdk = Arc::new( @@ -1776,7 +2229,7 @@ mod tests { .expect("lock stays tracked") .status .clone(); - (error, status) + (error, status, manager, signer) } /// An AMBIGUOUS re-broadcast must not end the resume. A lock sitting at @@ -1787,7 +2240,8 @@ mod tests { /// to wait for the proof — here, until the 10ms test timeout. #[tokio::test] async fn built_resume_survives_an_ambiguous_rebroadcast_and_advances() { - let (error, status) = resume_built_lock_with(Arc::new(AlwaysMaybeSentBroadcaster)).await; + let (error, status, ..) = + resume_built_lock_with(Arc::new(AlwaysMaybeSentBroadcaster)).await; assert!( matches!(error, PlatformWalletError::FinalityTimeout(_)), @@ -1818,25 +2272,53 @@ mod tests { /// and release, may emit 26. #[tokio::test] async fn built_resume_of_a_rejected_rebroadcast_reports_an_unknown_outcome() { - let (error, status) = resume_built_lock_with(Arc::new(AlwaysRejectedBroadcaster)).await; - - assert!( - !matches!(error, PlatformWalletError::TransactionBroadcast(_)), - "a re-broadcast that never left the device is not evidence that an earlier \ - send was rejected, so it must not claim the definite-rejection contract \ - while the row and its reservation are kept: {error:?}" - ); + let (error, status, manager, signer) = + resume_built_lock_with(Arc::new(AlwaysRejectedBroadcaster)).await; + + let PlatformWalletError::TransactionBroadcastUnconfirmed(reason) = &error else { + panic!( + "a rejection that never dispatched must not be reported as a \ + definite rejection, whose contract is that the inputs were \ + released and a rebuild is safe: {error:?}" + ); + }; assert!( - matches!( - error, - PlatformWalletError::TransactionBroadcastUnconfirmed(_) - ), - "a rejected re-broadcast must fail the resume as an unknown outcome: {error:?}" + reason.contains("rejected before dispatch"), + "the failure must name the undispatched send: {reason}" ); assert_eq!( status, AssetLockStatus::Built, - "a send that never dispatched must leave the row resumable at Built" + "a send that never dispatched must leave the row resumable at \ + Built, with its inputs still reserved" + ); + + // "Inputs still reserved" is half of the contract the non-terminal + // verdict promises, and the status alone does not carry it: a resume + // that kept the row at `Built` while releasing its reservation would + // satisfy every assertion above and still hand the host exactly what + // the definite-rejection code was rejected for — a wallet whose funds + // look free for a rebuild while a possibly-live asset lock holds them. + // A second build on the same wallet is what tells the two apart. The + // fixture funds one UTXO, so the reservation leaves nothing to select + // and the build fails with the typed shortfall rather than any + // unrelated error. + let rebuild = manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::IdentityTopUp, + 4, + &signer, + ) + .await; + assert!( + matches!( + rebuild, + Err(PlatformWalletError::AssetLockInsufficientFunds { available: 0, .. }) + ), + "the rejected resume must keep the funding reservation, leaving a \ + same-wallet rebuild zero available funds, got {rebuild:?}" ); } @@ -2336,7 +2818,9 @@ mod tests { /// mempool. Refusing to broadcast or wait on that evidence returned the /// contested verdict on every resume and every launch for a lock whose /// own funding transaction was sitting in history chain-locked, ready - /// to settle. The resume must run and take the proof. + /// to settle. The resume must run and take the proof — which it now + /// does before the transport is consulted at all, since the record is + /// already final; the sighting must not stand in the way of that either. #[tokio::test] async fn a_standing_conflict_never_costs_the_lock_a_proof_that_has_arrived() { let fixture = ConflictFixture::new().await; @@ -2376,13 +2860,17 @@ mod tests { /// dispatches must still take a proof that has already arrived. /// /// This is the launch catch-up shape: `catchUpStuckAssetLocks` resumes - /// a restored row before SPV connects, so the re-broadcast draws the + /// a restored row before SPV connects, where every send draws the /// DEFINITE `Rejected` (unstarted client / zero peers), while history /// carries both a restored spender of the lock's input and the lock's /// own chain-locked funding record. Returning the rejection there /// skipped the record entirely — an already-final lock failed on every /// launch until connectivity returned, and it failed as the FFI's code /// 26, whose released-reservation contract this path does not honour. + /// The record is now read before a send is even considered, so this + /// fixture's rejecting broadcaster is never reached; what the test still + /// pins is that the standing sighting does not cost the lock the proof + /// its own record already holds. #[tokio::test] async fn a_rejected_rebroadcast_of_a_conflicted_built_lock_still_takes_an_arrived_proof() { let fixture = ConflictFixture::rejecting().await; @@ -2550,9 +3038,10 @@ mod tests { let stub = Arc::new(InterleavedPersistence::new( wallet_manager, wallet_id, - // Lookup 0 is the expiring proof wait's own miss; lookup - // 1 is the verdict's probe, the gap under test. - 1, + // Lookup 0 is the pre-transport local-finality probe's + // miss and lookup 1 is the expiring proof wait's own; + // lookup 2 is the verdict's probe, the gap under test. + 2, move |info| { let transaction = handle .lock() @@ -2643,9 +3132,10 @@ mod tests { let stub = Arc::new(InterleavedPersistence::new( wallet_manager, wallet_id, - // Lookup 0 is the expiring proof wait's own miss; lookup - // 1 is the verdict's probe, the gap under test. - 1, + // Lookup 0 is the pre-transport local-finality probe's + // miss and lookup 1 is the expiring proof wait's own; + // lookup 2 is the verdict's probe, the gap under test. + 2, move |info| { let transaction = handle .lock() @@ -2757,7 +3247,11 @@ mod tests { let stub = Arc::new(InterleavedPersistence::new( wallet_manager, wallet_id, - 1, + // Lookup 0 is the pre-transport local-finality probe's + // miss and lookup 1 is the expiring proof wait's own; + // lookup 2 is the verdict's probe, the gap the settling + // has to land in. + 2, |info| { let (out_point, lock) = info .tracked_asset_locks @@ -2832,10 +3326,11 @@ mod tests { let stub = Arc::new(InterleavedPersistence::new( wallet_manager, wallet_id, - // Lookup 0 is the rejection's own local-proof probe, - // lookup 1 the expiring wait, lookup 2 the verdict's + // Lookup 0 is the pre-transport local-finality probe's + // miss, lookup 1 the rejection's own local-proof probe, + // lookup 2 the expiring wait, lookup 3 the verdict's // probe — the gap the retraction has to land in. - 2, + 3, move |info| { let transaction = handle .lock() @@ -3453,6 +3948,52 @@ mod tests { ); } + /// A tracked lock at `status` on a funded wallet, plus the manager that + /// resumes it through `broadcaster` and the handles a test needs to + /// read the row back afterwards. + struct TrackedLockFixture { + manager: AssetLockManager, + wallet_manager: Arc>>, + wallet_id: WalletId, + out_point: OutPoint, + } + + impl TrackedLockFixture { + /// The lock's tracked status right now (`None` = untracked). + async fn status(&self) -> Option { + self.wallet_manager + .read() + .await + .get_wallet_info(&self.wallet_id) + .expect("wallet") + .tracked_asset_locks + .get(&self.out_point) + .map(|lock| lock.status.clone()) + } + + /// The lock's attached proof right now. + async fn proof(&self) -> Option { + self.wallet_manager + .read() + .await + .get_wallet_info(&self.wallet_id) + .expect("wallet") + .tracked_asset_locks + .get(&self.out_point) + .and_then(|lock| lock.proof.clone()) + } + + /// Whether a readiness-deferred retry is currently registered for + /// the lock. + fn retry_deferred(&self) -> bool { + self.manager + .deferred_resumes + .lock() + .expect("deferred set mutex") + .contains(&self.out_point) + } + } + /// Builds a tracked lock at `status` on a funded wallet and resumes it /// through `broadcaster` with the given `timeout`, returning the resume /// error and the lock's tracked state afterwards (`None` = untracked). @@ -3461,6 +4002,22 @@ mod tests { status: AssetLockStatus, timeout: Option, ) -> (PlatformWalletError, Option) { + let fixture = tracked_lock_at(broadcaster, status).await; + let error = fixture + .manager + .resume_asset_lock(&fixture.out_point, timeout) + .await + .expect_err("no proof event is ever delivered in these cases"); + let tracked = fixture.status().await; + (error, tracked) + } + + /// Builds a tracked lock at `status` on a funded wallet, resumable + /// through `broadcaster`. + async fn tracked_lock_at( + broadcaster: Arc, + status: AssetLockStatus, + ) -> TrackedLockFixture { let (wallet_manager, wallet_id, _balance, signer) = funded_wallet_manager(StandardAccountType::BIP44Account).await; let sdk = Arc::new( @@ -3507,82 +4064,370 @@ mod tests { }, ); } - - let error = manager - .resume_asset_lock(&out_point, timeout) - .await - .expect_err("no proof event is ever delivered in these cases"); - let tracked = wallet_manager - .read() - .await - .get_wallet_info(&wallet_id) - .expect("wallet") - .tracked_asset_locks - .get(&out_point) - .map(|lock| lock.status.clone()); - (error, tracked) + TrackedLockFixture { + manager, + wallet_manager, + wallet_id, + out_point, + } } - /// Regression: an ambiguous re-broadcast on the UNBOUNDED resume path - /// must not hang. - /// - /// `MaybeSent` is the broadcaster's verdict for a genuinely rejected - /// transaction as much as for an accepted one — `DapiBroadcaster` - /// classifies every failure that way, and the SPV broadcaster reaches - /// `Rejected` only on `NotConnected`. So advancing to `Broadcast` and - /// then waiting with `wait_for_proof(None)` — which is what the three - /// `resume_asset_lock(.., None)` production call sites do — turned a - /// broadcast failure that used to surface in ~30s into a wait that never - /// ends, because no proof can arrive for a tx that was never accepted. - /// - /// `start_paused` auto-advances the substituted bound, so this asserts - /// termination *and* that the caller gets the pre-#4367 typed error back. + /// Readiness polls before a transport that misses the bounded wait + /// comes up: the ceiling is `BROADCAST_TRANSPORT_READY_WAIT` of 5ms + /// polls, and this lands a little after it. + const TRANSPORT_COMES_UP_AFTER_THE_CEILING: u32 = + (BROADCAST_TRANSPORT_READY_WAIT.as_millis() / 5) as u32 + 600; + + /// A transport that misses the bounded wait must not strand the lock + /// for the session. The in-tree hosts start the catch-up only while + /// loading a wallet, so a resume that spent its one send into a dead + /// transport and reported the unknown outcome used to leave the row + /// exactly as it was — the original failure, back again, whenever SPV + /// startup took longer than the ceiling. The resume keeps its bounded + /// wait (the host thread cannot afford more) and hands the lock to a + /// retry off that thread, which sends once the transport is up. #[tokio::test(start_paused = true)] - async fn unbounded_resume_of_an_ambiguous_rebroadcast_terminates() { - let (error, status) = resume_lock_at( - Arc::new(AlwaysMaybeSentBroadcaster), - AssetLockStatus::Built, - None, - ) - .await; + async fn a_missed_transport_wait_defers_a_retry_until_the_transport_comes_up() { + let broadcaster = Arc::new(StartingUpBroadcaster::comes_up_after( + TRANSPORT_COMES_UP_AFTER_THE_CEILING, + )); + let fixture = tracked_lock_at(broadcaster.clone(), AssetLockStatus::Built).await; + let caller_budget = Some(Duration::from_secs(300)); + let error = fixture + .manager + .resume_asset_lock(&fixture.out_point, caller_budget) + .await + .expect_err("the transport is down for the whole bounded wait"); assert!( matches!( error, PlatformWalletError::TransactionBroadcastUnconfirmed(_) ), - "an unbounded resume whose re-broadcast was ambiguous must end as a \ - broadcast-unconfirmed failure rather than hang, got {error:?}" + "the caller still gets the attempt's own verdict: {error:?}" ); - assert_eq!( - status, - Some(AssetLockStatus::Broadcast), - "the advance itself is still correct — the row stays resumable so a \ - later pass can pick up a proof that does eventually arrive" + let attempts = broadcaster.attempts(); + assert_eq!(attempts.len(), 1, "the bounded attempt spends its one send"); + assert!( + !attempts[0].1, + "and that send went into the transport as it was" + ); + assert_eq!(fixture.status().await, Some(AssetLockStatus::Built)); + assert!( + fixture.retry_deferred(), + "a row that still needs a send after a missed wait has a retry deferred" ); - } - /// The bounded callers must be untouched by the fix above. The shielded - /// seed pool passes its own timeout and treats `FinalityTimeout` as a - /// pacing signal (pause, resume the lock later), so re-typing the error - /// for every caller would have broken a working flow to fix a different - /// one. - #[tokio::test] - async fn bounded_resume_of_an_ambiguous_rebroadcast_still_reports_finality_timeout() { - let (error, status) = resume_lock_at( - Arc::new(AlwaysMaybeSentBroadcaster), - AssetLockStatus::Built, - Some(Duration::from_millis(10)), - ) - .await; + // Let the paused clock run the deferred retry: the transport comes + // up a few seconds in, and the retry sends into it. + tokio::time::sleep(Duration::from_secs(60)).await; + let attempts = broadcaster.attempts(); + assert_eq!( + attempts.len(), + 2, + "the deferred retry makes exactly one further send" + ); assert!( - matches!(error, PlatformWalletError::FinalityTimeout(_)), - "a caller-supplied timeout must keep its FinalityTimeout semantics: {error:?}" + attempts[1].1, + "and holds it until the transport is actually up" ); - assert_eq!(status, Some(AssetLockStatus::Broadcast)); - } - + assert_eq!( + attempts[1].0, attempts[0].0, + "the retry re-sends the ORIGINAL transaction, never a rebuild" + ); + assert_eq!( + fixture.status().await, + Some(AssetLockStatus::Broadcast), + "a send that dispatched advances the row" + ); + assert_eq!( + broadcaster.readiness_budgets(), + vec![ + BROADCAST_TRANSPORT_READY_WAIT, + DEFERRED_RESUME_TRANSPORT_WAIT, + BROADCAST_TRANSPORT_READY_WAIT, + ], + "the host's wait keeps its ceiling; only the off-thread retry \ + waits under the longer one, and its own resume then finds the \ + transport ready at once" + ); + + // The retry's proof wait runs out the caller's budget and the task + // ends, releasing the outpoint for a later deferral. + tokio::time::sleep(Duration::from_secs(300)).await; + assert!( + !fixture.retry_deferred(), + "a finished retry no longer holds the outpoint" + ); + assert_eq!( + broadcaster.attempts().len(), + 2, + "and nothing sends again on its own" + ); + } + + /// The deferred retry acts only on a row that is exactly as the + /// deferring resume left it. A host that resumed the same lock again + /// meanwhile — with a transport that had come up by then — made the + /// send itself and advanced the row, and the retry must not add a + /// send of its own on top of that. + #[tokio::test(start_paused = true)] + async fn a_deferred_retry_stands_down_when_the_row_moved_on_meanwhile() { + let broadcaster = Arc::new(StartingUpBroadcaster::comes_up_after( + TRANSPORT_COMES_UP_AFTER_THE_CEILING, + )); + let fixture = tracked_lock_at(broadcaster.clone(), AssetLockStatus::Built).await; + + let _ = fixture + .manager + .resume_asset_lock(&fixture.out_point, Some(Duration::from_secs(300))) + .await; + assert!(fixture.retry_deferred()); + + // A second host-driven resume before the retry's own wait has + // observed readiness: its bounded wait is what sees the transport + // come up, and its send advances the row. + let _ = fixture + .manager + .resume_asset_lock(&fixture.out_point, Some(Duration::from_secs(300))) + .await; + assert_eq!(fixture.status().await, Some(AssetLockStatus::Broadcast)); + let attempts = broadcaster.attempts(); + assert_eq!(attempts.len(), 2); + assert!( + attempts[1].1, + "the second resume's send waited for the transport" + ); + + tokio::time::sleep(Duration::from_secs(400)).await; + assert_eq!( + broadcaster.attempts().len(), + 2, + "the deferred retry saw a row that had moved on and sent nothing" + ); + assert!(!fixture.retry_deferred()); + } + + /// A lock consumed while its resume is suspended stays consumed. + /// + /// The resume snapshots the row at `Built`, then waits for the + /// transport; an explicit funding flow can consume the same lock in + /// that interval. The stale resume's `Built` → `Broadcast` advance + /// must not overwrite the tombstone — a `Consumed` row that reads + /// `Broadcast` again is a spent lock the wallet would offer for reuse + /// until the next restart — and the resume must report the lock as + /// consumed rather than hand out a proof for it. + #[tokio::test(start_paused = true)] + async fn a_lock_consumed_while_the_resume_awaits_the_transport_keeps_its_tombstone() { + let broadcaster = Arc::new(StartingUpBroadcaster::comes_up_after( + TRANSPORT_COMES_UP_AFTER_POLLS, + )); + let fixture = tracked_lock_at(broadcaster.clone(), AssetLockStatus::Built).await; + + let resume = fixture + .manager + .resume_asset_lock(&fixture.out_point, Some(Duration::from_secs(300))); + let consume_meanwhile = async { + // One poll into the transport wait: the resume has taken its + // snapshot and is suspended. + tokio::time::sleep(Duration::from_millis(1)).await; + fixture + .manager + .consume_asset_lock(&fixture.out_point) + .await + .expect("consume") + }; + let (outcome, _) = tokio::join!(resume, consume_meanwhile); + + let error = outcome.expect_err("a consumed lock yields no proof"); + assert!( + matches!( + error, + PlatformWalletError::AssetLockAlreadyConsumed(actual) if actual == fixture.out_point + ), + "expected AssetLockAlreadyConsumed, got {error:?}" + ); + assert_eq!( + fixture.status().await, + Some(AssetLockStatus::Consumed), + "the tombstone survives the stale resume's status advance" + ); + assert!( + fixture.proof().await.is_none(), + "and no proof is re-attached to a consumed lock" + ); + assert!( + fixture + .manager + .resume_dispatch_claims + .lock() + .expect("claims mutex") + .is_empty(), + "consumption releases the resume's dispatch claim — there is no \ + cleanup left for it to exclude" + ); + assert!( + !fixture.retry_deferred(), + "a consumed lock is not a lock that needs a send" + ); + } + + /// `Consumed` absorbs every later transition, whatever proof it comes + /// with: the proof-attaching advance a resume makes after its wait is + /// the other place a stale resume could resurrect a spent lock. + #[tokio::test] + async fn a_consumed_tombstone_absorbs_every_later_status_advance() { + use dpp::identity::state_transition::asset_lock_proof::chain::ChainAssetLockProof; + + let fixture = tracked_lock_at( + Arc::new(RecordingBroadcaster::default()), + AssetLockStatus::Broadcast, + ) + .await; + fixture + .manager + .consume_asset_lock(&fixture.out_point) + .await + .expect("consume"); + + let proof = dpp::prelude::AssetLockProof::Chain(ChainAssetLockProof { + core_chain_locked_height: 78, + out_point: fixture.out_point, + }); + let error = fixture + .manager + .advance_asset_lock_status( + &fixture.out_point, + AssetLockStatus::ChainLocked, + Some(proof), + ) + .await + .expect_err("a tombstone is never overwritten"); + assert!(matches!( + error, + PlatformWalletError::AssetLockAlreadyConsumed(actual) if actual == fixture.out_point + )); + assert_eq!(fixture.status().await, Some(AssetLockStatus::Consumed)); + assert!(fixture.proof().await.is_none()); + } + + /// A conditional advance whose premise no longer holds leaves the row + /// alone: a resume recording "this send dispatched" must not regress a + /// row a concurrent resume already carried to a proof-bearing status. + #[tokio::test] + async fn a_conditional_advance_leaves_a_row_that_moved_on_alone() { + use dpp::identity::state_transition::asset_lock_proof::chain::ChainAssetLockProof; + + let fixture = tracked_lock_at( + Arc::new(RecordingBroadcaster::default()), + AssetLockStatus::Built, + ) + .await; + let proof = dpp::prelude::AssetLockProof::Chain(ChainAssetLockProof { + core_chain_locked_height: 78, + out_point: fixture.out_point, + }); + fixture + .manager + .advance_asset_lock_status( + &fixture.out_point, + AssetLockStatus::ChainLocked, + Some(proof.clone()), + ) + .await + .expect("the concurrent resume's proof lands first"); + + let advanced = fixture + .manager + .advance_asset_lock_status_if( + &fixture.out_point, + |current| *current == AssetLockStatus::Built, + AssetLockStatus::Broadcast, + None, + ) + .await + .expect("a stale premise is not an error"); + assert!( + advanced.is_none(), + "nothing to persist for a skipped advance" + ); + assert_eq!(fixture.status().await, Some(AssetLockStatus::ChainLocked)); + assert_eq!(fixture.proof().await, Some(proof)); + + let advanced = fixture + .manager + .advance_asset_lock_status_if( + &fixture.out_point, + |current| *current == AssetLockStatus::ChainLocked, + AssetLockStatus::ChainLocked, + None, + ) + .await + .expect("a premise that holds applies"); + assert!(advanced.is_some()); + } + + /// Regression: an ambiguous re-broadcast on the UNBOUNDED resume path + /// must not hang. + /// + /// `MaybeSent` is the broadcaster's verdict for a genuinely rejected + /// transaction as much as for an accepted one — `DapiBroadcaster` + /// classifies every failure that way, and the SPV broadcaster reaches + /// `Rejected` only on `NotConnected`. So advancing to `Broadcast` and + /// then waiting with `wait_for_proof(None)` — which is what the three + /// `resume_asset_lock(.., None)` production call sites do — turned a + /// broadcast failure that used to surface in ~30s into a wait that never + /// ends, because no proof can arrive for a tx that was never accepted. + /// + /// `start_paused` auto-advances the substituted bound, so this asserts + /// termination *and* that the caller gets the pre-#4367 typed error back. + #[tokio::test(start_paused = true)] + async fn unbounded_resume_of_an_ambiguous_rebroadcast_terminates() { + let (error, status) = resume_lock_at( + Arc::new(AlwaysMaybeSentBroadcaster), + AssetLockStatus::Built, + None, + ) + .await; + + assert!( + matches!( + error, + PlatformWalletError::TransactionBroadcastUnconfirmed(_) + ), + "an unbounded resume whose re-broadcast was ambiguous must end as a \ + broadcast-unconfirmed failure rather than hang, got {error:?}" + ); + assert_eq!( + status, + Some(AssetLockStatus::Broadcast), + "the advance itself is still correct — the row stays resumable so a \ + later pass can pick up a proof that does eventually arrive" + ); + } + + /// The bounded callers must be untouched by the fix above. The shielded + /// seed pool passes its own timeout and treats `FinalityTimeout` as a + /// pacing signal (pause, resume the lock later), so re-typing the error + /// for every caller would have broken a working flow to fix a different + /// one. + #[tokio::test] + async fn bounded_resume_of_an_ambiguous_rebroadcast_still_reports_finality_timeout() { + let (error, status) = resume_lock_at( + Arc::new(AlwaysMaybeSentBroadcaster), + AssetLockStatus::Built, + Some(Duration::from_millis(10)), + ) + .await; + + assert!( + matches!(error, PlatformWalletError::FinalityTimeout(_)), + "a caller-supplied timeout must keep its FinalityTimeout semantics: {error:?}" + ); + assert_eq!(status, Some(AssetLockStatus::Broadcast)); + } + /// A rejected defensive re-broadcast must not be reported as a rejection /// of the ORIGINAL transaction — and the row SURVIVES it. /// @@ -3638,8 +4483,8 @@ mod tests { ); } - /// A definite rejection must consult the LOCAL record before failing - /// the resume. + /// A resume must consult the LOCAL record rather than fail on a + /// transport that cannot carry a send it does not need. /// /// A row can sit at `Broadcast` while its transaction record already /// carries finality: `LockNotifyHandler` only wakes waiters, so an @@ -3647,11 +4492,14 @@ mod tests { /// but never advances the tracked status, and `enrich_from_record` /// upgrades only `InChainLockedBlock` records on scan paths — an /// `InstantSend` context is invisible to it. On the next launch - /// `catchUpStuckAssetLocks` resumes the row before SPV connects, the - /// defensive re-broadcast draws `Rejected` (unstarted client / zero - /// peers), and the pre-fix arm failed the resume even though - /// `wait_for_proof` would have returned the proof on its first - /// iteration, straight from the record, without any network at all. + /// `catchUpStuckAssetLocks` resumes the row before SPV connects, where + /// every send draws `Rejected` (unstarted client / zero peers), and the + /// pre-fix arm failed the resume even though `wait_for_proof` would have + /// returned the proof on its first iteration, straight from the record, + /// without any network at all. The record is now read before the send is + /// even considered, so this broadcaster's rejection is never reached — + /// the outcome the test pins is the same one either ordering owes the + /// caller. #[tokio::test] async fn definite_rejection_on_a_broadcast_lock_yields_the_local_proof() { use dashcore::ephemerealdata::instant_lock::InstantLock; @@ -3754,86 +4602,956 @@ mod tests { ); } - /// Regression: the `Broadcast` arm's proof wait must terminate on the - /// UNBOUNDED resume path too. + /// The `Built` twin of the test above — the same offline relaunch, one + /// status earlier. /// - /// The first revision of this fix bounded only the `Built` arm. Its own - /// retained behavior — advance an ambiguous `Built` lock to `Broadcast` - /// and leave the row there — routes exactly that lock into this arm on - /// the next resume pass, where a bare `wait_for_proof(None)` waits on - /// `Notify` forever. The hang was deferred by one pass, not removed, and - /// under the FFI's `runtime().block_on(...)` it pins a host thread. + /// A row sits at `Built` after a successful broadcast too, whenever the + /// app died between the send and the status advance; the finality that + /// followed is then filed on the funding account by the ordinary SPV scan + /// while nothing is waiting for it. On the next launch the catch-up + /// resumes the row before SPV connects, where every send draws + /// `Rejected` (unstarted client), and the resume must complete from the + /// record it already holds instead of failing a lock that is finished. /// - /// `start_paused` auto-advances the substituted bound, so this asserts - /// termination *and* the typed error the caller gets on expiry. - #[tokio::test(start_paused = true)] - async fn unbounded_resume_of_a_broadcast_row_terminates() { - let (error, status) = resume_lock_at( - Arc::new(AlwaysMaybeSentBroadcaster), - AssetLockStatus::Broadcast, - None, - ) - .await; - - assert!( - matches!( - error, - PlatformWalletError::TransactionBroadcastUnconfirmed(_) - ), - "an unbounded resume of a Broadcast row must end as a \ - broadcast-unconfirmed failure rather than hang, got {error:?}" - ); - assert_eq!( - status, - Some(AssetLockStatus::Broadcast), - "the row must stay exactly where it was so a proof arriving after the \ - bound is picked up by the next resume" - ); - } - - /// The bounded callers of the `Broadcast` arm keep their semantics, the - /// same way the `Built` arm's do: `or` is the identity on `Some`, and the - /// re-typing is gated on the caller having asked for an unbounded wait. + /// What it must NOT do on the way is send at all, or advance the row to + /// `Broadcast` first. That advance is the record of a send that reached + /// the network, and the only send this pass made never left the device — + /// writing it would tell the next reader that an undispatched attempt + /// was broadcast. The final status is no evidence either way, since a + /// `Broadcast` step would be overwritten by the same + /// `InstantSendLocked` a moment later, so it is the persisted trail that + /// is asserted. #[tokio::test] - async fn bounded_resume_of_a_broadcast_row_still_reports_finality_timeout() { - let (error, status) = resume_lock_at( - Arc::new(AlwaysMaybeSentBroadcaster), - AssetLockStatus::Broadcast, - Some(Duration::from_millis(10)), - ) - .await; + async fn definite_rejection_on_a_built_lock_yields_the_local_proof() { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::managed_account::managed_account_trait::ManagedAccountTrait; + use key_wallet::managed_account::transaction_record::{ + TransactionDirection, TransactionRecord, + }; + use key_wallet::transaction_checking::{TransactionContext, TransactionType}; - assert!( - matches!(error, PlatformWalletError::FinalityTimeout(_)), - "a caller-supplied timeout must keep its FinalityTimeout semantics: {error:?}" + 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"), ); - assert_eq!(status, Some(AssetLockStatus::Broadcast)); - } - - /// The `RecoveredFromChain` proof-less fallback is bounded for the same - /// reason, even though its wait resolves immediately whenever the - /// chain-locked record it reads is still present. The accident that - /// leaves a `RecoveredFromChain` row without its persisted proof can take - /// the record too, and then the "resolves immediately by construction" - /// argument yields an unbounded `Notify` loop. No re-typing: nothing is - /// broadcast on this arm, so `FinalityTimeout` is the honest verdict. - #[tokio::test(start_paused = true)] - async fn unbounded_resume_of_a_proofless_recovered_row_terminates() { - let (error, status) = resume_lock_at( - Arc::new(AlwaysRejectedBroadcaster), - AssetLockStatus::RecoveredFromChain, - None, - ) - .await; - - assert!( - matches!(error, PlatformWalletError::FinalityTimeout(_)), - "a proof-less RecoveredFromChain resume must terminate, got {error:?}" + let broadcaster = Arc::new(CountingRejectedBroadcaster::default()); + let persistence = Arc::new(RecordingPersistence::default()); + let manager = AssetLockManager::new( + sdk, + Arc::clone(&wallet_manager), + wallet_id, + Arc::new(Notify::new()), + Arc::clone(&broadcaster) as Arc, + WalletPersister::new(wallet_id, Arc::clone(&persistence) as Arc<_>), ); - assert_eq!( - status, - Some(AssetLockStatus::RecoveredFromChain), - "the row keeps its status — the resume proved nothing new about it" + let (transaction, _path) = manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::AssetLockAddressTopUp, + 4, + &signer, + ) + .await + .expect("build asset lock"); + let out_point = OutPoint::new(transaction.txid(), 0); + { + let mut wm = wallet_manager.write().await; + let info = wm + .get_wallet_info_mut(&wallet_id) + .expect("wallet must remain registered"); + // The finality that arrived while nobody was waiting: an + // IS-locked record for the funding tx, filed under the BIP44 + // account the lock was built from. + let record = TransactionRecord::new( + transaction.clone(), + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InstantSend(InstantLock::default()), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + 0, + ); + info.core_wallet + .accounts + .standard_bip44_accounts + .get_mut(&0) + .expect("funded wallet has BIP44 account 0") + .transactions_mut() + .insert(record.txid, record); + info.tracked_asset_locks.insert( + out_point, + TrackedAssetLock { + out_point, + transaction, + account_index: 0, + funding_type: AssetLockFundingType::AssetLockAddressTopUp, + identity_index: 4, + amount: 1_000_000, + status: AssetLockStatus::Built, + proof: None, + }, + ); + } + + let (proof, _path) = manager + .resume_asset_lock(&out_point, None) + .await + .expect("a locally-proven Built lock must survive a rejected re-broadcast"); + assert!( + matches!(proof, dpp::prelude::AssetLockProof::Instant(_)), + "the proof must come from the record's InstantSend context: {proof:?}" + ); + assert_eq!( + wallet_manager + .read() + .await + .get_wallet_info(&wallet_id) + .expect("wallet") + .tracked_asset_locks + .get(&out_point) + .expect("lock stays tracked") + .status, + AssetLockStatus::InstantSendLocked, + "the resume must advance the row exactly as a waited-for proof would" + ); + assert_eq!( + broadcaster.sends(), + 0, + "an already-final row needs no send at all: the record is read \ + before the transport is even consulted, so nothing is dispatched \ + and nothing waits for a transport to dispatch it" + ); + let persisted: Vec = persistence + .stored + .lock() + .expect("recording mutex") + .iter() + .filter_map(|cs| cs.asset_locks.as_ref()) + .filter_map(|al| al.asset_locks.get(&out_point)) + .map(|entry| entry.status.clone()) + .collect(); + assert_eq!( + persisted, + vec![AssetLockStatus::InstantSendLocked], + "the row must go straight to its proven status: an intermediate \ + Broadcast write would durably claim a send that never left the \ + device" + ); + } + + /// Regression: the `Broadcast` arm's proof wait must terminate on the + /// UNBOUNDED resume path too. + /// + /// The first revision of this fix bounded only the `Built` arm. Its own + /// retained behavior — advance an ambiguous `Built` lock to `Broadcast` + /// and leave the row there — routes exactly that lock into this arm on + /// the next resume pass, where a bare `wait_for_proof(None)` waits on + /// `Notify` forever. The hang was deferred by one pass, not removed, and + /// under the FFI's `runtime().block_on(...)` it pins a host thread. + /// + /// `start_paused` auto-advances the substituted bound, so this asserts + /// termination *and* the typed error the caller gets on expiry. + #[tokio::test(start_paused = true)] + async fn unbounded_resume_of_a_broadcast_row_terminates() { + let (error, status) = resume_lock_at( + Arc::new(AlwaysMaybeSentBroadcaster), + AssetLockStatus::Broadcast, + None, + ) + .await; + + assert!( + matches!( + error, + PlatformWalletError::TransactionBroadcastUnconfirmed(_) + ), + "an unbounded resume of a Broadcast row must end as a \ + broadcast-unconfirmed failure rather than hang, got {error:?}" + ); + assert_eq!( + status, + Some(AssetLockStatus::Broadcast), + "the row must stay exactly where it was so a proof arriving after the \ + bound is picked up by the next resume" + ); + } + + /// The bounded callers of the `Broadcast` arm keep their semantics, the + /// same way the `Built` arm's do: `or` is the identity on `Some`, and the + /// re-typing is gated on the caller having asked for an unbounded wait. + #[tokio::test] + async fn bounded_resume_of_a_broadcast_row_still_reports_finality_timeout() { + let (error, status) = resume_lock_at( + Arc::new(AlwaysMaybeSentBroadcaster), + AssetLockStatus::Broadcast, + Some(Duration::from_millis(10)), + ) + .await; + + assert!( + matches!(error, PlatformWalletError::FinalityTimeout(_)), + "a caller-supplied timeout must keep its FinalityTimeout semantics: {error:?}" + ); + assert_eq!(status, Some(AssetLockStatus::Broadcast)); + } + + /// The `RecoveredFromChain` proof-less fallback is bounded for the same + /// reason, even though its wait resolves immediately whenever the + /// chain-locked record it reads is still present. The accident that + /// leaves a `RecoveredFromChain` row without its persisted proof can take + /// the record too, and then the "resolves immediately by construction" + /// argument yields an unbounded `Notify` loop. No re-typing: nothing is + /// broadcast on this arm, so `FinalityTimeout` is the honest verdict. + #[tokio::test(start_paused = true)] + async fn unbounded_resume_of_a_proofless_recovered_row_terminates() { + let (error, status) = resume_lock_at( + Arc::new(AlwaysRejectedBroadcaster), + AssetLockStatus::RecoveredFromChain, + None, + ) + .await; + + assert!( + matches!(error, PlatformWalletError::FinalityTimeout(_)), + "a proof-less RecoveredFromChain resume must terminate, got {error:?}" + ); + assert_eq!( + status, + Some(AssetLockStatus::RecoveredFromChain), + "the row keeps its status — the resume proved nothing new about it" + ); + } + /// How long the fake transport takes to "come up", expressed in + /// readiness polls so the tests stay deterministic under a paused clock. + const TRANSPORT_COMES_UP_AFTER_POLLS: u32 = 3; + + /// Broadcaster whose transport comes up partway through the resume, the + /// way the SPV client does while an app is launching. + /// + /// Until it is up, `broadcast` reproduces `SpvRuntime`'s pre-send + /// verdict for an unstarted client: `Rejected` — definitively never + /// sent. Each attempt is recorded together with the readiness state at + /// the moment it was made, so a test can tell "held the send until the + /// transport was up" apart from "sent into a dead transport and got + /// lucky". + struct StartingUpBroadcaster { + /// Readiness polls before the transport comes up; `None` never + /// comes up at all. + comes_up_after: Option, + polls: AtomicU32, + ready: AtomicBool, + /// Every budget handed to `wait_until_ready`, in call order. + readiness_budgets: Mutex>, + attempts: Mutex>, + } + + impl StartingUpBroadcaster { + fn comes_up_after(polls: u32) -> Self { + Self { + comes_up_after: Some(polls), + polls: AtomicU32::new(0), + ready: AtomicBool::new(false), + readiness_budgets: Mutex::new(Vec::new()), + attempts: Mutex::new(Vec::new()), + } + } + + fn never_comes_up() -> Self { + Self { + comes_up_after: None, + polls: AtomicU32::new(0), + ready: AtomicBool::new(false), + readiness_budgets: Mutex::new(Vec::new()), + attempts: Mutex::new(Vec::new()), + } + } + + fn readiness_budgets(&self) -> Vec { + self.readiness_budgets + .lock() + .expect("readiness budget mutex") + .clone() + } + + 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: Duration) -> bool { + self.readiness_budgets + .lock() + .expect("readiness budget mutex") + .push(timeout); + tokio::time::timeout(timeout, async { + let comes_up_after = match self.comes_up_after { + Some(polls) => polls, + // Never comes up. Park rather than spin, so a paused + // clock jumps straight to the budget's expiry however + // large that budget is. + None => match std::future::pending::().await {}, + }; + loop { + let polls = self.polls.fetch_add(1, Ordering::SeqCst) + 1; + if polls >= comes_up_after { + self.ready.store(true, Ordering::SeqCst); + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .is_ok() + } + } + + /// The launch catch-up resumes a `Built` lock while the SPV client is + /// still starting. The resume gets exactly one send, and a send into an + /// unstarted client draws `Rejected` — the never-sent verdict, which + /// ends the `Built` arm. Nothing reschedules the catch-up, so a lock + /// that loses that race stays at `Built` with its funds behind it, and + /// every later session repeats the identical race. The send has to be + /// held until the transport is actually up. + #[tokio::test(start_paused = true)] + async fn built_resume_holds_the_broadcast_until_the_transport_is_up() { + let broadcaster = Arc::new(StartingUpBroadcaster::comes_up_after( + TRANSPORT_COMES_UP_AFTER_POLLS, + )); + let (error, status) = resume_lock_at( + broadcaster.clone(), + AssetLockStatus::Built, + Some(Duration::from_secs(300)), + ) + .await; + + let attempts = broadcaster.attempts(); + assert_eq!(attempts.len(), 1, "the resume gets exactly one send"); + let (_broadcast_tx, was_ready) = &attempts[0]; + assert!( + was_ready, + "the send must land after the transport came up, not into a \ + client that cannot dispatch it" + ); + assert_eq!( + status, + Some(AssetLockStatus::Broadcast), + "a send that actually dispatched must advance the row: {error:?}" + ); + assert!( + matches!(error, PlatformWalletError::FinalityTimeout(_)), + "the resume must go on to the proof wait, not fail on the \ + broadcast: {error:?}" + ); + } + + /// The transport wait must stay a delay and never become a park. + /// + /// Hosts drive the catch-up through the FFI's blocking entry point, + /// several locks at a time, on a fixed-width thread pool — the same + /// pool the host needs in order to reach `start_spv` and bring the + /// transport up. A wait with no ceiling there parks the transport's own + /// start, and the `None` these production call sites pass would be + /// exactly that ceiling-free wait if the caller's budget were the only + /// bound. `None` must draw the constant instead, and the resume must + /// still terminate with the verdict it would have reported without any + /// wait at all. + /// + /// That verdict is the non-terminal one. A pre-dispatch rejection says + /// only that *this* attempt never left the device: a row sits at `Built` + /// after a successful broadcast too, so an earlier send may be in a + /// mempool or already mined. The resume keeps the row and its input + /// reservation, which is exactly what the definite-rejection code + /// promises it has released — a host honouring that promise would + /// rebuild from other UTXOs and create a second asset lock beside a + /// possibly-live one. + #[tokio::test(start_paused = true)] + async fn transport_wait_is_bounded_even_when_the_caller_named_no_timeout() { + let broadcaster = Arc::new(StartingUpBroadcaster::never_comes_up()); + let (error, status) = + resume_lock_at(broadcaster.clone(), AssetLockStatus::Built, None).await; + + assert_eq!( + broadcaster.readiness_budgets().first(), + Some(&BROADCAST_TRANSPORT_READY_WAIT), + "an unbounded caller must still hand the transport wait a ceiling" + ); + let PlatformWalletError::TransactionBroadcastUnconfirmed(reason) = &error else { + panic!( + "once the wait expires the send is attempted anyway, and the \ + pre-dispatch rejection it draws must surface as an unknown \ + outcome — never as the definite rejection whose contract is \ + that the inputs were released and a rebuild is safe: {error:?}" + ); + }; + assert!( + reason.contains("rejected before dispatch"), + "the failure must name the undispatched send, so it is not \ + confused with the proof wait running out on a send that did \ + dispatch: {reason}" + ); + assert_eq!( + status, + Some(AssetLockStatus::Built), + "the row and its reservation stay exactly as they were — the \ + attempt proved nothing about any earlier send" + ); + let attempts = broadcaster.attempts(); + assert_eq!( + attempts.len(), + 1, + "an expired wait still costs exactly ONE send. A second attempt \ + here would be a retry of a send whose outcome is unknown, which \ + is the double-broadcast this arm's contract exists to prevent" + ); + assert!( + !attempts[0].1, + "and that one send goes into the transport as it actually is — \ + still down. The ceiling bounds the race, it does not resolve it" + ); + } + + /// A caller that named its own budget keeps it: the transport wait is + /// capped by the constant, and whatever it consumes is deducted so the + /// resume's total stays inside what the caller asked for. + #[tokio::test(start_paused = true)] + async fn transport_wait_is_capped_and_deducted_from_the_caller_budget() { + let broadcaster = Arc::new(StartingUpBroadcaster::never_comes_up()); + let caller_budget = Duration::from_secs(300); + let started = tokio::time::Instant::now(); + let (_error, _status) = resume_lock_at( + broadcaster.clone(), + AssetLockStatus::Broadcast, + Some(caller_budget), + ) + .await; + let elapsed = started.elapsed(); + + assert_eq!( + broadcaster.readiness_budgets().first(), + Some(&BROADCAST_TRANSPORT_READY_WAIT), + "the caller's 300s must not become a 300s transport wait" + ); + assert!( + elapsed <= caller_budget, + "the transport wait must come out of the caller's budget, not \ + on top of it: {elapsed:?} > {caller_budget:?}" + ); + } + + /// The deduction is only observable once the transport actually comes + /// up: the resume then goes on to the proof wait, and what that wait + /// gets is the caller's budget MINUS what readiness spent. + /// + /// Without the subtraction the proof wait starts a fresh full budget + /// and the resume overruns the bound its caller asked for — by the + /// transport wait's whole length, up to 15s of a budget the host sized + /// for its own timeout. Under a paused clock the total is exact, so it + /// is the total that is asserted: readiness plus proof, inside the + /// caller's budget, to the millisecond. + #[tokio::test(start_paused = true)] + async fn a_ready_transport_leaves_the_rest_of_the_budget_to_the_proof_wait() { + let broadcaster = Arc::new(StartingUpBroadcaster::comes_up_after( + TRANSPORT_COMES_UP_AFTER_POLLS, + )); + let caller_budget = Duration::from_secs(300); + let started = tokio::time::Instant::now(); + let (error, _status) = resume_lock_at( + broadcaster.clone(), + AssetLockStatus::Built, + Some(caller_budget), + ) + .await; + let elapsed = started.elapsed(); + + assert!( + broadcaster + .attempts() + .first() + .is_some_and(|(_, ready)| *ready), + "the premise: the transport came up and the send dispatched, so \ + the resume reached the proof wait this test measures" + ); + assert!( + matches!(error, PlatformWalletError::FinalityTimeout(_)), + "and that proof wait is what ran out, not the broadcast: {error:?}" + ); + assert_eq!( + elapsed, caller_budget, + "readiness plus proof must total exactly the caller's budget — \ + a proof wait handed the undiminished budget overruns it by \ + however long readiness took" + ); + } + + /// A caller whose budget is already below the ceiling gets its own + /// number, not the ceiling. + /// + /// `min(caller, 15s)` is the whole policy: the constant is an upper + /// bound on the wait, never a floor under it. A host that asks for a 5s + /// resume has usually sized that against something of its own — a + /// foreground refresh, a watchdog — and spending 15s in the transport + /// wait alone would blow through it three times over before the send is + /// even attempted. + #[tokio::test(start_paused = true)] + async fn a_caller_budget_under_the_ceiling_is_the_whole_transport_wait() { + let broadcaster = Arc::new(StartingUpBroadcaster::never_comes_up()); + let caller_budget = Duration::from_secs(5); + let started = tokio::time::Instant::now(); + let (error, _status) = resume_lock_at( + broadcaster.clone(), + AssetLockStatus::Built, + Some(caller_budget), + ) + .await; + let elapsed = started.elapsed(); + + assert_eq!( + broadcaster.readiness_budgets().first(), + Some(&caller_budget), + "a caller under the ceiling hands the transport wait its own \ + budget — the constant caps the wait, it never extends it" + ); + assert_eq!( + elapsed, caller_budget, + "and the resume ends inside that budget rather than at the \ + ceiling: {error:?}" + ); + } + + /// The `Broadcast` arm's defensive re-broadcast is worth attempting only + /// once the transport can carry it: rejected before dispatch it proves + /// nothing, ends the resume as an unknown outcome, and leaves the row to + /// be retried on some later launch. It has to wait too. + #[tokio::test(start_paused = true)] + async fn broadcast_resume_also_holds_its_rebroadcast_for_the_transport() { + let broadcaster = Arc::new(StartingUpBroadcaster::comes_up_after( + TRANSPORT_COMES_UP_AFTER_POLLS, + )); + let (error, status) = resume_lock_at( + broadcaster.clone(), + AssetLockStatus::Broadcast, + Some(Duration::from_secs(300)), + ) + .await; + + let attempts = broadcaster.attempts(); + assert_eq!(attempts.len(), 1, "one defensive re-broadcast"); + assert!( + attempts[0].1, + "the defensive re-broadcast must wait for the transport too — \ + rejected before dispatch it proves nothing and only ends the \ + resume early" + ); + assert!( + matches!(error, PlatformWalletError::FinalityTimeout(_)), + "a dispatched re-broadcast leaves the resume in the proof wait: \ + {error:?}" + ); + assert_eq!(status, Some(AssetLockStatus::Broadcast)); + } + + /// An arm that broadcasts nothing must not consult the transport at all. + /// + /// The proof-less `RecoveredFromChain` fallback resolves from an + /// already chain-locked record, and four production call sites reach it + /// with `timeout: None`. Gating it would buy nothing and cost every one + /// of them the full transport wait on a device whose SPV client is + /// down. + #[tokio::test(start_paused = true)] + async fn a_resume_that_broadcasts_nothing_never_waits_for_the_transport() { + let broadcaster = Arc::new(StartingUpBroadcaster::never_comes_up()); + let (error, _status) = resume_lock_at( + broadcaster.clone(), + AssetLockStatus::RecoveredFromChain, + None, + ) + .await; + + assert!( + broadcaster.readiness_budgets().is_empty(), + "a proof-only arm must not wait on a transport it never uses" + ); + assert!( + broadcaster.attempts().is_empty(), + "a proof-only arm must not broadcast" + ); + assert!( + matches!(error, PlatformWalletError::FinalityTimeout(_)), + "unchanged verdict for a record that never becomes final: {error:?}" + ); + } + + /// Resume a row at `status` whose funding record is ALREADY IS-locked, + /// over a transport that never comes up, and report what the resume + /// cost in virtual time. + /// + /// The IS-locked record is the ordinary state of a lock whose proof + /// arrived while nothing was waiting for it: `LockNotifyHandler` only + /// wakes waiters, so the record is enriched and the tracked status is + /// left where it was. + async fn locally_final_resume_over_a_down_transport( + status: AssetLockStatus, + ) -> ( + dpp::prelude::AssetLockProof, + Duration, + Arc, + AssetLockStatus, + ) { + use dashcore::ephemerealdata::instant_lock::InstantLock; + + 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 broadcaster = Arc::new(StartingUpBroadcaster::never_comes_up()); + let manager = AssetLockManager::new( + sdk, + Arc::clone(&wallet_manager), + wallet_id, + Arc::new(Notify::new()), + Arc::clone(&broadcaster) as Arc, + WalletPersister::new(wallet_id, Arc::new(RecordingPersistence::default())), + ); + let (transaction, _path) = manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::AssetLockAddressTopUp, + 4, + &signer, + ) + .await + .expect("build asset lock"); + let out_point = OutPoint::new(transaction.txid(), 0); + { + let mut wm = wallet_manager.write().await; + let info = wm + .get_wallet_info_mut(&wallet_id) + .expect("wallet must remain registered"); + let record = TransactionRecord::new( + transaction.clone(), + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InstantSend(InstantLock::default()), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + 0, + ); + info.core_wallet + .accounts + .standard_bip44_accounts + .get_mut(&0) + .expect("funded wallet has BIP44 account 0") + .transactions_mut() + .insert(record.txid, record); + info.tracked_asset_locks.insert( + out_point, + TrackedAssetLock { + out_point, + transaction, + account_index: 0, + funding_type: AssetLockFundingType::AssetLockAddressTopUp, + identity_index: 4, + amount: 1_000_000, + status, + proof: None, + }, + ); + } + + let started = tokio::time::Instant::now(); + let (proof, _path) = manager + .resume_asset_lock(&out_point, None) + .await + .expect("an already-final lock must resume with no transport at all"); + let elapsed = started.elapsed(); + let tracked = wallet_manager + .read() + .await + .get_wallet_info(&wallet_id) + .expect("wallet") + .tracked_asset_locks + .get(&out_point) + .expect("lock stays tracked") + .status + .clone(); + (proof, elapsed, broadcaster, tracked) + } + + /// Broadcaster whose transport comes up carrying the finality the record + /// was missing, and whose sends never leave the device anyway. + /// + /// This is the second half of the launch window: the resume's opening + /// probe reads a record with no proof in it, SPV connects and delivers + /// the lock event while the resume is waiting for exactly that, and the + /// send that follows still draws the pre-dispatch rejection because the + /// client has no peers yet. Filing the record from inside + /// `wait_until_ready` is what puts finality in that gap deterministically. + struct FinalityArrivesWithTheTransport { + wallet_manager: Arc>>, + wallet_id: WalletId, + transaction: Mutex>, + sends: AtomicU32, + } + + impl FinalityArrivesWithTheTransport { + fn sends(&self) -> u32 { + self.sends.load(Ordering::SeqCst) + } + } + + #[async_trait] + impl TransactionBroadcaster for FinalityArrivesWithTheTransport { + async fn broadcast(&self, _transaction: &Transaction) -> Result { + self.sends.fetch_add(1, Ordering::SeqCst); + Err(BroadcastError::Rejected { + reason: "SPV broadcast not sent: no connected peers".to_string(), + }) + } + + async fn wait_until_ready(&self, _timeout: Duration) -> bool { + let transaction = self + .transaction + .lock() + .expect("funding transaction slot") + .clone() + .expect("the fixture files the transaction before resuming"); + let mut wm = self.wallet_manager.write().await; + let info = wm + .get_wallet_info_mut(&self.wallet_id) + .expect("wallet must remain registered"); + insert_record( + info, + record_for( + transaction, + TransactionContext::InstantSend( + dashcore::ephemerealdata::instant_lock::InstantLock::default(), + ), + ), + ); + true + } + } + + /// Regression: finality that lands DURING the transport wait must still + /// complete the resume, even though the send it makes afterwards is + /// rejected before dispatch. + /// + /// The opening probe is not the only place a proof can appear. The + /// transport wait is what SPV needs in order to connect, and connecting + /// is what delivers the lock event — so on the launch pass the record is + /// routinely enriched inside the very wait that precedes the send, while + /// the send itself still draws the never-sent verdict from a client with + /// no peers. Reading the record again after that rejection is the only + /// thing standing between an already-final lock and a resume that fails + /// it every launch. + async fn finality_landing_in_the_transport_wait_completes_the_resume(status: AssetLockStatus) { + 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 broadcaster = Arc::new(FinalityArrivesWithTheTransport { + wallet_manager: Arc::clone(&wallet_manager), + wallet_id, + transaction: Mutex::new(None), + sends: AtomicU32::new(0), + }); + let manager = AssetLockManager::new( + sdk, + Arc::clone(&wallet_manager), + wallet_id, + Arc::new(Notify::new()), + Arc::clone(&broadcaster) as Arc, + WalletPersister::new(wallet_id, Arc::new(RecordingPersistence::default())), + ); + let (transaction, _path) = manager + .build_asset_lock_transaction( + 1_000_000, + 0, + AssetLockFundingType::AssetLockAddressTopUp, + 4, + &signer, + ) + .await + .expect("build asset lock"); + let out_point = OutPoint::new(transaction.txid(), 0); + *broadcaster + .transaction + .lock() + .expect("funding transaction slot") = Some(transaction.clone()); + { + 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, + account_index: 0, + funding_type: AssetLockFundingType::AssetLockAddressTopUp, + identity_index: 4, + amount: 1_000_000, + status, + proof: None, + }, + ); + } + + let (proof, _path) = manager + .resume_asset_lock(&out_point, None) + .await + .expect("a lock made final during the transport wait must still resume"); + assert!( + matches!(proof, dpp::prelude::AssetLockProof::Instant(_)), + "the proof must come from the record the transport wait filed: {proof:?}" + ); + assert_eq!( + broadcaster.sends(), + 1, + "the resume must have gone through the send — this is the arm that \ + reads the record AFTER a rejection, not the opening probe" + ); + assert_eq!( + wallet_manager + .read() + .await + .get_wallet_info(&wallet_id) + .expect("wallet") + .tracked_asset_locks + .get(&out_point) + .expect("lock stays tracked") + .status, + AssetLockStatus::InstantSendLocked, + "the resume must advance the row exactly as a waited-for proof would" + ); + } + + #[tokio::test(start_paused = true)] + async fn finality_landing_in_the_transport_wait_completes_a_built_resume() { + finality_landing_in_the_transport_wait_completes_the_resume(AssetLockStatus::Built).await; + } + + #[tokio::test(start_paused = true)] + async fn finality_landing_in_the_transport_wait_completes_a_broadcast_resume() { + finality_landing_in_the_transport_wait_completes_the_resume(AssetLockStatus::Broadcast) + .await; + } + + /// Regression: a `Built` row that is already final locally must complete + /// offline, without paying the transport-readiness ceiling first. + /// + /// This is the canonical launch shape — `catchUpStuckAssetLocks` resumes + /// rows before `start_spv` — and for a row whose proof is already in the + /// record there is nothing for the transport to carry. Consulting + /// readiness first cost that resume the full ceiling on every launch of + /// an offline device and then completed it from the same record anyway. + /// + /// `start_paused` makes the cost measurable: the assertion is that + /// virtual time does not advance at all, which no arrangement of a + /// sleeping wait can satisfy. + #[tokio::test(start_paused = true)] + async fn a_locally_final_built_row_completes_without_touching_the_transport() { + let (proof, elapsed, broadcaster, status) = + locally_final_resume_over_a_down_transport(AssetLockStatus::Built).await; + + assert!( + matches!(proof, dpp::prelude::AssetLockProof::Instant(_)), + "the proof must come from the record's InstantSend context: {proof:?}" + ); + assert!( + broadcaster.readiness_budgets().is_empty(), + "a row that needs no send must not wait for a transport to carry \ + it: readiness was consulted with {:?}", + broadcaster.readiness_budgets() + ); + assert!( + broadcaster.attempts().is_empty(), + "and it must not broadcast either — the transaction is already final" + ); + assert_eq!( + elapsed, + Duration::ZERO, + "the whole resume must cost no time at all; anything else is the \ + readiness ceiling being paid by a lock that is already finished" + ); + assert_eq!( + status, + AssetLockStatus::InstantSendLocked, + "the row is advanced exactly as a waited-for proof would advance it" + ); + } + + /// The `Broadcast` twin: the defensive re-broadcast exists to revive a + /// transaction no node still holds, which an already-final record proves + /// is not this one. + #[tokio::test(start_paused = true)] + async fn a_locally_final_broadcast_row_completes_without_touching_the_transport() { + let (proof, elapsed, broadcaster, status) = + locally_final_resume_over_a_down_transport(AssetLockStatus::Broadcast).await; + + assert!( + matches!(proof, dpp::prelude::AssetLockProof::Instant(_)), + "the proof must come from the record's InstantSend context: {proof:?}" + ); + assert!( + broadcaster.readiness_budgets().is_empty(), + "the defensive re-broadcast is skipped outright, so its transport \ + wait must be skipped with it: readiness was consulted with {:?}", + broadcaster.readiness_budgets() + ); + assert!( + broadcaster.attempts().is_empty(), + "nothing to revive: the record already holds finality" + ); + assert_eq!( + elapsed, + Duration::ZERO, + "the whole resume must cost no time at all; anything else is the \ + readiness ceiling being paid by a lock that is already finished" + ); + assert_eq!( + status, + AssetLockStatus::InstantSendLocked, + "the row is advanced exactly as a waited-for proof would advance it" ); } } diff --git a/packages/rs-platform-wallet/src/wallet/asset_lock/sync/tracking.rs b/packages/rs-platform-wallet/src/wallet/asset_lock/sync/tracking.rs index 9a08a5930d9..a127cdf2b17 100644 --- a/packages/rs-platform-wallet/src/wallet/asset_lock/sync/tracking.rs +++ b/packages/rs-platform-wallet/src/wallet/asset_lock/sync/tracking.rs @@ -2,6 +2,7 @@ use crate::broadcaster::TransactionBroadcaster; use dashcore::OutPoint; +use std::collections::BTreeMap; use crate::changeset::changeset::AssetLockChangeSet; use crate::changeset::changeset::PlatformWalletChangeSet; @@ -11,7 +12,95 @@ use crate::error::PlatformWalletError; use super::super::manager::AssetLockManager; use super::super::tracked::{AssetLockStatus, TrackedAssetLock}; +/// One resume's hold on an outpoint's cleanup-exclusion window. +/// +/// Held from the moment the resume snapshots the tracked row until it has +/// sent the transaction and recorded that send by advancing the row — the +/// span in which the resume is committed to broadcasting a transaction it +/// has already read out of the map. While it stands, +/// [`untrack_asset_lock`](AssetLockManager::untrack_asset_lock) refuses to +/// remove the row, which is what keeps the initial build's rejection +/// cleanup from releasing the funding reservation and the in-broadcast +/// fence under a send that is still coming. +/// +/// A claim releases on drop until the resume observes local finality or +/// enters a broadcast that may have side effects. From that point it becomes +/// sticky: cancellation cannot prove that releasing the inputs is safe, so +/// the exclusion survives until a status transition makes the `Built` +/// cleanup inapplicable. Exits proven to be pre-dispatch restore ordinary +/// RAII release. +pub(crate) struct ResumeDispatchClaim<'a> { + claims: &'a std::sync::Mutex>, + out_point: OutPoint, + release_on_drop: bool, +} + +impl ResumeDispatchClaim<'_> { + /// Preserve cleanup exclusion if this future is cancelled before it can + /// record the transaction's send or proof. + pub(crate) fn preserve_on_drop(&mut self) { + self.release_on_drop = false; + } + + /// Restore ordinary RAII release after the current attempt is proven not + /// to have left the device and no local finality was observed. + pub(crate) fn release_on_drop(&mut self) { + self.release_on_drop = true; + } +} + +impl Drop for ResumeDispatchClaim<'_> { + fn drop(&mut self) { + if !self.release_on_drop { + return; + } + // Recover from poisoning rather than skipping the release: a claim + // that outlived its resume would block the rejection cleanup — and + // with it the funding reservation's release — for the process's + // remaining lifetime. + let mut claims = self.claims.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(count) = claims.get_mut(&self.out_point) { + *count -= 1; + if *count == 0 { + claims.remove(&self.out_point); + } + } + } +} + impl AssetLockManager { + /// Claim `out_point`'s dispatch window for a resume that is about to + /// broadcast. + /// + /// MUST be called while the resume still holds the wallet read guard it + /// snapshotted the tracked row under. That is where the serialization + /// against the rejection cleanup comes from: the cleanup reads the claim + /// under the wallet WRITE guard, so a claim taken under the read guard is + /// either already visible to it — and the row is kept — or the cleanup + /// went first, removed the row, and the snapshot the claim would have + /// protected never happened. + pub(crate) fn claim_resume_dispatch(&self, out_point: OutPoint) -> ResumeDispatchClaim<'_> { + *self + .resume_dispatch_claims + .lock() + .unwrap_or_else(|e| e.into_inner()) + .entry(out_point) + .or_insert(0) += 1; + ResumeDispatchClaim { + claims: &self.resume_dispatch_claims, + out_point, + release_on_drop: true, + } + } + + /// Whether active or sticky resume state excludes rejected-build cleanup. + fn resume_cleanup_excluded(&self, out_point: &OutPoint) -> bool { + self.resume_dispatch_claims + .lock() + .unwrap_or_else(|e| e.into_inner()) + .contains_key(out_point) + } + /// Snapshot the funding role, bound index and status used to authorize an /// existing-lock resume. Taking all three under one read lock avoids a /// role/status time-of-check/time-of-use split in the resolver. @@ -59,17 +148,41 @@ impl AssetLockManager { /// re-broadcast a transaction whose inputs may have been re-spent. /// /// Idempotent: returns an empty changeset if the outpoint is not - /// tracked. Guarded on the row still being - /// [`Built`](AssetLockStatus::Built): if a concurrent flow advanced it - /// (e.g. a `resume_asset_lock` that re-broadcast in the window between - /// the rejected broadcast and this cleanup), the progress is kept - /// rather than clobbered. The caller queues the changeset (call sites - /// live in `asset_lock/build.rs`, inside the module). + /// tracked. Guarded twice over, and both guards mean the same thing — + /// that the transaction may be live or committed to dispatch, so the row + /// and everything that pins its inputs must stay: + /// + /// 1. The row must still be [`Built`](AssetLockStatus::Built). A resume + /// that already re-broadcast advanced it, and that advance is + /// positive evidence the transaction reached the network. + /// 2. No active or sticky cleanup exclusion may remain + /// ([`claim_resume_dispatch`](Self::claim_resume_dispatch)). A resume + /// that has snapshotted the row but not yet sent is still `Built`, and + /// cancellation after a possible send or observed proof can leave it + /// there. Guard 1 alone cannot distinguish either case from a clean + /// pre-dispatch exit. + /// + /// Refusing on either guard is what the caller reads back out of the + /// changeset: an empty `removed` set keeps the funding reservation and + /// the in-broadcast fence held and turns the definite-rejection verdict + /// into the unknown outcome, whose contract — row tracked, inputs + /// reserved, do not retry — is the one that actually holds. The caller + /// queues the changeset (call sites live in `asset_lock/build.rs`, + /// inside the module). pub(crate) async fn untrack_asset_lock(&self, out_point: &OutPoint) -> AssetLockChangeSet { let mut wm = self.wallet_manager.write().await; + // Read under the write guard, which is what makes this atomic + // against a resume claiming the window under its read guard. + let cleanup_excluded = self.resume_cleanup_excluded(out_point); let mut cs = AssetLockChangeSet::default(); if let Some(info) = wm.get_wallet_info_mut(&self.wallet_id) { match info.tracked_asset_locks.get(out_point) { + Some(_) if cleanup_excluded => tracing::warn!( + outpoint = %out_point, + "untrack_asset_lock: resume state excludes cleanup of this Built lock — \ + leaving it tracked, since its transaction may already be live or \ + committed to dispatch" + ), Some(entry) if entry.status == AssetLockStatus::Built => { info.tracked_asset_locks.remove(out_point); cs.removed.insert(*out_point); @@ -159,6 +272,14 @@ impl AssetLockManager { entry.status = AssetLockStatus::Consumed; entry.proof = None; // one-shot — never relevant after consumption cs.asset_locks.insert(*out_point, (&*entry).into()); + // A tombstone is beyond rejected-build cleanup, so no + // resume claim — active or sticky — has anything left + // to exclude. Same release `advance_asset_lock_status` + // performs when a row leaves `Built`. + self.resume_dispatch_claims + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(out_point); } Some(_) => { tracing::debug!( @@ -271,12 +392,45 @@ impl AssetLockManager { /// /// Returns an [`AssetLockChangeSet`] carrying a full snapshot of the /// updated entry. + /// + /// [`Consumed`](AssetLockStatus::Consumed) is absorbing: a tombstone is + /// never overwritten, and the attempt fails as + /// [`PlatformWalletError::AssetLockAlreadyConsumed`]. A resume snapshots + /// the row and then suspends — on the transport, the send and the proof + /// wait — and an explicit funding flow can consume the very same lock in + /// that interval. Letting the stale resume land its `Broadcast` / + /// `InstantSendLocked` / `ChainLocked` afterwards would resurrect a + /// spent lock in memory, and the wallet would offer it for reuse until + /// the next restart or reconciliation put the tombstone back. pub(crate) async fn advance_asset_lock_status( &self, out_point: &OutPoint, new_status: AssetLockStatus, proof: Option, ) -> Result { + self.advance_asset_lock_status_if(out_point, |_| true, new_status, proof) + .await + .map(|advanced| advanced.expect("an unconditional advance always applies")) + } + + /// [`advance_asset_lock_status`](Self::advance_asset_lock_status) gated + /// on the row's CURRENT status: the transition applies only when + /// `expected(¤t)` holds, and returns `Ok(None)` — row untouched — + /// when it does not. `Consumed` stays absorbing regardless of the + /// predicate. + /// + /// For a transition whose evidence is bound to a particular prior state. + /// A resume's `Built` → `Broadcast` advance records "this send + /// dispatched", which a row that a concurrent resume has meanwhile + /// carried to a proof-bearing status must not be regressed to; its + /// predicate is therefore "still `Built`". + pub(crate) async fn advance_asset_lock_status_if( + &self, + out_point: &OutPoint, + expected: impl FnOnce(&AssetLockStatus) -> bool, + new_status: AssetLockStatus, + proof: Option, + ) -> Result, PlatformWalletError> { let mut wm = self.wallet_manager.write().await; let info = wm .get_wallet_info_mut(&self.wallet_id) @@ -287,6 +441,25 @@ impl AssetLockManager { out_point )) })?; + if entry.status == AssetLockStatus::Consumed { + tracing::warn!( + outpoint = %out_point, + attempted = ?new_status, + "advance_asset_lock_status: the lock was consumed while this \ + transition was in flight; keeping the tombstone" + ); + return Err(PlatformWalletError::AssetLockAlreadyConsumed(*out_point)); + } + if !expected(&entry.status) { + tracing::info!( + outpoint = %out_point, + current = ?entry.status, + attempted = ?new_status, + "advance_asset_lock_status: the row moved on while this \ + transition was in flight; leaving it where it is" + ); + return Ok(None); + } entry.status = new_status; if proof.is_some() { entry.proof = proof; @@ -294,6 +467,15 @@ impl AssetLockManager { let mut cs = AssetLockChangeSet::default(); cs.asset_locks.insert(*out_point, (&*entry).into()); - Ok(cs) + if entry.status != AssetLockStatus::Built { + // The status now excludes rejected-build cleanup on its own. + // Clear both active claims and sticky claims left by cancelled + // resumes; their eventual drops tolerate the absent entry. + self.resume_dispatch_claims + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(out_point); + } + Ok(Some(cs)) } } diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift index 8b7025e72b3..8093bab0c03 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManager.swift @@ -2010,6 +2010,17 @@ 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. + /// + /// Safe to call before SPV is up, and hosts must not add a + /// readiness gate of their own here. Load runs before + /// `platform_wallet_manager_start_spv`, so a lock still at + /// `Built` would otherwise take a never-sent rejection and — with + /// nothing rescheduling the catch-up — stay stranded for the + /// session. Rust's `resume_asset_lock` gives the SPV transport a + /// bounded chance to come up before it broadcasts. Gating here + /// would delay the locks that need no broadcast at all, and a gate + /// that waited on the same thread pool the host needs to *reach* + /// `startSpv` would delay the transport it is waiting for. public func catchUpStuckAssetLocks(wallets: [ManagedPlatformWallet]) { guard let persistenceHandler = persistenceHandler else { return } for wallet in wallets {