Skip to content
Draft
Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
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
23 changes: 23 additions & 0 deletions packages/rs-platform-wallet/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,29 @@ pub enum PlatformWalletError {
#[error("Asset lock proof waiting failed: {0}")]
AssetLockProofWait(String),

/// The operation was issued through an `AssetLockManager` whose wallet
/// has since been removed from the `PlatformWalletManager`.
///
/// Wallet ids are deterministic in (seed, network), so re-importing the
/// same mnemonic re-creates the very same id against a *fresh*
/// `PlatformWalletInfo` and a *fresh* `AssetLockManager`. A handle
/// retained across the removal (an FFI `asset_lock_manager` handle the
/// host never destroyed, or an in-flight resume task) resolves through
/// the shared `WalletManager` by id alone, so without this guard it
/// would silently start mutating and persisting the replacement
/// wallet's rows under a different `status_persist_serial` than the
/// live manager — reintroducing the very stale-snapshot reversal the
/// ordering mutex closes within one instance.
///
/// Always a stale-handle bug on the caller's side; the fix is to
/// re-acquire the manager from the current `PlatformWallet`.
#[error(
"Asset lock manager for wallet {0} is no longer active — its wallet was \
removed from the manager; re-acquire the asset lock manager from the \
current wallet handle"
)]
AssetLockManagerInactive(String),
Comment thread
thepastaclaw marked this conversation as resolved.

#[error("SDK error: {0}")]
Sdk(#[from] dash_sdk::Error),

Expand Down
388 changes: 388 additions & 0 deletions packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs

Large diffs are not rendered by default.

1,728 changes: 1,713 additions & 15 deletions packages/rs-platform-wallet/src/wallet/asset_lock/build.rs

Large diffs are not rendered by default.

363 changes: 363 additions & 0 deletions packages/rs-platform-wallet/src/wallet/asset_lock/manager.rs

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
Expand Up @@ -6,4 +6,8 @@

mod proof;
mod recovery;
mod tracking;
/// `pub(super)` so the create path in `build.rs` — the other
/// `Built` → `Broadcast` writer — can name
/// [`BuiltPromotion`](tracking::BuiltPromotion) and share the same
/// compare-and-set instead of writing the status unconditionally.
pub(super) mod tracking;
156 changes: 145 additions & 11 deletions packages/rs-platform-wallet/src/wallet/asset_lock/sync/recovery.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
//! resolving status from wallet info, resuming interrupted locks,
//! and re-deriving private keys.

use crate::broadcaster::TransactionBroadcaster;
use crate::broadcaster::{BroadcastError, TransactionBroadcaster};
use std::time::Duration;

use dashcore::Address as DashAddress;
Expand All @@ -18,6 +18,7 @@ use crate::error::PlatformWalletError;

use super::super::manager::AssetLockManager;
use super::super::tracked::{AssetLockStatus, TrackedAssetLock};
use super::tracking::BuiltPromotion;

// ---------------------------------------------------------------------------
// Blocking accessor (for synchronous / evo-tool contexts)
Expand All @@ -33,6 +34,22 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
/// on-chain context from `ManagedWalletInfo` to determine the correct
/// status (and constructs a `ChainAssetLockProof` if the TX is in a
/// chain-locked block).
///
/// # Lifecycle
///
/// Silently no-ops (with a warning) when the owning wallet has been
/// removed from the `PlatformWalletManager`. This path returns `()`
/// — it is a best-effort catch-up whose every other failure mode is
/// already logged-and-dropped — so a stale handle is refused the
/// same way rather than by a signature change.
///
/// Like the async mutators, the authoritative check happens AFTER
/// `status_persist_serial` is taken and before the commit-phase
/// wallet lookup: phase 2 resolves status without any lock held and
/// may call into host persistence, which is more than enough time
/// for a removal (and a re-import re-creating the same
/// deterministic id) to land. The advisory pre-check up front only
/// saves that work.
#[allow(clippy::too_many_arguments)]
pub fn recover_asset_lock_blocking(
&self,
Expand All @@ -44,6 +61,16 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
out_point: OutPoint,
proof: Option<dpp::prelude::AssetLockProof>,
) {
if let Err(e) = self.ensure_active() {
tracing::warn!(
outpoint = %out_point,
error = %e,
"recover_asset_lock_blocking: refusing to recover through a \
retired asset-lock manager"
);
return;
}

// Phase 1 (lock held): claim the tracked-asset-lock slot and
// pull the in-memory record out so the lookup work is
// bounded to a single hashmap fetch.
Expand Down Expand Up @@ -85,7 +112,31 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
None => self.resolve_status_with_in_memory(in_memory_record, account_index, &out_point),
};

// Phase 3 (lock held): commit the tracked-asset-lock entry.
// Phase 3 (locks held): commit the tracked-asset-lock entry and
// enqueue it as one serialized unit. Without the ordering mutex
// a concurrent flow could finalize this row and enqueue its
// proof-bearing snapshot in the window between the insert below
// and the enqueue, leaving the older recovered snapshot durable
// (see `status_persist_serial`). `blocking_lock` matches the
// `blocking_write` already used here — this method is documented
// as callable only OUTSIDE a tokio async context.
let _serial = self.status_persist_serial.blocking_lock();

// Authoritative stale-handle check, under the same mutex
// `deactivate` must hold to retire this manager — so a removal
// racing phase 2 either finished (and this insert is refused)
// or is still waiting for this critical section to end.
if let Err(e) = self.ensure_active_under_serial(&_serial) {
tracing::warn!(
outpoint = %out_point,
error = %e,
"recover_asset_lock_blocking: wallet was removed while resolving \
the lock's status — dropping the recovery instead of writing to \
replacement wallet state"
);
return;
}

// We re-check `tracked_asset_locks.contains_key` because
// another caller could have raced in during phase 2 — first
// writer wins.
Expand Down Expand Up @@ -218,8 +269,16 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
) -> Result<(dpp::prelude::AssetLockProof, DerivationPath), PlatformWalletError> {
tracing::info!(outpoint = %out_point, ?timeout, "resume_asset_lock: entered");

// Fail a stale handle before doing any work. Advisory only —
// this call goes on to await a broadcast and a proof, so the
// removal it is meant to catch can equally land afterwards. The
// guarantee comes from the same check inside
// `promote_built_to_broadcast` / `advance_asset_lock_status`,
// taken under `status_persist_serial`.
self.ensure_active()?;

// 1. Look up the tracked lock — snapshot the fields we need.
let (tx, status, existing_proof, account_index) = {
let (tx, mut status, mut existing_proof, account_index) = {
let wm = self.wallet_manager.read().await;
let info = wm
.get_wallet_info(&self.wallet_id)
Expand Down Expand Up @@ -251,15 +310,89 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
)
};

// Test-only pause between the read-locked snapshot above and the
// write-locked compare-and-set below, so a test can deterministically
// hold this resume on a stale `Built` snapshot while another flow
// finalizes the same row. No-op unless a test installed the gate.
#[cfg(test)]
{
let gate = self
.resume_pre_promote_gate
.lock()
.expect("resume pre-promote gate mutex")
.clone();
if let Some(gate) = gate {
// Signal first: the snapshot above is taken, so whatever the
// test does next is guaranteed to race a stale `Built`.
gate.arrived.notify_one();
gate.release.notified().await;
}
}

// 1b. Promote `Built` → `Broadcast` BEFORE calling `broadcast(&tx)`.
// The snapshot above dropped the read lock, so a concurrent
// create-path Rejected cleanup can race the re-broadcast: if the row
// is still `Built` when `untrack_asset_lock` runs, the guard doesn't
// fire, the row is deleted, and the funding reservation is released
// while this call is still handing the same transaction to the
// network. Advancing first pushes the status past `Built` under the
// write lock, so either (a) we win and the untrack guard preserves
// the row + reservation, or (b) untrack ran first, the row is
// already gone, and this promotion fails before we ever call
// `broadcast(&tx)`.
//
// The promotion is a compare-and-set rather than an unconditional
// write because that same dropped read lock lets TWO resumes both
// snapshot `Built`. If the first one broadcasts, obtains a proof and
// finalizes the row to `InstantSendLocked` / `ChainLocked` (step 3),
// an unconditional write from this delayed second caller would
// downgrade the finalized row to `Broadcast` while leaving the proof
// attached, and persist that inconsistent state. Instead we re-read
// the row's current status and proof under the write lock and
// re-dispatch from there — the arms below then take the already-have-
// a-proof path instead of waiting again for a proof we already hold.
if status == AssetLockStatus::Built {
match self.promote_built_to_broadcast(out_point).await? {
// The promotion queued its own changeset, atomically with
// the compare-and-set — see `status_persist_serial`.
BuiltPromotion::Promoted(_cs) => {}
BuiltPromotion::AlreadyAdvanced {
current_status,
current_proof,
} => {
tracing::info!(
outpoint = %out_point,
status = ?current_status,
has_proof = current_proof.is_some(),
"resume_asset_lock: row advanced past Built concurrently — \
re-dispatching from its current state"
);
status = current_status;
existing_proof = current_proof;
}
}
}

// 2. Resume from the current status.
let proof = match status {
AssetLockStatus::Built => {
// Re-broadcast and wait for proof.
self.broadcaster.broadcast(&tx).await?;
let cs = self
.advance_asset_lock_status(out_point, AssetLockStatus::Broadcast, None)
.await?;
self.queue_asset_lock_changeset(cs);
// Promoted to `Broadcast` in step 1b — this arm owns that
// promotion, so it is the one that re-broadcasts.
match self.broadcaster.broadcast(&tx).await {
Ok(_) => {}
Err(e @ BroadcastError::Rejected { .. }) => {
// Keep `Broadcast`: a concurrent successful resume
// may own that status, and the `Broadcast` arm
// defensively re-broadcasts on later resumes.
return Err(e.into());
}
Err(e @ BroadcastError::MaybeSent { .. }) => {
// Outcome unknown — the tx may already be
// propagating. Keep `Broadcast` so a later resume
// can defensively re-broadcast and wait for proof.
return Err(e.into());
Comment thread
thepastaclaw marked this conversation as resolved.
}
}
let proof = self.wait_for_proof(out_point, timeout).await?;
self.validate_or_upgrade_proof(proof, account_index, out_point)
.await?
Expand Down Expand Up @@ -330,10 +463,11 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
dpp::prelude::AssetLockProof::Instant(_) => AssetLockStatus::InstantSendLocked,
dpp::prelude::AssetLockProof::Chain(_) => AssetLockStatus::ChainLocked,
};
let cs = self
// Queued by `advance_asset_lock_status` itself, atomically with
// the in-memory write.
let _cs = self
.advance_asset_lock_status(out_point, new_status, Some(proof.clone()))
.await?;
self.queue_asset_lock_changeset(cs);

// 4. Re-derive the one-time credit-output derivation path.
let path = {
Expand Down
Loading
Loading