fix(key-wallet): stop a never-broadcast transaction from crediting money that does not exist

dash-spv-ffi/src/bin/ffi_cli.rs

@@ -226,6 +226,34 @@ extern "C" fn on_transaction_detected(
     );
 }
 
+extern "C" fn on_transactions_swept(
+    wallet_id: *const c_char,
+    txids: *const [u8; 32],
+    txids_count: usize,
+    superseded_by: *const [u8; 32],
+    balance: *const FFIBalance,
+    _account_balances: *const dash_spv_ffi::FFIAccountBalance,
+    _account_balances_count: u32,
+    _user_data: *mut c_void,
+) {
+    let wallet_short = short_wallet(wallet_id);
+    if txids.is_null() || superseded_by.is_null() {
+        println!("[Wallet] TXs swept: wallet={}..., null payload", wallet_short);
+        return;
+    }
+    let list = unsafe { std::slice::from_raw_parts(txids, txids_count) };
+    let winner = unsafe { &*superseded_by };
+    let b = read_balance(balance);
+    println!(
+        "[Wallet] TXs swept: wallet={}..., removed=[{}], superseded_by={}, balance[confirmed={}, unconfirmed={}]",
+        wallet_short,
+        list.iter().map(hex::encode).collect::<Vec<_>>().join(","),
+        hex::encode(winner),
+        b.confirmed,
+        b.unconfirmed,
+    );
+}
+
 extern "C" fn on_transaction_instant_locked(
     wallet_id: *const c_char,
     txid: *const [u8; 32],
@@ -548,6 +576,7 @@ fn main() {
             wallet: FFIWalletEventCallbacks {
                 on_transaction_detected: Some(on_transaction_detected),
                 on_transaction_instant_locked: Some(on_transaction_instant_locked),
+                on_transactions_swept: Some(on_transactions_swept),
                 on_block_processed: Some(on_wallet_block_processed),
                 on_sync_height_advanced: Some(on_sync_height_advanced),
                 on_chain_lock_processed: Some(on_wallet_chain_lock_processed),

dash-spv-ffi/src/callbacks.rs

@@ -752,6 +752,37 @@ pub type OnTransactionDetectedCallback = Option<
     ),
 >;
 
+/// Callback for `WalletEvent::TransactionsSwept`.
+///
+/// Fires when the wallet removes transactions that a later, final transaction
+/// provably beat to one of their inputs: they can never confirm, so their
+/// outputs are gone from the UTXO set and their records deleted.
+///
+/// **The only removal-shaped wallet callback.** Every other one is additive,
+/// so a consumer mirroring wallet state to disk must act on this — delete the
+/// named transactions and any UTXO they created. Ignoring it leaves the dead
+/// rows in the mirror, which replays them on the next load and re-creates a
+/// balance the wallet has already corrected.
+///
+/// `txids` points to `txids_count` consecutive 32-byte txids.
+/// `superseded_by` is the transaction whose arrival settled the inputs.
+/// All pointer parameters are borrowed and only valid for the duration of the
+/// callback. `balance` is the wallet's balance *after* the removal;
+/// `account_balances` follows the same contract as on
+/// [`OnTransactionDetectedCallback`].
+pub type OnTransactionsSweptCallback = Option<
+    extern "C" fn(
+        wallet_id: *const c_char,
+        txids: *const [u8; 32],
+        txids_count: usize,
+        superseded_by: *const [u8; 32],
+        balance: *const FFIBalance,
+        account_balances: *const FFIAccountBalance,
+        account_balances_count: u32,
+        user_data: *mut c_void,
+    ),
+>;
+
 /// Callback for `WalletEvent::TransactionInstantLocked`.
 ///
 /// Fires when an InstantSend lock is applied to a previously-seen off-chain
@@ -909,6 +940,7 @@ pub type OnWalletChainLockProcessedCallback = Option<
 pub struct FFIWalletEventCallbacks {
     pub on_transaction_detected: OnTransactionDetectedCallback,
     pub on_transaction_instant_locked: OnTransactionInstantLockedCallback,
+    pub on_transactions_swept: OnTransactionsSweptCallback,
     pub on_block_processed: OnWalletBlockProcessedCallback,
     pub on_sync_height_advanced: OnSyncHeightAdvancedCallback,
     pub on_chain_lock_processed: OnWalletChainLockProcessedCallback,
@@ -924,6 +956,7 @@ impl Default for FFIWalletEventCallbacks {
         Self {
             on_transaction_detected: None,
             on_transaction_instant_locked: None,
+            on_transactions_swept: None,
             on_block_processed: None,
             on_sync_height_advanced: None,
             on_chain_lock_processed: None,
@@ -1021,6 +1054,52 @@ impl FFIWalletEventCallbacks {
     /// Dispatch a WalletEvent to the appropriate callback.
     pub fn dispatch(&self, event: &WalletEvent) {
         match event {
+            WalletEvent::TransactionsSwept {
+                wallet_id,
+                txids,
+                superseded_by,
+                balance,
+                account_balances,
+            } => {
+                if let Some(cb) = self.on_transactions_swept {
+                    let wallet_id_hex = hex::encode(wallet_id);
+                    let c_wallet_id = CString::new(wallet_id_hex).unwrap_or_default();
+                    let raw_txids: Vec<[u8; 32]> =
+                        txids.iter().map(|t| t.to_byte_array()).collect();
+                    let raw_superseded_by = superseded_by.to_byte_array();
+                    let ffi_balance = FFIBalance::from(*balance);
+                    let ffi_account_balances = FFIAccountBalance::from_map(account_balances);
+                    let account_balances_ptr = if ffi_account_balances.is_empty() {
+                        ptr::null()
+                    } else {
+                        ffi_account_balances.as_ptr()
+                    };
+
+                    cb(
+                        c_wallet_id.as_ptr(),
+                        raw_txids.as_ptr(),
+                        raw_txids.len(),
+                        &raw_superseded_by as *const [u8; 32],
+                        &ffi_balance as *const FFIBalance,
+                        account_balances_ptr,
+                        ffi_account_balances.len() as u32,
+                        self.user_data,
+                    );
+
+                    drop(ffi_account_balances);
+                } else {
+                    // Deliberately loud: every other wallet callback is
+                    // additive, so a consumer that leaves this one unset keeps
+                    // transactions the wallet has already dropped.
+                    tracing::warn!(
+                        wallet_id = %hex::encode(wallet_id),
+                        swept = txids.len(),
+                        %superseded_by,
+                        "no on_transactions_swept callback set; the consumer will keep \
+                         mirroring transactions the wallet removed"
+                    );
+                }
+            }
             WalletEvent::TransactionDetected {
                 wallet_id,
                 record,

dash-spv-ffi/tests/dashd_sync/callbacks.rs

@@ -622,6 +622,8 @@ pub(super) fn create_wallet_callbacks(tracker: &Arc<CallbackTracker>) -> FFIWall
         on_block_processed: Some(on_wallet_block_processed),
         on_sync_height_advanced: Some(on_sync_height_advanced),
         on_chain_lock_processed: None,
+        // Not exercised by these tests: they never build a conflicting spend.
+        on_transactions_swept: None,
         user_data: Arc::as_ptr(tracker) as *mut c_void,
     }
 }

key-wallet-manager/src/events.rs

@@ -222,6 +222,29 @@ pub enum WalletEvent {
         /// full balance after the change — not a delta.
         account_balances: BTreeMap<AccountType, WalletCoreBalance>,
     },
+    /// Transactions were removed from the wallet: each was a recorded spend
+    /// that a later, final transaction provably beat to one of its inputs, so
+    /// it can never confirm. Their outputs are gone from the UTXO set and
+    /// their records deleted.
+    ///
+    /// The only removal-shaped event on this bus. A consumer mirroring wallet
+    /// state to disk must act on it — every other variant is additive, so
+    /// without this the mirror keeps the dead rows and replays them on the
+    /// next load, re-creating a balance the wallet has already corrected.
+    TransactionsSwept {
+        /// ID of the affected wallet.
+        wallet_id: WalletId,
+        /// Transactions removed. Delete these rows and any UTXO they created.
+        txids: Vec<Txid>,
+        /// The transaction whose arrival settled the inputs, for provenance.
+        superseded_by: Txid,
+        /// Wallet balance after the removal.
+        balance: WalletCoreBalance,
+        /// Post-event balance **snapshots** for accounts whose balance
+        /// changed as a result of this event. Each value is the account's
+        /// full balance after the change — not a delta.
+        account_balances: BTreeMap<AccountType, WalletCoreBalance>,
+    },
     /// A block was processed for a wallet. Carries records bucketed by what
     /// happened to them in this block, plus the post-block balance.
     /// `inserted` is records first stored in this block, `updated` is
@@ -332,6 +355,10 @@ impl WalletEvent {
                 wallet_id,
                 ..
             }
+            | WalletEvent::TransactionsSwept {
+                wallet_id,
+                ..
+            }
             | WalletEvent::TransactionInstantLocked {
                 wallet_id,
                 ..
@@ -382,6 +409,20 @@ impl fmt::Display for WalletEvent {
                 balance,
                 format_account_balances(account_balances),
             ),
+            WalletEvent::TransactionsSwept {
+                txids,
+                superseded_by,
+                balance,
+                account_balances,
+                ..
+            } => write!(
+                f,
+                "TransactionsSwept(count={}, superseded_by={}, balance={}, account_balances={})",
+                txids.len(),
+                superseded_by,
+                balance,
+                format_account_balances(account_balances),
+            ),
             WalletEvent::BlockProcessed {
                 height,
                 chain_lock,

key-wallet-manager/src/lib.rs

@@ -24,14 +24,16 @@ pub use events::{DerivedAddress, WalletEvent};
 pub use matching::{check_compact_filters_for_elements, FilterMatchKey};
 pub use wallet_interface::{BlockProcessingResult, MempoolTransactionResult, WalletInterface};
 
-use dashcore::blockdata::transaction::Transaction;
+use dashcore::blockdata::transaction::{OutPoint, Transaction};
 use dashcore::prelude::CoreBlockHeight;
+use dashcore::Txid;
 use key_wallet::account::AccountCollection;
 use key_wallet::managed_account::transaction_record::TransactionRecord;
 use key_wallet::transaction_checking::{DerivedAddressInfo, TransactionContext};
 use key_wallet::wallet::managed_wallet_info::coin_selection::SelectionStrategy;
 use key_wallet::wallet::managed_wallet_info::transaction_building::AccountTypePreference;
 use key_wallet::wallet::managed_wallet_info::wallet_info_interface::WalletInfoInterface;
+use key_wallet::wallet::managed_wallet_info::AbandonOutcome;
 use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
 use key_wallet::{AccountType, Address, ExtendedPrivKey, Mnemonic, Network, Wallet};
 use key_wallet::{ExtendedPubKey, WalletCoreBalance};
@@ -94,6 +96,12 @@ pub struct CheckTransactionsResult {
     /// Records whose state was updated by this check (confirmation or
     /// InstantSend lock on a previously stored record), grouped by wallet.
     pub per_wallet_updated_records: BTreeMap<WalletId, Vec<TransactionRecord>>,
+    /// Transactions this check *removed*, grouped by wallet: recorded spends
+    /// the arriving transaction provably beat to one of its inputs. See
+    /// [`crate::events::WalletEvent::TransactionsSwept`]
+    /// — a consumer mirroring wallet state must delete these rows, since no
+    /// other signal on the bus reports a removal.
+    pub per_wallet_swept: BTreeMap<WalletId, Vec<Txid>>,
 }
 
 impl CheckTransactionsResult {
@@ -628,6 +636,19 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
                     }
                 }
 
+                // Gathered outside the relevance branch above: a sweep can
+                // fire for a transaction this wallet finds irrelevant — the
+                // shared input is gone from `utxos` and the winner may pay
+                // only external addresses — and the removal still has to
+                // reach the consumer.
+                if !check_result.swept_transactions.is_empty() {
+                    result
+                        .per_wallet_swept
+                        .entry(*wallet_id)
+                        .or_default()
+                        .extend(check_result.swept_transactions);
+                }
+
                 if !check_result.new_addresses.is_empty() {
                     result
                         .new_addresses
@@ -662,6 +683,51 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
 }
 
 impl WalletManager<ManagedWalletInfo> {
+    /// Abandon `root` in `wallet_id`, and every recorded transaction
+    /// descending from it, then recompute the balance.
+    ///
+    /// The manager-level entry point for
+    /// [`ManagedWalletInfo::abandon_transaction_with_spends`] — the only
+    /// production path that clears a transaction the network never accepted.
+    /// The conflict sweep cannot reach that case: it needs a competing final
+    /// spend to prove the loser dead, and a transaction nobody ever saw has
+    /// no competitor. Its outputs would otherwise be credited forever, and as
+    /// trusted self-send change they are counted confirmed and are spendable.
+    ///
+    /// `external_spends` maps an outpoint to the transaction a caller's
+    /// persistence mirror recorded as spending it, for descendants whose own
+    /// records the load path never restored. Pass an empty map to walk only
+    /// the recorded transactions.
+    ///
+    /// **This asserts the root is dead; it does not establish it.** There is
+    /// no negative signal on the p2p network — Dash Core removed BIP61
+    /// `reject` — so silence is not proof, and abandoning a transaction that
+    /// is merely quiet re-exposes its inputs to coin selection. Settled roots
+    /// are refused, but the judgement otherwise belongs to the caller that
+    /// owns broadcast policy.
+    ///
+    /// Returns `None` when the wallet is unknown.
+    pub fn abandon_transaction(
+        &mut self,
+        wallet_id: &WalletId,
+        root: Txid,
+        external_spends: &BTreeMap<OutPoint, Txid>,
+    ) -> Option<AbandonOutcome> {
+        let info = self.get_wallet_info_mut(wallet_id)?;
+        let outcome = info.abandon_transaction_with_spends(root, external_spends);
+        if !outcome.is_empty() {
+            info.update_balance();
+            tracing::info!(
+                %root,
+                abandoned = outcome.abandoned.len(),
+                records_removed = outcome.records_removed,
+                utxos_removed = outcome.utxos_removed,
+                "Abandoned a dead transaction and everything built on it"
+            );
+        }
+        Some(outcome)
+    }
+
     /// Get receive address from a specific wallet and account
     pub fn next_receive_address(
         &mut self,

key-wallet-manager/src/process_block.rs

@@ -84,6 +84,26 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
             for (wallet_id, records) in check_result.per_wallet_updated_records {
                 per_wallet_updated.entry(wallet_id).or_default().extend(records);
             }
+            // Emitted per transaction rather than batched into the block
+            // event: a sweep names the transaction that superseded the
+            // removed ones, and that attribution is lost once the block's
+            // transactions are folded together.
+            for (wallet_id, txids) in check_result.per_wallet_swept {
+                if txids.is_empty() {
+                    continue;
+                }
+                let Some(info) = self.wallet_infos.get(&wallet_id) else {
+                    continue;
+                };
+                let event = WalletEvent::TransactionsSwept {
+                    wallet_id,
+                    txids,
+                    superseded_by: tx.txid(),
+                    balance: info.balance(),
+                    account_balances: BTreeMap::new(),
+                };
+                self.emit_event(event);
+            }
         }
 
         self.finalize_block_advance(
@@ -185,6 +205,29 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
             }
         }
 
+        // Removals, before the additive events: a consumer applying these in
+        // order sees the dead rows deleted first, so a replacement paying the
+        // same address cannot be clobbered by the delete that follows it.
+        for (wallet_id, txids) in std::mem::take(&mut check_result.per_wallet_swept) {
+            if txids.is_empty() {
+                continue;
+            }
+            let Some(info) = self.wallet_infos.get(&wallet_id) else {
+                continue;
+            };
+            let event = WalletEvent::TransactionsSwept {
+                wallet_id,
+                txids,
+                superseded_by: tx.txid(),
+                balance: info.balance(),
+                account_balances: per_wallet_account_diff
+                    .get(&wallet_id)
+                    .cloned()
+                    .unwrap_or_default(),
+            };
+            self.emit_event(event);
+        }
+
         if let Some(lock) = instant_lock {
             for (wallet_id, records) in per_wallet_updated_records {
                 if records.is_empty() {