Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions packages/rs-platform-wallet-ffi/src/asset_lock/sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,14 @@ pub unsafe extern "C" fn asset_lock_manager_resume(
/// this on a background queue — `runtime().block_on(...)` parks the
/// calling thread for up to `timeout_secs` (or **indefinitely** when
/// `timeout_secs == 0`, since a ChainLock is guaranteed finality).
///
/// Safe to call before SPV is up. Hosts drive this at wallet load,
/// which typically precedes `platform_wallet_manager_start_spv`, so a
/// lock that still needs broadcasting would otherwise take a
/// definitive never-sent rejection and — nothing retries the catch-up
/// — stay stranded for the session. `resume_asset_lock` therefore
/// waits for the SPV transport (within `timeout_secs`) before
/// broadcasting; callers need no readiness gate of their own.
#[no_mangle]
pub unsafe extern "C" fn asset_lock_manager_catch_up_blocking(
handle: Handle,
Expand Down
34 changes: 34 additions & 0 deletions packages/rs-platform-wallet/src/broadcaster.rs
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,23 @@ pub trait TransactionBroadcaster: Send + Sync {
/// transport ambiguity, or unverifiable response must be
/// [`BroadcastError::MaybeSent`].
async fn broadcast(&self, transaction: &Transaction) -> Result<Txid, BroadcastError>;

/// Resolve once this broadcaster's transport can actually reach the
/// network, or when `timeout` elapses (`None` waits indefinitely).
/// Returns whether readiness was reached.
///
/// Callers that resume queued work at app start use this to avoid
/// racing a transport that is still coming up: a transport-not-ready
/// failure is definitive ([`BroadcastError::Rejected`]) and the resume
/// paths do not retry, so losing that race strands the transaction for
/// the session.
///
/// The default is "always ready" — correct for any broadcaster with no
/// startup phase of its own, such as [`DapiBroadcaster`], whose gRPC
/// requests carry their own connection handling.
async fn wait_until_ready(&self, _timeout: Option<Duration>) -> bool {
true
}
}

/// Broadcasts transactions via Platform's DAPI gRPC endpoint.
Expand Down Expand Up @@ -148,6 +165,11 @@ trait SpvChannel: Send + Sync {
transaction: &Transaction,
timeout: Option<Duration>,
) -> Result<BroadcastResult, BroadcastError>;

/// Resolve once the SPV client is started and has at least one
/// connected peer — the two conditions whose absence makes
/// `broadcast_and_wait` fail before any send.
async fn wait_until_ready(&self, timeout: Option<Duration>) -> bool;
}

#[async_trait]
Expand All @@ -160,6 +182,10 @@ impl SpvChannel for SpvRuntime {
self.broadcast_transaction_and_wait(transaction, timeout)
.await
}

async fn wait_until_ready(&self, timeout: Option<Duration>) -> bool {
SpvRuntime::wait_until_ready(self, timeout).await
}
}

/// Broadcasts purely over the SPV P2P network — no DAPI involvement.
Expand Down Expand Up @@ -219,6 +245,10 @@ impl TransactionBroadcaster for SpvBroadcaster {
Err(other) => Err(other),
}
}

async fn wait_until_ready(&self, timeout: Option<Duration>) -> bool {
self.spv.wait_until_ready(timeout).await
Comment thread
shumkov marked this conversation as resolved.
Outdated
}
}

#[cfg(test)]
Expand Down Expand Up @@ -256,6 +286,10 @@ mod tests {
.take()
.expect("one acceptance check")
}

async fn wait_until_ready(&self, _timeout: Option<Duration>) -> bool {
true
}
}

fn transaction() -> Transaction {
Expand Down
44 changes: 44 additions & 0 deletions packages/rs-platform-wallet/src/spv/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,10 @@ const SPV_CLIENT_STOP_BUDGET: Duration = Duration::from_secs(15);
/// graceful timeout above was meant to escape.
const SPV_ABORT_GRACE: Duration = Duration::from_secs(2);

/// How often [`SpvRuntime::wait_until_ready`] re-checks for a started client
/// with connected peers.
const SPV_READINESS_POLL_INTERVAL: Duration = Duration::from_millis(250);

/// Join a stopped SPV runner, escalating to cancellation after `timeout`.
///
/// Returns `None` once Tokio has confirmed the task terminated. Returns
Expand Down Expand Up @@ -196,6 +200,46 @@ impl SpvRuntime {
self.client.try_read().map(|c| c.is_some()).unwrap_or(false)
}

/// Whether a broadcast issued right now could reach the network: the
/// client is started *and* at least one peer is connected.
///
/// Both halves are required because both are pre-send rejections in
/// [`broadcast_transaction_and_wait`](Self::broadcast_transaction_and_wait)
/// — an unstarted client and dash-spv's zero-connected-peers check.
async fn is_broadcast_ready(&self) -> bool {
self.client.read().await.is_some() && !self.peer_tracker.snapshot().is_empty()
}

/// Resolve once a broadcast could actually reach the network, or when
/// `timeout` elapses. `None` waits indefinitely.
///
/// Returns whether readiness was reached. This closes the launch race
/// where work resumed at app start (asset-lock catch-up in particular)
/// broadcasts into a client that has not finished starting, takes the
/// definitive `Rejected { "client not started" }` verdict, and — having
/// no retry — strands the transaction for the whole session.
///
/// Readiness is polled rather than pushed: "started" is a `client`
/// transition and "has peers" arrives as a dash-spv `PeersUpdated`
/// event, with no combined signal to subscribe to. The poll interval is
/// irrelevant next to the network latency being waited on.
pub async fn wait_until_ready(&self, timeout: Option<Duration>) -> bool {
let deadline = timeout.map(|t| tokio::time::Instant::now() + t);
Comment thread
shumkov marked this conversation as resolved.
Outdated
loop {
if self.is_broadcast_ready().await {
return true;
}
let now = tokio::time::Instant::now();
match deadline {
None => tokio::time::sleep(SPV_READINESS_POLL_INTERVAL).await,
Some(deadline) if now < deadline => {
tokio::time::sleep_until(deadline.min(now + SPV_READINESS_POLL_INTERVAL)).await
}
Some(_) => return false,
}
}
}

/// Broadcast a transaction through SPV peers and wait for dash-spv's
/// network-acceptance verdict.
///
Expand Down
Loading
Loading