diff --git a/packages/rs-platform-wallet-ffi/src/persistence.rs b/packages/rs-platform-wallet-ffi/src/persistence.rs index efb7e3d20d..a43db2c188 100644 --- a/packages/rs-platform-wallet-ffi/src/persistence.rs +++ b/packages/rs-platform-wallet-ffi/src/persistence.rs @@ -1131,6 +1131,13 @@ impl FFIPersister { // wallet-event adapter would couple the flip to a round that // silently drops it: the accepted-and-ignored shape the sweep // bit's own gating exists to prevent, reproduced one channel over. + // Unlike `CORE_SWEEP_REMOVAL` below, this bit does NOT fold in the + // begin/end pair: its contract is per-callback durability of the + // overlay rows, which holds on a non-atomic host too. The + // round-coupling the adapter's staging needs is expressed as the + // `ROUND_COUPLED_PAYMENT_FLIPS` composite (this bit plus + // `ATOMIC_CHANGESETS`), so atomicity stays attested once, by the + // bit that owns it. if self.callbacks.on_persist_dashpay_payments_fn.is_some() { capabilities = capabilities.union(PersistenceCapabilities::DASHPAY_PAYMENTS); } @@ -6352,8 +6359,13 @@ mod tests { /// vtable leaves `on_persist_dashpay_payments_fn` unset, so even a /// host blindly OR-ing the bit must read as payments-blind: the /// wallet-event adapter keys the sweep's Failed-flip staging on this - /// bit, and an accepted-and-dropped overlay is exactly the shape the - /// gating exists to prevent. + /// bit (composed with `ATOMIC_CHANGESETS` — the + /// `ROUND_COUPLED_PAYMENT_FLIPS` composite — since the staging also + /// needs the round to commit as one unit), and an accepted-and-dropped + /// overlay is exactly the shape the gating exists to prevent. The bit + /// itself deliberately stays atomicity-free: it attests per-callback + /// durability, and the positive case below is such a host — one the + /// adapter now refuses to stage round-coupled overlays for. #[test] fn dashpay_payments_requires_the_slot_and_the_declaration() { fn persister_with( diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 72e7b62299..e306449cab 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -158,6 +158,11 @@ struct BatchDiagnostics { /// Wallets in this drain that are faulted — whether they entered faulted /// or were faulted by it. Each wallet counts at most once per drain. faulted: usize, + /// Wallets whose `store()` this drain REJECTED outright (the `Err` + /// arm) — not the nominal-success sweep-capability freeze, whose round + /// did store. The caller uses this to roll back the in-memory payment + /// flips whose durable half the rejection discarded. + rejected_wallets: std::collections::BTreeSet, } impl BatchDiagnostics { @@ -300,6 +305,33 @@ async fn run_wallet_event_adapter

( // once per session rather than once per faulted batch. let mut freeze_logged = false; + // Whether the backend can give a payment flip the round-coupled + // durability this staging exists to provide — which takes BOTH bits + // of `ROUND_COUPLED_PAYMENT_FLIPS`. `DASHPAY_PAYMENTS` proves the + // overlay rows are durably applied: a sweep never re-emits once its + // round is durable, so handing the overlay to a host that silently + // drops it (Android deliberately keeps payment recording + // in-memory-only, its payments slot unwired) would leave this + // adapter believing a flip persisted — the accepted-and-ignored + // shape the sweep capability's own gating exists to prevent, one + // channel over. `ATOMIC_CHANGESETS` proves the round the flip rides + // commits or rolls back as one unit: on a host whose callbacks + // commit independently, the Core record and watermark can land + // durably and the process stop before the payments write — and for + // a one-shot chainlocked reinstatement nothing ever re-emits, so + // the reinstatement would stay durably recorded beside a payment + // durably `Failed`. Payments durability without the atomic round + // therefore gives neither the coupling nor the fail-closed + // watermark backstop, and such a host is treated exactly like a + // payments-blind one here: it still gets the in-memory flip (the + // truthful session state; the transaction IS dead) with nothing + // round-coupled — funds-safe, since payment entries are display + // metadata, and consistent with every other payment write on such + // hosts. + let payments_attested = persister + .persistence_capabilities() + .contains(PersistenceCapabilities::ROUND_COUPLED_PAYMENT_FLIPS); + loop { // Block for the first event of a batch. Everything already sitting in // the channel behind it is folded in below without another await, so a @@ -320,6 +352,17 @@ async fn run_wallet_event_adapter

( }; let mut batch: BTreeMap = BTreeMap::new(); + // This drain's sent-payment round journal: per wallet, per + // `(owner, txid)`, the durable pre-round entry and the entry the + // round currently intends to leave staged. The store overlay + // (materialized just before commit), the rejected-round rollback, + // and the same-fold retraction are all projections of this ONE + // structure — see [`fold_payment_flips`] — so a staged row and + // its undo cannot describe different histories. Kept OUTSIDE the + // batch because `commit_batch` consumes the batch, and the + // rollback only runs for wallets whose store was rejected (see + // below). + let mut payment_journal: BTreeMap = BTreeMap::new(); let mut closed = false; { let wallet_id = event.wallet_id(); @@ -329,9 +372,32 @@ async fn run_wallet_event_adapter

( // read lock on the manager. let core = build_core_changeset(&wallet_manager, &event).await; let asset_locks = reconstruct_asset_locks_for_event(&wallet_manager, &event).await; + let flips = swept_payment_flips_for_event(&wallet_manager, &event).await; let entry = batch.entry(wallet_id).or_default(); + retract_reinstated_payment_flips( + &wallet_manager, + &mut payment_journal, + wallet_id, + &core.records, + ) + .await; + // After the same-fold retraction, so a sweep undone by this + // very event reads the durable pre-round state and the + // ordered confirm below takes it straight to `Confirmed` on + // this event's round. + let confirm_flips = + crate::wallet::identity::network::confirm_final_sent_payments_for_store( + &wallet_manager, + &wallet_id, + &event, + ) + .await; entry.core.merge(core); entry.asset_locks.merge(asset_locks); + if payments_attested { + fold_payment_flips(&mut payment_journal, wallet_id, flips); + fold_payment_flips(&mut payment_journal, wallet_id, confirm_flips); + } } // Fold in whatever else is already buffered. `try_recv` never waits, @@ -345,9 +411,31 @@ async fn run_wallet_event_adapter

( let core = build_core_changeset(&wallet_manager, &event).await; let asset_locks = reconstruct_asset_locks_for_event(&wallet_manager, &event).await; + let flips = swept_payment_flips_for_event(&wallet_manager, &event).await; let entry = batch.entry(wallet_id).or_default(); + retract_reinstated_payment_flips( + &wallet_manager, + &mut payment_journal, + wallet_id, + &core.records, + ) + .await; + // See the first-fold site: after the retraction, so a + // same-fold-swept entry reads the durable pre-round + // state for the ordered confirm. + let confirm_flips = + crate::wallet::identity::network::confirm_final_sent_payments_for_store( + &wallet_manager, + &wallet_id, + &event, + ) + .await; entry.core.merge(core); entry.asset_locks.merge(asset_locks); + if payments_attested { + fold_payment_flips(&mut payment_journal, wallet_id, flips); + fold_payment_flips(&mut payment_journal, wallet_id, confirm_flips); + } folded += 1; } Err(TryRecvError::Empty) => break, @@ -358,9 +446,49 @@ async fn run_wallet_event_adapter

( } } + // Materialize each wallet's payment overlay FROM the journal — + // the overlay's only writer. One row per `(owner, txid)`: the + // round's final staged entry. Because the store payload and the + // rollback below are projections of the same journal entry, the + // undo always matches what this round actually offers the store. + for (wallet_id, ledger) in &payment_journal { + if ledger.is_empty() { + continue; + } + let entry = batch.entry(*wallet_id).or_default(); + for ((owner, txid), round) in ledger { + entry + .payments_overlay + .entry(*owner) + .or_default() + .insert(txid.clone(), round.staged.clone()); + } + } + // Commit the folded batch. The channel is lossless, so the only way a // watermark is held back is a rejected `store()` (the fail-closed // backstop inside `commit_batch`). + // + // No commit-time re-validation of the staged payment rows is + // needed, and none is done: every sent-payment verdict writer + // either IS this task (the sweep flip and the ordered confirm + // above, applied in emission order and coalesced into this + // drain's round journal, where a later flip updates the staged + // entry per `(owner, txid)`), or persists memory-and-store + // atomically under + // a continuous hold of the manager WRITE lock with `Pending`-only + // evidence (the reconcile pass via `resolve_sent_payment_by_txid` + // → `record_dashpay_payment`). A staged row always asserts + // `Failed` or `Confirmed`, so between this drain's fold and its + // store no other writer can move the entry the row describes — + // the reconcile pass's evidence excludes both states — and its + // write-lock hold across its own store means the store order of + // the two writers matches their memory order. When verdicts also + // ran on the EventHandler broadcast (unordered spawned tasks + // persisting on their own rounds), a commit-time re-validation + // under a held read lock was required here to drop staged rows a + // hook had outrun; routing every verdict through this drain is + // what retired it. let diag = commit_batch( &*persister, batch, @@ -370,6 +498,40 @@ async fn run_wallet_event_adapter

( &mut freeze_logged, ); + // A rejected round leaves NOTHING durable — the loser's record and + // the payment flip alike — so memory must return to the durable + // state or the replayed sweep (re-emitted by the re-scan, since the + // rejection kept the loser's record) would find the entries already + // `Failed` in memory, skip them as ineligible, and the store would + // never learn. Only rejected wallets roll back: a stored round — + // including one that stored but froze the watermark for a + // non-attesting sweep backend — has the flip durably applied. + for wallet_id in &diag.rejected_wallets { + if let Some(ledger) = payment_journal.remove(wallet_id) { + // One guarded undo per `(owner, txid)`: restore the + // durable pre-round entry while the round's final staged + // status still stands in memory — never an intermediate + // flip's snapshot. + let rollback: Vec = ledger + .into_iter() + .map(|((owner, txid), round)| { + crate::wallet::identity::network::PaymentFlipUndo { + owner, + txid, + wrote: round.staged.status, + previous: round.previous, + } + }) + .collect(); + crate::wallet::identity::network::rollback_payment_flips( + &wallet_manager, + wallet_id, + rollback, + ) + .await; + } + } + // One structured line per drain via the `log` facade so a tester // logcat is unambiguous about whether the watermark is advancing. // Every field reports an observed outcome — see [`BatchDiagnostics`]. @@ -447,6 +609,7 @@ fn commit_wallet

( let WalletBatch { mut core, asset_locks, + payments_overlay, } = wallet_batch; { // Hold this wallet's durable watermark at the last fully persisted @@ -466,7 +629,10 @@ fn commit_wallet

( diag.record_frozen(h); } } - if core.is_empty_no_records() && Merge::is_empty(&asset_locks) { + if core.is_empty_no_records() + && Merge::is_empty(&asset_locks) + && payments_overlay.is_empty() + { // SyncHeightAdvanced for an unknown wallet, empty BlockProcessed, a // watermark-only batch stripped by the fault guard above, etc. — // nothing to persist. Skip the round-trip. @@ -511,6 +677,9 @@ fn commit_wallet

( // same store round-trip so the row and the record that // implies it land atomically. asset_locks: (!Merge::is_empty(&asset_locks)).then_some(asset_locks), + // The sweep-failed payments ride the same atomic round as the + // sweep that proved them dead — see `WalletBatch::payments_overlay`. + dashpay_payments_overlay: (!payments_overlay.is_empty()).then_some(payments_overlay), ..PlatformWalletChangeSet::default() }; match persister.store(wallet_id, cs) { @@ -555,6 +724,9 @@ fn commit_wallet

( // A rejected changeset means these rows are not on disk. Fault // THIS wallet's watermark so it can't outrun them; the next // scan re-emits and the idempotent upserts recover the state. + // Reported to the caller so the in-memory payment flips whose + // durable half this rejection discarded are rolled back. + diag.rejected_wallets.insert(wallet_id); if fault_and_freeze( diag, offered_height, @@ -635,6 +807,56 @@ fn freeze_synced_height_if_faulted(core: &mut CoreChangeSet, persistence_faulted struct WalletBatch { core: CoreChangeSet, asset_locks: AssetLockChangeSet, + /// Sent DashPay payment verdicts this fold's events produced — a + /// sweep's `Failed` rows and the ordered confirm's `Confirmed` rows — + /// riding the SAME `store()` as the events that proved them. This is + /// a flip's only durability: a sweep never re-emits once its round is + /// durable, and a chainlocked reinstatement can be a one-shot, so a + /// separately persisted flip whose store failed was lost for good — + /// while here a rejection keeps the proving event's rows with it, the + /// wallet faults, the re-scan re-emits the event, and the replay + /// recomputes the flip (after [`run_wallet_event_adapter`] rolls the + /// in-memory half back). NOT written during the fold: it is + /// materialized from the drain's round journal (see + /// [`fold_payment_flips`]) just before commit — one row per + /// `(owner, txid)`, the round's final staged verdict, matching + /// `PlatformWalletChangeSet::merge`'s last-write-wins overlay rule — + /// which is also what makes a fold containing both a confirmation + /// and a later eviction commit only the newer verdict. + payments_overlay: std::collections::BTreeMap< + dpp::prelude::Identifier, + std::collections::BTreeMap, + >, +} + +/// One wallet's coalesced sent-payment journal for a single drain round, +/// keyed — like the store overlay — per `(owner, txid)`. See +/// [`fold_payment_flips`]. +type PaymentRoundLedger = BTreeMap<(dpp::prelude::Identifier, String), PaymentRoundEntry>; + +/// What one drain round did to one sent payment — exactly what any +/// consumer of the round needs, and the ONLY record of it: +/// +/// - the store overlay row is `staged` (materialized before commit); +/// - the rejected-round undo is "restore `previous` while `staged.status` +/// still stands"; +/// - the same-fold retraction drops the whole entry and applies that +/// same undo. +/// +/// `previous` is pinned by the round's FIRST flip and never overwritten, +/// so it is always the durable pre-round entry (at fold start memory +/// equals the durable state: the adapter is the only staging writer, a +/// stored round made its flips durable, a rejected one was rolled back, +/// and the reconcile pass persists memory-and-store atomically). +/// `staged` follows the round's LAST flip. Intermediate flips leave no +/// trace — there is no history to unwind, which is what makes rollback +/// and retraction structurally unable to disagree with the staged row. +#[derive(Debug, Clone)] +struct PaymentRoundEntry { + /// The entry as durably stored before this round's first flip. + previous: crate::wallet::identity::PaymentEntry, + /// The entry this round currently intends to leave staged. + staged: crate::wallet::identity::PaymentEntry, } /// Rebuild missing tracked asset locks from the records an event @@ -725,6 +947,165 @@ async fn reconstruct_asset_locks_for_event( reconstruction::reconstruct_tracked_asset_locks(wallet_manager, &wallet_id, &candidates).await } +/// The payment half of a sweep: flip the losers' sent DashPay payments +/// (`Pending` — or `Confirmed`, when this sweep postdates the entry's +/// confirmation) to `Failed` in memory and hand back the flips the drain +/// loop journals into the sweep's own store round. +/// Every other event is a no-op. See [`WalletBatch::payments_overlay`] +/// for why this rides the round rather than persisting on its own. +async fn swept_payment_flips_for_event( + wallet_manager: &Arc>>, + event: &WalletEvent, +) -> crate::wallet::identity::network::SweptPaymentFlips { + match event { + WalletEvent::TransactionsSwept { + wallet_id, txids, .. + } => { + crate::wallet::identity::network::flip_swept_sent_payments_for_store( + wallet_manager, + wallet_id, + txids, + ) + .await + } + _ => crate::wallet::identity::network::SweptPaymentFlips::default(), + } +} + +/// The batch-level half of the reinstatement invariant: **a merged +/// changeset must never carry a sweep-derived assertion about a txid the +/// same fold reinstates.** Each sweep-derived channel enforces it where +/// that channel folds: +/// +/// - `core.sweeps.txids` — `CoreChangeSet::merge` retracts reinstated +/// txids from folded batches; +/// - `core.sweeps.released_outpoints` — deliberately NOT retracted; every +/// backend withholds an outpoint a surviving record claims, so the +/// reinstated transaction's own entries are inert (documented at the +/// merge); +/// - `asset_locks.removed` — `AssetLockChangeSet::merge` cancels a folded +/// sweep tombstone when the reinstating reconstruction upsert lands; +/// - the payment round journal (from which `payments_overlay` is +/// materialized at commit) — lives at DRAIN level, not inside any +/// sub-changeset's `Merge`, so its retraction lives here. Any future +/// sweep-derived channel carried on [`WalletBatch`] must get its +/// retraction in this function too. +/// +/// `reinstated` is exactly `core.records` of the event being folded — the +/// same set `CoreChangeSet::merge` keys its own retraction on, taken from +/// the same projection, so the two can never diverge. Without this, a +/// buffered `[TransactionsSwept(X), BlockProcessed(chainlocked X)]` fold +/// would commit X's reinstated record beside a stale `Failed` overlay +/// row. +/// +/// Two moves per reinstated txid whose round-final staged status is +/// `Failed`, both before the journal can be materialized into a store: +/// drop the journal entry whole (its staged row and its undo are the same +/// record — a later rejection of this round must not replay a dead undo), +/// and restore the DURABLE pre-round entry to memory through the guarded +/// [`rollback_payment_flips`], so the ordered confirm running right after +/// this retraction +/// ([`confirm_final_sent_payments_for_store`](crate::wallet::identity::network)) +/// re-derives `Confirmed` from the reinstating record on this same round +/// and journals a fresh `previous → Confirmed` entry. Restoring the +/// pre-round entry — not any intermediate flip's snapshot — is what keeps +/// a `[finality(X), sweep(X), reinstating record(X)]` fold sound: the +/// entry returns to `Pending`, the confirm genuinely transitions, and the +/// reinstated record commits WITH its payment correction. An entry whose +/// staged status is `Confirmed` asserts exactly what the reinstating +/// record says and stays on the round. +/// +/// This function only sees records THIS drain captured, and that is +/// enough: a reinstating record queued after `try_recv` stopped folding +/// is simply the NEXT drain's event — the adapter is the only +/// sent-payment verdict writer that stages rows, so nothing can supersede +/// this batch's staged rows between its fold and its store (see the +/// commit-site comment in [`run_wallet_event_adapter`]), and the later +/// drain's `Failed → Confirmed` is exactly the transition the shared +/// table permits. +async fn retract_reinstated_payment_flips( + wallet_manager: &Arc>>, + payment_journal: &mut BTreeMap, + wallet_id: WalletId, + records: &[TransactionRecord], +) { + if records.is_empty() { + return; + } + let Some(ledger) = payment_journal.get_mut(&wallet_id) else { + return; + }; + if ledger.is_empty() { + return; + } + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + + let reinstated: std::collections::HashSet = records + .iter() + .map(|record| record.txid.to_string()) + .collect(); + + // Only rounds whose FINAL staged verdict is the sweep-derived + // `Failed` are retracted; a staged `Confirmed` stays on the round. + let mut undo = Vec::new(); + ledger.retain(|(owner, txid), round| { + if reinstated.contains(txid) && round.staged.status == PaymentStatus::Failed { + undo.push(crate::wallet::identity::network::PaymentFlipUndo { + owner: *owner, + txid: txid.clone(), + wrote: round.staged.status, + previous: round.previous.clone(), + }); + false + } else { + true + } + }); + if !undo.is_empty() { + crate::wallet::identity::network::rollback_payment_flips(wallet_manager, &wallet_id, undo) + .await; + } +} + +/// Fold one event's sent-payment flips into the drain's round journal — +/// the ONE structure the round's store overlay, rejected-round rollback, +/// and same-fold retraction are all projections of. Per `(owner, txid)`: +/// the first flip of the round pins `previous` (the durable pre-round +/// entry — the fold never overwrites it), and every flip updates `staged` +/// (last write wins, matching `PlatformWalletChangeSet::merge`'s overlay +/// rule). A repeated flip therefore coalesces instead of appending +/// history: whatever a rejected round must restore is always `previous`, +/// guarded by the `staged` status the round actually offers the store — +/// there is no intermediate undo for a rollback to skip or a retraction +/// to half-apply, and a future writer folding more flips through here +/// cannot reintroduce that divergence. The inverse — a later event in +/// the same fold reinstating a flipped txid — is +/// [`retract_reinstated_payment_flips`]' job, which the drain runs for +/// every record-bearing event before merging it. +fn fold_payment_flips( + payment_journal: &mut BTreeMap, + wallet_id: WalletId, + flips: crate::wallet::identity::network::SweptPaymentFlips, +) { + if flips.is_empty() { + return; + } + let ledger = payment_journal.entry(wallet_id).or_default(); + for flip in flips.flips { + match ledger.entry((flip.owner, flip.txid)) { + std::collections::btree_map::Entry::Occupied(mut occupied) => { + occupied.get_mut().staged = flip.updated; + } + std::collections::btree_map::Entry::Vacant(vacant) => { + vacant.insert(PaymentRoundEntry { + previous: flip.previous, + staged: flip.updated, + }); + } + } + } +} + /// Project an upstream [`WalletEvent`] into a [`CoreChangeSet`] suitable /// for atomic persistence. async fn build_core_changeset( @@ -2289,6 +2670,8 @@ mod tests { n_records: usize, n_asset_locks: usize, n_asset_locks_removed: usize, + n_payment_overlay_rows: usize, + n_payment_overlay_confirmed: usize, rejected: bool, } @@ -2354,6 +2737,24 @@ mod tests { .as_ref() .map(|a| a.removed.len()) .unwrap_or(0), + n_payment_overlay_rows: changeset + .dashpay_payments_overlay + .as_ref() + .map(|o| o.values().map(|rows| rows.len()).sum()) + .unwrap_or(0), + n_payment_overlay_confirmed: changeset + .dashpay_payments_overlay + .as_ref() + .map(|o| { + o.values() + .flat_map(|rows| rows.values()) + .filter(|entry| { + entry.status + == crate::wallet::identity::types::dashpay::payment::PaymentStatus::Confirmed + }) + .count() + }) + .unwrap_or(0), rejected, }); if rejected { @@ -3245,6 +3646,1516 @@ mod tests { handle.await.expect("adapter task joins"); } + /// A backend that never attested `DASHPAY_PAYMENTS` — Android, whose + /// payments slot is deliberately unwired — must not be handed the + /// sweep's Failed flip on the round at all: it would accept the round, + /// silently drop the overlay, and leave this adapter believing a flip + /// persisted that no store ever applied — the accepted-and-ignored + /// shape the sweep capability's own gating exists to prevent, one + /// channel over. The withhold keeps the in-memory flip (the truthful + /// session state) with nothing round-coupled. + #[tokio::test] + async fn a_payments_blind_backend_is_not_handed_the_sweeps_flip_on_the_round() { + use dpp::identity::v0::IdentityV0; + use dpp::identity::Identity; + use dpp::prelude::Identifier; + use key_wallet::account::account_type::StandardAccountType; + + use super::spawn_wallet_event_adapter; + use crate::test_support::{funded_wallet_manager, NoopTestPersister}; + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + use crate::wallet::persister::WalletPersister; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xB9; 32]); + let txid_key = txid.to_string(); + + let noop = WalletPersister::new( + wallet_id, + Arc::new(NoopTestPersister) as Arc, + ); + { + let mut wm = wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("wallet info"); + info.identity_manager + .add_identity( + Identity::V0(IdentityV0 { + id: owner, + public_keys: std::collections::BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &noop, + ) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid_key.clone(), + PaymentEntry::new_sent(contact, 50_000, None), + &noop, + ) + .expect("record pending sent"); + } + + let (obs_tx, mut obs_rx) = unbounded_channel(); + // Sweep-capable but payments-blind: the exact Android shape. + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL, + )); + let (event_tx, event_rx) = unbounded_channel(); + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xBA; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + + let observed = obs_rx.recv().await.expect("sweep store"); + assert!(!observed.rejected); + assert_eq!( + observed.n_payment_overlay_rows, 0, + "an overlay a payments-blind backend would silently drop must be withheld \ + from its round" + ); + + { + let wm = wallet_manager.read().await; + let status = wm + .get_wallet_info(&wallet_id) + .expect("info") + .identity_manager + .managed_identity(&owner) + .expect("managed") + .dashpay() + .payments + .get(&txid_key) + .expect("entry") + .status; + assert_eq!( + status, + PaymentStatus::Failed, + "the in-memory flip still happens — the truthful session state" + ); + } + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// A backend that attests `DASHPAY_PAYMENTS` but NOT `ATOMIC_CHANGESETS` + /// must be treated exactly like a payments-blind one: the whole point + /// of staging a flip onto the triggering record's round is that the two + /// land or fail together, and a host whose callbacks commit + /// independently gives neither the coupling nor the fail-closed + /// watermark backstop. It can commit the Core record and watermark and + /// then stop before the payments write — and a one-shot chainlocked + /// reinstatement never re-emits, so its payment would stay durably + /// `Failed` beside a durably recorded reinstatement. Staging requires + /// the full `ROUND_COUPLED_PAYMENT_FLIPS` composite; this host keeps + /// the in-memory flip with nothing round-coupled. + #[tokio::test] + async fn an_atomicity_blind_backend_is_not_handed_payment_flips_on_the_round() { + use dpp::identity::v0::IdentityV0; + use dpp::identity::Identity; + use dpp::prelude::Identifier; + use key_wallet::account::account_type::StandardAccountType; + + use super::spawn_wallet_event_adapter; + use crate::test_support::{funded_wallet_manager, NoopTestPersister}; + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + use crate::wallet::persister::WalletPersister; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xC9; 32]); + let txid_key = txid.to_string(); + + let noop = WalletPersister::new( + wallet_id, + Arc::new(NoopTestPersister) as Arc, + ); + { + let mut wm = wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("wallet info"); + info.identity_manager + .add_identity( + Identity::V0(IdentityV0 { + id: owner, + public_keys: std::collections::BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &noop, + ) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid_key.clone(), + PaymentEntry::new_sent(contact, 50_000, None), + &noop, + ) + .expect("record pending sent"); + } + + let (obs_tx, mut obs_rx) = unbounded_channel(); + // Sweep-capable and payments-durable, but with no atomic round: + // each callback commits on its own, so the flip and the record + // cannot be made to land or fail together. + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS), + )); + let (event_tx, event_rx) = unbounded_channel(); + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xCA; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + + let observed = obs_rx.recv().await.expect("sweep store"); + assert!(!observed.rejected); + assert_eq!( + observed.n_payment_overlay_rows, 0, + "an overlay that cannot ride an atomic round must be withheld from a host \ + whose callbacks commit independently" + ); + + { + let wm = wallet_manager.read().await; + let status = wm + .get_wallet_info(&wallet_id) + .expect("info") + .identity_manager + .managed_identity(&owner) + .expect("managed") + .dashpay() + .payments + .get(&txid_key) + .expect("entry") + .status; + assert_eq!( + status, + PaymentStatus::Failed, + "the in-memory flip still happens — the truthful session state" + ); + } + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// The sweep's payment flip is durable BECAUSE it rides the sweep's own + /// atomic store round: a sweep never re-emits once its round is + /// durable, so a separately persisted flip whose store failed was lost + /// for good. End to end through the real adapter loop: the flip's + /// overlay lands in the SAME `store()` as the sweep; a rejected round + /// rolls the in-memory flip back to the durable state (`Pending`), so + /// the replayed sweep finds the entry eligible and recomputes it; and + /// the replay's round carries the overlay again. + #[tokio::test] + async fn swept_payment_flip_rides_the_sweeps_round_and_rolls_back_on_rejection() { + use dpp::identity::v0::IdentityV0; + use dpp::identity::Identity; + use dpp::prelude::Identifier; + use key_wallet::account::account_type::StandardAccountType; + + use super::spawn_wallet_event_adapter; + use crate::test_support::{funded_wallet_manager, NoopTestPersister}; + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + use crate::wallet::persister::WalletPersister; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + let tx1 = dashcore::Txid::from([0xB1; 32]); + let tx2 = dashcore::Txid::from([0xB2; 32]); + + // Seed the identity and two Pending sent entries through a noop + // persister so the probe's observation stream carries ONLY the + // adapter's own stores. + let noop = WalletPersister::new( + wallet_id, + Arc::new(NoopTestPersister) as Arc, + ); + { + let mut wm = wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("wallet info"); + info.identity_manager + .add_identity( + Identity::V0(IdentityV0 { + id: owner, + public_keys: std::collections::BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &noop, + ) + .expect("add owner"); + let managed = info + .identity_manager + .managed_identity_mut(&owner) + .expect("managed"); + for (txid, amount) in [(tx1, 50_000u64), (tx2, 10_000u64)] { + managed + .record_dashpay_payment( + txid.to_string(), + PaymentEntry::new_sent(contact, amount, None), + &noop, + ) + .expect("record pending sent"); + } + } + + async fn status( + wallet_manager: &Arc>>, + wallet_id: &WalletId, + owner: &dpp::prelude::Identifier, + txid: &str, + ) -> crate::wallet::identity::types::dashpay::payment::PaymentStatus { + let wm = wallet_manager.read().await; + wm.get_wallet_info(wallet_id) + .expect("info") + .identity_manager + .managed_identity(owner) + .expect("managed") + .dashpay() + .payments + .get(txid) + .expect("entry") + .status + } + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + let (event_tx, event_rx) = unbounded_channel(); + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + let swept = |txid: dashcore::Txid| WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xC1; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }; + + // Leg 1: the flip's overlay rides the sweep's own store. + event_tx.send(swept(tx1)).expect("send sweep 1"); + let observed = obs_rx.recv().await.expect("sweep 1 store"); + assert!(!observed.rejected); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the Failed flip must ride the same store() as the sweep that proved it" + ); + assert_eq!( + status(&wallet_manager, &wallet_id, &owner, &tx1.to_string()).await, + PaymentStatus::Failed + ); + + // Leg 2: a rejected round rolls the in-memory flip back to the + // durable state, so the replayed sweep can recompute it. + persister.fail_next(wallet_id); + event_tx.send(swept(tx2)).expect("send sweep 2"); + let observed = obs_rx.recv().await.expect("sweep 2 store attempt"); + assert!(observed.rejected, "the probe rejects this round"); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the attempt carried the flip" + ); + // The rollback runs right after commit in the same drain iteration; + // bounded-poll memory rather than racing it. + let mut rolled_back = false; + for _ in 0..50 { + if status(&wallet_manager, &wallet_id, &owner, &tx2.to_string()).await + == PaymentStatus::Pending + { + rolled_back = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!( + rolled_back, + "a rejected round must roll the in-memory flip back to Pending — the \ + durable state, and the replayed sweep's eligibility" + ); + + // Leg 3: the replayed sweep (the re-scan re-emits it, because the + // rejected round kept the loser's record too) recomputes the flip + // and its round carries the overlay again. + event_tx.send(swept(tx2)).expect("send sweep 2 replay"); + let observed = obs_rx.recv().await.expect("replayed sweep store"); + assert!(!observed.rejected); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the replayed sweep must recompute the flip the rollback undid" + ); + assert_eq!( + status(&wallet_manager, &wallet_id, &owner, &tx2.to_string()).await, + PaymentStatus::Failed + ); + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// The payment channel's half of the reinstatement invariant, end to + /// end: a buffered `[TransactionsSwept(X), BlockProcessed(chainlocked + /// X)]` pair folds into ONE store round, and that round must carry X's + /// reinstated record with NO sweep-derived `Failed` overlay row beside + /// it — `CoreChangeSet::merge` retracts the sweep, and + /// `retract_reinstated_payment_flips` must retract the payment flip + /// keyed on the very same record set. The in-memory flip is undone + /// with it, so the entry reads `Pending` for the ordered confirm the + /// reinstated record drives on this same fold. Without the retraction + /// the fold would commit a stale `Failed` assertion beside the record + /// that disproves it. + /// + /// Both events are queued BEFORE the adapter task spawns, which is + /// what makes the single-fold deterministic: the first `recv` takes + /// the sweep and the backlog `try_recv` folds the record. + #[tokio::test] + async fn a_reinstating_record_in_the_same_fold_retracts_the_payment_flip() { + use dpp::identity::v0::IdentityV0; + use dpp::identity::Identity; + use dpp::prelude::Identifier; + use key_wallet::account::account_type::StandardAccountType; + use key_wallet::account::AccountType; + use key_wallet::managed_account::transaction_record::{ + TransactionDirection, TransactionRecord, + }; + use key_wallet::transaction_checking::transaction_router::TransactionType; + use key_wallet::transaction_checking::{BlockInfo, TransactionContext}; + + use super::spawn_wallet_event_adapter; + use crate::test_support::{funded_wallet_manager, NoopTestPersister}; + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + use crate::wallet::persister::WalletPersister; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + + // X: the transaction that is swept and then returns chainlocked in + // the same buffered fold. + let tx = dashcore::Transaction { + version: 1, + lock_time: 0, + input: vec![dashcore::TxIn { + previous_output: dashcore::OutPoint::new(dashcore::Txid::from([0xD0; 32]), 0), + ..Default::default() + }], + output: Vec::new(), + special_transaction_payload: None, + }; + let record = TransactionRecord::new( + tx, + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InChainLockedBlock(BlockInfo::new( + 4321, + { + use dashcore::hashes::Hash as _; + dashcore::BlockHash::all_zeros() + }, + 1_650_000_000, + )), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + 0, + ); + let txid = record.txid; + let txid_key = txid.to_string(); + + let noop = WalletPersister::new( + wallet_id, + Arc::new(NoopTestPersister) as Arc, + ); + { + let mut wm = wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("wallet info"); + info.identity_manager + .add_identity( + Identity::V0(IdentityV0 { + id: owner, + public_keys: std::collections::BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &noop, + ) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid_key.clone(), + PaymentEntry::new_sent(contact, 50_000, None), + &noop, + ) + .expect("record pending sent"); + } + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + let (event_tx, event_rx) = unbounded_channel(); + + // Queue BOTH events before the adapter runs, so they land in one + // fold: the sweep of X, then the chainlocked record reinstating X. + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xD1; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + event_tx + .send(WalletEvent::BlockProcessed { + wallet_id, + height: 4321, + chain_lock: None, + inserted: vec![record], + updated: vec![], + matured: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + addresses_derived: vec![], + }) + .expect("send reinstating record"); + + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + let observed = obs_rx.recv().await.expect("the folded store"); + assert!(!observed.rejected); + assert_eq!( + observed.n_records, 1, + "the reinstated record must ride the fold's store" + ); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the fold must carry exactly the reinstating record's own verdict" + ); + assert_eq!( + observed.n_payment_overlay_confirmed, 1, + "a merged changeset must never carry a sweep-derived Failed about a \ + txid the same fold reinstates — the retraction drops it and the \ + ordered confirm stages Confirmed in its place" + ); + + { + let wm = wallet_manager.read().await; + let status = wm + .get_wallet_info(&wallet_id) + .expect("info") + .identity_manager + .managed_identity(&owner) + .expect("managed") + .dashpay() + .payments + .get(&txid_key) + .expect("entry") + .status; + assert_eq!( + status, + PaymentStatus::Confirmed, + "the retraction undoes the in-memory flip and the reinstating \ + record's ordered confirm decides the entry on the same fold" + ); + } + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// Seed `owner` with one `Pending` sent payment under `txid`, through + /// a noop persister so a probe's observation stream carries only the + /// adapter's own stores. Shared by the ordering regressions below. + async fn seed_pending_sent_payment( + wallet_manager: &Arc>>, + wallet_id: WalletId, + owner: dpp::prelude::Identifier, + contact: dpp::prelude::Identifier, + txid: dashcore::Txid, + ) -> crate::wallet::persister::WalletPersister { + use dpp::identity::v0::IdentityV0; + use dpp::identity::Identity; + + use crate::test_support::NoopTestPersister; + use crate::wallet::identity::types::dashpay::payment::PaymentEntry; + use crate::wallet::persister::WalletPersister; + + let noop = WalletPersister::new( + wallet_id, + Arc::new(NoopTestPersister) as Arc, + ); + let mut wm = wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("wallet info"); + info.identity_manager + .add_identity( + Identity::V0(IdentityV0 { + id: owner, + public_keys: std::collections::BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &noop, + ) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid.to_string(), + PaymentEntry::new_sent(contact, 50_000, None), + &noop, + ) + .expect("record pending sent"); + noop + } + + async fn sent_payment_status( + wallet_manager: &Arc>>, + wallet_id: &WalletId, + owner: &dpp::prelude::Identifier, + txid: &str, + ) -> crate::wallet::identity::types::dashpay::payment::PaymentStatus { + let wm = wallet_manager.read().await; + wm.get_wallet_info(wallet_id) + .expect("info") + .identity_manager + .managed_identity(owner) + .expect("managed") + .dashpay() + .payments + .get(txid) + .expect("entry") + .status + } + + /// THE regression from the dashpay/platform#4442 review — interleaving + /// A: a pre-sweep confirmation hook is parked until the newer sweep + /// has been emitted (and durably applied), then released. + /// + /// Upstream emits `TransactionInstantLocked(X)` and later — the + /// chainlocked-conflict eviction upstream explicitly permits — + /// `TransactionsSwept(X)`. The adapter applies both in emission + /// order: `Confirmed` rides the IS-lock's round, the newer sweep + /// demotes to `Failed` on its own round. The EventHandler hook task + /// for the OLD IS-lock event — parked past both rounds, which the + /// lossy, unordered broadcast path genuinely allows — is then + /// released, running exactly the code the handler's spawned task runs + /// (`run_dashpay_payment_hooks`). The hook no longer writes + /// sent-payment verdicts, so the stale pre-sweep evidence cannot + /// resurrect the dead payment. Before the fix, the released hook + /// confirmed from `Failed` (`LIVE_CONFIRM_EVIDENCE` admitted it) and + /// durably landed `Confirmed` for a transaction upstream had proven + /// dead — with no later sweep re-emission to repair it. + #[tokio::test] + async fn a_parked_pre_sweep_confirmation_hook_cannot_resurrect_the_swept_payment() { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::account::account_type::StandardAccountType; + + use super::spawn_wallet_event_adapter; + use crate::test_support::funded_wallet_manager; + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = dpp::prelude::Identifier::from([0xAA; 32]); + let contact = dpp::prelude::Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xF1; 32]); + let noop = + seed_pending_sent_payment(&wallet_manager, wallet_id, owner, contact, txid).await; + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + let (event_tx, event_rx) = unbounded_channel(); + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + // The pre-sweep confirmation event. Its HOOK task is parked (not + // run) until after the sweep below is durable. + let is_lock_event = WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }; + + // The adapter applies the IS-lock in emission order: the ordered + // confirm stages `Confirmed` on the event's own round. + event_tx + .send(is_lock_event.clone()) + .expect("send IS-lock event"); + let observed = obs_rx.recv().await.expect("IS-lock round"); + assert!(!observed.rejected); + assert_eq!( + observed.n_payment_overlay_confirmed, 1, + "the ordered confirm must ride the IS-lock event's own round" + ); + assert_eq!( + sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await, + PaymentStatus::Confirmed + ); + + // The newer sweep — upstream evicted X for a chainlocked conflict + // — demotes the confirmation on its own round. + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xF2; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + let observed = obs_rx.recv().await.expect("sweep round"); + assert!(!observed.rejected); + assert_eq!(observed.n_payment_overlay_rows, 1); + assert_eq!( + observed.n_payment_overlay_confirmed, 0, + "the newer sweep's Failed row must overrule the older confirmation" + ); + assert_eq!( + sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await, + PaymentStatus::Failed + ); + + // Release the parked hook — the delayed task for the pre-sweep + // IS-lock event finally runs, after the newer sweep is durable. + crate::wallet::identity::network::run_dashpay_payment_hooks( + &wallet_manager, + &wallet_id, + &noop, + &is_lock_event, + ) + .await; + assert_eq!( + sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await, + PaymentStatus::Failed, + "a parked pre-sweep confirmation hook must not resurrect the swept payment" + ); + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// The same review finding's interleaving B: the pre-sweep + /// confirmation RUNS just before the sweep is staged. Pre-fix, the + /// hook confirmed `Pending → Confirmed` on its own round and the + /// sweep's eligibility check then skipped the entry (`Confirmed` was + /// terminal), leaving the dead payment durably `Confirmed`. Post-fix + /// the released hook writes nothing, the confirmation is the + /// adapter's own ordered write on the IS-lock round, and the sweep — + /// the newer verdict — demotes it on its round. + #[tokio::test] + async fn a_sweep_staged_after_a_confirmation_still_fails_the_dead_payment() { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::account::account_type::StandardAccountType; + + use super::spawn_wallet_event_adapter; + use crate::test_support::funded_wallet_manager; + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = dpp::prelude::Identifier::from([0xAA; 32]); + let contact = dpp::prelude::Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xF3; 32]); + let noop = + seed_pending_sent_payment(&wallet_manager, wallet_id, owner, contact, txid).await; + + let is_lock_event = WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }; + + // The hook task runs FIRST — just before the adapter stages + // anything. Post-fix it writes no sent-payment verdict. + crate::wallet::identity::network::run_dashpay_payment_hooks( + &wallet_manager, + &wallet_id, + &noop, + &is_lock_event, + ) + .await; + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + let (event_tx, event_rx) = unbounded_channel(); + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + // The adapter's ordered confirm owns the flip instead. + event_tx + .send(is_lock_event.clone()) + .expect("send IS-lock event"); + let observed = obs_rx.recv().await.expect("IS-lock round"); + assert_eq!( + observed.n_payment_overlay_confirmed, 1, + "the confirmation must be the adapter's ordered write, not the hook's" + ); + + // The newer sweep must still fail the dead payment — pre-fix the + // hook's earlier Confirmed made this a skipped, durable wrong + // terminal. + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xF4; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + let observed = obs_rx.recv().await.expect("sweep round"); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the sweep must demote the earlier confirmation on its own round" + ); + assert_eq!(observed.n_payment_overlay_confirmed, 0); + assert_eq!( + sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await, + PaymentStatus::Failed, + "the newer sweep's verdict must win over the pre-sweep confirmation" + ); + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// The two events of the race folded into ONE drain: `[IS-lock(X), + /// Swept(X)]` buffered together must commit a single round whose + /// payment row is the LAST verdict in emission order — `Failed` — + /// because the fold overwrites the staged `Confirmed` row per + /// `(owner, txid)` when the later sweep demotes it in memory. + #[tokio::test] + async fn a_confirmation_and_its_eviction_in_one_fold_commit_the_newer_verdict() { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::account::account_type::StandardAccountType; + + use super::spawn_wallet_event_adapter; + use crate::test_support::funded_wallet_manager; + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = dpp::prelude::Identifier::from([0xAA; 32]); + let contact = dpp::prelude::Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xF5; 32]); + seed_pending_sent_payment(&wallet_manager, wallet_id, owner, contact, txid).await; + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + let (event_tx, event_rx) = unbounded_channel(); + + // Queue BOTH events before the adapter task spawns, which is what + // makes the single-fold deterministic: the first `recv` takes the + // IS-lock and the backlog `try_recv` folds the sweep. + event_tx + .send(WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send IS-lock event"); + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xF6; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + let observed = obs_rx.recv().await.expect("the folded round"); + assert!(!observed.rejected); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "one row per (owner, txid): the later flip overwrites the earlier" + ); + assert_eq!( + observed.n_payment_overlay_confirmed, 0, + "the fold must commit the newer verdict — Failed, not the folded-over Confirmed" + ); + assert_eq!( + sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await, + PaymentStatus::Failed + ); + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// REGRESSION (dashpay/platform#4442 review, rejected-store leg): a + /// buffered `[IS-lock(X), Swept(X)]` round from durably `Pending` + /// flips the entry twice (`Pending → Confirmed → Failed`) but must + /// journal ONE coalesced verdict — so when the store rejects the + /// round, the rollback restores the durable pre-round state, + /// `Pending`, and the replayed sweep finds the entry eligible again. + /// + /// Before the fix the rollback ledger appended every intermediate + /// undo: the forward guarded replay skipped the first (`wrote = + /// Confirmed` no longer stood) and applied only the sweep's, leaving + /// memory at the in-round `Confirmed` — a state the store never held + /// — where the replayed sweep's `Confirmed → Failed` flip would + /// commit a demotion of a confirmation that never existed durably. + #[tokio::test] + async fn a_rejected_flip_round_restores_the_pre_round_durable_state() { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::account::account_type::StandardAccountType; + + use super::spawn_wallet_event_adapter; + use crate::test_support::funded_wallet_manager; + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = dpp::prelude::Identifier::from([0xAA; 32]); + let contact = dpp::prelude::Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xF7; 32]); + seed_pending_sent_payment(&wallet_manager, wallet_id, owner, contact, txid).await; + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + persister.fail_next(wallet_id); + let (event_tx, event_rx) = unbounded_channel(); + + // Queue BOTH events before the adapter task spawns, which is what + // makes the single-fold deterministic: the first `recv` takes the + // IS-lock and the backlog `try_recv` folds the sweep. + event_tx + .send(WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send IS-lock event"); + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xF8; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + let observed = obs_rx.recv().await.expect("the rejected fold"); + assert!(observed.rejected, "the probe rejects this round"); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the round coalesces to one staged verdict per (owner, txid)" + ); + assert_eq!(observed.n_payment_overlay_confirmed, 0); + + // The rollback runs right after commit in the same drain iteration; + // bounded-poll memory rather than racing it. + let mut rolled_back = false; + for _ in 0..50 { + if sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await + == PaymentStatus::Pending + { + rolled_back = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!( + rolled_back, + "a rejected round must restore the durable PRE-ROUND state (Pending) — \ + not any intermediate flip's snapshot" + ); + + // The replayed sweep (the re-scan re-emits it, because the + // rejected round kept the loser's record too) finds the entry + // eligible and recomputes the flip on a clean round. + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xF8; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep replay"); + let observed = obs_rx.recv().await.expect("replayed sweep store"); + assert!(!observed.rejected); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the replayed sweep must recompute the flip the rollback undid" + ); + assert_eq!( + sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await, + PaymentStatus::Failed + ); + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// REGRESSION (dashpay/platform#4442 review, three-event leg): + /// `[IS-lock(X), Swept(X), BlockProcessed(chainlocked X)]` in ONE + /// fold. The reinstating record's retraction must return the entry to + /// the durable pre-round `Pending` — dropping the coalesced journal + /// entry whole — so the ordered confirm genuinely transitions + /// `Pending → Confirmed` and the fold commits the reinstated record + /// WITH its payment correction. + /// + /// Before the fix the retraction unwound only the sweep's own undo, + /// whose captured `previous` was the same round's in-memory + /// `Confirmed`: memory came back `Confirmed`, the confirm no-opped + /// (same-state moves are not transitions), and — the earlier + /// `Confirmed` overlay row having been overwritten by the sweep's, + /// then retracted — the round committed the reinstated record with NO + /// payment row at all, leaving the payment durably `Pending` beside a + /// chainlocked record while memory claimed `Confirmed`. + #[tokio::test] + async fn a_three_event_reinstatement_fold_commits_the_confirmed_row() { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::account::account_type::StandardAccountType; + use key_wallet::account::AccountType; + use key_wallet::managed_account::transaction_record::{ + TransactionDirection, TransactionRecord, + }; + use key_wallet::transaction_checking::transaction_router::TransactionType; + use key_wallet::transaction_checking::{BlockInfo, TransactionContext}; + + use super::spawn_wallet_event_adapter; + use crate::test_support::funded_wallet_manager; + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = dpp::prelude::Identifier::from([0xAA; 32]); + let contact = dpp::prelude::Identifier::from([0xBB; 32]); + + // X: IS-locked, swept, and returned chainlocked — all in one + // buffered fold. + let tx = dashcore::Transaction { + version: 1, + lock_time: 0, + input: vec![dashcore::TxIn { + previous_output: dashcore::OutPoint::new(dashcore::Txid::from([0xD2; 32]), 0), + ..Default::default() + }], + output: Vec::new(), + special_transaction_payload: None, + }; + let record = TransactionRecord::new( + tx, + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InChainLockedBlock(BlockInfo::new( + 4321, + { + use dashcore::hashes::Hash as _; + dashcore::BlockHash::all_zeros() + }, + 1_650_000_000, + )), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + 0, + ); + let txid = record.txid; + seed_pending_sent_payment(&wallet_manager, wallet_id, owner, contact, txid).await; + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + let (event_tx, event_rx) = unbounded_channel(); + + // Queue all THREE events before the adapter task spawns: the + // first `recv` takes the IS-lock and the backlog `try_recv` folds + // the sweep, then the reinstating chainlocked record. + event_tx + .send(WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send IS-lock event"); + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: dashcore::Txid::from([0xD3; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + event_tx + .send(WalletEvent::BlockProcessed { + wallet_id, + height: 4321, + chain_lock: None, + inserted: vec![record], + updated: vec![], + matured: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + addresses_derived: vec![], + }) + .expect("send reinstating record"); + + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + let observed = obs_rx.recv().await.expect("the folded store"); + assert!(!observed.rejected); + assert_eq!( + observed.n_records, 1, + "the reinstated record must ride the fold's store" + ); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the fold must carry the reinstating record's payment correction" + ); + assert_eq!( + observed.n_payment_overlay_confirmed, 1, + "the retraction restores the durable pre-round Pending, so the \ + ordered confirm re-stages Confirmed — the reinstated record must \ + never commit without its payment correction" + ); + assert_eq!( + sent_payment_status(&wallet_manager, &wallet_id, &owner, &txid.to_string()).await, + PaymentStatus::Confirmed + ); + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + + /// The one-shot reinstatement gets the round's durability, end to end + /// through the real adapter loop: a chainlocked reinstating record + /// arriving in a LATER drain than the sweep finds the entry durably + /// `Failed`, and the record re-arrives already final, so no further + /// detection follows it and the reconcile pass (`Pending`-only by + /// construction) cannot cover it — a separately persisted correction + /// whose store was rejected would have left a durable `Failed` for a + /// transaction that survived. The adapter therefore owns the + /// correction: `confirm_final_sent_payments_for_store` flips the + /// entry and rides the `Confirmed` row on the SAME store round as the + /// reinstated record. + /// + /// Three legs mirror the sweep-flip test: the flip rides the record's + /// round; a rejected round rolls the in-memory `Confirmed` back to + /// `Failed` (the durable state) so the replay can recompute it; and + /// the replayed record's round carries the row again. + #[tokio::test] + async fn a_chainlocked_reinstatement_rides_the_records_round_and_survives_rejection() { + use dpp::identity::v0::IdentityV0; + use dpp::identity::Identity; + use dpp::prelude::Identifier; + use key_wallet::account::account_type::StandardAccountType; + use key_wallet::account::AccountType; + use key_wallet::managed_account::transaction_record::{ + TransactionDirection, TransactionRecord, + }; + use key_wallet::transaction_checking::transaction_router::TransactionType; + use key_wallet::transaction_checking::{BlockInfo, TransactionContext}; + + use super::spawn_wallet_event_adapter; + use crate::test_support::{funded_wallet_manager, NoopTestPersister}; + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + use crate::wallet::persister::WalletPersister; + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + + // Two transactions, each later reinstated by its own chainlocked + // record: X drives the ride leg, Y the rejection-and-replay legs. + let chainlocked_record = |input_byte: u8| { + let tx = dashcore::Transaction { + version: 1, + lock_time: 0, + input: vec![dashcore::TxIn { + previous_output: dashcore::OutPoint::new( + dashcore::Txid::from([input_byte; 32]), + 0, + ), + ..Default::default() + }], + output: Vec::new(), + special_transaction_payload: None, + }; + TransactionRecord::new( + tx, + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InChainLockedBlock(BlockInfo::new( + 4321, + { + use dashcore::hashes::Hash as _; + dashcore::BlockHash::all_zeros() + }, + 1_650_000_000, + )), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + 0, + ) + }; + let record_x = chainlocked_record(0xD0); + let record_y = chainlocked_record(0xD1); + let (txid_x, txid_y) = (record_x.txid, record_y.txid); + + let noop = WalletPersister::new( + wallet_id, + Arc::new(NoopTestPersister) as Arc, + ); + { + let mut wm = wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("wallet info"); + info.identity_manager + .add_identity( + Identity::V0(IdentityV0 { + id: owner, + public_keys: std::collections::BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &noop, + ) + .expect("add owner"); + let managed = info + .identity_manager + .managed_identity_mut(&owner) + .expect("managed"); + for txid in [txid_x, txid_y] { + managed + .record_dashpay_payment( + txid.to_string(), + PaymentEntry::new_sent(contact, 50_000, None), + &noop, + ) + .expect("record pending sent"); + } + } + + async fn status( + wallet_manager: &Arc>>, + wallet_id: &WalletId, + owner: &dpp::prelude::Identifier, + txid: &str, + ) -> crate::wallet::identity::types::dashpay::payment::PaymentStatus { + let wm = wallet_manager.read().await; + wm.get_wallet_info(wallet_id) + .expect("info") + .identity_manager + .managed_identity(owner) + .expect("managed") + .dashpay() + .payments + .get(txid) + .expect("entry") + .status + } + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS) + .union(crate::changeset::PersistenceCapabilities::ATOMIC_CHANGESETS), + )); + let (event_tx, event_rx) = unbounded_channel(); + let cancel = CancellationToken::new(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let handle = spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::clone(&persister), + event_rx, + Arc::clone(&sync_fault), + cancel.clone(), + ); + + // The sweep lands durably in its own drain: both entries flip to + // `Failed`, the overlay rides the sweep's round. + event_tx + .send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid_x, txid_y], + superseded_by: dashcore::Txid::from([0xDD; 32]), + released_outpoints: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .expect("send sweep"); + let observed = obs_rx.recv().await.expect("sweep round"); + assert!(!observed.rejected); + assert_eq!(observed.n_payment_overlay_rows, 2); + assert_eq!(observed.n_payment_overlay_confirmed, 0); + + let reinstating = |record: TransactionRecord| WalletEvent::BlockProcessed { + wallet_id, + height: 4321, + chain_lock: None, + inserted: vec![record], + updated: vec![], + matured: vec![], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + addresses_derived: vec![], + }; + + // Leg 1: X's reinstating record arrives in a LATER drain (the + // sweep's round above is already durable). Its round must carry + // the reinstated record AND the `Confirmed` correction — the flip + // rides the same atomic store as the record that justifies it. + event_tx + .send(reinstating(record_x.clone())) + .expect("send reinstating record for X"); + let observed = obs_rx.recv().await.expect("X's reinstating round"); + assert!(!observed.rejected); + assert_eq!(observed.n_records, 1, "the reinstated record rides"); + assert_eq!( + observed.n_payment_overlay_rows, 1, + "the reinstatement confirmation must ride the record's own round — \ + there is no later detection to retry from" + ); + assert_eq!( + observed.n_payment_overlay_confirmed, 1, + "and the row asserts Confirmed, not a stale Failed" + ); + assert_eq!( + status(&wallet_manager, &wallet_id, &owner, &txid_x.to_string()).await, + PaymentStatus::Confirmed + ); + + // Leg 2: a rejected reinstating round rolls the in-memory flip + // back to `Failed` — the durable state — so the replayed record + // can recompute it. Without the rollback, memory would read + // `Confirmed` ahead of the store and the replay's eligibility + // check would skip the entry. + persister.fail_next(wallet_id); + event_tx + .send(reinstating(record_y.clone())) + .expect("send reinstating record for Y"); + let observed = obs_rx.recv().await.expect("Y's rejected round"); + assert!(observed.rejected); + assert_eq!( + observed.n_payment_overlay_confirmed, 1, + "the attempt carried the correction" + ); + let mut rolled_back = false; + for _ in 0..50 { + if status(&wallet_manager, &wallet_id, &owner, &txid_y.to_string()).await + == PaymentStatus::Failed + { + rolled_back = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!( + rolled_back, + "a rejected round must roll the in-memory Confirmed back to Failed, \ + the durable state the rejection left in the store" + ); + + // Leg 3: the replayed record (the rejected round froze the + // watermark, so the re-scan re-emits the chainlocked block) + // recomputes the flip and its round carries the correction again. + event_tx + .send(reinstating(record_y)) + .expect("send replayed record for Y"); + let observed = obs_rx.recv().await.expect("Y's replayed round"); + assert!(!observed.rejected); + assert_eq!(observed.n_records, 1); + assert_eq!( + observed.n_payment_overlay_confirmed, 1, + "the replay must recompute the correction the rollback undid" + ); + assert_eq!( + status(&wallet_manager, &wallet_id, &owner, &txid_y.to_string()).await, + PaymentStatus::Confirmed + ); + + cancel.cancel(); + handle.await.expect("adapter task joins"); + } + /// The coalesced sweep-then-chainlocked-reinstatement fold, driven /// through the REAL producers rather than hand-built changesets: the /// sweep arm removes the tracked entry and emits its tombstone, the @@ -3579,6 +5490,7 @@ mod tests { super::WalletBatch { core: CoreChangeSet::default(), asset_locks, + payments_overlay: BTreeMap::new(), }, ); commit_batch( @@ -3637,6 +5549,7 @@ mod tests { WalletBatch { core, asset_locks: AssetLockChangeSet::default(), + payments_overlay: BTreeMap::new(), }, ); batch diff --git a/packages/rs-platform-wallet/src/changeset/persistence_capabilities.rs b/packages/rs-platform-wallet/src/changeset/persistence_capabilities.rs index 41839ecf66..7d09af5d7c 100644 --- a/packages/rs-platform-wallet/src/changeset/persistence_capabilities.rs +++ b/packages/rs-platform-wallet/src/changeset/persistence_capabilities.rs @@ -84,6 +84,15 @@ impl PersistenceCapabilities { /// `CORE_SWEEP_REMOVAL`. On the FFI surface Rust honours the /// declaration only when `on_persist_dashpay_payments_fn` is actually /// wired. + /// + /// This bit alone attests only per-callback durability. The adapter's + /// round-coupled staging additionally requires `ATOMIC_CHANGESETS` + /// (see [`Self::ROUND_COUPLED_PAYMENT_FLIPS`]): on a host whose + /// callbacks commit independently, the Core record and watermark can + /// become durable while the process stops before the payments + /// callback — and a one-shot chainlocked reinstatement never + /// re-emits, so its payment would stay durably `Failed` beside a + /// durably recorded reinstatement. pub const DASHPAY_PAYMENTS: Self = Self(1 << 11); /// Capabilities required before exporting and funding an invitation voucher. @@ -103,6 +112,22 @@ impl PersistenceCapabilities { pub const ASSET_LOCK_RECONCILIATION: Self = Self(Self::ATOMIC_CHANGESETS.0 | Self::TRACKED_ASSET_LOCKS.0 | Self::WALLET_RESTORE.0); + /// Capabilities required before the wallet-event adapter stages a + /// sweep's `Failed` flip or a reinstatement's `Confirmed` correction + /// onto the triggering record's own store round. The point of that + /// staging is that the flip and the record land or fail together — + /// `DASHPAY_PAYMENTS` proves the overlay rows are durably applied, + /// and `ATOMIC_CHANGESETS` proves the round commits or rolls back as + /// one unit. A payments-durable host without the atomic round gives + /// neither the coupling nor the fail-closed watermark backstop: it + /// can commit the Core record and watermark, then stop before the + /// payments write — and a one-shot reinstatement never re-emits to + /// retry the orphaned flip. Such a host is treated as payments-blind + /// for staging (the in-memory flip still happens; funds-safe, as + /// payment entries are display metadata). + pub const ROUND_COUPLED_PAYMENT_FLIPS: Self = + Self(Self::ATOMIC_CHANGESETS.0 | Self::DASHPAY_PAYMENTS.0); + pub const fn from_bits_retain(bits: u64) -> Self { Self(bits) } @@ -208,6 +233,10 @@ mod tests { PersistenceCapabilities::ASSET_LOCK_RECONCILIATION.bits(), 0x281 ); + assert_eq!( + PersistenceCapabilities::ROUND_COUPLED_PAYMENT_FLIPS.bits(), + 0x801 + ); } #[test] diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index 1e64401db2..85b8f06bb2 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/packages/rs-platform-wallet/src/manager/mod.rs @@ -467,9 +467,11 @@ impl PlatformWalletManager

{ // with SPV's write lock. let lock_handler = Arc::new(LockNotifyHandler::new(Arc::clone(&lock_notify))); let balance_handler = Arc::new(BalanceUpdateHandler::new(Arc::clone(&wallets))); - // DashPayPaymentHandler records incoming DashPay payments and - // confirms sent ones off the wallet-event fan-out, keeping that - // domain logic out of the generic core-changeset bridge. It holds + // DashPayPaymentHandler records incoming DashPay payments off the + // wallet-event fan-out, keeping that domain logic out of the + // generic core-changeset bridge. (Sent-payment verdicts do NOT + // run here: they are the wallet-event adapter's to apply in + // emission order — see `payment_handler`'s module docs.) It holds // the wallet-manager (for the in-memory payment state it mutates) // and the persister (to write the resulting payment rows). let dashpay_payment_handler = Arc::new(DashPayPaymentHandler::new( diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs b/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs index 752fee202c..b45a2e4649 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs @@ -47,14 +47,14 @@ pub use invitation::{ }; mod payment_handler; pub(crate) use payment_handler::DashPayPaymentHandler; -// Re-exported for the payments unit tests, which drive the hooks -// directly; the handler itself calls it module-locally. +// Re-exported for the core-bridge ordering regressions, which release a +// "parked" hook task directly; the handler itself calls it module-locally. #[cfg(test)] pub(crate) use payment_handler::run_dashpay_payment_hooks; mod payments; pub(crate) use payments::{ - confirm_sent_dashpay_payment, confirm_sent_dashpay_payment_by_txid, - record_incoming_dashpay_payments, + confirm_final_sent_payments_for_store, flip_swept_sent_payments_for_store, + record_incoming_dashpay_payments, rollback_payment_flips, PaymentFlipUndo, SweptPaymentFlips, }; mod profile; pub(crate) mod sdk_writer; diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/payment_handler.rs b/packages/rs-platform-wallet/src/wallet/identity/network/payment_handler.rs index e5ec944104..343253f159 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/payment_handler.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/payment_handler.rs @@ -1,4 +1,4 @@ -//! Event handler that drives the DashPay payment hooks off upstream +//! Event handler that records incoming DashPay payments off upstream //! `WalletEvent`s. //! //! Registered as one of the [`PlatformEventHandler`]s in @@ -6,22 +6,35 @@ //! keeps the DashPay-payment domain logic out of the generic //! core-changeset bridge ([`spawn_wallet_event_adapter`]): the bridge //! projects every event into a `CoreChangeSet` and persists it, while -//! this handler independently records incoming payments and confirms -//! sent ones. +//! this handler independently records incoming payments. +//! +//! # No sent-payment verdicts here +//! +//! Sent-payment status writes (`Confirmed` on finality, `Failed` on a +//! sweep) live EXCLUSIVELY on the wallet-event adapter's single ordered +//! drain (`confirm_final_sent_payments_for_store`, +//! `payments::SweptPaymentFlips`), never on this handler. This handler +//! rides dash-spv's bounded, lossy broadcast and spawns one independent +//! task per event, so execution order does not preserve emission order — +//! a pre-sweep confirmation task delayed past a newer sweep would +//! resurrect a dead payment as durably `Confirmed`, and a task that ran +//! just before the sweep staged would make the sweep skip the entry. +//! Verdicts are only safe where they apply in emission order. Incoming +//! recording is different in kind: it is an idempotent insert keyed by +//! txid with no state machine to race, and the recurring reconcile sweep +//! backfills anything the lossy broadcast dropped. //! //! # Why it spawns //! //! [`PlatformEventHandler::on_wallet_event`] is synchronous and is //! dispatched from dash-spv's wallet-event broadcast monitor, which can -//! fire while SPV holds the wallet-manager write lock. The payment hooks -//! are async and take that same write lock, so they cannot run inline. -//! The handler therefore captures an owned copy of the event and spawns -//! a task that queues on the write lock and runs once SPV releases it. -//! Every hook path is idempotent per txid (re-detections converge and -//! the recurring reconcile sweep backfills anything a lagged broadcast -//! dropped), so running off the core-store bridge's ordering is safe — -//! a payment row's only foreign key is to its `identities` parent, never -//! to a core transaction row. +//! fire while SPV holds the wallet-manager write lock. The recording +//! hook is async and takes that same write lock, so it cannot run +//! inline. The handler therefore captures an owned copy of the event and +//! spawns a task that queues on the write lock and runs once SPV +//! releases it. Recording is idempotent per txid, so running off the +//! core-store bridge's ordering is safe — a payment row's only foreign +//! key is to its `identities` parent, never to a core transaction row. use std::sync::Arc; use std::{future::Future, sync::Mutex}; @@ -35,8 +48,9 @@ use crate::changeset::traits::PlatformWalletPersistence; use crate::events::PlatformEventHandler; use crate::wallet::platform_wallet::PlatformWalletInfo; -/// Records incoming DashPay payments and confirms sent ones in response -/// to upstream `WalletEvent`s. +/// Records incoming DashPay payments in response to upstream +/// `WalletEvent`s. (Sent-payment verdicts live on the wallet-event +/// adapter's ordered drain — see the module docs.) /// /// Holds the manager's `wallet_manager` (for the in-memory identity / /// payment state the hooks mutate) and an `Arc` @@ -227,25 +241,20 @@ impl EventHandler for DashPayPaymentHandler { impl PlatformEventHandler for DashPayPaymentHandler {} /// Transaction records carried by `event` that should drive the DashPay -/// payment hooks (live incoming-record recording + sent-payment confirm). +/// incoming-payment recording. /// /// [`WalletEvent::TransactionDetected`] is the first off-chain sighting of -/// a transaction — mempool, or a direct InstantSend lock — so its -/// `record.context` is not yet block-confirmed. +/// a transaction — mempool, or a direct InstantSend lock. /// [`WalletEvent::BlockProcessed`] carries the records a block changed: -/// `inserted` (first stored in this block) and `updated` -/// (previously-known records that this block confirmed). A wallet sees its -/// *own* broadcast in the mempool first, so that transaction reaches a -/// confirmed context only via `BlockProcessed.updated` — routing solely -/// `TransactionDetected` is the gap that left sent payments stuck -/// `Pending`: the confirm hook early-returns on the unconfirmed mempool -/// sighting and never sees the confirming block. `matured` is +/// `inserted` (first stored in this block — how a payment first seen in a +/// block lands) and `updated` (previously-known records the block +/// confirmed — a second chance to record a payment whose first-sighting +/// broadcast was dropped by the lossy bus). `matured` is /// coinbase-maturity only — never a DashPay payment — so it is excluded. fn dashpay_payment_records(event: &WalletEvent) -> Vec<&TransactionRecord> { // Exhaustive on purpose (no `_` arm): a new upstream `WalletEvent` // variant that carries transaction records must fail to compile here - // rather than be silently dropped — routing only `TransactionDetected` - // is exactly the gap that left sent payments stuck `Pending`. + // rather than be silently dropped. match event { WalletEvent::TransactionDetected { record, .. } => vec![record.as_ref()], WalletEvent::BlockProcessed { @@ -253,12 +262,16 @@ fn dashpay_payment_records(event: &WalletEvent) -> Vec<&TransactionRecord> { } => inserted.iter().chain(updated.iter()).collect(), // `TransactionsSwept` carries txids, not records: the wallet has // already dropped the records these name. Its payment consequence - // — failing the matching `Pending` sent payments, since a swept - // transaction can never confirm — is NOT this handler's to apply: - // a sweep never re-emits once its round is durable, so the flip - // must ride the sweep's own atomic store round, which belongs to - // the wallet-event adapter. Routing it here would persist the - // flip on a separate round with no replay if that round fails. + // — failing the matching sent payments, since a swept transaction + // can never confirm — is NOT this handler's to apply: sent-payment + // verdicts must apply in emission order, so the wallet-event + // adapter owns them (see `payments::SweptPaymentFlips` and the + // module docs). `TransactionInstantLocked` likewise: it is + // sent-payment finality evidence (no record, only a txid), owned + // by the adapter's ordered confirm + // (`confirm_final_sent_payments_for_store`) — a confirmation run + // from this handler's unordered task could land after a newer + // sweep and durably resurrect a dead payment. WalletEvent::TransactionInstantLocked { .. } | WalletEvent::TransactionsSwept { .. } | WalletEvent::SyncHeightAdvanced { .. } @@ -268,59 +281,49 @@ fn dashpay_payment_records(event: &WalletEvent) -> Vec<&TransactionRecord> { /// Whether `event` is worth spawning a payment-hook task for. /// -/// Covers the record-bearing events ([`dashpay_payment_records`]) plus -/// [`WalletEvent::TransactionInstantLocked`], which drives the sent-payment -/// confirm by txid alone (no record). A `BlockProcessed` that changed no -/// records — the common case while syncing past empty blocks — has no -/// payment work, so it is skipped rather than spawning a task that would -/// only take and release the wallet-manager write lock for nothing. +/// Exactly the record-bearing events ([`dashpay_payment_records`]): only +/// they can carry an incoming payment to record. A `BlockProcessed` that +/// changed no records — the common case while syncing past empty blocks — +/// has no payment work, so it is skipped rather than spawning a task that +/// would only take and release the wallet-manager write lock for nothing. /// Allocation-free. fn drives_payment_hooks(event: &WalletEvent) -> bool { match event { - WalletEvent::TransactionDetected { .. } | WalletEvent::TransactionInstantLocked { .. } => { - true - } + WalletEvent::TransactionDetected { .. } => true, WalletEvent::BlockProcessed { inserted, updated, .. } => !inserted.is_empty() || !updated.is_empty(), // No records to route (see `dashpay_payment_records`), so a task // here would take and release the wallet-manager write lock for - // nothing. The sweep's payment consequence belongs on the - // wallet-event adapter's own store round — see `dashpay_payment_records`. - WalletEvent::TransactionsSwept { .. } + // nothing. Sent-payment verdicts — the sweep's `Failed` AND every + // finality confirmation, including `TransactionInstantLocked`'s — + // ride the wallet-event adapter's ordered drain instead. + WalletEvent::TransactionInstantLocked { .. } + | WalletEvent::TransactionsSwept { .. } | WalletEvent::SyncHeightAdvanced { .. } | WalletEvent::ChainLockProcessed { .. } => false, } } /// Run the DashPay payment hooks for `event`: record any incoming DashPay -/// payment, then advance a matching sent payment from `Pending` to -/// `Confirmed` once its transaction reaches finality (mined or -/// InstantSend-locked). The opposite terminal — `Failed`, when a sweep -/// proves the transaction never can confirm — is deliberately not applied -/// here: it belongs on the sweep's own atomic store round in the -/// wallet-event adapter (see `dashpay_payment_records`). All paths are -/// idempotent per txid, so re-detections and repeated block-processing -/// rounds converge without duplicating entries. +/// payment the event's records carry. Idempotent per txid, so +/// re-detections and repeated block-processing rounds converge without +/// duplicating entries. +/// +/// Deliberately does NOT touch sent-payment status. Both verdicts — a +/// sweep's `Failed` and a finality `Confirmed` (including the +/// reinstatement correction) — are applied by the wallet-event adapter in +/// emission order on the events' own store rounds (see +/// `payments::SweptPaymentFlips` and +/// `confirm_final_sent_payments_for_store`). This function runs on an +/// unordered spawned task off a lossy broadcast; a verdict written here +/// could overrule a newer one it never saw. pub(crate) async fn run_dashpay_payment_hooks( wallet_manager: &Arc>>, wallet_id: &WalletId, persister: &crate::wallet::persister::WalletPersister, event: &WalletEvent, ) { - // An InstantSend lock applied to a previously-seen transaction carries - // no record — only a txid — and is final for DashPay display, so - // confirm the matching sent payment directly. - if let WalletEvent::TransactionInstantLocked { txid, .. } = event { - crate::wallet::identity::network::confirm_sent_dashpay_payment_by_txid( - wallet_manager, - wallet_id, - persister, - txid, - ) - .await; - return; - } for record in dashpay_payment_records(event) { crate::wallet::identity::network::record_incoming_dashpay_payments( wallet_manager, @@ -329,13 +332,6 @@ pub(crate) async fn run_dashpay_payment_hooks( record, ) .await; - crate::wallet::identity::network::confirm_sent_dashpay_payment( - wallet_manager, - wallet_id, - persister, - record, - ) - .await; } } @@ -398,13 +394,12 @@ mod tests { } } - /// `BlockProcessed` is the path by which a wallet's own broadcast - /// confirms (`updated`), and the path by which a payment first seen in a - /// block lands (`inserted`); both must drive the DashPay payment hooks. - /// `matured` is coinbase-maturity only and carries no DashPay payment, so - /// it is excluded. A regression that re-narrows routing to - /// `TransactionDetected` — the original sent-payment-stuck-`Pending` bug — - /// drops the `updated` record and fails this test. + /// `BlockProcessed` is the path by which a payment first seen in a + /// block lands (`inserted`), and `updated` is the second chance to + /// record a payment whose first-sighting broadcast the lossy bus + /// dropped; both must drive the incoming-recording hook. `matured` is + /// coinbase-maturity only and carries no DashPay payment, so it is + /// excluded. #[test] fn dashpay_payment_records_covers_block_processed_inserted_and_updated() { let event = block_processed(vec![record(0x01)], vec![record(0x02)], vec![record(0x03)]); @@ -419,7 +414,7 @@ mod tests { assert!( txids.contains(&record(0x02).txid), "updated (just-confirmed) record must drive the payment hooks — \ - this is how a sent payment flips Pending → Confirmed" + the backfill chance for a dropped first sighting" ); assert!( !txids.contains(&record(0x03).txid), @@ -458,11 +453,14 @@ mod tests { assert!(!drives_payment_hooks(&event)); } - /// `TransactionInstantLocked` carries no record but DOES drive the - /// payment hooks — it confirms a sent payment by txid alone (an - /// InstantSend lock is final for DashPay display). + /// `TransactionInstantLocked` must NOT drive the payment hooks: it is + /// sent-payment finality evidence, and every sent-payment verdict is + /// the wallet-event adapter's to apply in emission order + /// (`confirm_final_sent_payments_for_store`). A hook task here would + /// confirm on an unordered, lossy path — the exact shape that let a + /// stale pre-sweep confirmation durably resurrect a dead payment. #[test] - fn instant_locked_drives_payment_hooks_without_a_record() { + fn instant_locked_does_not_drive_payment_hooks() { use dashcore::ephemerealdata::instant_lock::InstantLock; let event = WalletEvent::TransactionInstantLocked { wallet_id: [0u8; 32], @@ -471,18 +469,17 @@ mod tests { balance: WalletCoreBalance::default(), account_balances: std::collections::BTreeMap::new(), }; - // No record to route, but the event must still drive the hooks. assert!(dashpay_payment_records(&event).is_empty()); - assert!(drives_payment_hooks(&event)); + assert!(!drives_payment_hooks(&event)); } /// `TransactionsSwept` must NOT drive the payment hooks: its payment - /// consequence — failing the losers' `Pending` sent payments — belongs - /// on the wallet-event adapter's own atomic store round, because a - /// sweep never re-emits once its round is durable and a separately - /// persisted flip that failed its store would be lost for good. - /// Spawning a hook task here would race a second write against that - /// round. + /// consequence — failing the losers' `Pending` sent payments — rides + /// the wallet-event adapter's own atomic store round (see + /// `payments::SweptPaymentFlips`), because a sweep never re-emits once + /// its round is durable and a separately persisted flip that failed + /// its store would be lost for good. Spawning a hook task here would + /// race a second write against that round. #[test] fn transactions_swept_does_not_drive_payment_hooks() { let event = WalletEvent::TransactionsSwept { diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs b/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs index c084ea667a..025d5fba47 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs @@ -654,15 +654,16 @@ impl DashPayView<'_, B> { /// persisted core transaction record reports the transaction final. /// /// Recovery path for sent-payment confirmation. The live confirm path - /// ([`confirm_sent_dashpay_payment`](super::confirm_sent_dashpay_payment)) - /// flips a sent payment the moment its block / InstantSend-lock event - /// arrives, but that is a single live event: if it is missed — a lagged - /// wallet-event broadcast, or a relaunch after the transaction confirmed - /// but before the flip was captured — the entry would otherwise stay - /// `Pending` forever (received payments self-heal from receival-account - /// UTXOs; sent payments have no such ground truth). This sweep consults - /// the persisted core tx record (txid + context) and flips any `Pending` - /// `Sent` entry whose transaction is mined or InstantSend-locked. + /// ([`confirm_final_sent_payments_for_store`](super::confirm_final_sent_payments_for_store), + /// on the wallet-event adapter's ordered drain) flips a sent payment + /// the moment its block / InstantSend-lock event is applied, but that + /// is a single live event: if it is missed — a relaunch after the + /// transaction confirmed but before the flip was captured — the entry + /// would otherwise stay `Pending` forever (received payments + /// self-heal from receival-account UTXOs; sent payments have no such + /// ground truth). This sweep consults the persisted core tx record + /// (txid + context) and flips any `Pending` `Sent` entry whose + /// transaction is mined or InstantSend-locked. /// /// Runs as a local-only step of `dashpay_sync()` — one persister read /// per pending sent payment, no network round-trips. Idempotent: a @@ -719,14 +720,25 @@ impl DashPayView<'_, B> { if sent_payment_status_for_record(&record) != PaymentStatus::Confirmed { continue; } - // Flip in place via the shared confirm path (re-checks the - // entry is still a `Pending` `Sent` under its own write lock, - // so it stays correct if a live event raced this sweep). + // Flip in place via the shared confirm path, declaring what + // this sweep's evidence can speak for: the record was read + // AFTER a snapshot that saw the entry `Pending`, so it proves + // nothing about an entry that has since moved. In particular + // the wallet-event adapter can flip the entry to `Failed` + // anywhere in the snapshot→confirm span (this pass runs on its + // own task, off the adapter's ordered drain, and the sweep + // deletes the record on the adapter task) — this pass's record + // read may predate that verdict, and confirming from it would + // land a dead payment `Confirmed` against a newer sweep. The + // resolver re-checks under its own write lock against exactly + // this evidence set, so an entry no longer `Pending` is left + // for the adapter, whose verdicts apply in emission order. confirm_sent_payment_by_txid( &self.wallet_manager, &self.wallet_id, &self.persister, &txid_str, + RECONCILE_CONFIRM_EVIDENCE, ) .await; confirmed += 1; @@ -916,107 +928,548 @@ fn sent_payment_status_for_record( } } -/// Advance a sender's `Sent` [`PaymentEntry`] from `Pending` to -/// `Confirmed` once its broadcast transaction reaches finality. +/// What the reconcile confirm's evidence can speak for — the from-states +/// it is entitled to advance. The transition table +/// ([`sent_status_transition_allowed`]) says which moves the machine +/// permits; this says which of them this caller's evidence actually +/// supports, and the resolver requires both. /// -/// [`IdentityWallet::send_payment`] records the outgoing entry as -/// `Pending` at broadcast time and nothing else advances it. The wallet -/// re-emits the sender's own transaction as it moves through mempool → -/// InstantSend → in-block → chain-locked, so when a re-detection reports -/// the transaction final the matching entry is flipped in place. +/// `Pending`-only, by construction: reconcile evidence is a +/// persisted-record read made after a snapshot that saw the entry +/// `Pending`, on a task with no ordering against the wallet-event +/// adapter. If the entry has since moved to `Failed`, the read raced the +/// adapter's sweep (which deletes the record on its own ordered drain) +/// and may predate the verdict — confirming from it would land a dead +/// payment `Confirmed` against a newer sweep. Every OTHER sent-payment +/// verdict is applied by the adapter itself in emission order +/// ([`confirm_final_sent_payments_for_store`], +/// [`flip_swept_sent_payments_for_store`]) and needs no evidence class: +/// on the ordered drain, the event being applied postdates every verdict +/// already applied. +const RECONCILE_CONFIRM_EVIDENCE: + &[crate::wallet::identity::types::dashpay::payment::PaymentStatus] = + &[crate::wallet::identity::types::dashpay::payment::PaymentStatus::Pending]; + +/// Flip the `Sent` [`PaymentEntry`] under `txid` (if any) to `Confirmed`, +/// in place, preserving amount/memo/counterparty. The reconcile sweep's +/// recovery flip — every live/ordered confirmation is the wallet-event +/// adapter's job ([`confirm_final_sent_payments_for_store`]). /// -/// An **InstantSend lock counts as final** for DashPay display: it is -/// effectively irreversible, so the user sees `Confirmed` without waiting -/// for the surrounding block. A bare mempool re-detection (no IS lock, not -/// yet mined) leaves the entry `Pending` — which it genuinely still is. -/// Idempotent: once `Confirmed`, later re-detections find nothing to -/// change and skip the persistence round. -pub(crate) async fn confirm_sent_dashpay_payment( +/// No-op when no entry exists for `txid`, it is not a `Sent` entry, it is +/// already `Confirmed` (so repeated reconcile passes are idempotent and +/// skip the persistence round), or its current state is outside what +/// `evidence` can speak for — a `Failed` entry never advances here, +/// because a reconcile pass's record read can predate the sweep verdict +/// entirely (see [`RECONCILE_CONFIRM_EVIDENCE`]). +async fn confirm_sent_payment_by_txid( wallet_manager: &Arc>>, wallet_id: &WalletId, persister: &crate::wallet::persister::WalletPersister, - record: &key_wallet::managed_account::transaction_record::TransactionRecord, + txid: &str, + evidence: &[crate::wallet::identity::types::dashpay::payment::PaymentStatus], ) { - use key_wallet::transaction_checking::TransactionContext; - // Mined (InBlock / InChainLockedBlock) OR InstantSend-locked advances - // the entry. A plain mempool sighting does not. - let is_instant_send = matches!(record.context, TransactionContext::InstantSend(_)); - if !record.is_confirmed() && !is_instant_send { - return; - } - confirm_sent_payment_by_txid( + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + // Log-and-continue is sound here: the flip rolled back in memory with + // the failed store, and the entry is still `Pending`, so the next + // reconcile pass — and any adapter-applied signal for the same + // transaction — re-drives the confirmation. The one-shot cases with + // no later signal (a sweep's `Failed`, a reinstating record's + // `Failed → Confirmed`) never come through here: they ride their + // event's own store round on the adapter with the round's fail-closed + // machinery, so a store failure HERE is only ever a lost recovery + // write, never the last chance. + if let Err(e) = resolve_sent_payment_by_txid( wallet_manager, wallet_id, persister, - &record.txid.to_string(), + txid, + PaymentStatus::Confirmed, + evidence, ) - .await; + .await + { + tracing::warn!( + error = %e, + "Failed to persist sent-payment confirmation; will retry on next detection" + ); + } +} + +/// The in-memory `Failed` flips for a sweep's losers, packaged for the +/// wallet-event adapter to ride on the sweep's OWN store round. +/// +/// A swept transaction was provably beaten to one of its inputs, so it can +/// never confirm — exactly the "transaction was dropped" case +/// [`PaymentStatus::Failed`](crate::wallet::identity::types::dashpay::payment::PaymentStatus::Failed) +/// documents — and the sweep deletes the record that was the last thing +/// `reconcile_sent_payments_from_tx_history` could have resolved the entry +/// from. Durability is why this is a changeset payload rather than a hook +/// that persists on its own: a sweep never re-emits once its round is +/// durable, so a separately persisted flip that failed its store was lost +/// for good (a bounded retry only narrowed the window). Riding the same +/// atomic `store()` as the core sweep gives the flip the round's own +/// fail-closed machinery — a rejection keeps the loser's record too, the +/// wallet faults, and the re-scan re-detects the conflict and re-emits the +/// sweep, recomputing the flip. +/// +/// A `Confirmed` entry IS demoted here (the shared transition table's +/// `(Confirmed, Failed)` edge): the only confirmations that can exist +/// when a sweep is applied were written from evidence the sweep +/// postdates — every verdict writer is either this same ordered drain or +/// re-validates under the manager lock against `Pending`-only evidence — +/// and upstream never emits a sweep for a transaction that is currently +/// final (a confirmed record is never a sweep loser). The one way the +/// verdict reverses — a reinstatement — re-emits the record final, which +/// the adapter's ordered confirm accepts from `Failed` on a later event. +/// +/// Despite the name, this is the carrier for BOTH round-riding verdict +/// flips: the sweep's `Failed` rows and the ordered confirm's +/// `Confirmed` rows ([`confirm_final_sent_payments_for_store`]). +#[derive(Debug, Default)] +pub(crate) struct SweptPaymentFlips { + /// The flips applied to memory, in application order. Each is the + /// SINGLE record of its flip — the store overlay row, the + /// rejected-round undo, and the same-fold retraction are all derived + /// from it by the wallet-event adapter's round journal (see + /// `fold_payment_flips` in the core bridge), never bookkept apart. + pub flips: Vec, +} + +/// One applied in-memory sent-payment flip: the entry as it stood before +/// (`previous`) and as memory now holds it (`updated`). Everything a +/// round needs is a projection of this pair: the staged overlay row is +/// `updated`, and the guarded undo is "restore `previous` while +/// `updated.status` still stands" ([`PaymentFlipUndo`]). +#[derive(Debug, Clone)] +pub(crate) struct PaymentFlip { + pub owner: Identifier, + pub txid: String, + /// The entry as it stood before this flip. + pub previous: crate::wallet::identity::PaymentEntry, + /// The entry as memory holds it after the flip. + pub updated: crate::wallet::identity::PaymentEntry, +} + +/// One staged payment flip's undo: what to restore if the store round the +/// flip rode is rejected, and the status the flip wrote — the undo applies +/// only while that write still stands (see [`rollback_payment_flips`]). +/// +/// Derived, never bookkept: the adapter projects it out of its coalesced +/// round journal (`previous` = the durable pre-round entry, `wrote` = the +/// status the round last staged), so an undo can never describe a +/// different history than the staged row it guards. +#[derive(Debug, Clone)] +pub(crate) struct PaymentFlipUndo { + pub owner: Identifier, + pub txid: String, + /// The entry as it stood before the flip. + pub previous: crate::wallet::identity::PaymentEntry, + /// The status the flip wrote over it. + pub wrote: crate::wallet::identity::types::dashpay::payment::PaymentStatus, +} + +impl SweptPaymentFlips { + pub(crate) fn is_empty(&self) -> bool { + self.flips.is_empty() + } + + /// Project every flip into its guarded undo, in application order — + /// what [`rollback_payment_flips`] takes. Production rollbacks go + /// through the adapter's coalesced round journal instead (one undo + /// per `(owner, txid)`); this uncoalesced projection exists for the + /// single-flip unit tests. + #[cfg(test)] + pub(crate) fn into_undos(self) -> Vec { + self.flips + .into_iter() + .map(|flip| PaymentFlipUndo { + owner: flip.owner, + txid: flip.txid, + wrote: flip.updated.status, + previous: flip.previous, + }) + .collect() + } } -/// Confirm a sender's `Sent` [`PaymentEntry`] by txid alone, for a -/// [`WalletEvent::TransactionInstantLocked`](key_wallet_manager::WalletEvent::TransactionInstantLocked) -/// that applies an InstantSend lock to a previously-seen transaction. -/// That event carries no [`TransactionRecord`](key_wallet::managed_account::transaction_record::TransactionRecord), -/// only the txid; an IS lock is treated as final for DashPay display, so -/// this flips a matching `Pending` `Sent` entry to `Confirmed`. Idempotent -/// (the underlying flip skips entries already past `Pending`). -pub(crate) async fn confirm_sent_dashpay_payment_by_txid( +/// Flip the `Sent` entries under `txids` to `Failed` in memory and return +/// the store payload + rollback. Persists NOTHING itself — the caller +/// owns the store round (see [`SweptPaymentFlips`]). +/// +/// Eligibility is the shared transition table: `Pending` AND `Confirmed` +/// entries flip. Demoting `Confirmed` is what makes a sweep that +/// postdates a confirmation win — upstream permits a chainlocked +/// transaction to evict an IS-locked conflict, so a payment confirmed on +/// IS-lock evidence can genuinely die afterwards — and it is sound only +/// because this runs on the adapter's ordered drain, where every already +/// applied confirmation predates the sweep being applied (see the table's +/// `(Confirmed, Failed)` edge for the full argument). `Failed` entries +/// are skipped, which keeps a replayed sweep idempotent. +pub(crate) async fn flip_swept_sent_payments_for_store( wallet_manager: &Arc>>, wallet_id: &WalletId, - persister: &crate::wallet::persister::WalletPersister, - txid: &dashcore::Txid, + txids: &[dashcore::Txid], +) -> SweptPaymentFlips { + use crate::wallet::identity::types::dashpay::payment::{PaymentDirection, PaymentStatus}; + + let mut flips = SweptPaymentFlips::default(); + if txids.is_empty() { + return flips; + } + let mut wm = wallet_manager.write().await; + let Some(info) = wm.get_wallet_info_mut(wallet_id) else { + return flips; + }; + for txid in txids { + let key = txid.to_string(); + 'owners: for owner in info.identity_manager.identity_ids() { + let Some(managed) = info.identity_manager.managed_identity_mut(&owner) else { + continue; + }; + let previous = match managed.dashpay().payments.get(&key) { + Some(entry) + if entry.direction == PaymentDirection::Sent + && sent_status_transition_allowed(entry.status, PaymentStatus::Failed) => + { + entry.clone() + } + _ => continue, + }; + let mut updated = previous.clone(); + updated.status = PaymentStatus::Failed; + tracing::info!( + owner = %owner, + txid = %key, + "Failing sent DashPay payment on its sweep's own store round" + ); + managed + .dashpay_payments_mut() + .insert(key.clone(), updated.clone()); + flips.flips.push(PaymentFlip { + owner, + txid: key.clone(), + previous, + updated, + }); + // txid is unique — only one identity can hold this entry. + break 'owners; + } + } + flips +} + +/// The adapter-owned sent-payment confirmation: flip the `Sent` entries +/// whose transaction `event` proves final to `Confirmed` in memory, and +/// return the flips for the wallet-event adapter to journal onto the +/// event's OWN store round. Persists NOTHING itself — the caller owns +/// the round. +/// +/// This is the ONLY live confirmation path. It runs on the adapter's +/// single ordered drain of the lossless persistence channel, which is +/// what makes each verdict safe against every other: the event being +/// applied postdates every verdict already applied, so a confirmation +/// here can never overrule a newer sweep — and a sweep folded later +/// ([`flip_swept_sent_payments_for_store`]) rightly overrules this +/// confirmation. The confirmation must NOT also run off the EventHandler +/// broadcast (the old payment-hook path): that bus is bounded, lossy, and +/// drained by a detached task with no ordering against this drain, so a +/// stale pre-sweep confirmation could execute after the sweep and durably +/// resurrect a dead payment. +/// +/// Finality is the shared gate ([`sent_payment_status_for_record`]: +/// mined or IS-locked context), plus +/// [`TransactionInstantLocked`](key_wallet_manager::WalletEvent::TransactionInstantLocked), +/// which carries no record — an IS lock is final for DashPay display, so +/// the txid alone confirms. Both `Pending` and `Failed` entries advance: +/// `Failed → Confirmed` is the reinstatement correction, and final +/// evidence delivered on this ordered drain necessarily postdates the +/// sweep that failed the entry (upstream never re-emits final evidence +/// for a txid it still considers dead). Riding the event's own atomic +/// `store()` gives the flip the round's fail-closed machinery — needed +/// because a chainlocked reinstatement can be a one-shot with no later +/// signal to retry from — and keeps store order equal to emission order, +/// so a round carrying this `Confirmed` row can never land after a newer +/// sweep's `Failed` row. The `Pending`-only reconcile pass +/// ([`RECONCILE_CONFIRM_EVIDENCE`]) remains as recovery for confirmations +/// this drain missed (relaunch, lossy start-up). +pub(crate) async fn confirm_final_sent_payments_for_store( + wallet_manager: &Arc>>, + wallet_id: &WalletId, + event: &key_wallet_manager::WalletEvent, +) -> SweptPaymentFlips { + use crate::wallet::identity::types::dashpay::payment::{PaymentDirection, PaymentStatus}; + use key_wallet_manager::WalletEvent; + + // The txids this event proves final. Exhaustive on purpose (no `_` + // arm): a new upstream `WalletEvent` variant that can carry finality + // evidence must fail to compile here rather than be silently dropped. + let final_txids: Vec = match event { + // No record, only a txid — final for DashPay display. + WalletEvent::TransactionInstantLocked { txid, .. } => vec![txid.to_string()], + // First off-chain sighting: final only when the record already + // carries IS-lock context (a plain mempool sighting is not). + WalletEvent::TransactionDetected { record, .. } => std::iter::once(record.as_ref()) + .filter(|r| sent_payment_status_for_record(r) == PaymentStatus::Confirmed) + .map(|r| r.txid.to_string()) + .collect(), + // `inserted` (first stored in this block — including a swept + // transaction's reinstating re-detection) and `updated` + // (previously-known records this block confirmed). `matured` is + // coinbase maturity — never a DashPay payment. + WalletEvent::BlockProcessed { + inserted, updated, .. + } => inserted + .iter() + .chain(updated.iter()) + .filter(|r| sent_payment_status_for_record(r) == PaymentStatus::Confirmed) + .map(|r| r.txid.to_string()) + .collect(), + // The sweep's verdict is the opposite flip + // ([`flip_swept_sent_payments_for_store`]); the rest carry no + // finality evidence for a sent payment. + WalletEvent::TransactionsSwept { .. } + | WalletEvent::SyncHeightAdvanced { .. } + | WalletEvent::ChainLockProcessed { .. } => Vec::new(), + }; + + let mut flips = SweptPaymentFlips::default(); + if final_txids.is_empty() { + return flips; + } + // Fast path under the READ lock: almost every final record names no + // sent payment at all — don't make the drain take the write lock per + // event just to discover that. Missing a concurrent flip is not + // possible: the sweep flip and the rejected-round undo both run on + // this same adapter task, and the write-lock re-check below still + // gates the flip itself, so an entry the reconcile pass confirms in + // the gap is simply found already `Confirmed` and skipped. + { + let wm = wallet_manager.read().await; + let Some(info) = wm.get_wallet_info(wallet_id) else { + return flips; + }; + let any_candidate = final_txids.iter().any(|key| { + info.identity_manager + .identity_ids() + .into_iter() + .any(|owner| { + info.identity_manager + .managed_identity(&owner) + .and_then(|managed| managed.dashpay().payments.get(key)) + .is_some_and(|entry| { + entry.direction == PaymentDirection::Sent + && matches!( + entry.status, + PaymentStatus::Pending | PaymentStatus::Failed + ) + }) + }) + }); + if !any_candidate { + return flips; + } + } + let mut wm = wallet_manager.write().await; + let Some(info) = wm.get_wallet_info_mut(wallet_id) else { + return flips; + }; + for key in &final_txids { + 'owners: for owner in info.identity_manager.identity_ids() { + let Some(managed) = info.identity_manager.managed_identity_mut(&owner) else { + continue; + }; + let previous = match managed.dashpay().payments.get(key) { + Some(entry) + if entry.direction == PaymentDirection::Sent + && sent_status_transition_allowed( + entry.status, + PaymentStatus::Confirmed, + ) => + { + entry.clone() + } + _ => continue, + }; + let mut updated = previous.clone(); + updated.status = PaymentStatus::Confirmed; + tracing::info!( + owner = %owner, + txid = %key, + previous_status = ?previous.status, + "Confirming sent DashPay payment on its event's own store round" + ); + managed + .dashpay_payments_mut() + .insert(key.clone(), updated.clone()); + flips.flips.push(PaymentFlip { + owner, + txid: key.clone(), + previous, + updated, + }); + // txid is unique — only one identity can hold this entry. + break 'owners; + } + } + flips +} + +/// Undo round-riding in-memory payment flips — after the round they rode +/// was rejected, or after the same fold reinstated their transaction and +/// the adapter retracted the staged overlay row. Memory returns to the +/// durable state (the entry the rejection left untouched in the store), +/// which is what lets the replayed event — re-emitted by the re-scan, +/// because the rejected round kept its rows too — find the entries +/// eligible and recompute the flip. Without this, memory would read ahead +/// of the store, the replay's eligibility check would skip the entries, +/// and the store would never learn. +/// +/// An undo is NOT a forward transition, so it does not go through +/// [`sent_status_transition_allowed`] — but it obeys the same authority: +/// it may only revert the flip's own still-standing write +/// ([`PaymentFlipUndo::wrote`]). An entry that moved on — a later flip in +/// the same fold overwriting this one, or the reconcile pass confirming a +/// still-`Pending` entry concurrently — outranks the undo; restoring the +/// captured state over what another writer's round may already hold +/// durably would demote a verdict this undo has no authority over. +pub(crate) async fn rollback_payment_flips( + wallet_manager: &Arc>>, + wallet_id: &WalletId, + rollback: Vec, ) { - confirm_sent_payment_by_txid(wallet_manager, wallet_id, persister, &txid.to_string()).await; + if rollback.is_empty() { + return; + } + let mut wm = wallet_manager.write().await; + let Some(info) = wm.get_wallet_info_mut(wallet_id) else { + return; + }; + for undo in rollback { + let Some(managed) = info.identity_manager.managed_identity_mut(&undo.owner) else { + continue; + }; + let payments = managed.dashpay_payments_mut(); + match payments.get(&undo.txid) { + Some(current) if current.status == undo.wrote => { + payments.insert(undo.txid, undo.previous); + } + _ => {} + } + } } -/// Flip the `Pending` `Sent` [`PaymentEntry`] under `txid` (if any) to -/// `Confirmed`, in place, preserving amount/memo/counterparty. +/// The sent-payment state machine's one transition table, shared by every +/// writer so the confirm path and the sweep flip can never drift. Every +/// legal edge is enumerated explicitly — no wildcard — so adding a +/// `PaymentStatus` variant forces a review of this machine instead of +/// silently admitting new transitions, and a same-state "move" is never a +/// transition (that no-op is what lets re-delivered evidence skip the +/// persistence round). /// -/// No-op when no entry exists for `txid`, it is not a `Sent` entry, or it -/// is already past `Pending` (so repeated confirmed re-detections are -/// idempotent and skip the persistence round). Separated from the event -/// glue above so the state transition is unit-testable without -/// constructing a full `TransactionRecord`. -async fn confirm_sent_payment_by_txid( +/// - `Pending → Confirmed`: the transaction reached finality (mined or +/// IS-locked). +/// - `Pending → Failed`: a sweep proved the transaction can never confirm +/// (beaten to one of its inputs). +/// - `Failed → Confirmed`: the reinstatement correction — the swept +/// transaction re-arrived final. +/// - `Confirmed → Failed`: the eviction correction — upstream permits a +/// chainlocked transaction to evict an IS-locked conflict, so a payment +/// confirmed on IS-lock (or provisional-block) evidence can genuinely +/// die afterwards. This edge is sound ONLY because sent-payment +/// verdicts are applied in emission order on the wallet-event adapter's +/// single drain: when the sweep is applied, every `Confirmed` it can +/// see was written from evidence the sweep postdates (the adapter's own +/// earlier events, or a reconcile pass's read of a record state the +/// sweep is about to delete), and no sweep can follow the evidence that +/// would make `Confirmed` truly final, because upstream never sweeps a +/// confirmed record. Its only legitimate writer is the adapter's sweep +/// flip ([`flip_swept_sent_payments_for_store`]); any new writer with +/// `to = Failed` must run on that same ordered drain. +pub(crate) fn sent_status_transition_allowed( + from: crate::wallet::identity::types::dashpay::payment::PaymentStatus, + to: crate::wallet::identity::types::dashpay::payment::PaymentStatus, +) -> bool { + use crate::wallet::identity::types::dashpay::payment::PaymentStatus; + matches!( + (from, to), + (PaymentStatus::Pending, PaymentStatus::Confirmed) + | (PaymentStatus::Pending, PaymentStatus::Failed) + | (PaymentStatus::Failed, PaymentStatus::Confirmed) + | (PaymentStatus::Confirmed, PaymentStatus::Failed) + ) +} + +/// The reconcile pass's flip: move the `Sent` [`PaymentEntry`] under +/// `txid` to `to`, in place, preserving amount/memo/counterparty, and +/// persist it through its own store round — memory advance and store +/// under ONE continuous hold of the manager write lock, which is what +/// orders this writer against the adapter's staged rounds (see the +/// commit-site comment in the wallet-event adapter). (Adapter-applied +/// verdicts do NOT come through here — they ride their event's atomic +/// round; see [`SweptPaymentFlips`].) +/// +/// Eligibility is [`sent_status_transition_allowed`] (shared with the +/// adapter's flips so the state machine cannot drift) INTERSECTED with +/// the caller's declared `evidence_from` (see +/// [`RECONCILE_CONFIRM_EVIDENCE`]); every ineligible combination is a +/// no-op, which is what keeps repeated passes idempotent, skipping the +/// persistence round, and a stale reconcile snapshot unable to overrule +/// a sweep verdict it never saw. Separated from the sweep glue so the +/// transition is unit-testable without constructing a full +/// `TransactionRecord`. +/// +/// A no-op resolution (no entry, not `Sent`, not eligible) is `Ok(())`; +/// `Err` means the flip was found, attempted, and its store rejected — the +/// in-memory overwrite has already been rolled back +/// (`record_dashpay_payment`'s contract), so the caller may retry or +/// accept per its own signal model (a confirmation is re-driven by every +/// later signal for its transaction). +async fn resolve_sent_payment_by_txid( wallet_manager: &Arc>>, wallet_id: &WalletId, persister: &crate::wallet::persister::WalletPersister, txid: &str, -) { - use crate::wallet::identity::types::dashpay::payment::{PaymentDirection, PaymentStatus}; + to: crate::wallet::identity::types::dashpay::payment::PaymentStatus, + evidence_from: &[crate::wallet::identity::types::dashpay::payment::PaymentStatus], +) -> Result<(), crate::changeset::PersistenceError> { + use crate::wallet::identity::types::dashpay::payment::PaymentDirection; let mut wm = wallet_manager.write().await; let Some(info) = wm.get_wallet_info_mut(wallet_id) else { - return; + return Ok(()); }; // The sent transaction belongs to one managed identity; find the - // `Pending` `Sent` entry under this txid and confirm it in place. + // eligible `Sent` entry under this txid and resolve it in place. for owner in info.identity_manager.identity_ids() { let Some(managed) = info.identity_manager.managed_identity_mut(&owner) else { continue; }; - let confirmed = match managed.dashpay().payments.get(txid) { - Some(entry) - if entry.direction == PaymentDirection::Sent - && entry.status == PaymentStatus::Pending => - { + let resolved = match managed.dashpay().payments.get(txid) { + Some(entry) if entry.direction == PaymentDirection::Sent => { + // Both gates, deliberately: the table says the machine + // permits the move, `evidence_from` says this caller's + // evidence supports it. The re-check under this write lock + // is what turns a caller's stale snapshot into a safe + // no-op — an entry that moved outside the declared set + // means the evidence predates another writer's verdict. + if !evidence_from.contains(&entry.status) + || !sent_status_transition_allowed(entry.status, to) + { + continue; + } let mut updated = entry.clone(); - updated.status = PaymentStatus::Confirmed; + updated.status = to; updated } _ => continue, }; - tracing::info!(owner = %owner, %txid, "Confirming sent DashPay payment"); - if let Err(e) = managed.record_dashpay_payment(txid.to_string(), confirmed, persister) { - tracing::warn!( - error = %e, - "Failed to persist sent-payment confirmation; will retry on next detection" - ); - } - // txid is unique — only one identity can hold this entry. - break; + tracing::info!(owner = %owner, %txid, status = ?to, "Resolving sent DashPay payment"); + // txid is unique — only one identity can hold this entry, so the + // first eligible hit decides the call's result either way. + return managed.record_dashpay_payment(txid.to_string(), resolved, persister); } + Ok(()) } // --------------------------------------------------------------------------- @@ -1519,6 +1972,9 @@ mod tests { #[derive(Default)] struct RecordingPersister { stores: Mutex>, + /// Fail the next N `store` calls with an injected backend error + /// before recording resumes — the shape of a transient rejection. + fail_next_stores: Mutex, } impl PlatformWalletPersistence for RecordingPersister { @@ -1527,6 +1983,13 @@ mod tests { wallet_id: WalletId, changeset: PlatformWalletChangeSet, ) -> Result<(), PersistenceError> { + { + let mut budget = self.fail_next_stores.lock().unwrap(); + if *budget > 0 { + *budget -= 1; + return Err(PersistenceError::backend("injected store failure")); + } + } self.stores.lock().unwrap().push((wallet_id, changeset)); Ok(()) } @@ -2801,13 +3264,20 @@ mod tests { } /// A `Sent` payment must advance `Pending → Confirmed` once its - /// transaction confirms on-chain. `send_payment` records it `Pending` - /// and nothing else moved it, so before the confirm path was wired the - /// entry was stuck `Pending` forever (sent payments never showed - /// confirmed). Pins the flip, idempotency on re-detection, and that - /// amount/memo are preserved. + /// transaction reaches finality. `send_payment` records it `Pending` + /// and only the wallet-event adapter's ordered confirm + /// ([`confirm_final_sent_payments_for_store`]) moves it live, so + /// before that path was wired the entry was stuck `Pending` forever + /// (sent payments never showed confirmed). Pins the flip riding the + /// event's own round (the `previous`/`updated` pair the adapter + /// journals), idempotency on re-delivery, and that amount/memo are + /// preserved. #[tokio::test] async fn confirm_flips_sent_payment_pending_to_confirmed() { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::WalletCoreBalance; + use key_wallet_manager::WalletEvent; + use crate::wallet::identity::types::dashpay::payment::{ PaymentDirection, PaymentEntry, PaymentStatus, }; @@ -2815,7 +3285,8 @@ mod tests { let (manager, persister, wallet_id) = make_wallet().await; let owner = Identifier::from([0xAA; 32]); let contact = Identifier::from([0xBB; 32]); - let txid = "a".repeat(64); + let txid = dashcore::Txid::from([0x5a; 32]); + let txid_key = txid.to_string(); let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); let iw = wallet.identity(); @@ -2831,7 +3302,7 @@ mod tests { .managed_identity_mut(&owner) .expect("managed") .record_dashpay_payment( - txid.clone(), + txid_key.clone(), PaymentEntry::new_sent(contact, 50_000, Some("dinner".into())), &p, ) @@ -2858,48 +3329,528 @@ mod tests { } assert_eq!( - read_entry(iw, &wallet_id, &owner, &txid).await.status, + read_entry(iw, &wallet_id, &owner, &txid_key).await.status, PaymentStatus::Pending, "precondition: entry starts Pending" ); - // A confirmed detection flips it to Confirmed, preserving fields. - super::confirm_sent_payment_by_txid(&iw.wallet_manager, &wallet_id, &p, &txid).await; - let entry = read_entry(iw, &wallet_id, &owner, &txid).await; + // The IS-lock event (final for DashPay display) flips it to + // Confirmed, staging the row for the event's own store round. + let event = WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: std::collections::BTreeMap::new(), + }; + let flips = + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &event) + .await; + assert_eq!(flips.flips.len(), 1); + assert_eq!(flips.flips[0].owner, owner); + assert_eq!(flips.flips[0].txid, txid_key); + assert_eq!( + flips.flips[0].updated.status, + PaymentStatus::Confirmed, + "the Confirmed row must ride the event's own round" + ); + assert_eq!( + flips.flips[0].previous.status, + PaymentStatus::Pending, + "the flip records the pre-flip entry its undo may restore" + ); + let entry = read_entry(iw, &wallet_id, &owner, &txid_key).await; assert_eq!( entry.status, PaymentStatus::Confirmed, - "a confirmed tx must flip the Sent entry to Confirmed" + "a final tx must flip the Sent entry to Confirmed" ); assert_eq!(entry.direction, PaymentDirection::Sent); assert_eq!(entry.amount_duffs, 50_000); assert_eq!(entry.memo.as_deref(), Some("dinner"), "memo preserved"); - // Idempotent: a second confirmed re-detection changes nothing. - super::confirm_sent_payment_by_txid(&iw.wallet_manager, &wallet_id, &p, &txid).await; + // Idempotent: re-delivered finality evidence changes nothing and + // stages nothing (the same-state move is not a transition). + let flips = + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &event) + .await; + assert!( + flips.is_empty(), + "re-delivered evidence must skip the persistence round" + ); + assert_eq!( + read_entry(iw, &wallet_id, &owner, &txid_key).await.status, + PaymentStatus::Confirmed + ); + } + + /// A sweep naming a `Pending` sent payment's transaction must fail the + /// entry: the transaction was provably beaten to one of its inputs and + /// can never confirm, and the same sweep deletes the record that was + /// the last thing reconciliation could have resolved the entry from — + /// so without this transition the sender's payment sat `Pending` + /// forever with no terminal state. Driven through the flip the + /// wallet-event adapter stages onto the sweep's own store round. Also + /// pins the guard rails: a re-emitted sweep on an already-`Failed` + /// entry is an idempotent no-op; the reinstatement reversal — the + /// swept transaction re-arriving final — advances `Failed` to + /// `Confirmed` on the reinstating event's round; and a sweep + /// delivered AFTER a confirmation demotes it (`Confirmed → Failed`): + /// on the adapter's ordered drain a delivered sweep always postdates + /// every applied confirmation — upstream never sweeps a currently + /// final record — so the newer verdict must win. + #[tokio::test] + async fn swept_sent_payment_fails_and_a_reinstating_confirmation_recovers_it() { + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + + let (manager, persister, wallet_id) = make_wallet().await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xAB; 32]); + let txid_key = txid.to_string(); + + let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); + let iw = wallet.identity(); + let p = WalletPersister::new(wallet_id, Arc::clone(&persister) as _); + { + let mut wm = iw.wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("info"); + info.identity_manager + .add_identity(bare_identity([0xAA; 32]), 0, wallet_id, &p) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid_key.clone(), + PaymentEntry::new_sent(contact, 50_000, Some("dinner".into())), + &p, + ) + .expect("record pending sent"); + } + + async fn status( + iw: &crate::wallet::identity::IdentityWallet, + wallet_id: &WalletId, + owner: &Identifier, + txid: &str, + ) -> PaymentStatus { + let wm = iw.wallet_manager.read().await; + let info = wm.get_wallet_info(wallet_id).expect("info"); + info.identity_manager + .managed_identity(owner) + .unwrap() + .dashpay() + .payments + .get(txid) + .expect("entry") + .status + } + + // The sweep's flip: memory moves to Failed and the overlay carries + // exactly the flipped row for the sweep's own store round. + let flips = + super::flip_swept_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &[txid]) + .await; + assert_eq!(flips.flips.len(), 1); + assert_eq!(flips.flips[0].owner, owner); + assert_eq!(flips.flips[0].txid, txid_key); + assert_eq!( + flips.flips[0].updated.status, + PaymentStatus::Failed, + "the flip must carry the Failed row for the sweep's own round" + ); + assert_eq!( + status(iw, &wallet_id, &owner, &txid_key).await, + PaymentStatus::Failed, + "a swept transaction can never confirm — its sent payment must fail" + ); + + // Re-emitted sweep: idempotent no-op (nothing eligible, empty flip). + let flips = + super::flip_swept_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &[txid]) + .await; + assert!( + flips.is_empty(), + "a re-emitted sweep must find nothing to flip" + ); + assert_eq!( + status(iw, &wallet_id, &owner, &txid_key).await, + PaymentStatus::Failed + ); + + // The reinstatement: the swept transaction re-arrives final (here + // as its IS-lock event); the ordered confirm must correct the + // `Failed` verdict on the reinstating event's own round. + let reinstating = { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::WalletCoreBalance; + use key_wallet_manager::WalletEvent; + WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: std::collections::BTreeMap::new(), + } + }; + let flips = super::confirm_final_sent_payments_for_store( + &iw.wallet_manager, + &wallet_id, + &reinstating, + ) + .await; + assert_eq!( + flips.flips[0].updated.status, + PaymentStatus::Confirmed, + "the reinstatement correction must ride the event's own round" + ); + assert_eq!( + status(iw, &wallet_id, &owner, &txid_key).await, + PaymentStatus::Confirmed, + "a reinstated, final transaction must recover the payment" + ); + + // A sweep delivered AFTER the confirmation demotes it: on the + // ordered drain the sweep postdates the applied confirmation, and + // upstream never sweeps a currently final record — so this sweep + // is genuinely newer truth (the reinstated transaction lost a new + // conflict) and must win. + let flips = + super::flip_swept_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &[txid]) + .await; assert_eq!( - read_entry(iw, &wallet_id, &owner, &txid).await.status, + flips.flips[0].updated.status, + PaymentStatus::Failed, + "a newer sweep must demote a Confirmed entry — its Failed row rides its round" + ); + assert_eq!( + status(iw, &wallet_id, &owner, &txid_key).await, + PaymentStatus::Failed, + "the newer sweep's verdict wins over the older confirmation" + ); + } + + /// The transition table enumerates its legal edges explicitly — every + /// same-state pair is a no-op (that skip is what keeps re-delivered + /// evidence off the persistence round), `Failed → Pending` and + /// `Confirmed → Pending` do not exist, and the four legal edges are + /// exactly the machine's verdicts. A wildcard `(Pending, _)` arm — the + /// shape this replaces — silently admits `Pending → Pending` and any + /// future `PaymentStatus` variant without a state-machine review. + #[test] + fn sent_status_transition_table_enumerates_exactly_the_legal_edges() { + use crate::wallet::identity::types::dashpay::payment::PaymentStatus::*; + + let allowed = [ + (Pending, Confirmed), + (Pending, Failed), + (Failed, Confirmed), + (Confirmed, Failed), + ]; + for from in [Pending, Confirmed, Failed] { + for to in [Pending, Confirmed, Failed] { + assert_eq!( + super::sent_status_transition_allowed(from, to), + allowed.contains(&(from, to)), + "transition {from:?} -> {to:?} must be {}", + if allowed.contains(&(from, to)) { + "allowed" + } else { + "rejected — a same-state move is not a transition, and \ + nothing returns to Pending" + } + ); + } + } + } + + /// The rejected-round contract: rolling the flip back returns memory to + /// the durable state (`Pending`), which is exactly what lets the + /// replayed sweep — re-emitted by the re-scan, because the rejected + /// round kept the loser's record too — find the entry eligible and + /// recompute the flip. Without the rollback, memory would read `Failed` + /// ahead of the store and the replay's eligibility check would skip the + /// entry forever. The adapter-level rejection wiring is pinned + /// end to end in `core_bridge`. + #[tokio::test] + async fn a_rolled_back_flip_is_recomputed_by_the_replayed_sweep() { + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + + let (manager, persister, wallet_id) = make_wallet().await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xAC; 32]); + let txid_key = txid.to_string(); + + let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); + let iw = wallet.identity(); + let p = WalletPersister::new(wallet_id, Arc::clone(&persister) as _); + { + let mut wm = iw.wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("info"); + info.identity_manager + .add_identity(bare_identity([0xAA; 32]), 0, wallet_id, &p) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid_key.clone(), + PaymentEntry::new_sent(contact, 10_000, None), + &p, + ) + .expect("record pending sent"); + } + + let flips = + super::flip_swept_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &[txid]) + .await; + assert!(!flips.is_empty()); + super::rollback_payment_flips(&iw.wallet_manager, &wallet_id, flips.into_undos()).await; + + { + let wm = iw.wallet_manager.read().await; + let info = wm.get_wallet_info(&wallet_id).expect("info"); + let entry = info + .identity_manager + .managed_identity(&owner) + .unwrap() + .dashpay() + .payments + .get(&txid_key) + .expect("entry") + .clone(); + assert_eq!( + entry.status, + PaymentStatus::Pending, + "the rollback must return memory to the durable state" + ); + assert_eq!( + entry.amount_duffs, 10_000, + "the previous entry is restored whole" + ); + } + + // The replayed sweep finds the entry eligible again and recomputes + // the flip — the durability loop closes. + let flips = + super::flip_swept_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &[txid]) + .await; + assert_eq!( + flips.flips[0].updated.status, + PaymentStatus::Failed, + "the replayed sweep must recompute the flip the rollback undid" + ); + } + + /// The stale-evidence race, frozen at its worst point: the reconcile + /// sweep snapshots an entry as `Pending` and reads its persisted + /// record, the adapter's sweep flip lands `Failed` mid-flight (the + /// reconcile pass runs on its own task, off the adapter's ordered + /// drain, and the sweep deletes the record on the adapter), and the + /// reconciler then confirms from evidence that predates the verdict — + /// landing a dead payment `Confirmed` against a newer sweep, with no + /// later sweep re-emission to repair it. The reconciler's declared + /// evidence (`RECONCILE_CONFIRM_EVIDENCE`, `Pending`-only) makes the + /// resolver's write-lock re-check turn exactly that into a no-op, + /// while reinstatement evidence applied on the ordered drain — which + /// postdates the sweep by delivery order — still recovers the entry. + #[tokio::test] + async fn a_stale_reconcile_snapshot_cannot_confirm_a_swept_payment() { + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + + let (manager, persister, wallet_id) = make_wallet().await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xAF; 32]); + let txid_key = txid.to_string(); + + let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); + let iw = wallet.identity(); + let p = WalletPersister::new(wallet_id, Arc::clone(&persister) as _); + { + let mut wm = iw.wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("info"); + info.identity_manager + .add_identity(bare_identity([0xAA; 32]), 0, wallet_id, &p) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid_key.clone(), + PaymentEntry::new_sent(contact, 50_000, None), + &p, + ) + .expect("record pending sent"); + } + + async fn status( + iw: &crate::wallet::identity::IdentityWallet, + wallet_id: &WalletId, + owner: &Identifier, + txid: &str, + ) -> PaymentStatus { + let wm = iw.wallet_manager.read().await; + let info = wm.get_wallet_info(wallet_id).expect("info"); + info.identity_manager + .managed_identity(owner) + .unwrap() + .dashpay() + .payments + .get(txid) + .expect("entry") + .status + } + + // The reconciler snapshotted the entry Pending; before it confirms, + // the sweep's verdict lands. + let flips = + super::flip_swept_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &[txid]) + .await; + assert!(!flips.is_empty()); + assert_eq!( + status(iw, &wallet_id, &owner, &txid_key).await, + PaymentStatus::Failed + ); + + // The racing reconciler now confirms from its stale read, declaring + // exactly the evidence the production sweep declares. + super::confirm_sent_payment_by_txid( + &iw.wallet_manager, + &wallet_id, + &p, + &txid_key, + super::RECONCILE_CONFIRM_EVIDENCE, + ) + .await; + assert_eq!( + status(iw, &wallet_id, &owner, &txid_key).await, + PaymentStatus::Failed, + "evidence read before the sweep's verdict must not confirm the dead payment" + ); + + // Reinstatement evidence applied on the ordered drain — which + // postdates the sweep by delivery order — still recovers the + // entry. + let reinstating = { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::WalletCoreBalance; + use key_wallet_manager::WalletEvent; + WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: std::collections::BTreeMap::new(), + } + }; + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &reinstating) + .await; + assert_eq!( + status(iw, &wallet_id, &owner, &txid_key).await, PaymentStatus::Confirmed ); } + /// The rollback may only revert the sweep flip's own still-standing + /// `Failed` write. A later flip can land between the sweep flip and + /// its undo — the same-fold reinstatement's ordered confirm advancing + /// the entry to `Confirmed` before a rejected round replays the + /// ledger. An unconditional restore would clobber it back to the + /// captured `Pending`, demoting a verdict the undo has no authority + /// over. + #[tokio::test] + async fn rollback_does_not_clobber_a_concurrently_confirmed_entry() { + use crate::wallet::identity::types::dashpay::payment::{PaymentEntry, PaymentStatus}; + + let (manager, persister, wallet_id) = make_wallet().await; + let owner = Identifier::from([0xAA; 32]); + let contact = Identifier::from([0xBB; 32]); + let txid = dashcore::Txid::from([0xAE; 32]); + let txid_key = txid.to_string(); + + let wallet = manager.get_wallet(&wallet_id).await.expect("wallet"); + let iw = wallet.identity(); + let p = WalletPersister::new(wallet_id, Arc::clone(&persister) as _); + { + let mut wm = iw.wallet_manager.write().await; + let info = wm.get_wallet_info_mut(&wallet_id).expect("info"); + info.identity_manager + .add_identity(bare_identity([0xAA; 32]), 0, wallet_id, &p) + .expect("add owner"); + info.identity_manager + .managed_identity_mut(&owner) + .expect("managed") + .record_dashpay_payment( + txid_key.clone(), + PaymentEntry::new_sent(contact, 50_000, None), + &p, + ) + .expect("record pending sent"); + } + + let flips = + super::flip_swept_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &[txid]) + .await; + assert!(!flips.is_empty()); + + // The reinstated transaction's ordered confirmation lands before + // the undo — Failed → Confirmed, the table's permitted correction. + let reinstating = { + use dashcore::ephemerealdata::instant_lock::InstantLock; + use key_wallet::WalletCoreBalance; + use key_wallet_manager::WalletEvent; + WalletEvent::TransactionInstantLocked { + wallet_id, + txid, + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: std::collections::BTreeMap::new(), + } + }; + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &reinstating) + .await; + + // The undo arrives late (rejected round or same-fold retraction); + // it must find its own write gone and leave the terminal alone. + super::rollback_payment_flips(&iw.wallet_manager, &wallet_id, flips.into_undos()).await; + + let wm = iw.wallet_manager.read().await; + let status = wm + .get_wallet_info(&wallet_id) + .expect("info") + .identity_manager + .managed_identity(&owner) + .expect("managed") + .dashpay() + .payments + .get(&txid_key) + .expect("entry") + .status; + assert_eq!( + status, + PaymentStatus::Confirmed, + "an undo may only revert the sweep's own still-standing Failed write — \ + never a concurrently confirmed terminal" + ); + } + /// A sent payment confirmed by a block must flip `Pending → Confirmed`. /// /// The wallet sees its *own* broadcast in the mempool first - /// (`TransactionDetected`, context `Mempool`), where the confirm hook - /// early-returns because the transaction is not yet confirmed. The - /// transaction reaches a confirmed context only when a block mines it — - /// delivered as [`key_wallet_manager::WalletEvent::BlockProcessed`] with - /// the record in `updated` (a previously-known record that just - /// confirmed). Routing the payment hooks only for `TransactionDetected` - /// would leave the entry `Pending` forever. This drives the real adapter - /// dispatch - /// ([`run_dashpay_payment_hooks`](crate::wallet::identity::network::run_dashpay_payment_hooks)) - /// with a `BlockProcessed` event and pins the flip end-to-end, so a - /// regression that re-narrows the routing to `TransactionDetected` is - /// caught here. Also pins idempotency across a repeated block-processing - /// round and that the `matured` bucket (coinbase maturity) never - /// confirms a payment. + /// (`TransactionDetected`, context `Mempool`), which carries no + /// finality. The transaction reaches a confirmed context only when a + /// block mines it — delivered as + /// [`key_wallet_manager::WalletEvent::BlockProcessed`] with the record + /// in `updated` (a previously-known record that just confirmed). + /// Routing the ordered confirm + /// ([`confirm_final_sent_payments_for_store`]) only for + /// `TransactionDetected` would leave the entry `Pending` forever; + /// this pins the `BlockProcessed.updated` routing, idempotency across + /// a repeated block-processing round, and that the `matured` bucket + /// (coinbase maturity) never confirms a payment. #[tokio::test] async fn block_processed_confirms_sent_payment() { use dashcore::blockdata::transaction::Transaction; @@ -2991,13 +3942,7 @@ mod tests { addresses_derived: Vec::new(), }; - crate::wallet::identity::network::run_dashpay_payment_hooks( - &iw.wallet_manager, - &wallet_id, - &p, - &event, - ) - .await; + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &event).await; // Read the entry under a short-lived read lock so the re-fire below // can take the write lock. @@ -3029,13 +3974,7 @@ mod tests { // Idempotent: a repeated block-processing round for the same txid // changes nothing (the confirm path skips entries past `Pending`). - crate::wallet::identity::network::run_dashpay_payment_hooks( - &iw.wallet_manager, - &wallet_id, - &p, - &event, - ) - .await; + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &event).await; assert_eq!( read_status(iw, &wallet_id, &owner, &txid.to_string()) .await @@ -3046,7 +3985,7 @@ mod tests { // A confirmed record arriving only in the `matured` bucket (coinbase // maturity) must NOT confirm a payment — `matured` is never a DashPay - // payment, so it is excluded from the payment hooks. + // payment, so the ordered confirm excludes it. let matured_tx = Transaction { version: 2, lock_time: 0, @@ -3102,10 +4041,9 @@ mod tests { account_balances: std::collections::BTreeMap::new(), addresses_derived: Vec::new(), }; - crate::wallet::identity::network::run_dashpay_payment_hooks( + super::confirm_final_sent_payments_for_store( &iw.wallet_manager, &wallet_id, - &p, &matured_event, ) .await; @@ -3121,8 +4059,8 @@ mod tests { /// An InstantSend lock applied to a previously-seen sent payment /// confirms it without waiting for a block. The lock arrives as /// `WalletEvent::TransactionInstantLocked` (no record, just a txid); an - /// IS lock is final for DashPay display, so the entry flips - /// `Pending → Confirmed`. Drives the real adapter dispatch. + /// IS lock is final for DashPay display, so the ordered confirm flips + /// the entry `Pending → Confirmed`. #[tokio::test] async fn instant_send_lock_confirms_sent_payment() { use dashcore::ephemerealdata::instant_lock::InstantLock; @@ -3163,13 +4101,7 @@ mod tests { balance: WalletCoreBalance::default(), account_balances: std::collections::BTreeMap::new(), }; - crate::wallet::identity::network::run_dashpay_payment_hooks( - &iw.wallet_manager, - &wallet_id, - &p, - &event, - ) - .await; + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &event).await; let wm = iw.wallet_manager.read().await; let info = wm.get_wallet_info(&wallet_id).expect("info"); @@ -3269,13 +4201,7 @@ mod tests { account_balances: std::collections::BTreeMap::new(), addresses_derived: Vec::new(), }; - crate::wallet::identity::network::run_dashpay_payment_hooks( - &iw.wallet_manager, - &wallet_id, - &p, - &event, - ) - .await; + super::confirm_final_sent_payments_for_store(&iw.wallet_manager, &wallet_id, &event).await; let wm = iw.wallet_manager.read().await; let info = wm.get_wallet_info(&wallet_id).expect("info");