fix(key-wallet): out-of-order UTXO spend causes history divergence (#649)

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

@@ -575,7 +575,15 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
                 if !scanned_wallets.is_empty() {
                     let mut wallet = self.wallet.write().await;
                     for wallet_id in &scanned_wallets {
-                        wallet.update_wallet_synced_height(wallet_id, end);
+                        // Contiguity guard: a batch extends a wallet's certified
+                        // coverage only if the wallet was already certified up to
+                        // the batch's start. A checkpoint rewound after this batch
+                        // was scanned (an account added mid-flight) stays behind
+                        // and is picked up by the tick rescan instead of being
+                        // silently clobbered forward (dashpay/rust-dashcore#649).
+                        if wallet.wallet_synced_height(wallet_id).saturating_add(1) >= batch_start {
+                            wallet.update_wallet_synced_height(wallet_id, end);
+                        }
                     }
                 }
             }
@@ -1373,6 +1381,154 @@ mod tests {
         assert_eq!(multi.read().await.wallet_synced_height(&wallet_b), 0);
     }
 
+    /// Contiguity guard (dashpay/rust-dashcore#649): a batch scanned before an
+    /// account-add rewinds the wallet's checkpoint must NOT clobber the rewound
+    /// value forward at commit time — otherwise the account-addition rescan is
+    /// silently cancelled. The wallet stays behind and the tick picks it up.
+    #[tokio::test]
+    async fn mid_flight_account_add_does_not_clobber_rescan_floor() {
+        let wallet_a: WalletId = [0xAA; 32];
+        let multi = Arc::new(RwLock::new(MultiMockWallet::new()));
+        {
+            let mut w = multi.write().await;
+            // Already synced up to 4999, so the next batch legitimately starts at 5000.
+            w.insert_wallet(
+                wallet_a,
+                MockWalletState {
+                    synced_height: 4999,
+                    ..MockWalletState::default()
+                },
+            );
+        }
+        let mut manager = create_multi_test_manager(multi.clone()).await;
+        manager.set_state(SyncState::Syncing);
+
+        // Batch [5000..9999] scanned with wallet_a recorded, ready to commit.
+        let mut batch = FiltersBatch::new(5000, 9999, HashMap::new());
+        batch.set_pending_blocks(0);
+        batch.mark_scanned();
+        batch.mark_rescan_complete();
+        batch.set_scanned_wallets(BTreeSet::from([wallet_a]));
+        manager.active_batches.insert(5000, batch);
+
+        // Simulate an account being added mid-flight: the wallet's checkpoint is
+        // rewound to just below birth, far below this batch's start.
+        multi.write().await.wallet_mut(&wallet_a).synced_height = 49;
+
+        manager.try_commit_batches().await.unwrap();
+
+        // committed_height still advances (chain progress), but the wallet's
+        // checkpoint is NOT dragged forward over the gap it must rescan.
+        assert_eq!(manager.progress.committed_height(), 9999);
+        assert_eq!(
+            multi.read().await.wallet_synced_height(&wallet_a),
+            49,
+            "the rewound checkpoint must survive commit — the batch is non-contiguous with it"
+        );
+        // The wallet is still behind, so the next tick rescans it.
+        assert!(
+            multi.read().await.wallets_behind(9999).contains(&wallet_a),
+            "the wallet remains behind and is picked up by the tick rescan"
+        );
+    }
+
+    /// The contiguity guard is transparent in normal operation: contiguous
+    /// ascending batches advance each wallet's checkpoint exactly as before.
+    #[tokio::test]
+    async fn contiguity_guard_permits_normal_advance() {
+        let wallet_a: WalletId = [0xCC; 32];
+        let multi = Arc::new(RwLock::new(MultiMockWallet::new()));
+        multi.write().await.insert_wallet(wallet_a, MockWalletState::default());
+        let mut manager = create_multi_test_manager(multi.clone()).await;
+        manager.set_state(SyncState::Syncing);
+
+        // First batch starts at 0 = synced_height (0) + ... the wallet is fresh,
+        // so this batch is contiguous and advances the checkpoint.
+        let mut batch1 = FiltersBatch::new(0, 4999, HashMap::new());
+        batch1.set_pending_blocks(0);
+        batch1.mark_scanned();
+        batch1.mark_rescan_complete();
+        batch1.set_scanned_wallets(BTreeSet::from([wallet_a]));
+        manager.active_batches.insert(0, batch1);
+
+        manager.try_commit_batches().await.unwrap();
+        assert_eq!(multi.read().await.wallet_synced_height(&wallet_a), 4999);
+
+        // Second batch starts exactly at synced_height + 1 = 5000: still
+        // contiguous, so it advances too.
+        let mut batch2 = FiltersBatch::new(5000, 9999, HashMap::new());
+        batch2.set_pending_blocks(0);
+        batch2.mark_scanned();
+        batch2.mark_rescan_complete();
+        batch2.set_scanned_wallets(BTreeSet::from([wallet_a]));
+        manager.active_batches.insert(5000, batch2);
+
+        manager.try_commit_batches().await.unwrap();
+        assert_eq!(
+            multi.read().await.wallet_synced_height(&wallet_a),
+            9999,
+            "contiguous ascending batches advance the checkpoint unchanged by the guard"
+        );
+    }
+
+    /// Marvin QA-001 round 2: two wallets in the SAME batch, only one of them
+    /// rewound mid-flight (simulating an account add on wallet_a only). The
+    /// contiguity guard must block wallet_a's advance while still advancing
+    /// wallet_b normally — a per-wallet leak here would either strand a
+    /// healthy wallet or silently clobber the rewound one.
+    #[tokio::test]
+    async fn contiguity_guard_is_per_wallet_not_batch_wide() {
+        let wallet_a: WalletId = [0xAA; 32];
+        let wallet_b: WalletId = [0xBB; 32];
+        let multi = Arc::new(RwLock::new(MultiMockWallet::new()));
+        {
+            let mut w = multi.write().await;
+            w.insert_wallet(
+                wallet_a,
+                MockWalletState {
+                    synced_height: 4999,
+                    ..MockWalletState::default()
+                },
+            );
+            w.insert_wallet(
+                wallet_b,
+                MockWalletState {
+                    synced_height: 4999,
+                    ..MockWalletState::default()
+                },
+            );
+        }
+        let mut manager = create_multi_test_manager(multi.clone()).await;
+        manager.set_state(SyncState::Syncing);
+
+        // Batch [5000..9999] scanned with BOTH wallets recorded.
+        let mut batch = FiltersBatch::new(5000, 9999, HashMap::new());
+        batch.set_pending_blocks(0);
+        batch.mark_scanned();
+        batch.mark_rescan_complete();
+        batch.set_scanned_wallets(BTreeSet::from([wallet_a, wallet_b]));
+        manager.active_batches.insert(5000, batch);
+
+        // Only wallet_a gets an account added mid-flight (rewound). wallet_b
+        // is untouched and remains legitimately contiguous with this batch.
+        multi.write().await.wallet_mut(&wallet_a).synced_height = 49;
+
+        manager.try_commit_batches().await.unwrap();
+
+        assert_eq!(manager.progress.committed_height(), 9999);
+        assert_eq!(
+            multi.read().await.wallet_synced_height(&wallet_a),
+            49,
+            "wallet_a's rewound checkpoint must survive commit"
+        );
+        assert_eq!(
+            multi.read().await.wallet_synced_height(&wallet_b),
+            9999,
+            "wallet_b was legitimately contiguous and must still advance despite \
+             sharing a batch with a rewound wallet"
+        );
+    }
+
     /// `scan_batch` with two wallets at different `synced_height` values:
     /// only the wallet whose synced_height is below the matching block's
     /// height should be attributed.

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

@@ -0,0 +1,74 @@
+//! 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, Network, 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 dummy address (`ext_id` selects a distinct address).
+pub fn spend_tx(outpoint: OutPoint, value: u64, ext_id: usize) -> Transaction {
+    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: Address::dummy(Network::Testnet, ext_id).script_pubkey(),
+        }],
+        special_transaction_payload: None,
+    }
+}

key-wallet-manager/tests/observed_spent_large_block_stress_test.rs

@@ -0,0 +1,143 @@
+//! 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`, and — for transactions it judges irrelevant —
+//! also calls `remove_spent_from_accounts`, which allocates a
+//! `Vec<&mut ManagedCoreFundsAccount>` via `all_funding_accounts_mut()` and
+//! attempts a UTXO-map removal per input, for every one of those irrelevant
+//! transactions.
+//!
+//! The sibling `many_orphaned_spend_blocks_stay_bounded_and_linear`
+//! (`key-wallet/src/tests/observed_spent_outpoints_tests.rs`) spreads growth
+//! across many separate `check_transaction` calls at distinct heights. That
+//! does not exercise what a real matched block looks like: many transactions
+//! sharing ONE height/block, most of them entirely unrelated to the wallet.
+//! 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:?}"
+    );
+}