fix(key-wallet): out-of-order UTXO spend recorded nowhere — phantom unspent balance (#649)

dash-spv/src/sync/filters/batch.rs

@@ -1,7 +1,7 @@
 use dashcore::bip158::BlockFilter;
 use dashcore::ScriptBuf;
 use key_wallet_manager::{FilterMatchKey, WalletId};
-use std::collections::{BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
 /// A completed batch of compact block filters ready for verification.
 ///
@@ -24,10 +24,15 @@ pub(super) struct FiltersBatch {
     pending_blocks: u32,
     /// Whether rescan has been completed for this batch.
     rescan_complete: bool,
-    /// Wallets that were behind for this batch's height range at scan time and
-    /// therefore need their `synced_height` advanced when the batch commits.
-    /// Already-synced wallets must not be touched.
-    scanned_wallets: BTreeSet<WalletId>,
+    /// Wallets that were behind for this batch's height range at scan time —
+    /// and therefore need their `synced_height` advanced when the batch
+    /// commits — each mapped to the wallet's `account_generation` at scan
+    /// time. Commit refuses to advance a wallet whose generation has changed
+    /// since the scan: an account added mid-flight means the scan did not test
+    /// the new account's scripts, so the batch cannot certify coverage for the
+    /// current account set (dashpay/rust-dashcore#649). Already-synced wallets
+    /// must not be touched.
+    scanned_wallets: BTreeMap<WalletId, u64>,
     /// Cached scriptPubKeys discovered during block processing that still
     /// need rescan, attributed per wallet so we can rerun matching only
     /// against the wallet that produced each new script.
@@ -49,7 +54,7 @@ impl FiltersBatch {
             scanned: false,
             pending_blocks: 0,
             rescan_complete: false,
-            scanned_wallets: BTreeSet::new(),
+            scanned_wallets: BTreeMap::new(),
             collected_scripts: HashMap::new(),
         }
     }
@@ -118,13 +123,14 @@ impl FiltersBatch {
     pub(super) fn take_collected_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
         std::mem::take(&mut self.collected_scripts)
     }
-    /// Record the set of wallets that were behind for this batch at scan time.
-    pub(super) fn set_scanned_wallets(&mut self, wallets: BTreeSet<WalletId>) {
+    /// Record the wallets that were behind for this batch at scan time, each
+    /// with its `account_generation` snapshot.
+    pub(super) fn set_scanned_wallets(&mut self, wallets: BTreeMap<WalletId, u64>) {
         self.scanned_wallets = wallets;
     }
-    /// Wallets that were behind at scan time and must have their synced_height
-    /// advanced when this batch commits.
-    pub(super) fn scanned_wallets(&self) -> &BTreeSet<WalletId> {
+    /// Wallets that were behind at scan time (with their generation snapshot)
+    /// and must have their synced_height advanced when this batch commits.
+    pub(super) fn scanned_wallets(&self) -> &BTreeMap<WalletId, u64> {
         &self.scanned_wallets
     }
 }

key-wallet-manager/src/process_block.rs

@@ -274,6 +274,10 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
         self.wallet_infos.get(wallet_id).map(|info| info.synced_height()).unwrap_or(0)
     }
 
+    fn wallet_account_generation(&self, wallet_id: &WalletId) -> u64 {
+        self.wallet_infos.get(wallet_id).map(|info| info.account_generation()).unwrap_or(0)
+    }
+
     fn update_wallet_synced_height(&mut self, wallet_id: &WalletId, height: CoreBlockHeight) {
         if let Some(info) = self.wallet_infos.get_mut(wallet_id) {
             if height > info.synced_height() {

key-wallet-manager/src/test_utils/mock_wallet.rs

@@ -380,6 +380,9 @@ pub struct MockWalletState {
     pub addresses: Vec<Address>,
     pub synced_height: CoreBlockHeight,
     pub last_processed_height: CoreBlockHeight,
+    /// Mirrors `ManagedWalletInfo::account_generation`: bumped by tests to
+    /// simulate an account being added to this wallet mid-flight.
+    pub account_generation: u64,
 }
 
 /// Multi-wallet mock that holds independent state for several wallet IDs,
@@ -492,6 +495,10 @@ impl WalletInterface for MultiMockWallet {
         self.wallets.get(wallet_id).map(|s| s.synced_height).unwrap_or(0)
     }
 
+    fn wallet_account_generation(&self, wallet_id: &WalletId) -> u64 {
+        self.wallets.get(wallet_id).map(|s| s.account_generation).unwrap_or(0)
+    }
+
     fn update_wallet_synced_height(&mut self, wallet_id: &WalletId, height: CoreBlockHeight) {
         if let Some(state) = self.wallets.get_mut(wallet_id) {
             if height > state.synced_height {

key-wallet-manager/src/wallet_interface.rs

@@ -137,6 +137,17 @@ pub trait WalletInterface: Send + Sync + 'static {
     /// Return the per-wallet committed sync checkpoint, or `0` if unknown.
     fn wallet_synced_height(&self, wallet_id: &WalletId) -> CoreBlockHeight;
 
+    /// Return the generation of one wallet's account set — a counter bumped
+    /// whenever an account is added to that wallet (`0` if unknown). Filter
+    /// sync snapshots this per wallet when scanning a range and refuses to
+    /// advance `wallet_synced_height` at commit time if it changed in between:
+    /// the scan did not test the added account's scripts, so it cannot certify
+    /// coverage for the current account set (dashpay/rust-dashcore#649). The
+    /// default (constant `0`) opts an implementation out of the check.
+    fn wallet_account_generation(&self, _wallet_id: &WalletId) -> u64 {
+        0
+    }
+
     /// Advance one wallet's committed sync checkpoint. Implementations must
     /// only advance forward (a value below the current is silently ignored).
     fn update_wallet_synced_height(&mut self, wallet_id: &WalletId, height: CoreBlockHeight);

key-wallet-manager/tests/common/mod.rs

@@ -0,0 +1,87 @@
+//! Shared helpers for the observed-spent-outpoint guard integration tests
+//! (dashpay/rust-dashcore#649).
+//!
+//! Each test binary pulls this in via `mod common;` and uses a subset of the
+//! helpers, so unused-item warnings are expected per binary and silenced here.
+#![allow(dead_code)]
+
+use dashcore::blockdata::block::Block;
+use dashcore::blockdata::transaction::OutPoint;
+use dashcore::{Address, ScriptBuf, Transaction, TxIn, TxOut, Witness};
+use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
+use key_wallet_manager::{WalletId, WalletInterface, WalletManager};
+use std::collections::BTreeSet;
+
+/// Deliver `block` at `height` to the given `wallets`.
+pub async fn process_block_for(
+    manager: &mut WalletManager<ManagedWalletInfo>,
+    block: &Block,
+    height: u32,
+    wallets: &BTreeSet<WalletId>,
+) {
+    manager.process_block_for_wallets(block, block.block_hash(), height, wallets).await;
+}
+
+/// Deliver `block` at `height` to every wallet the manager holds.
+pub async fn process_block_all_wallets(
+    manager: &mut WalletManager<ManagedWalletInfo>,
+    block: &Block,
+    height: u32,
+) {
+    let wallet_ids: BTreeSet<WalletId> = manager.list_wallets().into_iter().copied().collect();
+    process_block_for(manager, block, height, &wallet_ids).await;
+}
+
+/// A transaction paying `value` to `address` from one synthetic, unrelated
+/// input (`input_seed` makes its txid deterministic and distinct). Only the
+/// transaction's own txid/vout matter to the tests, as the coin later spent.
+pub fn funding_tx(address: &Address, value: u64, input_seed: u8) -> Transaction {
+    Transaction {
+        version: 2,
+        lock_time: 0,
+        input: vec![TxIn {
+            previous_output: OutPoint::new(dashcore::Txid::from([input_seed; 32]), 0),
+            script_sig: ScriptBuf::new(),
+            sequence: 0xffffffff,
+            witness: Witness::new(),
+        }],
+        output: vec![TxOut {
+            value,
+            script_pubkey: address.script_pubkey(),
+        }],
+        special_transaction_payload: None,
+    }
+}
+
+/// A transaction spending `outpoint` and paying `value` to an unrelated
+/// external payee (`ext_id` selects a distinct script).
+///
+/// The payee script is built directly (a P2PKH shape seeded by `ext_id`) rather
+/// than via `Address::dummy`, so the helper needs no `Network`: the
+/// observed-spend logic keys on the input outpoint, never on the payee, so the
+/// payee's network is irrelevant to every scenario here. `Transaction::dummy`
+/// does not fit this helper — it derives its inputs from an id range and cannot
+/// spend a caller-specified `outpoint`, which is the whole point of a spend.
+pub fn spend_tx(outpoint: OutPoint, value: u64, ext_id: usize) -> Transaction {
+    // OP_DUP OP_HASH160 <20-byte push> OP_EQUALVERIFY OP_CHECKSIG — a well-formed
+    // P2PKH scriptPubKey whose hash160 is seeded by `ext_id` so each payee is
+    // distinct and never collides with a wallet-owned script.
+    let mut payee = vec![0x76, 0xa9, 0x14];
+    payee.extend_from_slice(&[ext_id as u8; 20]);
+    payee.extend_from_slice(&[0x88, 0xac]);
+    Transaction {
+        version: 2,
+        lock_time: 0,
+        input: vec![TxIn {
+            previous_output: outpoint,
+            script_sig: ScriptBuf::new(),
+            sequence: 0xffffffff,
+            witness: Witness::new(),
+        }],
+        output: vec![TxOut {
+            value,
+            script_pubkey: ScriptBuf::from_bytes(payee),
+        }],
+        special_transaction_payload: None,
+    }
+}

key-wallet-manager/tests/observed_spent_large_block_stress_test.rs

@@ -0,0 +1,136 @@
+//! Adversarial stress test for the observed-spent-outpoint guard
+//! (dashpay/rust-dashcore#649) at the *single-block* granularity, as opposed
+//! to spreading growth across many small blocks.
+//!
+//! `process_block_for_wallets` (`key-wallet-manager/src/process_block.rs`)
+//! iterates every transaction in `block.txdata` and calls
+//! `check_transaction_in_wallets` for EACH ONE, unconditionally — a matched
+//! block is delivered whole once ANY of its transactions matches a wallet's
+//! compact filter. `ManagedWalletInfo::check_core_transaction`
+//! (`key-wallet/src/transaction_checking/wallet_checker.rs`) then records
+//! every input of every transaction it sees in block context into
+//! `observed_spent_outpoints`, whether or not the transaction is relevant to the
+//! wallet — so a single matched block grows the map by one entry per input over
+//! the WHOLE block, not just the relevant tx.
+//!
+//! This test constructs a single `Block` with 5,000 unrelated 3-input
+//! transactions plus one wallet-relevant spend, and verifies both correctness
+//! (the bug is still caught when the relevant tx is buried in a large noisy
+//! block) and the actual growth/cost this produces.
+
+mod common;
+
+use common::process_block_all_wallets;
+use dashcore::blockdata::block::Block;
+use dashcore::blockdata::transaction::OutPoint;
+use dashcore::{Network, ScriptBuf, Transaction, TxIn, TxOut, Witness};
+use key_wallet::wallet::initialization::WalletAccountCreationOptions;
+use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
+use key_wallet_manager::WalletManager;
+use std::time::Instant;
+
+const NOISE_TXS_PER_BLOCK: usize = 5_000;
+const INPUTS_PER_NOISE_TX: usize = 3;
+
+/// An unrelated 3-input transaction paying an external address, with inputs
+/// deterministically derived from `seed` so every one is distinct and none
+/// resolves to any outpoint the wallet will ever fund.
+fn noise_tx(seed: u32) -> Transaction {
+    let input = (0..INPUTS_PER_NOISE_TX as u32)
+        .map(|v| {
+            let mut bytes = [0u8; 32];
+            bytes[..4].copy_from_slice(&seed.to_le_bytes());
+            bytes[4] = v as u8;
+            TxIn {
+                previous_output: OutPoint::new(dashcore::Txid::from(bytes), v),
+                script_sig: ScriptBuf::new(),
+                sequence: 0xffffffff,
+                witness: Witness::new(),
+            }
+        })
+        .collect();
+    Transaction {
+        version: 2,
+        lock_time: 0,
+        input,
+        output: vec![TxOut {
+            value: 1_000,
+            script_pubkey: dashcore::Address::dummy(Network::Testnet, (seed % 90) as usize + 1)
+                .script_pubkey(),
+        }],
+        special_transaction_payload: None,
+    }
+}
+
+#[tokio::test]
+async fn wallet_relevant_spend_buried_in_large_noisy_block_still_caught() {
+    let mut manager = WalletManager::<ManagedWalletInfo>::new(Network::Testnet);
+    let wallet_id = manager
+        .create_wallet_with_random_mnemonic(WalletAccountCreationOptions::Default)
+        .expect("failed to create wallet");
+    let funding_address =
+        manager.monitored_addresses().first().cloned().expect("wallet must have addresses");
+
+    let funding_value = 1_000_000u64;
+    let funding_tx = common::funding_tx(&funding_address, funding_value, 0xAB);
+    let funding_outpoint = OutPoint::new(funding_tx.txid(), 0);
+
+    let spend_tx = common::spend_tx(funding_outpoint, funding_value - 1_000, 99);
+
+    // One large block: 5,000 unrelated noise transactions with the single
+    // wallet-relevant spend buried in the middle — this is the realistic
+    // shape of "a matched block" (dash-spv delivers the whole block once any
+    // tx in it matches the filter), not 5,000 separate block deliveries.
+    let mut txdata: Vec<Transaction> = (0..NOISE_TXS_PER_BLOCK as u32).map(noise_tx).collect();
+    let mid = txdata.len() / 2;
+    txdata.insert(mid, spend_tx.clone());
+    let expected_inputs_in_spend_block =
+        NOISE_TXS_PER_BLOCK * INPUTS_PER_NOISE_TX + 1 /* spend_tx's own input */;
+
+    let spend_block = Block::dummy(200, txdata);
+
+    let start = Instant::now();
+    process_block_all_wallets(&mut manager, &spend_block, 200).await;
+    let large_block_elapsed = start.elapsed();
+
+    let funding_block = Block::dummy(100, vec![funding_tx.clone()]);
+    process_block_all_wallets(&mut manager, &funding_block, 100).await;
+
+    let utxos_after = manager.wallet_utxos(&wallet_id).expect("wallet must be registered");
+    let still_tracked = utxos_after.iter().any(|u| u.outpoint == funding_outpoint);
+    assert!(
+        !still_tracked,
+        "BUG #649 regression: the relevant spend buried among {NOISE_TXS_PER_BLOCK} unrelated \
+         transactions in the same block was not correctly reconciled against its \
+         out-of-order funding"
+    );
+
+    let observed_len =
+        manager.get_wallet_info(&wallet_id).expect("wallet info").observed_spent_outpoints().len();
+    let expected_total = expected_inputs_in_spend_block + 1 /* funding_tx's own input */;
+    assert_eq!(
+        observed_len, expected_total,
+        "a single matched block with {NOISE_TXS_PER_BLOCK} unrelated txs records one \
+         observed-spent entry per input in the WHOLE block, not just the relevant tx — \
+         set grew to {observed_len} entries from ONE block, confirming growth scales with \
+         transactions-per-block, not with block count"
+    );
+
+    eprintln!(
+        "single block with {} txs ({} inputs) processed in {:?} ({:.1} us/input); \
+         observed_spent_outpoints now holds {} entries after ONE block",
+        NOISE_TXS_PER_BLOCK + 1,
+        expected_inputs_in_spend_block,
+        large_block_elapsed,
+        large_block_elapsed.as_micros() as f64 / expected_inputs_in_spend_block as f64,
+        observed_len,
+    );
+
+    // Per-block processing cost, reported as a diagnostic rather than asserted:
+    // a wall-clock threshold flakes on shared CI runners, while correctness is
+    // already pinned above. A multi-second time here would flag the un-gated
+    // recording/removal seam as a production concern for busy blocks.
+    eprintln!(
+        "single block with {NOISE_TXS_PER_BLOCK} unrelated txs processed in {large_block_elapsed:?}"
+    );
+}