dashpay · thepastaclaw · Jul 6, 2026 · Jul 25, 2026 · Jul 25, 2026 · Jul 25, 2026
@@ -19,6 +19,33 @@ use key_wallet_manager::WalletManager;
 /// Default fee rate in duffs per kilobyte for asset lock transactions.
 pub(super) const DEFAULT_FEE_PER_KB: u64 = 1000;
 
+/// Test-only rendezvous for
+/// [`AssetLockManager::resume_pre_promote_gate`]. `arrived` fires once a
+/// resume has taken its read-locked status snapshot; the resume then
+/// blocks on `release`.
+#[cfg(test)]
+#[derive(Clone)]
+pub(super) struct ResumePrePromoteGate {
+    pub(super) arrived: Arc<Notify>,
+    pub(super) release: Arc<Notify>,
+}
+
+/// Test-only rendezvous for
+/// [`AssetLockManager::promote_post_cas_gate`]. `arrived` fires once a
+/// `Built` → `Broadcast` promotion has mutated the in-memory row but has
+/// NOT yet enqueued its changeset; the promoter then blocks on `release`.
+///
+/// This is precisely the window in which a stale promoter could enqueue
+/// its older `Broadcast + None` snapshot AFTER a concurrent flow already
+/// enqueued a newer proof-bearing one — the ordering hazard
+/// [`AssetLockManager::status_persist_serial`] exists to close.
+#[cfg(test)]
+#[derive(Clone)]
+pub(super) struct PromotePostCasGate {
+    pub(super) arrived: Arc<Notify>,
+    pub(super) release: Arc<Notify>,
+}
+
 /// Manages the full asset lock lifecycle: build, broadcast, proof, and tracking.
 ///
 /// Shared across sub-wallets via `Arc<AssetLockManager>` so that any sub-wallet
@@ -83,6 +110,67 @@ pub struct AssetLockManager<B: TransactionBroadcaster + ?Sized> {
     /// yet have collected its pool snapshot.
     #[cfg(test)]
     pub(super) build_serial_gate: std::sync::atomic::AtomicUsize,
+    /// Serializes every asset-lock **status mutation + persistence
+    /// enqueue** pair, so the order in which rows are mutated in memory
+    /// is the order in which their snapshots reach the persister.
+    ///
+    /// `wallet_manager`'s write lock alone is not enough. Each mutator
+    /// (`promote_built_to_broadcast`, `advance_asset_lock_status`, …)
+    /// takes it, mutates, and RELEASES it before returning the
+    /// changeset the caller then hands to `queue_asset_lock_changeset`.
+    /// In that post-mutation/pre-enqueue window another flow can acquire
+    /// the wallet lock, finalize the same row to `InstantSendLocked` /
+    /// `ChainLocked` with a proof, and enqueue that newer snapshot
+    /// first — after which the delayed writer enqueues its older
+    /// `Broadcast + None` snapshot LAST. Nothing downstream repairs the
+    /// inversion: `FFIPersister::store_round` only serializes rounds by
+    /// acquisition order (so it faithfully preserves the reversed
+    /// order), `AssetLockChangeSet::merge` is last-write-wins, and
+    /// Swift's `persistAssetLocks` upsert overwrites `statusRaw` /
+    /// `proofBytes` unconditionally. Memory stays finalized while the
+    /// durable row regresses to `Broadcast`, and because the load path
+    /// treats `statusRaw < 2` as still-pending, a restart resumes a lock
+    /// whose proof it already had.
+    ///
+    /// Held across mutation → enqueue, and deliberately NOT across the
+    /// unbounded awaits (`broadcast`, `wait_for_proof`) that surround
+    /// them: only the relative order of the two adjacent steps needs
+    /// serializing. A dedicated mutex rather than extending the
+    /// `wallet_manager` write guard because `queue_asset_lock_changeset`
+    /// calls the persister synchronously — on iOS that reenters Swift
+    /// via FFI — and holding the wallet lock across that would serialize
+    /// every unrelated wallet reader behind host I/O.
+    ///
+    /// Lock ordering: acquire this BEFORE `wallet_manager`, never the
+    /// reverse. Every holder follows the same
+    /// `status_persist_serial → wallet_manager.write() → drop(wallet) →
+    /// enqueue → drop(serial)` shape, so no cycle exists.
+    pub(super) status_persist_serial: tokio::sync::Mutex<()>,
+    /// Test-only pause point inside
+    /// [`resume_asset_lock`](Self::resume_asset_lock), between the
+    /// read-locked status snapshot and the write-locked `Built` →
+    /// `Broadcast` compare-and-set. When set, a resume signals `arrived`
+    /// and then awaits `release` at that point, which lets a test hold a
+    /// resume on a stale `Built` snapshot while another flow finalizes
+    /// the same row to `InstantSendLocked` / `ChainLocked` — the exact
+    /// interleave the compare-and-set exists to survive.
+    ///
+    /// Both halves matter for determinism: `arrived` proves the resume
+    /// really did snapshot `Built` BEFORE the test finalized the row (so
+    /// the snapshot under test is genuinely stale), and `release` holds
+    /// it there until the finalize has landed. `None` (the default)
+    /// makes the hook a no-op.
+    #[cfg(test)]
+    pub(super) resume_pre_promote_gate: std::sync::Mutex<Option<ResumePrePromoteGate>>,
+    /// Test-only pause point inside
+    /// [`promote_and_queue_built_to_broadcast`](Self::promote_and_queue_built_to_broadcast),
+    /// between the compare-and-set mutation and the persistence enqueue.
+    /// Lets a test hold a promoter in that window while another flow
+    /// finalizes and enqueues the same row, which is the interleave
+    /// [`Self::status_persist_serial`] exists to prevent. `None` (the
+    /// default) makes the hook a no-op.
+    #[cfg(test)]
+    pub(super) promote_post_cas_gate: std::sync::Mutex<Option<PromotePostCasGate>>,
 }
 
 impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
@@ -103,8 +191,13 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
             broadcaster,
             persister,
             build_persist_serial: tokio::sync::Mutex::new(()),
+            status_persist_serial: tokio::sync::Mutex::new(()),
             #[cfg(test)]
             build_serial_gate: std::sync::atomic::AtomicUsize::new(0),
+            #[cfg(test)]
+            resume_pre_promote_gate: std::sync::Mutex::new(None),
+            #[cfg(test)]
+            promote_post_cas_gate: std::sync::Mutex::new(None),
         }
     }
 

@@ -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;
@@ -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;
@@ -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)
@@ -85,7 +86,16 @@ 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();
+
         // We re-check `tracked_asset_locks.contains_key` because
         // another caller could have raced in during phase 2 — first
         // writer wins.
@@ -219,7 +229,7 @@ impl<B: TransactionBroadcaster + ?Sized> AssetLockManager<B> {
         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) = {
+        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)
@@ -251,15 +261,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());
+                    }
+                }
                 let proof = self.wait_for_proof(out_point, timeout).await?;
                 self.validate_or_upgrade_proof(proof, account_index, out_point)
                     .await?
@@ -330,10 +414,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 = {