fix(key-wallet): hold back catch-up receives from coin selection

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

@@ -273,6 +273,12 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         self.processing_height = scan_start;
         self.next_batch_to_store = download_start;
 
+        // Arm the spend-scan gate before any block is applied. A restored
+        // wallet starts scanning from its birth height with the tip thousands
+        // of blocks above it, and every receive found in between must be held
+        // back until the scan gets there.
+        self.publish_scan_target(self.progress.filter_header_tip_height()).await;
+
         // Check if already at target (nothing to download)
         if scan_start > self.progress.filter_header_tip_height() {
             // Park the idle pipeline at the download frontier so a later
@@ -1025,6 +1031,20 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         Ok(events)
     }
 
+    /// Tell the wallets how far this scan intends to reach.
+    ///
+    /// Coin selection uses the gap between a wallet's `synced_height` and this
+    /// target to hold back outputs discovered during catch-up, which a block
+    /// above the frontier may already have spent. Publishing a target of 0
+    /// would leave that gate open, so a not-yet-known tip is skipped rather
+    /// than published as 0.
+    async fn publish_scan_target(&self, tip_height: u32) {
+        if tip_height == 0 {
+            return;
+        }
+        self.wallet.write().await.update_scan_target_height(tip_height);
+    }
+
     /// Handle notification that new filter headers are available.
     /// Used by both FilterHeadersSyncComplete and FilterHeadersStored events.
     pub(super) async fn handle_new_filter_headers(
@@ -1034,6 +1054,7 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
     ) -> SyncResult<Vec<SyncEvent>> {
         self.progress.update_filter_header_tip_height(tip_height);
         self.update_target_height(tip_height);
+        self.publish_scan_target(tip_height).await;
 
         match self.state() {
             SyncState::Syncing | SyncState::Synced

key-wallet-manager/src/process_block.rs

@@ -278,6 +278,12 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
         self.wallet_infos.get(wallet_id).map(|info| info.account_generation()).unwrap_or(0)
     }
 
+    fn update_scan_target_height(&mut self, height: CoreBlockHeight) {
+        for info in self.wallet_infos.values_mut() {
+            info.update_scan_target_height(height);
+        }
+    }
+
     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/wallet_interface.rs

@@ -137,6 +137,18 @@ 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;
 
+    /// Publish the chain height the spend scan is working toward — the tip of
+    /// the filter-header chain the scanner has committed to covering. It is a
+    /// property of the chain rather than of any one wallet, so it applies to
+    /// all of them.
+    ///
+    /// Until a wallet's `synced_height` reaches this, that wallet is catching
+    /// up and cannot tell whether an output it just discovered was already
+    /// spent in a block it has not scanned yet, so such outputs are held back
+    /// from coin selection. The default is a no-op, leaving the gate open —
+    /// a scanner must call this for it to engage.
+    fn update_scan_target_height(&mut self, _height: 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

key-wallet/src/test_utils/wallet.rs

@@ -37,6 +37,19 @@ impl TestWalletContext {
     /// accounts.
     pub fn new_random_with_options(options: WalletAccountCreationOptions) -> Self {
         let wallet = Wallet::new_random(Network::Testnet, options).expect("Should create wallet");
+        Self::from_wallet(wallet)
+    }
+
+    /// Creates a testnet wallet from a fixed seed, so a failing test replays
+    /// with the same keys and addresses every run.
+    pub fn new_with_seed(seed: [u8; 64]) -> Self {
+        let wallet =
+            Wallet::from_seed_bytes(seed, Network::Testnet, WalletAccountCreationOptions::Default)
+                .expect("Should create wallet");
+        Self::from_wallet(wallet)
+    }
+
+    fn from_wallet(wallet: Wallet) -> Self {
         let mut managed_wallet =
             ManagedWalletInfo::from_wallet_with_name(&wallet, "Test".to_string(), 0);
 

key-wallet/src/tests/mod.rs

@@ -39,3 +39,5 @@ mod spent_outpoints_tests;
 mod unit_variant_wallet_tests;
 
 mod wallet_tests;
+
+mod spend_scan_frontier_tests;

key-wallet/src/tests/spend_scan_frontier_tests.rs

@@ -0,0 +1,211 @@
+//! Tests for the spend-scan frontier gate on coin selection.
+//!
+//! A wallet catching up on history applies blocks in ascending order, so a
+//! receive it discovers says nothing about whether some higher, not-yet-scanned
+//! block already spends it. Selecting such an output builds a transaction the
+//! network settled as a double-spend long ago; peers drop it silently, with no
+//! reject message, and whatever it was funding is stranded forever.
+//!
+//! The heights here are the ones from the incident that motivated this: a
+//! restored wallet found a 1000 DASH receive in block 758983 and funded an
+//! identity top-up from it three seconds later, while the block that had
+//! already spent that outpoint — height 1510203 — was still six minutes of
+//! scanning away.
+
+use dashcore::blockdata::transaction::{OutPoint, Transaction};
+use dashcore::hashes::Hash;
+use dashcore::{BlockHash, TxIn};
+
+use crate::test_utils::TestWalletContext;
+use crate::transaction_checking::{BlockInfo, TransactionContext};
+use crate::wallet::managed_wallet_info::coin_selection::{
+    CoinSelector, SelectionError, SelectionStrategy,
+};
+use crate::wallet::managed_wallet_info::fee::FeeRate;
+use crate::wallet::managed_wallet_info::wallet_info_interface::WalletInfoInterface;
+
+/// Fixed wallet seed so a failing test replays with the same keys every run.
+const SEED: [u8; 64] = [7; 64];
+
+/// Heights from the incident.
+const RECEIVE_HEIGHT: u32 = 758_983;
+const SPEND_HEIGHT: u32 = 1_510_203;
+const CHAIN_TIP: u32 = 2_200_000;
+
+fn block_at(height: u32) -> TransactionContext {
+    TransactionContext::InBlock(BlockInfo::new(
+        height,
+        BlockHash::from_slice(&[(height % 251) as u8; 32]).expect("hash"),
+        1_700_000_000,
+    ))
+}
+
+/// A transaction spending `outpoint`, standing in for the wallet's own earlier
+/// asset lock that consumed the coin long before this restore.
+fn spending_tx(outpoint: OutPoint) -> Transaction {
+    Transaction {
+        version: 1,
+        lock_time: 0,
+        input: vec![TxIn {
+            previous_output: outpoint,
+            ..Default::default()
+        }],
+        output: Vec::new(),
+        special_transaction_payload: None,
+    }
+}
+
+/// Ask coin selection for `target` duffs out of the wallet's BIP44 account,
+/// exactly as the transaction builder does.
+fn select(ctx: &TestWalletContext, target: u64) -> Result<u64, SelectionError> {
+    let account = ctx.bip44_account();
+    let utxos: Vec<_> = account.utxos.values().collect();
+    CoinSelector::new(SelectionStrategy::BranchAndBound)
+        .select_coins(utxos, target, FeeRate::normal(), ctx.managed_wallet.last_processed_height())
+        .map(|selection| selection.total_value)
+}
+
+/// Put the wallet in the state a restore lands in: scanning toward the chain
+/// tip from far below it, having applied a receive at `RECEIVE_HEIGHT`.
+async fn restored_wallet_mid_catch_up(amount: u64) -> (TestWalletContext, Transaction) {
+    let mut ctx = TestWalletContext::new_with_seed(SEED);
+    ctx.managed_wallet.update_scan_target_height(CHAIN_TIP);
+    ctx.managed_wallet.update_synced_height(RECEIVE_HEIGHT);
+    ctx.managed_wallet.update_last_processed_height(RECEIVE_HEIGHT);
+
+    let tx = Transaction::dummy(&ctx.receive_address, 0..1, &[amount]);
+    let result = ctx.check_transaction(&tx, block_at(RECEIVE_HEIGHT)).await;
+    assert!(result.is_relevant, "wallet must recognise its own receive");
+
+    (ctx, tx)
+}
+
+#[tokio::test]
+async fn receive_found_below_the_scan_target_is_tracked_but_not_selectable() {
+    let (ctx, tx) = restored_wallet_mid_catch_up(100_000_000_000).await;
+
+    let utxo = ctx.first_utxo();
+    assert_eq!(utxo.outpoint.txid, tx.txid(), "the receive is tracked");
+    assert!(utxo.is_confirmed, "and it is confirmed in a block");
+    assert!(!utxo.spend_scanned, "but the scan has not passed it yet");
+
+    assert!(
+        !ctx.managed_wallet.spend_scan_complete(),
+        "the wallet must report itself mid-catch-up"
+    );
+    assert!(
+        ctx.managed_wallet.get_spendable_utxos().is_empty(),
+        "nothing is spendable while the scan is behind"
+    );
+    assert!(
+        matches!(select(&ctx, 1_000_000), Err(SelectionError::NoUtxosAvailable)),
+        "coin selection must refuse to fund from an unscanned receive"
+    );
+}
+
+#[tokio::test]
+async fn scan_reaching_its_target_releases_the_held_back_receive() {
+    let (mut ctx, _tx) = restored_wallet_mid_catch_up(100_000_000_000).await;
+
+    // Catch-up finishes: every block up to the tip has been scanned and no
+    // spend of this outpoint turned up, so the coin is genuinely unspent.
+    ctx.managed_wallet.update_synced_height(CHAIN_TIP);
+    ctx.managed_wallet.update_last_processed_height(CHAIN_TIP);
+
+    assert!(ctx.managed_wallet.spend_scan_complete());
+    assert!(ctx.first_utxo().spend_scanned, "the scan has now covered it");
+    assert_eq!(ctx.managed_wallet.get_spendable_utxos().len(), 1, "and it becomes spendable");
+    assert_eq!(select(&ctx, 1_000_000).expect("selection succeeds"), 100_000_000_000);
+}
+
+/// The incident itself: the coin was already spent, by this wallet's own
+/// earlier transaction, in a block the restore had not reached yet.
+#[tokio::test]
+async fn coin_already_spent_above_the_frontier_is_never_selectable() {
+    let (mut ctx, tx) = restored_wallet_mid_catch_up(100_000_000_000).await;
+    let outpoint = ctx.first_utxo().outpoint;
+
+    // This is the window in which the top-up asset lock was built.
+    assert!(
+        select(&ctx, 1_000_000).is_err(),
+        "the double-spend must not be fundable during catch-up"
+    );
+
+    // The scan reaches the block that spent it.
+    ctx.managed_wallet.update_synced_height(SPEND_HEIGHT);
+    let spend = spending_tx(outpoint);
+    ctx.check_transaction(&spend, block_at(SPEND_HEIGHT)).await;
+
+    // ...and then finishes.
+    ctx.managed_wallet.update_synced_height(CHAIN_TIP);
+    ctx.managed_wallet.update_last_processed_height(CHAIN_TIP);
+
+    assert!(!ctx.bip44_account().utxos.contains_key(&outpoint), "the spend removed the coin");
+    assert!(
+        ctx.managed_wallet.get_spendable_utxos().is_empty(),
+        "so a completed scan releases nothing"
+    );
+    assert_ne!(tx.txid(), spend.txid());
+}
+
+#[tokio::test]
+async fn receive_at_the_scan_target_is_selectable_immediately() {
+    let mut ctx = TestWalletContext::new_with_seed(SEED);
+    ctx.managed_wallet.update_scan_target_height(CHAIN_TIP);
+    ctx.managed_wallet.update_synced_height(CHAIN_TIP);
+    ctx.managed_wallet.update_last_processed_height(CHAIN_TIP);
+
+    // A block at the tip: nothing above it exists to have spent this output,
+    // so a caught-up wallet must not be made to wait for the next batch commit.
+    let tx = Transaction::dummy(&ctx.receive_address, 0..1, &[500_000]);
+    ctx.check_transaction(&tx, block_at(CHAIN_TIP)).await;
+
+    assert!(ctx.first_utxo().spend_scanned);
+    assert_eq!(select(&ctx, 100_000).expect("selection succeeds"), 500_000);
+}
+
+#[tokio::test]
+async fn mempool_receive_during_catch_up_stays_selectable() {
+    let mut ctx = TestWalletContext::new_with_seed(SEED);
+    ctx.managed_wallet.update_scan_target_height(CHAIN_TIP);
+    ctx.managed_wallet.update_synced_height(RECEIVE_HEIGHT);
+    ctx.managed_wallet.update_last_processed_height(RECEIVE_HEIGHT);
+
+    // A mempool transaction is at the tip by definition — no historical block
+    // can have spent an output that does not exist historically.
+    let tx = Transaction::dummy(&ctx.receive_address, 0..1, &[500_000]);
+    ctx.check_transaction(&tx, TransactionContext::Mempool).await;
+
+    assert!(ctx.first_utxo().spend_scanned);
+    assert_eq!(select(&ctx, 100_000).expect("selection succeeds"), 500_000);
+}
+
+/// Consumers with no scanner never report a target, and must keep the
+/// pre-existing unconditional behavior.
+#[tokio::test]
+async fn without_a_reported_scan_target_the_gate_stays_open() {
+    let mut ctx = TestWalletContext::new_with_seed(SEED);
+    ctx.managed_wallet.update_last_processed_height(RECEIVE_HEIGHT);
+
+    let tx = Transaction::dummy(&ctx.receive_address, 0..1, &[500_000]);
+    ctx.check_transaction(&tx, block_at(RECEIVE_HEIGHT)).await;
+
+    assert_eq!(ctx.managed_wallet.scan_target_height(), 0);
+    assert!(ctx.managed_wallet.spend_scan_complete());
+    assert!(ctx.first_utxo().spend_scanned);
+    assert_eq!(select(&ctx, 100_000).expect("selection succeeds"), 500_000);
+}
+
+/// A UTXO deserialized from persistence written before `spend_scanned` existed
+/// must not become unspendable on upgrade.
+#[cfg(feature = "serde")]
+#[test]
+fn utxo_missing_the_field_deserializes_as_scanned() {
+    let utxo = crate::Utxo::dummy(1, 500_000, 100, false, true);
+    let mut value = serde_json::to_value(&utxo).expect("serialize");
+    value.as_object_mut().expect("object").remove("spend_scanned");
+
+    let restored: crate::Utxo = serde_json::from_value(value).expect("deserialize");
+    assert!(restored.spend_scanned, "legacy rows keep the old behavior");
+    assert!(restored.is_spendable(200));
+}

key-wallet/src/transaction_checking/wallet_checker.rs

@@ -258,6 +258,13 @@ impl WalletTransactionChecker for ManagedWalletInfo {
             );
         }
 
+        // Any UTXO just created from a block below the spend-scan target is
+        // only known-unspent as of that block, not as of the tip. Hold it back
+        // from coin selection until the scan proves it survived to the target.
+        if let Some(height) = block_height {
+            self.hold_back_unscanned_utxos(tx, height);
+        }
+
         if update_balance {
             self.update_balance();
         }

key-wallet/src/utxo.rs

@@ -39,6 +39,27 @@ pub struct Utxo {
     /// it is just our previously-tracked balance returning to us. Mirrors
     /// Bitcoin Core's `CWalletTx::IsTrusted` concept.
     pub is_trusted: bool,
+    /// Whether the wallet has scanned every block that could already have
+    /// spent this output.
+    ///
+    /// An output found while catching up on history is only *known* to be
+    /// unspent up to the scan frontier, which during catch-up sits far below
+    /// the chain tip. Spending it then builds a transaction the network has
+    /// long since seen double-spent, and peers drop it silently forever. So a
+    /// receive applied below the frontier starts unscanned, and is promoted in
+    /// bulk once the frontier reaches the height the scan is working toward.
+    ///
+    /// Defaults to `true` so wallets assembled by hand, and UTXOs
+    /// deserialized from persistence written before this field existed, keep
+    /// the previous unconditional behavior. Only a live scan demotes it.
+    #[cfg_attr(feature = "serde", serde(default = "spend_scanned_default"))]
+    pub spend_scanned: bool,
+}
+
+/// Serde default for [`Utxo::spend_scanned`]; see that field's docs.
+#[cfg(feature = "serde")]
+fn spend_scanned_default() -> bool {
+    true
 }
 
 impl Utxo {
@@ -60,6 +81,7 @@ impl Utxo {
             is_instantlocked: false,
             is_locked: false,
             is_trusted: false,
+            spend_scanned: true,
         }
     }
 
@@ -70,13 +92,14 @@ impl Utxo {
 
     /// Check if this UTXO can be spent at the given height.
     ///
-    /// A UTXO is spendable unless it is locked or (for coinbase)
+    /// A UTXO is spendable unless it is locked, not yet covered by the
+    /// wallet's spend scan (see [`Utxo::spend_scanned`]), or — for coinbase —
     /// immature. Mempool 0-conf outputs are spendable — callers that
     /// want to restrict to confirmed/InstantLocked UTXOs (e.g. the
     /// "spendable" balance bucket or conservative coin selection)
     /// should check `is_confirmed || is_instantlocked` themselves.
     pub fn is_spendable(&self, current_height: u32) -> bool {
-        if self.is_locked {
+        if self.is_locked || !self.spend_scanned {
             return false;
         }
         self.is_mature(current_height)

key-wallet/src/wallet/managed_wallet_info/asset_lock_builder.rs

@@ -595,6 +595,7 @@ mod tests {
             is_instantlocked: false,
             is_locked: false,
             is_trusted: false,
+            spend_scanned: true,
         };
         account.utxos.insert(outpoint, utxo);
         outpoint
@@ -639,6 +640,7 @@ mod tests {
             is_instantlocked: false,
             is_locked: false,
             is_trusted: false,
+            spend_scanned: true,
         };
         account.utxos.insert(outpoint, utxo);
         outpoint