From d166f896c00cef4616fc17d6eb6ccff654f520fd Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Thu, 10 Sep 2026 17:21:33 +0300 Subject: [PATCH 1/2] fix(platform-wallet): resolve a swept sent payment's verdict on the round that swept it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing in the wallet ever wrote `PaymentStatus::Failed` — the only two mentions were the FFI mappings. So a DashPay sent payment whose transaction lost a double-spend stayed `Pending` for good, and one already `Confirmed` whose InstantSend-locked transaction was then evicted by a chainlocked winner stayed `Confirmed`: a dead payment reported as good, with nothing able to correct either later. The reconcile pass cannot repair them because it resolves against the stored record, and the sweep has already deleted it. The verdict belongs to the wallet-event adapter rather than the payment hooks. The hooks run off dash-spv's lossy broadcast, so a sweep dropped under `RecvError::Lagged` during catch-up is gone with no ground truth to rebuild it from; the adapter drains the lossless persistence channel, in emission order, and already owns the round the sweep's removal commits on. Riding that round is also what keeps a verdict from being written against a removal that did not land. Two evidence classes drive one explicit transition table — `Swept` from `TransactionsSwept`, `Final` from an InstantSend lock or a record reaching a final context: Pending + Swept -> Failed nothing else writes Failed Confirmed + Swept -> Failed an evicted IS-locked payment is dead Pending + Final -> Confirmed the ordinary confirm Failed + Final -> Confirmed a chainlocked reinstatement repairs it The other two pairs are already-reached verdicts and emit no row. The match is exhaustive on the pair rather than closed with a wildcard, so a future `PaymentStatus` variant fails the build instead of silently taking an edge. The overlay rides the wallet's own `store()` through the existing last-write-wins merge, now shared with the adapter, so a transaction swept and then reinstated inside one drain reaches the store as the verdict the drain ended on. There is deliberately no round journal and no rollback ledger: a rejected round leaves the loser row and `Pending` in the store, and the next launch reloads memory from the store and re-emits the sweep from the frozen watermark. An in-session re-emit is not reachable — upstream selects losers from live in-memory records and deletes them in the same call — so there is no state to unwind that anything observes. Withheld from a host that does not attest `DASHPAY_PAYMENTS`, with a warning naming the wallet. Unlike a withheld sweep removal this does not freeze the watermark: a verdict is derived state that a host shipping the slot later re-derives, while a dropped removal has no such recovery. The payment hooks keep incoming payments only — idempotent inserts with no state machine to race — and their two sent-payment confirm paths are gone, each case now covered by the adapter's `Final` evidence. --- .../src/changeset/changeset.rs | 30 +- .../src/changeset/core_bridge.rs | 898 +++++++++++++++++- .../src/wallet/identity/network/mod.rs | 5 +- .../identity/network/payment_handler.rs | 162 ++-- .../src/wallet/identity/network/payments.rs | 162 ++-- 5 files changed, 1044 insertions(+), 213 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/changeset.rs b/packages/rs-platform-wallet/src/changeset/changeset.rs index f2f1b562330..09d17c948a0 100644 --- a/packages/rs-platform-wallet/src/changeset/changeset.rs +++ b/packages/rs-platform-wallet/src/changeset/changeset.rs @@ -1955,6 +1955,25 @@ pub fn upsert_pending_contact_crypto( // Top-Level PlatformWalletChangeSet // --------------------------------------------------------------------------- +/// DashPay payment rows keyed by owning identity, then by transaction id — +/// the shape [`PlatformWalletChangeSet::dashpay_payments_overlay`] carries. +pub(crate) type PaymentOverlay = BTreeMap>; + +/// Fold `other` into `target` with last-write-wins per `(owner, txid)`. +/// +/// The overlay is a set of whole rows, so the later write of a row is the +/// whole answer for it — there is nothing in an earlier row worth keeping. +/// Shared with the wallet-event adapter, which folds a drain's sent-payment +/// verdicts across events before they reach a changeset: coalescing per +/// `(owner, txid)` is what lets a transaction that is swept and then +/// reinstated inside one drain reach the store as the single verdict the +/// drain ended on, rather than as two rows the persister must order. +pub(crate) fn merge_payment_overlays(target: &mut PaymentOverlay, other: PaymentOverlay) { + for (id, payments) in other { + target.entry(id).or_default().extend(payments); + } +} + /// Delta of all wallet state changes from a single operation. /// /// `core` carries a [`CoreChangeSet`] — the platform-owned projection of @@ -2149,12 +2168,11 @@ impl Merge for PlatformWalletChangeSet { .extend(other_profiles); } if let Some(other_payments) = other.dashpay_payments_overlay { - let target = self - .dashpay_payments_overlay - .get_or_insert_with(Default::default); - for (id, payments) in other_payments { - target.entry(id).or_default().extend(payments); - } + merge_payment_overlays( + self.dashpay_payments_overlay + .get_or_insert_with(Default::default), + other_payments, + ); } // Wallet metadata: last-write-wins. `Network` doesn't // implement `Default`, so we can't lean on the `Option: diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 9b20a1658dc..73aeed927f7 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -51,12 +51,15 @@ use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use crate::changeset::changeset::{ - AssetLockChangeSet, CoreChangeSet, HighestUsedIndexes, PlatformWalletChangeSet, SweepBatch, + merge_payment_overlays, AssetLockChangeSet, CoreChangeSet, HighestUsedIndexes, PaymentOverlay, + PlatformWalletChangeSet, SweepBatch, }; use crate::changeset::merge::Merge; use crate::changeset::persistence_capabilities::PersistenceCapabilities; use crate::changeset::traits::PlatformWalletPersistence; use crate::wallet::asset_lock::sync::reconstruction; +use crate::wallet::identity::network::sent_payment_status_for_record; +use crate::wallet::identity::types::dashpay::payment::{PaymentDirection, PaymentStatus}; use crate::wallet::platform_wallet::PlatformWalletInfo; /// Maximum number of `WalletEvent`s folded into a single @@ -418,9 +421,11 @@ 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 payments = sent_payment_verdicts(&wallet_manager, &event).await; let entry = batch.entry(wallet_id).or_default(); entry.core.merge(core); entry.asset_locks.merge(asset_locks); + merge_payment_overlays(&mut entry.payments, payments); } // Fold in whatever else is already buffered. `try_recv` never waits, @@ -434,9 +439,14 @@ 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 payments = sent_payment_verdicts(&wallet_manager, &event).await; let entry = batch.entry(wallet_id).or_default(); entry.core.merge(core); entry.asset_locks.merge(asset_locks); + // Last-write-wins per `(owner, txid)`: a transaction swept + // and then reinstated inside one drain reaches the store as + // the verdict the drain ended on, never as two rows. + merge_payment_overlays(&mut entry.payments, payments); folded += 1; } Err(TryRecvError::Empty) => break, @@ -492,6 +502,7 @@ async fn run_wallet_event_adapter

( .filter(|(_, wallet_batch)| { !wallet_batch.core.is_empty_no_records() || !Merge::is_empty(&wallet_batch.asset_locks) + || !wallet_batch.payments.is_empty() }) .map(|(wallet_id, _)| *wallet_id) .collect(); @@ -708,8 +719,35 @@ fn commit_wallet

( let WalletBatch { mut core, asset_locks, + payments, } = wallet_batch; { + // Sent-payment verdicts reach a host only through the payment-overlay + // slot, which a persister advertises with `DASHPAY_PAYMENTS`. A host + // without it would take the round, return `Ok`, and drop the verdict + // on the floor — so withhold it and say so, once per round that had + // one. Unlike a withheld sweep this does not freeze the watermark: + // the verdict is derived state, so a host that later ships the slot + // re-derives it from the records and the reconcile pass, whereas a + // dropped removal has no such recovery. + let payments = if payments.is_empty() { + None + } else if persister + .persistence_capabilities() + .contains(PersistenceCapabilities::DASHPAY_PAYMENTS) + { + Some(payments) + } else { + tracing::warn!( + wallet_id = %hex::encode(wallet_id), + identities = payments.len(), + rows = payments.values().map(BTreeMap::len).sum::(), + "Persister does not advertise DASHPAY_PAYMENTS; withholding this round's \ + sent-payment verdicts. Swept sent payments stay as stored until the host \ + adopts the payment-overlay slot." + ); + None + }; // Hold this wallet's durable watermark at the last fully persisted // height once it has faulted. Records/UTXOs still persist — only the // height advance is suppressed. @@ -727,10 +765,11 @@ 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.is_none() { // 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. + // watermark-only batch stripped by the fault guard above, a verdict + // withheld from a payments-blind persister, etc. — nothing to + // persist. Skip the round-trip. return; } // The height this changeset OFFERS to the store. It is counted as @@ -781,6 +820,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 sent-payment verdicts this drain resolved, on the same round + // as the sweep removal or confirming record that justifies them. + dashpay_payments_overlay: payments, ..PlatformWalletChangeSet::default() }; let store_result = persister.store(wallet_id, cs); @@ -948,6 +990,11 @@ fn freeze_synced_height_if_faulted(core: &mut CoreChangeSet, persistence_faulted struct WalletBatch { core: CoreChangeSet, asset_locks: AssetLockChangeSet, + /// Sent-payment verdicts this drain resolved (see + /// [`sent_payment_verdicts`]). Rides the same `store()` as the rows that + /// justify it — a sweep's removal, or the record that confirmed it — + /// because neither event re-emits once its round is durable. + payments: PaymentOverlay, } /// Rebuild missing tracked asset locks from the records an event @@ -1283,6 +1330,220 @@ async fn build_core_changeset( } } +/// What one drained `WalletEvent` proved about the Core transaction behind a +/// `Sent` DashPay payment. +/// +/// Only two things are ever proven about a broadcast payment, and they are +/// exactly the two terminals a sent entry can reach — which is why the +/// evidence class, not the event variant, is what the transition table below +/// matches on. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SentPaymentEvidence { + /// The transaction lost a double-spend: a final rival claimed one of its + /// inputs, so it can never confirm and the wallet has already dropped its + /// record. + Swept, + /// The transaction reached a context that is final for DashPay display — + /// mined, or InstantSend-locked. + Final, +} + +/// The complete set of legal `Sent` payment transitions, and the only place a +/// status is chosen for a sent entry. +/// +/// | from | evidence | to | why this edge exists | +/// |-------------|----------|-------------|--------------------------------------------------------------------------------------| +/// | `Pending` | `Swept` | `Failed` | The broadcast lost; nothing else ever writes `Failed`, so without it the entry is stuck `Pending` for good. | +/// | `Confirmed` | `Swept` | `Failed` | An IS-locked payment evicted by a chainlocked winner is a dead payment reported as good — the worst of the two. | +/// | `Pending` | `Final` | `Confirmed` | The ordinary confirm: mempool → mined / IS-locked. | +/// | `Failed` | `Final` | `Confirmed` | A chainlock can reinstate a transaction a sweep removed; without this edge the repair is unreachable. | +/// +/// The two remaining pairs — `Failed` + `Swept` and `Confirmed` + `Final` — +/// return `None`: the entry already carries that verdict, so re-emitting it +/// would put an unchanged row on the store round for every re-detection. +/// +/// Exhaustive on purpose (no `_` arm): a future [`PaymentStatus`] variant must +/// fail to compile here and be given an explicit edge, rather than be silently +/// swept into — or excluded from — a verdict. +fn next_sent_payment_status( + current: PaymentStatus, + evidence: SentPaymentEvidence, +) -> Option { + use SentPaymentEvidence::{Final, Swept}; + match (current, evidence) { + (PaymentStatus::Pending, Swept) => Some(PaymentStatus::Failed), + (PaymentStatus::Confirmed, Swept) => Some(PaymentStatus::Failed), + (PaymentStatus::Failed, Swept) => None, + (PaymentStatus::Pending, Final) => Some(PaymentStatus::Confirmed), + (PaymentStatus::Failed, Final) => Some(PaymentStatus::Confirmed), + (PaymentStatus::Confirmed, Final) => None, + } +} + +/// The sent-payment evidence `event` carries, as `(txid, evidence)` pairs +/// keyed the way a [`PaymentEntry`](crate::wallet::identity::PaymentEntry) is +/// — by the transaction id's display string. +/// +/// Exhaustive on purpose: a new upstream `WalletEvent` variant that says +/// something about a broadcast transaction's fate must fail to compile here +/// rather than be silently dropped. +/// +/// `matured` is excluded from `BlockProcessed`: coinbase maturity is never a +/// DashPay payment, and a confirmed record in that bucket says nothing about +/// a sent one. +fn sent_payment_evidence(event: &WalletEvent) -> Vec<(String, SentPaymentEvidence)> { + /// A record is evidence only once its context is final for DashPay — + /// the same definition the reconcile sweep uses, so the live path and + /// the recovery path can never disagree about what "final" means. + fn finality<'a>( + records: impl Iterator, + ) -> Vec<(String, SentPaymentEvidence)> { + records + .filter(|record| sent_payment_status_for_record(record) == PaymentStatus::Confirmed) + .map(|record| (record.txid.to_string(), SentPaymentEvidence::Final)) + .collect() + } + + match event { + WalletEvent::TransactionsSwept { txids, .. } => txids + .iter() + .map(|txid| (txid.to_string(), SentPaymentEvidence::Swept)) + .collect(), + // Carries no record, only a txid — and an InstantSend lock is final + // for DashPay display, so the txid alone is the evidence. + WalletEvent::TransactionInstantLocked { txid, .. } => { + vec![(txid.to_string(), SentPaymentEvidence::Final)] + } + WalletEvent::TransactionDetected { record, .. } => { + finality(std::iter::once(record.as_ref())) + } + WalletEvent::BlockProcessed { + inserted, updated, .. + } => finality(inserted.iter().chain(updated.iter())), + WalletEvent::SyncHeightAdvanced { .. } | WalletEvent::ChainLockProcessed { .. } => { + Vec::new() + } + } +} + +/// Resolve `event`'s sent-payment evidence against the wallet's live payment +/// entries, flip the ones the transition table moves, and return them as a +/// ready overlay for this drain's `store()` round. +/// +/// # Why the adapter owns this and the payment handler does not +/// +/// The handler runs off dash-spv's *lossy* broadcast bus, which drops events +/// under `RecvError::Lagged` during catch-up. A sweep dropped there is +/// unrecoverable: `drop_conflicted_transactions` selects its losers from the +/// live in-memory records and deletes them in the same call, so the sweep +/// never re-emits and the reconcile pass — which resolves against a record +/// that no longer exists — gives up. This adapter drains the *lossless* +/// persistence channel, so the verdict rides the same `store()` as the row +/// removal that implies it. +/// +/// # Failure posture +/// +/// The flip lands in memory here and in the store when the round commits. A +/// rejected round leaves the two disagreeing until the next launch, and that +/// is deliberate: the wallet is faulted and its watermark frozen by the same +/// rejection, so the next launch reloads memory from the store and re-emits +/// the sweep from the frozen watermark, which re-derives the verdict. A +/// rollback ledger would only defend a divergence that cannot outlive the +/// session that caused it. +pub(crate) async fn sent_payment_verdicts( + wallet_manager: &Arc>>, + event: &WalletEvent, +) -> PaymentOverlay { + let mut overlay = PaymentOverlay::new(); + let evidence = sent_payment_evidence(event); + if evidence.is_empty() { + return overlay; + } + let wallet_id = event.wallet_id(); + + // Probe under a read lock first. During a catch-up nearly every block + // carries final records and almost none of them are DashPay payments, so + // this keeps the write lock — which contends with SPV's own wallet + // mutations on the hot path — for rounds that genuinely have a verdict to + // write. + { + let wm = wallet_manager.read().await; + let Some(info) = wm.get_wallet_info(&wallet_id) else { + return overlay; + }; + let any_verdict = info + .identity_manager + .identity_ids() + .into_iter() + .any(|owner| { + info.identity_manager + .managed_identity(&owner) + .is_some_and(|managed| { + evidence.iter().any(|(txid, evidence)| { + managed + .dashpay() + .payments + .get(txid) + .is_some_and(|entry| verdict_for(entry, *evidence).is_some()) + }) + }) + }); + if !any_verdict { + return overlay; + } + } + + let mut wm = wallet_manager.write().await; + let Some(info) = wm.get_wallet_info_mut(&wallet_id) else { + return overlay; + }; + for owner in info.identity_manager.identity_ids() { + let Some(managed) = info.identity_manager.managed_identity_mut(&owner) else { + continue; + }; + // The replay/restore accessor, deliberately: the live + // `record_dashpay_payment` writer persists on its own round, which is + // the one thing this fix exists to avoid. The overlay returned here + // carries the same row onto the adapter's round instead. + let payments = managed.dashpay_payments_mut(); + for (txid, evidence) in &evidence { + let Some(entry) = payments.get_mut(txid) else { + continue; + }; + let Some(next) = verdict_for(entry, *evidence) else { + continue; + }; + tracing::info!( + %owner, + %txid, + from = ?entry.status, + to = ?next, + "Sent DashPay payment verdict" + ); + entry.status = next; + overlay + .entry(owner) + .or_default() + .insert(txid.clone(), entry.clone()); + } + } + overlay +} + +/// The status `entry` moves to under `evidence`, or `None` if it does not +/// move. Received entries never move: their status is settled at the moment +/// they are recorded from an on-chain sighting, and a sweep of an unrelated +/// spend must not touch one. +fn verdict_for( + entry: &crate::wallet::identity::PaymentEntry, + evidence: SentPaymentEvidence, +) -> Option { + if entry.direction != PaymentDirection::Sent { + return None; + } + next_sent_payment_status(entry.status, evidence) +} + /// Rebuild the "addresses marked used" delta plus the post-batch /// highest-used watermarks for the accounts touched by `records`. /// @@ -2010,6 +2271,425 @@ mod swept_transaction_projection_tests { } } +#[cfg(test)] +mod sent_payment_verdict_tests { + //! Coverage for the sent-payment verdicts the adapter owns. + //! + //! Nothing else in the wallet writes `PaymentStatus::Failed`, and the + //! two events that prove one — a sweep, and the finality that can undo + //! it — never re-emit once their round is durable. So these pin both + //! terminals, and that the verdict comes back as a ready overlay for the + //! same `store()` round rather than a separate write. + + use super::*; + use dashcore::ephemerealdata::instant_lock::InstantLock; + use dashcore::hashes::Hash as _; + use dashcore::{BlockHash, Transaction, TxIn, Txid}; + use dpp::identity::{Identity, IdentityV0}; + use dpp::prelude::Identifier; + use key_wallet::account::account_type::StandardAccountType; + use key_wallet::managed_account::transaction_record::TransactionDirection; + use key_wallet::transaction_checking::{BlockInfo, TransactionType}; + use key_wallet::WalletCoreBalance; + + use crate::changeset::traits::PlatformWalletPersistence; + use crate::test_support::{funded_wallet_manager, NoopTestPersister}; + use crate::wallet::identity::{PaymentDirection, PaymentEntry, PaymentStatus}; + use crate::wallet::persister::WalletPersister; + + const OWNER: [u8; 32] = [0xAA; 32]; + const CONTACT: [u8; 32] = [0xBB; 32]; + + fn owner() -> Identifier { + Identifier::from(OWNER) + } + + /// The spend whose payment entry every test below flips. A real + /// `Transaction` rather than a bare txid, so the record used as finality + /// evidence and the entry agree on the same key. + fn sent_transaction() -> Transaction { + Transaction { + version: 2, + lock_time: 0, + input: vec![TxIn { + previous_output: OutPoint::new(Txid::from_byte_array([0x5f; 32]), 0), + ..Default::default() + }], + output: Vec::new(), + special_transaction_payload: None, + } + } + + /// A wallet holding one identity with a single `Sent` payment at + /// `status`, keyed by `sent_transaction()`'s txid. + async fn wallet_with_payment( + direction: PaymentDirection, + status: PaymentStatus, + ) -> ( + Arc>>, + WalletId, + String, + ) { + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + let txid = sent_transaction().txid().to_string(); + let persister = 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: BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &persister, + ) + .expect("add owner identity"); + let mut entry = match direction { + PaymentDirection::Sent => { + PaymentEntry::new_sent(Identifier::from(CONTACT), 50_000, Some("lunch".into())) + } + PaymentDirection::Received => PaymentEntry::new_received( + Identifier::from(CONTACT), + 50_000, + Some("lunch".into()), + ), + }; + entry.status = status; + // The replay accessor: seeding through the live writer would run + // its own persist round, which is the very thing under test. + info.identity_manager + .managed_identity_mut(&owner()) + .expect("managed identity") + .dashpay_payments_mut() + .insert(txid.clone(), entry); + } + (wallet_manager, wallet_id, txid) + } + + async fn stored_status( + wallet_manager: &Arc>>, + wallet_id: &WalletId, + txid: &str, + ) -> PaymentStatus { + let wm = wallet_manager.read().await; + wm.get_wallet_info(wallet_id) + .expect("wallet info") + .identity_manager + .managed_identity(&owner()) + .expect("managed identity") + .dashpay() + .payments + .get(txid) + .expect("entry under the sent txid") + .status + } + + fn sweep_of(wallet_id: WalletId, txid: Txid) -> WalletEvent { + WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![txid], + superseded_by: Txid::from_byte_array([0x77; 32]), + winner_mined_height: Some(1_499_050), + released_outpoints: Vec::new(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + } + } + + /// A `BlockProcessed` re-emitting the spend as chain-locked — the + /// evidence that reinstates a transaction a sweep removed. + fn chainlocked_reinstatement(wallet_id: WalletId) -> WalletEvent { + let record = TransactionRecord::new( + sent_transaction(), + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InChainLockedBlock(BlockInfo::new( + 1_499_060, + BlockHash::all_zeros(), + 0, + )), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + -50_000, + ); + WalletEvent::BlockProcessed { + wallet_id, + height: 1_499_060, + chain_lock: None, + inserted: Vec::new(), + updated: vec![record], + matured: Vec::new(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + addresses_derived: Vec::new(), + } + } + + /// One row, so a test can assert the overlay is exactly the verdict and + /// not a replay of the identity's whole payment history. + fn only_row(overlay: &PaymentOverlay, txid: &str) -> PaymentEntry { + assert_eq!(overlay.len(), 1, "exactly one identity: {overlay:?}"); + let rows = overlay.get(&owner()).expect("the owning identity"); + assert_eq!(rows.len(), 1, "exactly one row: {rows:?}"); + rows.get(txid).expect("the flipped row").clone() + } + + /// The defect: a swept sent payment stayed `Pending` forever because + /// nothing in the wallet ever wrote `Failed`. The sweep is the only + /// evidence that exists — the wallet has already deleted the loser's + /// record — so the verdict has to be taken here or not at all. + #[tokio::test] + async fn a_sweep_fails_a_pending_sent_payment() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Pending).await; + let event = sweep_of(wallet_id, sent_transaction().txid()); + + let overlay = sent_payment_verdicts(&wallet_manager, &event).await; + + assert_eq!( + only_row(&overlay, &txid).status, + PaymentStatus::Failed, + "the overlay must carry the Failed row for this drain's store()" + ); + assert_eq!( + only_row(&overlay, &txid).memo.as_deref(), + Some("lunch"), + "a verdict changes the status and nothing else" + ); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Failed, + "and the live entry must agree with what the round will store" + ); + } + + /// The worse half of the defect: an IS-locked payment already displayed + /// as `Confirmed`, then evicted by a chainlocked winner, is a dead + /// payment reported as good. `Confirmed -> Failed` is the only edge that + /// corrects it. + #[tokio::test] + async fn a_sweep_fails_an_already_confirmed_sent_payment() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Confirmed).await; + let event = sweep_of(wallet_id, sent_transaction().txid()); + + let overlay = sent_payment_verdicts(&wallet_manager, &event).await; + + assert_eq!(only_row(&overlay, &txid).status, PaymentStatus::Failed); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Failed + ); + } + + /// A sweep is not always the last word: a chainlock can reinstate the + /// transaction it removed. `Failed -> Confirmed` is what makes that + /// repair reachable — without it the entry would be stuck on a verdict + /// the chain has since overruled. + #[tokio::test] + async fn a_chainlocked_reinstatement_repairs_a_failed_sent_payment() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Failed).await; + + let overlay = + sent_payment_verdicts(&wallet_manager, &chainlocked_reinstatement(wallet_id)).await; + + assert_eq!( + only_row(&overlay, &txid).status, + PaymentStatus::Confirmed, + "a chainlocked record must overrule the sweep that failed it" + ); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Confirmed + ); + } + + /// An InstantSend lock carries no record, only a txid, and is final for + /// DashPay display — so the txid alone confirms the entry. This is the + /// path the payment handler used to own; it now belongs to the adapter, + /// which reaches it over the lossless channel. + #[tokio::test] + async fn an_instant_lock_confirms_a_pending_sent_payment() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Pending).await; + let event = WalletEvent::TransactionInstantLocked { + wallet_id, + txid: sent_transaction().txid(), + instant_lock: InstantLock::default(), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }; + + let overlay = sent_payment_verdicts(&wallet_manager, &event).await; + + assert_eq!(only_row(&overlay, &txid).status, PaymentStatus::Confirmed); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Confirmed + ); + } + + /// A received entry's status is settled when it is recorded from an + /// on-chain sighting. A sweep that happens to name its txid says nothing + /// about it, so it must not be touched. + #[tokio::test] + async fn a_sweep_never_touches_a_received_payment() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Received, PaymentStatus::Confirmed).await; + let event = sweep_of(wallet_id, sent_transaction().txid()); + + let overlay = sent_payment_verdicts(&wallet_manager, &event).await; + + assert!(overlay.is_empty(), "received entries carry no verdict"); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Confirmed + ); + } + + /// Idempotence, and the reason it matters: a re-emitted sweep (a relaunch + /// re-deriving from a frozen watermark) must produce no row at all, or + /// every re-detection would put an unchanged row on a store round. + #[tokio::test] + async fn a_verdict_already_reached_emits_no_row() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Failed).await; + let event = sweep_of(wallet_id, sent_transaction().txid()); + + let overlay = sent_payment_verdicts(&wallet_manager, &event).await; + + assert!(overlay.is_empty(), "no change, no row"); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Failed + ); + } + + /// A mempool sighting is not finality: the payment genuinely is still + /// pending, so a `TransactionDetected` at an unconfirmed context must + /// leave it alone. + #[tokio::test] + async fn a_mempool_sighting_is_not_finality() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Pending).await; + let record = TransactionRecord::new( + sent_transaction(), + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::Mempool, + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + -50_000, + ); + let event = WalletEvent::TransactionDetected { + wallet_id, + record: Box::new(record), + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + addresses_derived: Vec::new(), + }; + + let overlay = sent_payment_verdicts(&wallet_manager, &event).await; + + assert!(overlay.is_empty()); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Pending + ); + } + + /// The transition table, enumerated. Written out rather than derived so + /// that adding an edge means editing this list — the point of the table + /// is that every edge was chosen, not inferred. + #[test] + fn the_transition_table_admits_exactly_the_four_intended_edges() { + use SentPaymentEvidence::{Final, Swept}; + let expected = [ + ((PaymentStatus::Pending, Swept), Some(PaymentStatus::Failed)), + ( + (PaymentStatus::Confirmed, Swept), + Some(PaymentStatus::Failed), + ), + ((PaymentStatus::Failed, Swept), None), + ( + (PaymentStatus::Pending, Final), + Some(PaymentStatus::Confirmed), + ), + ( + (PaymentStatus::Failed, Final), + Some(PaymentStatus::Confirmed), + ), + ((PaymentStatus::Confirmed, Final), None), + ]; + for ((from, evidence), to) in expected { + assert_eq!( + next_sent_payment_status(from, evidence), + to, + "{from:?} + {evidence:?}" + ); + } + assert_eq!( + expected.iter().filter(|(_, to)| to.is_some()).count(), + 4, + "four edges move an entry; the other two are no-ops" + ); + } + + /// `matured` is coinbase maturity — never a DashPay payment — so a + /// confirmed record arriving only in that bucket is not evidence about a + /// sent one. + #[test] + fn the_matured_bucket_is_not_finality_evidence() { + let record = TransactionRecord::new( + sent_transaction(), + AccountType::Standard { + index: 0, + standard_account_type: StandardAccountType::BIP44Account, + }, + TransactionContext::InChainLockedBlock(BlockInfo::new( + 1_499_060, + BlockHash::all_zeros(), + 0, + )), + TransactionType::Standard, + TransactionDirection::Outgoing, + Vec::new(), + Vec::new(), + -50_000, + ); + let event = WalletEvent::BlockProcessed { + wallet_id: [0x01; 32], + height: 1_499_060, + chain_lock: None, + inserted: Vec::new(), + updated: Vec::new(), + matured: vec![record], + balance: WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + addresses_derived: Vec::new(), + }; + assert!(sent_payment_evidence(&event).is_empty()); + } +} + #[cfg(test)] mod contact_watch_only_projection_tests { //! Regression coverage for the persist-time projection of records @@ -3427,11 +4107,14 @@ mod tests { // lossless burst, a rejected `store()`, the per-wallet freeze, and // per-wallet batch folding. - use super::{run_wallet_event_adapter, AdapterFaultState, ADAPTER_STORE_BATCH_LIMIT}; + use super::{ + run_wallet_event_adapter, AdapterFaultState, PaymentOverlay, ADAPTER_STORE_BATCH_LIMIT, + }; use crate::changeset::changeset::PlatformWalletChangeSet; use crate::changeset::client_start_state::ClientStartState; use crate::changeset::traits::{PersistenceError, PlatformWalletPersistence}; use crate::wallet::platform_wallet::{PlatformWalletInfo, WalletId}; + use dpp::prelude::Identifier; use key_wallet::WalletCoreBalance; use key_wallet_manager::{WalletEvent, WalletManager}; use std::collections::{BTreeMap, HashSet}; @@ -3451,6 +4134,10 @@ mod tests { n_records: usize, n_asset_locks: usize, n_asset_locks_removed: usize, + /// The round's `dashpay_payments_overlay` verbatim — `None` when the + /// round carried none, which is also what a payments-blind persister + /// must see after the capability gate has withheld one. + dashpay_payments: Option, rejected: bool, } @@ -3551,6 +4238,7 @@ mod tests { .as_ref() .map(|a| a.removed.len()) .unwrap_or(0), + dashpay_payments: changeset.dashpay_payments_overlay.clone(), rejected, }); if rejected { @@ -4834,11 +5522,10 @@ mod tests { ); let (obs_tx, mut obs_rx) = unbounded_channel(); - // Attested for sweeps AND payments. Only the sweep half matters - // here: nothing in this PR writes `dashpay_payments_overlay`, so - // the payments bit is inert — it is declared so this fixture keeps - // describing a fully capable backend once the payment-flip coupling - // lands (dashpay/platform#4442) and starts staging that overlay. + // Attested for sweeps AND payments — a fully capable backend. Only + // the sweep half matters to this test's assertions; the payments bit + // keeps the fixture from silently withholding a sent-payment verdict + // if this event ever carries one. let persister = Arc::new(ProbePersister::with_capabilities( obs_tx, crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL @@ -5242,6 +5929,7 @@ mod tests { super::WalletBatch { core: CoreChangeSet::default(), asset_locks, + payments: Default::default(), }, ); commit_batch( @@ -5301,11 +5989,201 @@ mod tests { WalletBatch { core, asset_locks: AssetLockChangeSet::default(), + payments: Default::default(), }, ); batch } + /// One sent-payment verdict, in the shape the adapter folds into a + /// wallet's batch. + fn one_verdict(status: crate::wallet::identity::PaymentStatus) -> super::PaymentOverlay { + use crate::wallet::identity::PaymentEntry; + let mut entry = PaymentEntry::new_sent(Identifier::from([0xBB; 32]), 50_000, None); + entry.status = status; + super::PaymentOverlay::from([( + Identifier::from([0xAA; 32]), + BTreeMap::from([("deadbeef".to_string(), entry)]), + )]) + } + + /// A persister that attests `DASHPAY_PAYMENTS` gets the verdict on the + /// same round as everything else the drain folded. + #[test] + fn a_verdict_reaches_a_persister_that_attests_dashpay_payments() { + use crate::wallet::identity::PaymentStatus; + let wallet_id = [0x31u8; 32]; + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS, + ); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + let freeze_logged = AtomicBool::new(false); + + let mut batch = BTreeMap::new(); + batch.insert( + wallet_id, + WalletBatch { + core: watermark_with_rows(700, 700), + asset_locks: AssetLockChangeSet::default(), + payments: one_verdict(PaymentStatus::Failed), + }, + ); + let diag = commit_batch( + &persister, + batch, + 1, + &mut fault, + &sync_fault, + &freeze_logged, + &mut Vec::new(), + ); + + let observed = obs_rx.try_recv().expect("the round reaches store()"); + assert_eq!( + observed.dashpay_payments, + Some(one_verdict(PaymentStatus::Failed)), + "the verdict must ride the same store() as the rows that justify it" + ); + assert_eq!(diag.persisted, Some(700)); + assert_eq!(diag.faulted, 0, "a payments-capable host is not a fault"); + } + + /// A persister that never attested `DASHPAY_PAYMENTS` cannot apply the + /// overlay, so handing it one would let the round return `Ok` while the + /// verdict was silently dropped. Withhold it instead — and unlike a + /// withheld sweep this does NOT freeze the watermark: the verdict is + /// derived state a later host re-derives, whereas a dropped removal has + /// no recovery. + #[test] + fn a_verdict_is_withheld_from_a_persister_without_dashpay_payments() { + use crate::wallet::identity::PaymentStatus; + let wallet_id = [0x32u8; 32]; + let (obs_tx, mut obs_rx) = unbounded_channel(); + // No capabilities declared — the payments-blind host. + let persister = ProbePersister::new(obs_tx); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + let freeze_logged = AtomicBool::new(false); + + let mut batch = BTreeMap::new(); + batch.insert( + wallet_id, + WalletBatch { + core: watermark_with_rows(700, 700), + asset_locks: AssetLockChangeSet::default(), + payments: one_verdict(PaymentStatus::Failed), + }, + ); + let diag = commit_batch( + &persister, + batch, + 1, + &mut fault, + &sync_fault, + &freeze_logged, + &mut Vec::new(), + ); + + let observed = obs_rx.try_recv().expect("the rest of the round still runs"); + assert_eq!( + observed.dashpay_payments, None, + "a payments-blind persister must never be handed an overlay" + ); + assert_eq!( + diag.persisted, + Some(700), + "the rest of the round is unaffected — only the overlay is withheld" + ); + assert_eq!(diag.frozen, None, "a withheld verdict does not freeze"); + assert_eq!(diag.faulted, 0); + assert!(!sync_fault.load(Ordering::Relaxed)); + } + + /// A round whose ONLY content is a verdict the capability gate withholds + /// has nothing left to persist, so it must skip the store round-trip + /// entirely rather than send an empty changeset. + #[test] + fn a_withheld_verdict_alone_never_reaches_the_store() { + use crate::wallet::identity::PaymentStatus; + let wallet_id = [0x33u8; 32]; + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = ProbePersister::new(obs_tx); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + let freeze_logged = AtomicBool::new(false); + + let mut batch = BTreeMap::new(); + batch.insert( + wallet_id, + WalletBatch { + core: CoreChangeSet::default(), + asset_locks: AssetLockChangeSet::default(), + payments: one_verdict(PaymentStatus::Failed), + }, + ); + commit_batch( + &persister, + batch, + 1, + &mut fault, + &sync_fault, + &freeze_logged, + &mut Vec::new(), + ); + assert!( + obs_rx.try_recv().is_err(), + "nothing left to persist must not reach store()" + ); + } + + /// The mirror of the case above: a verdict is the ONLY thing a round + /// carries when the wallet's other projections are empty — a + /// `TransactionInstantLocked` for an already chain-locked txid projects + /// no core rows at all. That round must still reach the store, or the + /// verdict is lost with no event left to re-derive it from. + #[test] + fn a_verdict_alone_still_reaches_a_capable_store() { + use crate::wallet::identity::PaymentStatus; + let wallet_id = [0x34u8; 32]; + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = ProbePersister::with_capabilities( + obs_tx, + crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS, + ); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + let freeze_logged = AtomicBool::new(false); + + let mut batch = BTreeMap::new(); + batch.insert( + wallet_id, + WalletBatch { + core: CoreChangeSet::default(), + asset_locks: AssetLockChangeSet::default(), + payments: one_verdict(PaymentStatus::Confirmed), + }, + ); + commit_batch( + &persister, + batch, + 1, + &mut fault, + &sync_fault, + &freeze_logged, + &mut Vec::new(), + ); + let observed = obs_rx + .try_recv() + .expect("a verdict-only round must still be stored"); + assert_eq!( + observed.dashpay_payments, + Some(one_verdict(PaymentStatus::Confirmed)) + ); + } + /// The sweep guard strips the height BEFORE the store sees it, so a /// backend that never attested `CORE_SWEEP_REMOVAL` and returns `Ok` /// must report the height as FROZEN, not rejected: `rejected` would 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 fce9972f855..b32f7d88ccb 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs @@ -53,10 +53,7 @@ pub(crate) use payment_handler::DashPayPaymentHandler; #[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, -}; +pub(crate) use payments::{record_incoming_dashpay_payments, sent_payment_status_for_record}; mod profile; pub(crate) mod sdk_writer; mod seed_binding; 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 0c3cc79d3d1..695a0ae4bb1 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 @@ -6,8 +6,21 @@ //! 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. +//! +//! # Incoming only +//! +//! A *sent* payment's status is not this handler's to write. The broadcast +//! bus this handler runs off is lossy — it drops events under +//! `RecvError::Lagged` during catch-up — and the two events that decide a +//! sent payment's fate do not survive that: a sweep never re-emits once the +//! wallet has dropped the loser's record, so a dropped one is unrecoverable. +//! The adapter drains the lossless persistence channel instead, and resolves +//! every sent-payment verdict there so the flip rides the same `store()` +//! round as the rows that justify it (`sent_payment_verdicts` in +//! `crate::changeset::core_bridge`). Incoming payments have no such +//! constraint: they are idempotent inserts re-derivable from +//! receival-account UTXOs, so a dropped event costs nothing but latency. //! //! # Why it spawns //! @@ -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 belong to the wallet-event +/// adapter — see the module docs. /// /// Holds the manager's `wallet_manager` (for the in-memory identity / /// payment state the hooks mutate) and an `Arc` @@ -227,38 +241,31 @@ 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 recorder. /// /// [`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 +/// (previously-known records that this block confirmed); a payment first +/// seen in a block arrives only in `inserted`. `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 { inserted, updated, .. } => 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. + // Neither of the two sent-payment verdict carriers routes here. + // `TransactionsSwept` carries txids, not records — the wallet has + // already dropped the records it names — and + // `TransactionInstantLocked` carries only a txid. Both are resolved + // by the wallet-event adapter instead, on the lossless channel and + // on the same store round as the rows that justify the verdict + // (see the module docs). WalletEvent::TransactionInstantLocked { .. } | WalletEvent::TransactionsSwept { .. } | WalletEvent::SyncHeightAdvanced { .. } @@ -268,59 +275,43 @@ 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. -/// Allocation-free. +/// Exactly the record-bearing events ([`dashpay_payment_records`]). 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 { .. } + // No records to route (see `dashpay_payment_records`), so a task here + // would take and release the wallet-manager write lock for nothing. + // `TransactionInstantLocked` and `TransactionsSwept` are the two + // sent-payment verdict carriers and belong to the wallet-event + // adapter's own store round. + 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 records it carries pay to. +/// +/// Sent payments are not touched here — both terminals of a sent entry are +/// resolved by the wallet-event adapter, on the lossless channel and on the +/// same store round as the rows that justify them (see the module docs). +/// Idempotent per txid, so re-detections and repeated block-processing rounds +/// converge without duplicating entries. 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 +320,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 +382,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 the path by which a previously-seen one is + /// re-emitted on confirmation (`updated`); both must drive the DashPay + /// incoming recorder, whose inserts are idempotent per txid. `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 +402,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" + an incoming payment first matched on confirmation lands here" ); assert!( !txids.contains(&record(0x03).txid), @@ -428,8 +411,8 @@ mod tests { assert_eq!(txids.len(), 2, "exactly inserted ∪ updated"); } - /// The first mempool sighting still routes its single record (incoming - /// recording + the early-returning confirm probe). + /// The first mempool sighting still routes its single record — that is + /// where a live incoming payment is first recorded. #[test] fn dashpay_payment_records_covers_transaction_detected() { let event = WalletEvent::TransactionDetected { @@ -458,11 +441,16 @@ 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 + /// carries no record, so there is no incoming payment to recover from it, + /// and its one payment consequence — confirming a sent entry by txid + /// alone — belongs to the wallet-event adapter: this handler runs off the + /// lossy broadcast bus, while the adapter drains the lossless persistence + /// channel and can put the flip on the same store round as the rows that + /// justify it. Spawning a task here would take the wallet-manager write + /// lock for nothing and race a second write against that round. #[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 +459,16 @@ 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' 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. #[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 89f982f0db5..940b8da96ec 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs @@ -683,17 +683,23 @@ impl DashPayView<'_, B> { /// Flip `Pending` `Sent` [`PaymentEntry`]s to `Confirmed` when the /// 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` + /// Recovery path for sent-payment confirmation. The live verdict path + /// (`sent_payment_verdicts` in the wallet-event adapter) flips a sent + /// payment the moment its block / InstantSend-lock event arrives, but + /// that is a single live event: if its store round was rejected, or a + /// relaunch landed 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. /// + /// Its evidence class is deliberately only `Pending`: a durable `Failed` + /// is a verdict the adapter reached with evidence this sweep cannot see + /// (the loser's record is gone, so a read here returns nothing anyway), + /// and confirming out of `Failed` is reserved for the adapter's own + /// reinstatement edge. + /// /// Runs as a local-only step of `dashpay_sync()` — one persister read /// per pending sent payment, no network round-trips. Idempotent: a /// `Confirmed` entry is left alone, and a transaction not yet final is @@ -951,7 +957,16 @@ fn wallet_tx_table_digest(listed: &[crate::changeset::traits::ListedCoreTxid]) - sha256::Hash::from_engine(engine).to_byte_array() } -fn sent_payment_status_for_record( +/// Whether a transaction record is final enough for a `Sent` +/// [`PaymentEntry`] — the ONE definition of "final for DashPay display", +/// shared by the wallet-event adapter's live verdict path +/// (`sent_payment_verdicts`), the reconcile sweep, and the tx-history +/// reconstruction sweep, so the three can never disagree. +/// +/// An **InstantSend lock counts as final**: it is effectively irreversible, +/// so the user sees `Confirmed` without waiting for the surrounding block. A +/// bare mempool sighting does not — the payment genuinely is still pending. +pub(crate) fn sent_payment_status_for_record( record: &key_wallet::managed_account::transaction_record::TransactionRecord, ) -> crate::wallet::identity::types::dashpay::payment::PaymentStatus { use crate::wallet::identity::types::dashpay::payment::PaymentStatus; @@ -964,59 +979,6 @@ fn sent_payment_status_for_record( } } -/// Advance a sender's `Sent` [`PaymentEntry`] from `Pending` to -/// `Confirmed` once its broadcast transaction reaches finality. -/// -/// [`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. -/// -/// 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( - wallet_manager: &Arc>>, - wallet_id: &WalletId, - persister: &crate::wallet::persister::WalletPersister, - record: &key_wallet::managed_account::transaction_record::TransactionRecord, -) { - 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( - wallet_manager, - wallet_id, - persister, - &record.txid.to_string(), - ) - .await; -} - -/// 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( - wallet_manager: &Arc>>, - wallet_id: &WalletId, - persister: &crate::wallet::persister::WalletPersister, - txid: &dashcore::Txid, -) { - confirm_sent_payment_by_txid(wallet_manager, wallet_id, persister, &txid.to_string()).await; -} - /// Flip the `Pending` `Sent` [`PaymentEntry`] under `txid` (if any) to /// `Confirmed`, in place, preserving amount/memo/counterparty. /// @@ -3089,20 +3051,18 @@ mod tests { /// 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 + /// (`TransactionDetected`, context `Mempool`), which is not final. 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. + /// confirmed). Reading finality only from `TransactionDetected` would + /// leave the entry `Pending` forever. This drives the adapter's verdict + /// path with a `BlockProcessed` event and pins the flip end-to-end. + /// + /// Also pins that the incoming-payment handler leaves the sent entry + /// alone (the verdict is the adapter's, on the lossless channel), + /// 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; @@ -3194,6 +3154,10 @@ mod tests { addresses_derived: Vec::new(), }; + // The payment handler must NOT write a sent-payment verdict: it runs + // off the lossy broadcast bus, so a verdict it wrote could be lost + // for good. Running it here first pins that separation — the entry is + // still `Pending` afterwards. crate::wallet::identity::network::run_dashpay_payment_hooks( &iw.wallet_manager, &wallet_id, @@ -3201,6 +3165,15 @@ mod tests { &event, ) .await; + assert_eq!( + read_status(iw, &wallet_id, &owner, &txid.to_string()) + .await + .status, + PaymentStatus::Pending, + "the incoming-payment handler must not confirm a sent payment" + ); + + crate::changeset::core_bridge::sent_payment_verdicts(&iw.wallet_manager, &event).await; // Read the entry under a short-lived read lock so the re-fire below // can take the write lock. @@ -3231,14 +3204,9 @@ mod tests { assert_eq!(entry.memo.as_deref(), Some("lunch"), "memo preserved"); // 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; + // changes nothing (the transition table has no `Confirmed` + final + // edge, so no row is even emitted). + crate::changeset::core_bridge::sent_payment_verdicts(&iw.wallet_manager, &event).await; assert_eq!( read_status(iw, &wallet_id, &owner, &txid.to_string()) .await @@ -3249,7 +3217,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 it is excluded from the adapter's finality evidence. let matured_tx = Transaction { version: 2, lock_time: 0, @@ -3305,13 +3273,8 @@ 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, - &matured_event, - ) - .await; + crate::changeset::core_bridge::sent_payment_verdicts(&iw.wallet_manager, &matured_event) + .await; assert_eq!( read_status(iw, &wallet_id, &owner, &matured_txid.to_string()) .await @@ -3325,7 +3288,8 @@ mod tests { /// 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. + /// `Pending → Confirmed`. Drives the adapter's verdict path — the event + /// carries no record, so the txid alone is the evidence. #[tokio::test] async fn instant_send_lock_confirms_sent_payment() { use dashcore::ephemerealdata::instant_lock::InstantLock; @@ -3366,13 +3330,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; + crate::changeset::core_bridge::sent_payment_verdicts(&iw.wallet_manager, &event).await; let wm = iw.wallet_manager.read().await; let info = wm.get_wallet_info(&wallet_id).expect("info"); @@ -3472,13 +3430,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; + crate::changeset::core_bridge::sent_payment_verdicts(&iw.wallet_manager, &event).await; let wm = iw.wallet_manager.read().await; let info = wm.get_wallet_info(&wallet_id).expect("info"); From 62b162057713b2ec22b2abbb699f26094b03e3eb Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Mon, 14 Sep 2026 12:51:29 +0300 Subject: [PATCH 2/2] fix(platform-wallet-storage): patch a sent-payment verdict into the identity blob instead of shipping a snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous revision made the verdict durable by having the adapter ship a whole `IdentityEntry` alongside the overlay, because `load()` rehydrates a managed identity's payments from `identities.entry_blob` and nothing reads the overlay table back. That snapshot is captured under the wallet-manager write lock, but the lock is released long before the adapter reaches `store()` on its blocking thread — and in that window an ordinary `record_dashpay_payment` can take the lock and durably persist another payment. When the stale snapshot landed, `identities::apply_upserts` replaced `entry_blob` wholesale and the newer payment was gone from the authoritative restart state, memo and all. Apply only the rows that changed instead. The adapter no longer sends an identity it captured earlier — it carries the overlay alone — and the SQLite store patches the authoritative blob from that overlay inside the same write transaction: read the identity's `entry_blob`, insert or replace exactly the overlay's txids in its payments map, write it back. Every other payment and every other field of the entry is read out and written back untouched, and because the read-modify-write happens inside the persister's transaction it is atomic against every other writer on the file. Nothing is captured ahead of the commit, so there is no stale window left to lose an update in. The patch runs after `identities::apply_upserts` and before `apply_removals`, so a round legitimately carrying both a full identity and an overlay ends with the overlay on top, while a hard delete in the same round still wins. An identity row that is absent is skipped: the overlay's own foreign key guarantees it existed unless this round deleted it, and then the delete should stand. FFI hosts are unaffected. `persistence.rs` forwards the overlay to `on_persist_dashpay_payments_fn` and deliberately does not project payments out of `changeset.identities`, so the snapshot was already inert for them. The reviewer's reproducer lands as `tests/review_4651_concurrent_payment.rs`, unmodified: it pauses the sweep's store, records a second payment through the production writer, asserts it reached disk, then resumes and reopens. Red before this change on the final assertion. Disabling only the blob patch turns `a_swept_sent_payments_failed_verdict_survives_a_reopen` red, which pins the restart guarantee to the new mechanism — the two halves are jointly necessary. The unit tests that asserted the snapshot's presence are inverted rather than deleted, so its absence stays a regression guard. --- .../src/sqlite/schema/dashpay.rs | 75 +++++ .../tests/review_4651_concurrent_payment.rs | 186 ++++++++++++ .../src/changeset/core_bridge.rs | 285 ++++++------------ 3 files changed, 350 insertions(+), 196 deletions(-) create mode 100644 packages/rs-platform-wallet-storage/tests/review_4651_concurrent_payment.rs diff --git a/packages/rs-platform-wallet-storage/src/sqlite/schema/dashpay.rs b/packages/rs-platform-wallet-storage/src/sqlite/schema/dashpay.rs index e9dd5cdb6ae..0ec8c2bfe10 100644 --- a/packages/rs-platform-wallet-storage/src/sqlite/schema/dashpay.rs +++ b/packages/rs-platform-wallet-storage/src/sqlite/schema/dashpay.rs @@ -8,6 +8,24 @@ //! indexed queries. Round-trip pinned by //! `tests/sqlite_dashpay_overlay_contract.rs`. //! +//! # …which is why the payment overlay is also PATCHED into `entry_blob` +//! +//! Because `load()` reads only the identity blob, a payment row that reached +//! the overlay table alone is a write that does not survive a restart. The +//! wallet-event adapter is the single writer of swept sent-payment verdicts +//! and has no other durable channel for them, so [`apply`] finishes by +//! folding each overlay row into the owning identity's `entry_blob`. +//! +//! The fold is a read-modify-write of ONLY the `(txid -> PaymentEntry)` keys +//! the overlay names, run inside the persister's own write transaction: +//! payments the overlay does not mention, and every other field of the +//! entry, are read back and written out unchanged. That is what distinguishes +//! it from shipping a whole `IdentityEntry` from the caller — a snapshot +//! captured before the lock was released would wholesale replace the blob and +//! silently drop any payment another writer committed in between +//! (dashpay/platform#4651). Doing the merge here, in the same transaction, +//! makes it atomic against every other writer on the file. +//! //! # Precondition //! //! Every `identity_id` MUST already exist in `identities` and belong to the @@ -20,6 +38,7 @@ use std::collections::BTreeMap; use rusqlite::{params, Transaction}; use dpp::prelude::Identifier; +use platform_wallet::changeset::IdentityEntry; use platform_wallet::wallet::identity::{DashPayProfile, PaymentEntry}; use platform_wallet::wallet::platform_wallet::WalletId; @@ -81,7 +100,63 @@ pub fn apply( stmt.execute(params![identity_id.as_slice(), tx_id, payload])?; } } + patch_payments_into_entry_blobs(tx, payments)?; + } + } + Ok(()) +} + +/// Fold the overlay's rows into each owning identity's authoritative +/// `entry_blob`, one identity at a time, inside the caller's transaction. +/// +/// Read-modify-write, NOT a replace: the stored entry is decoded, only the +/// `(txid -> PaymentEntry)` keys this round names are inserted-or-replaced in +/// its `dashpay_payments` map, and the entry is written back. Every other +/// payment — including one another writer committed microseconds ago — and +/// every other field survive untouched. +/// +/// Runs AFTER `identities::apply_upserts` in `persister::apply_changeset`, so +/// a round that legitimately carries both a full identity snapshot and an +/// overlay ends with the overlay applied ON TOP of the snapshot, which is the +/// order the two mean: the snapshot is the round's view of the identity, the +/// overlay is the round's view of the payments that moved. +/// +/// A missing `identities` row is not an error here. The FK would have +/// rejected the overlay insert above, so by this point the row exists for +/// every identity in `payments` unless it was deleted inside this same +/// transaction; nothing is patched in that case and the delete stands. +fn patch_payments_into_entry_blobs( + tx: &Transaction<'_>, + payments: &BTreeMap>, +) -> Result<(), WalletStorageError> { + let mut read = tx.prepare_cached( + "SELECT length(entry_blob), entry_blob FROM identities WHERE identity_id = ?1", + )?; + let mut write = + tx.prepare_cached("UPDATE identities SET entry_blob = ?2 WHERE identity_id = ?1")?; + for (identity_id, by_tx) in payments { + if by_tx.is_empty() { + continue; + } + let stored: Option<(i64, Vec)> = read + .query_row(params![identity_id.as_slice()], |row| { + Ok((row.get(0)?, row.get(1)?)) + }) + .map(Some) + .or_else(|e| match e { + rusqlite::Error::QueryReturnedNoRows => Ok(None), + other => Err(other), + })?; + let Some((len, payload)) = stored else { + continue; + }; + blob::check_size(len)?; + let mut entry: IdentityEntry = blob::decode(&payload)?; + for (tx_id, row) in by_tx { + entry.dashpay_payments.insert(tx_id.clone(), row.clone()); } + let patched = blob::encode(&entry)?; + write.execute(params![identity_id.as_slice(), patched])?; } Ok(()) } diff --git a/packages/rs-platform-wallet-storage/tests/review_4651_concurrent_payment.rs b/packages/rs-platform-wallet-storage/tests/review_4651_concurrent_payment.rs new file mode 100644 index 00000000000..90e24ea2a15 --- /dev/null +++ b/packages/rs-platform-wallet-storage/tests/review_4651_concurrent_payment.rs @@ -0,0 +1,186 @@ +//! Review regression: a concurrent payment must survive an adapter snapshot. +mod common; + +use std::collections::BTreeMap; +use std::sync::atomic::AtomicBool; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use common::{ensure_wallet_meta, secure_tempdir}; +use dashcore::hashes::Hash; +use dpp::identity::{Identity, IdentityV0}; +use dpp::prelude::Identifier; +use key_wallet::account::account_type::StandardAccountType; +use platform_wallet::changeset::{ + spawn_wallet_event_adapter, ClientStartState, PersistenceCapabilities, PersistenceError, + PlatformWalletChangeSet, PlatformWalletPersistence, WalletMetadataEntry, +}; +use platform_wallet::key_wallet_manager::WalletEvent; +use platform_wallet::test_support::funded_wallet_manager; +use platform_wallet::wallet::identity::{PaymentEntry, PaymentStatus}; +use platform_wallet::wallet::persister::WalletPersister; +use platform_wallet_storage::{SqlitePersister, SqlitePersisterConfig}; + +struct PausedSweepStore { + inner: Arc, + entered: Mutex>>, + resume: Mutex>, +} + +impl PlatformWalletPersistence for PausedSweepStore { + fn persistence_capabilities(&self) -> PersistenceCapabilities { + self.inner.persistence_capabilities() + } + + fn store(&self, wallet: [u8; 32], cs: PlatformWalletChangeSet) -> Result<(), PersistenceError> { + if cs.core.as_ref().is_some_and(|core| !core.sweeps.is_empty()) { + if let Some(entered) = self.entered.lock().unwrap().take() { + entered.send(()).unwrap(); + self.resume + .lock() + .unwrap() + .recv_timeout(Duration::from_secs(10)) + .unwrap(); + } + } + self.inner.store(wallet, cs) + } + + fn flush(&self, wallet: [u8; 32]) -> Result<(), PersistenceError> { + self.inner.flush(wallet) + } + + fn load(&self) -> Result { + self.inner.load() + } +} + +#[tokio::test] +async fn should_preserve_a_payment_persisted_while_a_sweep_snapshot_is_in_flight() { + let tmp = secure_tempdir().unwrap(); + let path = tmp.path().join("wallet.db"); + let owner = Identifier::from([0xA7; 32]); + let contact = Identifier::from([0xB7; 32]); + let loser = dashcore::Txid::from_byte_array([0x5f; 32]); + let concurrent = dashcore::Txid::from_byte_array([0x6f; 32]).to_string(); + let (wm, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + { + let sqlite = Arc::new(SqlitePersister::open(SqlitePersisterConfig::new(&path)).unwrap()); + ensure_wallet_meta(&sqlite, &wallet_id); + sqlite + .store( + wallet_id, + PlatformWalletChangeSet { + wallet_metadata: Some(WalletMetadataEntry { + network: key_wallet::Network::Testnet, + wallet_group_id: wallet_id, + birth_height: 0, + }), + ..Default::default() + }, + ) + .unwrap(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = std::sync::mpsc::channel(); + let persister = Arc::new(PausedSweepStore { + inner: Arc::clone(&sqlite), + entered: Mutex::new(Some(entered_tx)), + resume: Mutex::new(resume_rx), + }); + let wp = WalletPersister::new(wallet_id, persister.clone()); + { + let mut manager = wm.write().await; + let info = manager.get_wallet_info_mut(&wallet_id).unwrap(); + info.identity_manager + .add_identity( + Identity::V0(IdentityV0 { + id: owner, + public_keys: BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &wp, + ) + .unwrap(); + info.identity_manager + .managed_identity_mut(&owner) + .unwrap() + .record_dashpay_payment( + loser.to_string(), + PaymentEntry::new_sent(contact, 50_000, Some("original".into())), + &wp, + ) + .unwrap(); + } + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + tx.send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![loser], + superseded_by: dashcore::Txid::from_byte_array([0x77; 32]), + winner_mined_height: Some(1_499_050), + released_outpoints: Vec::new(), + balance: key_wallet::WalletCoreBalance::default(), + account_balances: BTreeMap::new(), + }) + .unwrap(); + drop(tx); + let adapter = spawn_wallet_event_adapter( + Arc::clone(&wm), + Arc::downgrade(&persister), + rx, + Arc::new(AtomicBool::new(false)), + tokio_util::sync::CancellationToken::new(), + ); + tokio::time::timeout(Duration::from_secs(10), entered_rx) + .await + .unwrap() + .unwrap(); + + // The adapter has captured the old identity snapshot and released the + // manager lock. A normal production payment writer now commits B. + { + let mut manager = wm.write().await; + manager + .get_wallet_info_mut(&wallet_id) + .unwrap() + .identity_manager + .managed_identity_mut(&owner) + .unwrap() + .record_dashpay_payment( + concurrent.clone(), + PaymentEntry::new_sent(contact, 75_000, Some("new payment memo".into())), + &wp, + ) + .unwrap(); + } + let before = sqlite.load().unwrap(); + let before_identity = &before.wallets[&wallet_id] + .identity_manager + .wallet_identities[&wallet_id][&0]; + assert!( + before_identity.dashpay().payments.contains_key(&concurrent), + "the concurrent payment was durably recorded before the stale adapter write" + ); + + resume_tx.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(10), adapter) + .await + .unwrap() + .unwrap(); + sqlite.flush(wallet_id).unwrap(); + } + let reopened = SqlitePersister::open(SqlitePersisterConfig::new(&path)).unwrap(); + let loaded = reopened.load().unwrap(); + let identity = &loaded.wallets[&wallet_id] + .identity_manager + .wallet_identities[&wallet_id][&0]; + assert_eq!( + identity.dashpay().payments[&loser.to_string()].status, + PaymentStatus::Failed + ); + assert!(identity.dashpay().payments.contains_key(&concurrent), + "the adapter's stale full identity snapshot deleted a successfully persisted concurrent payment"); +} diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 4335765bd1c..6bdbccb8ced 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -51,9 +51,8 @@ use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use crate::changeset::changeset::{ - merge_payment_overlays, AssetLockChangeSet, CoreChangeSet, HighestUsedIndexes, - IdentityChangeSet, IdentityEntry, PaymentOverlay, PlatformWalletChangeSet, SweepBatch, - UtxoCreditVerdict, + merge_payment_overlays, AssetLockChangeSet, CoreChangeSet, HighestUsedIndexes, PaymentOverlay, + PlatformWalletChangeSet, SweepBatch, UtxoCreditVerdict, }; use crate::changeset::merge::Merge; use crate::changeset::persistence_capabilities::PersistenceCapabilities; @@ -426,7 +425,7 @@ async fn run_wallet_event_adapter

( let entry = batch.entry(wallet_id).or_default(); entry.core.merge(core); entry.asset_locks.merge(asset_locks); - entry.payments.merge(payments); + merge_payment_overlays(&mut entry.payments, payments); } // Fold in whatever else is already buffered. `try_recv` never waits, @@ -447,7 +446,7 @@ async fn run_wallet_event_adapter

( // Last-write-wins per `(owner, txid)`: a transaction swept // and then reinstated inside one drain reaches the store as // the verdict the drain ended on, never as two rows. - entry.payments.merge(payments); + merge_payment_overlays(&mut entry.payments, payments); folded += 1; } Err(TryRecvError::Empty) => break, @@ -731,30 +730,33 @@ fn commit_wallet

( // the verdict is derived state, so a host that later ships the slot // re-derives it from the records and the reconcile pass, whereas a // dropped removal has no such recovery. - let (payments, verdict_identities) = if payments.is_empty() { - (None, None) + let payments = if payments.is_empty() { + None } else if persister .persistence_capabilities() .contains(PersistenceCapabilities::DASHPAY_PAYMENTS) { - // Both carriers or neither. The identity snapshots are what make - // the verdict survive a restart (`load()` rehydrates payments from - // the identity blob, never from the overlay table); the overlay is - // the bounded row a delta-style persister projects. Splitting them - // across rounds is exactly the durability hole this pair closes, - // and `DASHPAY_PAYMENTS` is the one bit that says a host stores - // sent-payment state at all. - (Some(payments.overlay), Some(payments.identities)) + // The overlay is the ONLY carrier. A backend that attests + // `DASHPAY_PAYMENTS` owes durability for these rows — the SQLite + // one patches them into the authoritative identity blob inside its + // own write transaction, an FFI host stores them in its own + // payment rows. The adapter deliberately does not ship a whole + // `IdentityEntry` alongside: it captured the verdict under the + // wallet-manager write lock and releases it well before this + // `store()`, so such a snapshot would be stale on arrival and + // would erase any payment a concurrent `record_dashpay_payment` + // persisted in the gap (dashpay/platform#4651). + Some(payments) } else { tracing::warn!( wallet_id = %hex::encode(wallet_id), - identities = payments.overlay.len(), - rows = payments.overlay.values().map(BTreeMap::len).sum::(), + identities = payments.len(), + rows = payments.values().map(BTreeMap::len).sum::(), "Persister does not advertise DASHPAY_PAYMENTS; withholding this round's \ sent-payment verdicts. Swept sent payments stay as stored until the host \ adopts the payment-overlay slot." ); - (None, None) + None }; // Hold this wallet's durable watermark at the last fully persisted // height once it has faulted. Records/UTXOs still persist — only the @@ -773,11 +775,7 @@ fn commit_wallet

( diag.record_frozen(h); } } - if core.is_empty_no_records() - && Merge::is_empty(&asset_locks) - && payments.is_none() - && verdict_identities.is_none() - { + if core.is_empty_no_records() && Merge::is_empty(&asset_locks) && payments.is_none() { // SyncHeightAdvanced for an unknown wallet, empty BlockProcessed, a // watermark-only batch stripped by the fault guard above, a verdict // withheld from a payments-blind persister, etc. — nothing to @@ -832,14 +830,10 @@ 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 authoritative half of the same verdicts: a post-flip - // `IdentityEntry` per identity whose payments moved. `load()` - // rebuilds `dashpay_payments` from this snapshot and never reads - // the overlay table back, so without it the round's verdict is - // undone by the next restart. - identities: verdict_identities, // The sent-payment verdicts this drain resolved, on the same round // as the sweep removal or confirming record that justifies them. + // The persister is responsible for making these durable; the + // adapter never ships a whole identity of its own. dashpay_payments_overlay: payments, ..PlatformWalletChangeSet::default() }; @@ -1009,12 +1003,20 @@ struct WalletBatch { core: CoreChangeSet, asset_locks: AssetLockChangeSet, /// Sent-payment verdicts this drain resolved (see - /// [`sent_payment_verdicts`]), in both carriers: the bounded overlay rows - /// and the authoritative post-flip identity snapshots. Rides the same - /// `store()` as the rows that justify it — a sweep's removal, or the - /// record that confirmed it — because neither event re-emits once its - /// round is durable. - payments: SentPaymentVerdicts, + /// [`sent_payment_verdicts`]): the changed `(owner, txid)` rows, and + /// nothing else. Rides the same `store()` as the rows that justify it — + /// a sweep's removal, or the record that confirmed it — because neither + /// event re-emits once its round is durable. + /// + /// Deliberately NOT accompanied by an `IdentityEntry` snapshot. The + /// adapter captures its verdicts under the wallet-manager write lock but + /// calls `store()` long after releasing it, so a whole-identity snapshot + /// taken here would be stale by the time it landed and would wholesale + /// replace any payment a concurrent `record_dashpay_payment` had + /// persisted in between (dashpay/platform#4651). The authoritative + /// identity blob is instead patched row-by-row by the persister, inside + /// its own write transaction — see the SQLite backend's `dashpay` module. + payments: PaymentOverlay, } /// Rebuild missing tracked asset locks from the records an event @@ -1470,59 +1472,6 @@ fn sent_payment_evidence(event: &WalletEvent) -> Vec<(String, SentPaymentEvidenc } } -/// One event's sent-payment verdicts, in the two carriers a round needs. -/// -/// Both describe the same flips; neither is redundant. -/// -/// * `identities` is the **authoritative** one. `load()` rebuilds a managed -/// identity's `dashpay_payments` from the identity snapshot -/// (`identities.entry_blob` in the SQLite backend), so a verdict that does -/// not ride an [`IdentityEntry`] is silently replaced by the pre-verdict -/// status at the next launch — with the swept transaction gone and nothing -/// able to re-derive it. This is the same pair -/// `ManagedIdentity::record_dashpay_payment` writes on the path this -/// adapter replaced. -/// * `overlay` is the **bounded** one. A full snapshot replays the identity's -/// whole payment history on every flip, which delta-style persisters (the -/// FFI vtable) must not be handed per round; the single-row overlay is what -/// they project instead. -#[derive(Default)] -pub(crate) struct SentPaymentVerdicts { - /// The changed rows, keyed `(owner, txid)`. - overlay: PaymentOverlay, - /// A post-flip [`IdentityEntry`] snapshot per identity whose payments - /// moved. Merged with [`Merge`], which is keyed by identity id and folds - /// `dashpay_payments` last-write-wins per txid — so a drain touching one - /// identity twice still reaches the store as one entry. - identities: IdentityChangeSet, -} - -impl SentPaymentVerdicts { - /// True when nothing moved, so the round carries no verdict at all. The - /// two halves are populated together — an identity is snapshotted exactly - /// when at least one of its rows entered the overlay — so either one - /// answers, and both are asserted here to keep that coupling honest. - fn is_empty(&self) -> bool { - debug_assert_eq!( - self.overlay.is_empty(), - Merge::is_empty(&self.identities), - "a verdict's overlay row and its identity snapshot are written together" - ); - self.overlay.is_empty() && Merge::is_empty(&self.identities) - } - - /// Fold `other` in. The overlay takes last-write-wins per `(owner, txid)` - /// via [`merge_payment_overlays`]; the snapshots take - /// [`IdentityChangeSet`]'s own merge, whose `dashpay_payments` fold is - /// last-write-wins per txid as well — so a transaction swept and then - /// reinstated inside one drain reaches the store once, as the verdict the - /// drain ended on, in both carriers. - fn merge(&mut self, other: Self) { - merge_payment_overlays(&mut self.overlay, other.overlay); - self.identities.merge(other.identities); - } -} - /// Resolve `event`'s sent-payment evidence against the wallet's live payment /// entries, flip the ones the transition table moves, and return them as a /// ready overlay for this drain's `store()` round. @@ -1538,6 +1487,20 @@ impl SentPaymentVerdicts { /// persistence channel, so the verdict rides the same `store()` as the row /// removal that implies it. /// +/// # Only the changed rows, never a whole identity +/// +/// The flips below happen under the wallet-manager write lock, but the +/// adapter releases that lock and only then calls `store()` on a blocking +/// thread. Anything captured here that describes MORE than the rows it +/// changed — a whole `IdentityEntry`, say — would therefore be a stale +/// picture by the time it reached disk, and applying it wholesale would +/// delete whatever a concurrent `ManagedIdentity::record_dashpay_payment` +/// persisted in the gap (dashpay/platform#4651). So this returns exactly the +/// `(owner, txid)` rows that moved, and making them durable is the +/// persister's job: the SQLite backend patches them into the authoritative +/// identity blob inside its own write transaction, which is atomic against +/// every other writer. +/// /// # Failure posture /// /// The flip lands in memory here and in the store when the round commits. A @@ -1550,8 +1513,8 @@ impl SentPaymentVerdicts { pub(crate) async fn sent_payment_verdicts( wallet_manager: &Arc>>, event: &WalletEvent, -) -> SentPaymentVerdicts { - let mut overlay = SentPaymentVerdicts::default(); +) -> PaymentOverlay { + let mut overlay = PaymentOverlay::default(); let evidence = sent_payment_evidence(event); if evidence.is_empty() { return overlay; @@ -1603,7 +1566,6 @@ pub(crate) async fn sent_payment_verdicts( // the one thing this fix exists to avoid. The overlay returned here // carries the same row onto the adapter's round instead. let payments = managed.dashpay_payments_mut(); - let mut flipped_any = false; for (txid, evidence) in &evidence { let Some(entry) = payments.get_mut(txid) else { continue; @@ -1619,26 +1581,11 @@ pub(crate) async fn sent_payment_verdicts( "Sent DashPay payment verdict" ); entry.status = next; - flipped_any = true; overlay - .overlay .entry(owner) .or_default() .insert(txid.clone(), entry.clone()); } - if flipped_any { - // Taken AFTER the flips above, so the snapshot carries the verdict - // rather than the status it replaced. This is the authoritative - // half: `load()` rehydrates a managed identity's payments from the - // identity blob this entry encodes and never reads the overlay - // table back (see the sqlite `dashpay` module's own header), so a - // round carrying only the overlay is a verdict that does not - // survive a restart. - overlay - .identities - .identities - .insert(owner, IdentityEntry::from_managed(managed)); - } } overlay } @@ -2676,29 +2623,15 @@ mod sent_payment_verdict_tests { } /// One row, so a test can assert the overlay is exactly the verdict and - /// not a replay of the identity's whole payment history. - /// - /// Also asserts the authoritative carrier agrees: the round's identity - /// snapshot must hold the same post-flip entry, because that snapshot — - /// not the overlay table — is what `load()` rebuilds the payment map - /// from. - fn only_row(verdicts: &SentPaymentVerdicts, txid: &str) -> PaymentEntry { - let overlay = &verdicts.overlay; - assert_eq!(overlay.len(), 1, "exactly one identity: {overlay:?}"); - let rows = overlay.get(&owner()).expect("the owning identity"); + /// not a replay of the identity's whole payment history — and that it is + /// the ONLY thing the round carries about the identity. A whole + /// `IdentityEntry` alongside it is precisely the lost-update hazard the + /// adapter must not reintroduce (dashpay/platform#4651). + fn only_row(verdicts: &super::PaymentOverlay, txid: &str) -> PaymentEntry { + assert_eq!(verdicts.len(), 1, "exactly one identity: {verdicts:?}"); + let rows = verdicts.get(&owner()).expect("the owning identity"); assert_eq!(rows.len(), 1, "exactly one row: {rows:?}"); - let row = rows.get(txid).expect("the flipped row").clone(); - let snapshot = verdicts - .identities - .identities - .get(&owner()) - .expect("the round must carry the owning identity's snapshot"); - assert_eq!( - snapshot.dashpay_payments.get(txid), - Some(&row), - "the identity snapshot must carry the verdict, not the status it replaced" - ); - row + rows.get(txid).expect("the flipped row").clone() } /// The defect: a swept sent payment stayed `Pending` forever because @@ -4396,11 +4329,13 @@ mod tests { /// round carried none, which is also what a payments-blind persister /// must see after the capability gate has withheld one. dashpay_payments: Option, - /// The payment maps carried by the round's identity snapshots. This - /// is the carrier `load()` actually rebuilds `dashpay_payments` from, - /// so a verdict that reaches only `dashpay_payments` above does not - /// survive a restart. `None` when the round carried no `identities` - /// sub-changeset at all. + /// The payment maps carried by the round's identity snapshots. + /// `None` when the round carried no `identities` sub-changeset at + /// all, which is what a verdict round must look like: the adapter + /// captures its flips under a lock it has long released by the time + /// `store()` runs, so a whole-identity snapshot here would be a + /// stale wholesale replace (dashpay/platform#4651). Durability for + /// the overlay rows is the persister's job, not a second carrier's. dashpay_identity_payments: Option, rejected: bool, } @@ -6200,7 +6135,7 @@ mod tests { super::WalletBatch { core: CoreChangeSet::default(), asset_locks, - payments: Default::default(), + payments: super::PaymentOverlay::default(), }, ); commit_batch( @@ -6260,7 +6195,7 @@ mod tests { WalletBatch { core, asset_locks: AssetLockChangeSet::default(), - payments: Default::default(), + payments: super::PaymentOverlay::default(), }, ); batch @@ -6280,38 +6215,6 @@ mod tests { )]) } - /// One sent-payment verdict, in the shape the adapter folds into a - /// wallet's batch: the bounded overlay row plus the identity snapshot - /// that makes it survive a restart. - fn one_verdict(status: crate::wallet::identity::PaymentStatus) -> super::SentPaymentVerdicts { - let overlay = one_verdict_overlay(status); - let owner = Identifier::from([0xAA; 32]); - let mut identities = crate::changeset::IdentityChangeSet::default(); - identities.identities.insert( - owner, - crate::changeset::IdentityEntry { - id: owner, - balance: 0, - revision: 0, - identity_index: None, - last_updated_balance_block_time: None, - last_synced_keys_block_time: None, - dpns_names: Vec::new(), - contested_dpns_names: Vec::new(), - status: Default::default(), - wallet_id: None, - dashpay_profile: None, - dashpay_payments: overlay.get(&owner).cloned().unwrap_or_default(), - contact_profiles: Default::default(), - ignored_senders: Default::default(), - }, - ); - super::SentPaymentVerdicts { - overlay, - identities, - } - } - /// A persister that attests `DASHPAY_PAYMENTS` gets the verdict on the /// same round as everything else the drain folded. #[test] @@ -6333,7 +6236,7 @@ mod tests { WalletBatch { core: watermark_with_rows(700, 700), asset_locks: AssetLockChangeSet::default(), - payments: one_verdict(PaymentStatus::Failed), + payments: one_verdict_overlay(PaymentStatus::Failed), }, ); let diag = commit_batch( @@ -6353,10 +6256,10 @@ mod tests { "the verdict must ride the same store() as the rows that justify it" ); assert_eq!( - observed.dashpay_identity_payments, - Some(one_verdict_overlay(PaymentStatus::Failed)), - "and the identity snapshot must ride it too, or the verdict is undone \ - by the next load()" + observed.dashpay_identity_payments, None, + "and NOTHING else about the identity: a whole snapshot captured before \ + the manager lock was released would erase a concurrently persisted \ + payment when it landed (dashpay/platform#4651)" ); assert_eq!(diag.persisted, Some(700)); assert_eq!(diag.faulted, 0, "a payments-capable host is not a fault"); @@ -6385,7 +6288,7 @@ mod tests { WalletBatch { core: watermark_with_rows(700, 700), asset_locks: AssetLockChangeSet::default(), - payments: one_verdict(PaymentStatus::Failed), + payments: one_verdict_overlay(PaymentStatus::Failed), }, ); let diag = commit_batch( @@ -6405,8 +6308,8 @@ mod tests { ); assert_eq!( observed.dashpay_identity_payments, None, - "both carriers are withheld together — a snapshot smuggling the \ - verdict past the gate would make DASHPAY_PAYMENTS meaningless" + "the overlay is the verdict's only carrier, so the gate covers all of \ + it — nothing smuggles a verdict past DASHPAY_PAYMENTS" ); assert_eq!( diag.persisted, @@ -6437,7 +6340,7 @@ mod tests { WalletBatch { core: CoreChangeSet::default(), asset_locks: AssetLockChangeSet::default(), - payments: one_verdict(PaymentStatus::Failed), + payments: one_verdict_overlay(PaymentStatus::Failed), }, ); commit_batch( @@ -6479,7 +6382,7 @@ mod tests { WalletBatch { core: CoreChangeSet::default(), asset_locks: AssetLockChangeSet::default(), - payments: one_verdict(PaymentStatus::Confirmed), + payments: one_verdict_overlay(PaymentStatus::Confirmed), }, ); commit_batch( @@ -6498,10 +6401,7 @@ mod tests { observed.dashpay_payments, Some(one_verdict_overlay(PaymentStatus::Confirmed)) ); - assert_eq!( - observed.dashpay_identity_payments, - Some(one_verdict_overlay(PaymentStatus::Confirmed)) - ); + assert_eq!(observed.dashpay_identity_payments, None); } /// The sweep guard strips the height BEFORE the store sees it, so a @@ -6921,9 +6821,9 @@ mod tests { observed } - /// The row the round must carry, in both carriers, with the assertion - /// that there is exactly one of it: a verdict is a status flip, never a - /// replay of the identity's payment history. + /// The row the round must carry, with the assertion that there is exactly + /// one of it and nothing else: a verdict is a status flip, never a replay + /// of the identity's payment history. fn sole_verdict_row(observed: &StoreObserved, txid: &str) -> PaymentStatus { let overlay = observed .dashpay_payments @@ -6933,19 +6833,12 @@ mod tests { let rows = overlay.get(&owner()).expect("the owning identity"); assert_eq!(rows.len(), 1, "exactly one row: {rows:?}"); let status = rows.get(txid).expect("the flipped row").status; - - let snapshot = observed - .dashpay_identity_payments - .as_ref() - .expect("the round must carry the identity snapshot too"); assert_eq!( - snapshot - .get(&owner()) - .and_then(|p| p.get(txid)) - .map(|e| e.status), - Some(status), - "the authoritative carrier must agree with the overlay — `load()` \ - rebuilds payments from the snapshot, never from the overlay table" + observed.dashpay_identity_payments, None, + "and the round must carry NO identity snapshot — the adapter's lock is \ + long released by `store()`, so a snapshot would be stale on arrival \ + (dashpay/platform#4651). The persister patches the overlay into the \ + authoritative blob inside its own transaction instead." ); status }