fix(platform-wallet): resolve a swept sent payment's verdict on the round that swept it

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`.

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<Identifier, BTreeMap<String, PaymentEntry>>,
+) -> 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<u8>)> = 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(())
 }

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<SqlitePersister>,
+    entered: Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
+    resume: Mutex<std::sync::mpsc::Receiver<()>>,
+}
+
+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<ClientStartState, PersistenceError> {
+        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");
+}