diff --git a/Cargo.lock b/Cargo.lock index e2fcf55c579..bd993964162 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5355,6 +5355,8 @@ dependencies = [ "tempfile", "thiserror 1.0.69", "time", + "tokio", + "tokio-util", "tracing", "tracing-subscriber", "tracing-test", diff --git a/packages/rs-platform-wallet-storage/Cargo.toml b/packages/rs-platform-wallet-storage/Cargo.toml index 63e7491e009..fa375f50c5c 100644 --- a/packages/rs-platform-wallet-storage/Cargo.toml +++ b/packages/rs-platform-wallet-storage/Cargo.toml @@ -181,6 +181,17 @@ serial_test = "3" # so `secrets_mock_store_test_util.rs` proves the feature gate from the # outside — `cfg(test)` alone would satisfy it from within the crate. platform-wallet-storage = { path = ".", default-features = false, features = ["sqlite", "cli", "secrets", "kv", "__test-helpers", "test-util"] } +# `test-utils` exposes `platform_wallet::test_support::funded_wallet_manager`, +# which `sqlite_sent_payment_verdict_durability.rs` needs to stand up a real +# `WalletManager` and drive the wallet-event adapter against this crate's +# SQLite persister. Additive over the production dep above. +platform-wallet = { path = "../rs-platform-wallet", features = ["test-utils"] } +# The adapter is a tokio task; the durability test drives it and awaits its +# join handle. Current-thread runtime only — nothing here needs threads. +tokio = { version = "1", features = ["rt", "macros", "sync"] } +# The adapter's cancellation handle — the test passes a never-fired token, but +# the parameter is part of the public signature. +tokio-util = { version = "0.7", default-features = false } tempfile = "3" # `sqlite_hardening_3625.rs`, `sqlite_persist_roundtrip.rs`, and # `sqlite_load_reconstruction.rs` import `dash_sdk::platform::address_sync::AddressFunds`. 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-storage/tests/sqlite_sent_payment_verdict_durability.rs b/packages/rs-platform-wallet-storage/tests/sqlite_sent_payment_verdict_durability.rs new file mode 100644 index 00000000000..5dc90b1bec8 --- /dev/null +++ b/packages/rs-platform-wallet-storage/tests/sqlite_sent_payment_verdict_durability.rs @@ -0,0 +1,205 @@ +#![allow(clippy::field_reassign_with_default)] + +//! A sent-payment verdict must survive a process restart. +//! +//! The wallet-event adapter is the single writer of DashPay sent-payment +//! verdicts: when a sweep proves a broadcast lost a double-spend, the adapter +//! flips the payment to `Failed` on the same `store()` round that removes the +//! loser's record. That round is the ONLY chance — upstream selects 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, cannot repair it either. +//! +//! Which makes the durability question decisive rather than cosmetic. `load()` +//! rebuilds a managed identity's `dashpay_payments` from the identities +//! `entry_blob` and never reads `dashpay_payments_overlay` back (the +//! write-only overlay contract, pinned by +//! `sqlite_dashpay_overlay_contract.rs`). A round that carried the verdict +//! only on the overlay slot would therefore be undone by the next launch: the +//! payment reads `Pending` again, the transaction it names is gone, and +//! nothing can re-derive the verdict. +//! +//! So this test refuses to stop at the changeset. It stands up a real +//! `WalletManager`, seeds a `Pending` sent payment through the production +//! writer, drives the REAL adapter with a `TransactionsSwept` event against +//! this crate's SQLite persister, then closes the database and reopens it. +//! The assertion is on what `load()` hands back. + +mod common; + +use std::collections::BTreeMap; +use std::sync::atomic::AtomicBool; +use std::sync::Arc; + +use common::{ensure_wallet_meta, secure_tempdir}; +use dpp::identity::{Identity, IdentityV0}; +use dpp::prelude::Identifier; +use key_wallet::account::account_type::StandardAccountType; +use platform_wallet::changeset::{ + spawn_wallet_event_adapter, 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}; + +const OWNER: [u8; 32] = [0xA7; 32]; +const CONTACT: [u8; 32] = [0xB7; 32]; + +/// The transaction the sweep removes, and the key the payment entry is filed +/// under. A dummy txid is enough — the verdict path keys on the display +/// string, and nothing here inspects the transaction itself. +fn loser_txid() -> dashcore::Txid { + use dashcore::hashes::Hash as _; + dashcore::Txid::from_byte_array([0x5f; 32]) +} + +/// Read the sent payment's status out of a freshly `load()`ed start state — +/// the same path a relaunching client takes. +fn loaded_status( + persister: &SqlitePersister, + wallet_id: &[u8; 32], + txid: &str, +) -> Option { + let state = persister.load().expect("load must succeed after reopen"); + let wallet = state.wallets.get(wallet_id)?; + let managed = wallet + .identity_manager + .wallet_identities + .get(wallet_id)? + .get(&0)?; + managed + .dashpay() + .payments + .get(txid) + .map(|entry| entry.status) +} + +#[tokio::test] +async fn a_swept_sent_payments_failed_verdict_survives_a_reopen() { + let tmp = secure_tempdir().expect("tempdir"); + let path = tmp.path().join("wallet.db"); + let txid = loser_txid().to_string(); + + let (wallet_manager, wallet_id, _generation, _signer) = + funded_wallet_manager(StandardAccountType::BIP44Account).await; + + // ── Session one: an identity with a Pending sent payment, on disk. ── + { + let persister = Arc::new( + SqlitePersister::open(SqlitePersisterConfig::new(&path)).expect("open persister"), + ); + ensure_wallet_meta(&persister, &wallet_id); + let mut meta = PlatformWalletChangeSet::default(); + meta.wallet_metadata = Some(WalletMetadataEntry { + network: key_wallet::Network::Testnet, + wallet_group_id: wallet_id, + birth_height: 0, + }); + persister.store(wallet_id, meta).expect("store metadata"); + + let wallet_persister = WalletPersister::new( + wallet_id, + Arc::clone(&persister) 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: Identifier::from(OWNER), + public_keys: BTreeMap::new(), + balance: 0, + revision: 0, + }), + 0, + wallet_id, + &wallet_persister, + ) + .expect("add owner identity"); + // Seeded through the production writer, so the on-disk identity + // blob genuinely holds `Pending` before the sweep — the state the + // restart must NOT resurrect. + info.identity_manager + .managed_identity_mut(&Identifier::from(OWNER)) + .expect("managed identity") + .record_dashpay_payment( + txid.clone(), + PaymentEntry::new_sent(Identifier::from(CONTACT), 50_000, Some("lunch".into())), + &wallet_persister, + ) + .expect("record the pending payment"); + } + persister.flush(wallet_id).expect("flush session one"); + drop(wallet_persister); + + // ── The sweep, through the real adapter. ── + // + // Buffered before the adapter starts and the sender dropped straight + // after, so the drain folds the whole (one-event) backlog, commits it, + // and exits on `Disconnected`. No sleeping, no polling. + let (tx, rx) = tokio::sync::mpsc::unbounded_channel::(); + tx.send(WalletEvent::TransactionsSwept { + wallet_id, + txids: vec![loser_txid()], + superseded_by: { + use dashcore::hashes::Hash as _; + 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(), + }) + .expect("queue the sweep"); + drop(tx); + + spawn_wallet_event_adapter( + Arc::clone(&wallet_manager), + Arc::downgrade(&persister), + rx, + Arc::new(AtomicBool::new(false)), + tokio_util::sync::CancellationToken::new(), + ) + .await + .expect("the adapter exits cleanly once its backlog is committed"); + + persister.flush(wallet_id).expect("flush the verdict"); + // The round reached the store at all — asserted on the overlay table, + // NOT on `load()`. A verdict that only ever lands here is precisely + // the failure mode the reopen below exists to catch, so this check is + // deliberately blind to durability: it just rules out "the adapter + // never committed" as an explanation for a red reopen. + { + let conn = persister.lock_conn_for_test(); + let rows: i64 = conn + .query_row( + "SELECT COUNT(*) FROM dashpay_payments_overlay \ + WHERE identity_id = ?1 AND payment_id = ?2", + rusqlite::params![&OWNER[..], &txid], + |r| r.get(0), + ) + .expect("count overlay rows"); + assert_eq!(rows, 1, "the adapter's round must have reached store()"); + } + // The open-path registry refuses a second live persister on one file, + // so session one must be fully released before session two opens. + drop(persister); + } + + // ── Session two: a fresh process's view of the same file. ── + let reopened = + SqlitePersister::open(SqlitePersisterConfig::new(&path)).expect("reopen persister"); + assert_eq!( + loaded_status(&reopened, &wallet_id, &txid), + Some(PaymentStatus::Failed), + "a swept payment's Failed verdict must survive a restart. If this \ + reads Pending, the adapter's round carried the verdict only on \ + `dashpay_payments_overlay` — a table `load()` never reads — while \ + the authoritative identity blob kept the pre-sweep status, and the \ + swept transaction is gone with nothing able to re-derive it." + ); +} diff --git a/packages/rs-platform-wallet/src/changeset/changeset.rs b/packages/rs-platform-wallet/src/changeset/changeset.rs index 46dec66c4b3..775639f2bb2 100644 --- a/packages/rs-platform-wallet/src/changeset/changeset.rs +++ b/packages/rs-platform-wallet/src/changeset/changeset.rs @@ -2066,6 +2066,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 @@ -2260,12 +2279,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 3c7694d2e3f..6bdbccb8ced 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -51,13 +51,15 @@ use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use crate::changeset::changeset::{ - AssetLockChangeSet, CoreChangeSet, HighestUsedIndexes, PlatformWalletChangeSet, SweepBatch, - UtxoCreditVerdict, + merge_payment_overlays, AssetLockChangeSet, CoreChangeSet, HighestUsedIndexes, PaymentOverlay, + PlatformWalletChangeSet, SweepBatch, UtxoCreditVerdict, }; 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 @@ -419,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, @@ -435,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, @@ -493,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(); @@ -709,8 +719,45 @@ 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) + { + // 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.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. @@ -728,10 +775,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 @@ -782,6 +830,11 @@ 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. + // The persister is responsible for making these durable; the + // adapter never ships a whole identity of its own. + dashpay_payments_overlay: payments, ..PlatformWalletChangeSet::default() }; let store_result = persister.store(wallet_id, cs); @@ -949,6 +1002,21 @@ 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`]): 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 @@ -1308,6 +1376,234 @@ 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. +/// +/// # 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 +/// 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::default(); + 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`. /// @@ -2159,6 +2455,428 @@ 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]; + + pub(super) 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. + pub(super) 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. + pub(super) 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) + } + + pub(super) 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 + } + + pub(super) 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. + pub(super) 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 — 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:?}"); + 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 @@ -3576,11 +4294,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}; @@ -3600,6 +4321,22 @@ mod tests { n_records: usize, n_asset_locks: usize, n_asset_locks_removed: usize, + /// Swept transactions the round carries — the core change a verdict + /// must be coupled to, since the sweep never re-emits once its round + /// is durable. + n_sweeps: 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, + /// 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, } @@ -3700,6 +4437,14 @@ mod tests { .as_ref() .map(|a| a.removed.len()) .unwrap_or(0), + n_sweeps: core.map(|c| c.sweeps.len()).unwrap_or(0), + dashpay_payments: changeset.dashpay_payments_overlay.clone(), + dashpay_identity_payments: changeset.identities.as_ref().map(|ids| { + ids.identities + .iter() + .map(|(id, entry)| (*id, entry.dashpay_payments.clone())) + .collect() + }), rejected, }); if rejected { @@ -4983,11 +5728,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 @@ -5391,6 +6135,7 @@ mod tests { super::WalletBatch { core: CoreChangeSet::default(), asset_locks, + payments: super::PaymentOverlay::default(), }, ); commit_batch( @@ -5450,11 +6195,215 @@ mod tests { WalletBatch { core, asset_locks: AssetLockChangeSet::default(), + payments: super::PaymentOverlay::default(), }, ); batch } + /// The overlay half of [`one_verdict`], on its own — what a round's + /// `dashpay_payments_overlay` must look like. + fn one_verdict_overlay( + 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_overlay(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_overlay(PaymentStatus::Failed)), + "the verdict must ride the same store() as the rows that justify it" + ); + assert_eq!( + 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"); + } + + /// 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_overlay(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!( + observed.dashpay_identity_payments, None, + "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, + 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_overlay(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_overlay(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_overlay(PaymentStatus::Confirmed)) + ); + assert_eq!(observed.dashpay_identity_payments, None); + } + /// 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 @@ -5802,6 +6751,192 @@ mod tests { synced_height_frozen=None synced_height_rejected=Some(200) faulted=0" ); } + + // ── Verdicts through the real adapter drain ── + // + // The unit tests above call `sent_payment_verdicts` directly, and the + // `commit_batch` tests inject a ready `SentPaymentVerdicts` into a + // `WalletBatch`. Neither notices if the call disappears from one of the + // two fold sites in `run_wallet_event_adapter`, or if its result is + // dropped on the floor. These drive the loop itself and assert on what + // reaches the store. + // + // Determinism comes from buffering: every event is queued on the + // unbounded channel and the sender is dropped BEFORE the adapter is + // spawned, so the first `recv` and the `try_recv` fold that follows see + // the whole sequence and commit it as exactly one round. No sleeping, no + // racing the drain. + + use super::sent_payment_verdict_tests::{ + chainlocked_reinstatement, owner, sent_transaction, stored_status, sweep_of, + wallet_with_payment, + }; + use crate::wallet::identity::{PaymentDirection, PaymentStatus}; + + /// A probe that attests both bits this round needs: `CORE_SWEEP_REMOVAL` + /// so the sweep half is not withheld (which would fault the wallet and + /// strip the watermark), and `DASHPAY_PAYMENTS` so the verdict is not. + fn sweep_and_payments_probe(obs: UnboundedSender) -> Arc { + Arc::new(ProbePersister::with_capabilities( + obs, + crate::changeset::PersistenceCapabilities::CORE_SWEEP_REMOVAL + .union(crate::changeset::PersistenceCapabilities::DASHPAY_PAYMENTS), + )) + } + + /// Run `events` through the real adapter as one buffered drain and + /// return the single `store()` it produced. + async fn one_drained_round( + wallet_manager: Arc>>, + events: Vec, + ) -> StoreObserved { + let (tx, rx) = unbounded_channel::(); + for event in events { + tx.send(event).unwrap(); + } + // Dropped before the adapter starts: the backlog is already buffered, + // so the drain folds all of it and then exits on `Disconnected`. + drop(tx); + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = sweep_and_payments_probe(obs_tx); + let handle = tokio::spawn(run_wallet_event_adapter( + wallet_manager, + Arc::downgrade(&persister), + rx, + Arc::new(AtomicBool::new(false)), + CancellationToken::new(), + )); + handle.await.expect("the adapter exits on a closed channel"); + + let observed = obs_rx + .recv() + .await + .expect("the buffered drain must reach store()"); + assert!( + obs_rx.try_recv().is_err(), + "a buffered drain commits once — a second round would mean the \ + verdict and the rows that justify it were split across stores" + ); + observed + } + + /// 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 + .as_ref() + .expect("the round must carry the overlay"); + 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:?}"); + let status = rows.get(txid).expect("the flipped row").status; + assert_eq!( + 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 + } + + /// A sweep and the finality that overrules it, buffered into one drain in + /// that order. The store must see one row at the verdict the drain ended + /// on (`Confirmed -> Failed -> Confirmed` collapses to `Confirmed`), on + /// the same round as the record that justifies it. + /// + /// Starting at `Confirmed` rather than `Pending` is what makes this test + /// discriminate BOTH fold sites. `Confirmed + Final` is a no-op edge, so + /// an adapter that skipped the first event's verdict would emit no row at + /// all rather than the same one by a different route; and an adapter that + /// skipped the folded events' verdicts would stop at `Failed`. + #[tokio::test] + async fn a_buffered_sweep_then_final_reaches_the_store_as_one_confirmed_row() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Confirmed).await; + + let observed = one_drained_round( + Arc::clone(&wallet_manager), + vec![ + sweep_of(wallet_id, sent_transaction().txid()), + chainlocked_reinstatement(wallet_id), + ], + ) + .await; + + assert_eq!(observed.wallet_id, wallet_id); + assert_eq!( + sole_verdict_row(&observed, &txid), + PaymentStatus::Confirmed, + "last write wins inside a drain" + ); + assert_eq!( + observed.n_sweeps, 0, + "the reinstating record retracts the sweep inside the same core \ + merge (see `CoreChangeSet::merge`), so this round carries the \ + record and no removal — exactly what the Confirmed verdict says" + ); + assert_eq!( + observed.n_records, 1, + "the verdict rides the same round as the record that justifies it" + ); + assert_eq!( + observed.last_processed_height, + Some(1_499_060), + "and alongside the core watermark the same drain projected" + ); + assert!(!observed.rejected); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Confirmed, + "memory and the round must agree" + ); + } + + /// The same two events in the opposite order. `Failed -> Confirmed -> + /// Failed` collapses to `Failed`, which is the verdict that matters: + /// nothing else in the wallet ever writes it, and it must not be lost to + /// the confirm that preceded it in the same drain. + /// + /// Starting at `Failed` is the mirror of the test above: `Failed + Swept` + /// is the no-op edge here, so dropping either fold site's verdict changes + /// what reaches the store rather than arriving at it another way. + #[tokio::test] + async fn a_buffered_final_then_sweep_reaches_the_store_as_one_failed_row() { + let (wallet_manager, wallet_id, txid) = + wallet_with_payment(PaymentDirection::Sent, PaymentStatus::Failed).await; + + let observed = one_drained_round( + Arc::clone(&wallet_manager), + vec![ + chainlocked_reinstatement(wallet_id), + sweep_of(wallet_id, sent_transaction().txid()), + ], + ) + .await; + + assert_eq!(observed.wallet_id, wallet_id); + assert_eq!( + sole_verdict_row(&observed, &txid), + PaymentStatus::Failed, + "the sweep is the later evidence, so it is the drain's verdict" + ); + assert_eq!( + observed.n_sweeps, 1, + "the verdict must ride the sweep's own round — that coupling is \ + the whole point of resolving it here, since a sweep never \ + re-emits once its round is durable" + ); + assert_eq!(observed.last_processed_height, Some(1_499_060)); + assert!(!observed.rejected); + assert_eq!( + stored_status(&wallet_manager, &wallet_id, &txid).await, + PaymentStatus::Failed + ); + } } #[cfg(test)] diff --git a/packages/rs-platform-wallet/src/test_support.rs b/packages/rs-platform-wallet/src/test_support.rs index b0ec2af019c..b9446432ee0 100644 --- a/packages/rs-platform-wallet/src/test_support.rs +++ b/packages/rs-platform-wallet/src/test_support.rs @@ -178,7 +178,7 @@ impl ExtendedPubKeySigner for WalletSigner { /// rides on that one input, so a leaked reservation strands it. Returns /// the manager, the wallet id, the shared balance handle, and a soft /// signer over the wallet's seed. -pub(crate) async fn funded_wallet_manager( +pub async fn funded_wallet_manager( account_type: StandardAccountType, ) -> ( Arc>>, @@ -192,7 +192,7 @@ pub(crate) async fn funded_wallet_manager( /// Like [`funded_wallet_manager`] but with caller-chosen funding outputs — /// multiple outputs yield multiple spendable UTXOs, letting tests run /// concurrent asset-lock builds that each need their own input. -pub(crate) async fn funded_wallet_manager_with_outputs( +pub async fn funded_wallet_manager_with_outputs( account_type: StandardAccountType, outputs: &[u64], ) -> ( 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 6246e078725..d5bc97c6616 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");