feat(dash-spv): prune spent single-use CoinJoin addresses from the filter scan query

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

@@ -860,7 +860,12 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         let mut wallet_states: Vec<WalletScanState> = Vec::new();
         for wallet_id in &behind {
             let synced = wallet.wallet_synced_height(wallet_id);
-            let scripts = wallet.monitored_script_pubkeys_for(wallet_id);
+            // The scan query, not the full monitored set: spent single-use
+            // (CoinJoin) addresses are pruned so the per-filter match cost
+            // stays bounded by active UTXOs + gap lookahead instead of
+            // growing with every historical mixing round
+            // (dashpay/rust-dashcore#948).
+            let scripts = wallet.scan_script_pubkeys_for(wallet_id);
             // Bare owner/voting key hashes a compact filter carries beyond the
             // wallet's scriptPubKeys.
             let elements = wallet.monitored_filter_elements_for(wallet_id);
@@ -1931,6 +1936,64 @@ mod tests {
         assert!(!attr_70.contains(&wallet_high));
     }
 
+    /// `scan_batch` matches filters against the wallet's scan query
+    /// (`scan_script_pubkeys_for`), not the full monitored set: a monitored
+    /// script pruned from the scan query — a spent single-use CoinJoin
+    /// address (dashpay/rust-dashcore#948) — must not pull its block in.
+    #[tokio::test]
+    async fn test_scan_batch_uses_pruned_scan_query() {
+        let wallet_id: WalletId = [0x03; 32];
+        let dead_address = dashcore::Address::dummy(Network::Regtest, 1);
+        let live_address = dashcore::Address::dummy(Network::Regtest, 2);
+
+        let multi = Arc::new(RwLock::new(MultiMockWallet::new()));
+        {
+            let mut w = multi.write().await;
+            w.insert_wallet(
+                wallet_id,
+                MockWalletState {
+                    addresses: vec![dead_address.clone(), live_address.clone()],
+                    synced_height: 0,
+                    last_processed_height: 0,
+                    account_generation: 0,
+                },
+            );
+            // The scan query excludes the dead address.
+            w.set_scan_addresses(wallet_id, vec![live_address.clone()]);
+        }
+        let mut manager = create_multi_test_manager(multi).await;
+        manager.set_state(SyncState::Syncing);
+
+        let mut filters: HashMap<FilterMatchKey, BlockFilter> = HashMap::new();
+        let (key_dead, f_dead) = filter_for_address(30, &dead_address);
+        let (key_live, f_live) = filter_for_address(60, &live_address);
+        filters.insert(key_dead.clone(), f_dead);
+        filters.insert(key_live.clone(), f_live);
+
+        let mut batch = FiltersBatch::new(0, 99, filters);
+        batch.mark_verified();
+        manager.active_batches.insert(0, batch);
+        manager.progress.update_stored_height(99);
+
+        let events = manager.scan_batch(0).await.unwrap();
+
+        let blocks = events
+            .iter()
+            .find_map(|e| match e {
+                SyncEvent::BlocksNeeded {
+                    blocks,
+                } => Some(blocks),
+                _ => None,
+            })
+            .expect("BlocksNeeded event");
+
+        assert!(blocks.contains_key(&key_live), "block paying the scan-query address is needed");
+        assert!(
+            !blocks.contains_key(&key_dead),
+            "block paying only the pruned address must not be downloaded"
+        );
+    }
+
     /// `rescan_batch` with multiple wallets in `scripts_by_wallet`:
     /// each wallet's new scripts are matched independently and the
     /// attribution is correct in the emitted `BlocksNeeded`.

key-wallet-manager/Cargo.toml

@@ -39,6 +39,11 @@ key-wallet = { path = "../key-wallet", features = ["test-utils", "bincode"] }
 dashcore = { path = "../dash", features = ["test-utils"] }
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 hex = "0.4"
+criterion = "0.8.1"
+
+[[bench]]
+name = "filter_scan"
+harness = false
 
 [lints.rust]
 unexpected_cfgs = { level = "allow", check-cfg = ['cfg(bench)', 'cfg(fuzzing)'] }

key-wallet-manager/benches/filter_scan.rs

@@ -0,0 +1,174 @@
+//! Compact-filter matching cost: full monitored set vs the pruned
+//! forward-scan set for a mixing-heavy CoinJoin wallet
+//! (dashpay/rust-dashcore#948).
+//!
+//! Mimics a wallet mid-recovery after many mixing rounds. Every CoinJoin
+//! round pays a fresh single-use address, so the account accumulates `used`
+//! spent addresses, keeps a small set of still-funded denominations
+//! ([`LIVE_UTXOS`]), and watches the usual gap-limit lookahead on top. One
+//! scan batch of BIP158 filters is then matched with
+//! `monitored_script_pubkeys_for` (the pre-#948 query, which drags every
+//! historical address through SipHash + sort per filter) and with
+//! `scan_script_pubkeys_for` (the pruned query, bounded by live UTXOs + gap
+//! lookahead).
+//!
+//! BIP158 keys each filter's SipHashes off the block hash, so the whole
+//! query set is re-hashed and re-sorted per filter — which is exactly why
+//! the query size dominates and why nothing is cacheable across filters.
+//!
+//! Run with:
+//! `cargo bench -p key-wallet-manager --bench filter_scan`
+
+use std::collections::HashMap;
+use std::hint::black_box;
+
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+use dashcore::bip158::BlockFilter;
+use dashcore::hashes::Hash;
+use dashcore::{Address, Block, OutPoint, Transaction, TxOut, Txid};
+use key_wallet::account::ManagedAccountTrait;
+use key_wallet::wallet::initialization::WalletAccountCreationOptions;
+use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
+use key_wallet::{KeySource, ManagedAccountType, Network, Utxo};
+use key_wallet_manager::{
+    check_compact_filters_for_elements, FilterMatchKey, WalletInterface, WalletManager,
+};
+
+/// Deterministic test mnemonic (throwaway; same one the unit tests use).
+const MNEMONIC: &str =
+    "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about";
+
+/// Denominated coins still unspent in the CoinJoin account — the wallet's
+/// active mixing balance, which stays in the scan query.
+const LIVE_UTXOS: usize = 200;
+
+/// Filters matched per iteration — one scan batch.
+const FILTERS: u32 = 512;
+
+/// Historical single-use address counts to sweep, roughly
+/// `denominations x rounds` at different points of a recovery scan. The
+/// issue's reference wallet starts a mainnet recovery at a few hundred
+/// monitored scripts and ends at several thousand.
+const USED_ADDRESSES: [u32; 3] = [500, 2_000, 6_000];
+
+type Manager = WalletManager<ManagedWalletInfo>;
+
+/// Build a wallet whose CoinJoin account carries `used` spent single-use
+/// addresses, [`LIVE_UTXOS`] still-funded ones, and the default gap-limit
+/// lookahead of unused addresses above them.
+fn wallet_with_mixing_history(used: u32) -> (Manager, [u8; 32]) {
+    let mut manager = Manager::new(Network::Regtest);
+    let wallet_id = manager
+        .create_wallet_from_mnemonic(MNEMONIC, 0, WalletAccountCreationOptions::Default)
+        .expect("create wallet");
+
+    let key_source = KeySource::Public(
+        manager
+            .get_wallet(&wallet_id)
+            .expect("wallet")
+            .accounts
+            .coinjoin_accounts
+            .get(&0)
+            .expect("CoinJoin account 0")
+            .account_xpub,
+    );
+
+    let info = manager.get_wallet_info_mut(&wallet_id).expect("wallet info");
+    let coinjoin = info.accounts.coinjoin_accounts.get_mut(&0).expect("managed CoinJoin account");
+
+    // Extend the external (mixed-coin) branch so the pool holds `used`
+    // historical addresses plus the pre-generated gap window above them.
+    let addresses = {
+        let ManagedAccountType::CoinJoin {
+            external_addresses,
+            ..
+        } = coinjoin.managed_account_type_mut()
+        else {
+            panic!("expected CoinJoin managed account type");
+        };
+        external_addresses
+            .generate_addresses(used, &key_source, true)
+            .expect("derive CoinJoin addresses");
+        external_addresses.all_addresses()
+    };
+
+    // The first `used` indices each received one mixing round's payout...
+    let spent = &addresses[..used as usize];
+    for address in spent {
+        assert!(coinjoin.mark_address_used(address), "address should belong to the pool");
+    }
+    // ...and only the most recent LIVE_UTXOS denominations remain unspent.
+    for (i, address) in spent.iter().rev().take(LIVE_UTXOS).enumerate() {
+        let mut txid = [0u8; 32];
+        txid[..4].copy_from_slice(&(i as u32).to_le_bytes());
+        txid[31] = 0xc1;
+        let utxo = Utxo::new(
+            OutPoint::new(Txid::from_byte_array(txid), 0),
+            TxOut {
+                value: 100_001,
+                script_pubkey: address.script_pubkey(),
+            },
+            address.clone(),
+            100 + i as u32,
+            false,
+        );
+        coinjoin.utxos.insert(utxo.outpoint, utxo);
+    }
+
+    (manager, wallet_id)
+}
+
+/// One scan batch of realistic filters over blocks that do not pay the
+/// wallet. Each block's hash differs, so every filter re-keys its SipHashes
+/// — the property that forces the per-filter re-hash being measured.
+fn scan_batch_filters(count: u32) -> HashMap<FilterMatchKey, BlockFilter> {
+    (0..count)
+        .map(|height| {
+            let third_party = Address::dummy(Network::Regtest, 1_000_000 + height as usize);
+            let tx = Transaction::dummy(&third_party, 0..2, &[u64::from(height) + 1, 546]);
+            let block = Block::dummy(height, vec![tx]);
+            (FilterMatchKey::new(height, block.block_hash()), BlockFilter::dummy(&block))
+        })
+        .collect()
+}
+
+fn bench_filter_scan(c: &mut Criterion) {
+    let filters = scan_batch_filters(FILTERS);
+
+    let mut group = c.benchmark_group("filter_scan");
+    group.sample_size(10);
+    group.throughput(Throughput::Elements(u64::from(FILTERS)));
+
+    for used in USED_ADDRESSES {
+        let (manager, wallet_id) = wallet_with_mixing_history(used);
+        let monitored = manager.monitored_script_pubkeys_for(&wallet_id);
+        let pruned = manager.scan_script_pubkeys_for(&wallet_id);
+        assert!(
+            pruned.len() < monitored.len(),
+            "the scan query must shrink once CoinJoin addresses are spent"
+        );
+        println!(
+            "used={used}: monitored query = {} scripts, pruned scan query = {} scripts",
+            monitored.len(),
+            pruned.len()
+        );
+
+        for (name, scripts) in [("monitored", &monitored), ("pruned", &pruned)] {
+            group.bench_with_input(BenchmarkId::new(name, used), scripts, |b, scripts| {
+                b.iter(|| {
+                    check_compact_filters_for_elements(
+                        black_box(&filters),
+                        black_box(scripts),
+                        &[],
+                        0,
+                    )
+                })
+            });
+        }
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, bench_filter_scan);
+criterion_main!(benches);

key-wallet-manager/src/process_block.rs

@@ -230,6 +230,10 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
             .unwrap_or_default()
     }
 
+    fn scan_script_pubkeys_for(&self, wallet_id: &WalletId) -> Vec<ScriptBuf> {
+        self.wallet_infos.get(wallet_id).map(|info| info.scan_script_pubkeys()).unwrap_or_default()
+    }
+
     fn monitored_filter_elements_for(&self, wallet_id: &WalletId) -> Vec<Vec<u8>> {
         self.wallet_infos
             .get(wallet_id)
@@ -747,6 +751,34 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_scan_script_pubkeys_for_prunes_spent_coinjoin_addresses() {
+        use key_wallet::account::ManagedAccountTrait;
+
+        let (mut manager, wallet_id, _addr) = setup_manager_with_wallet();
+
+        // Untouched wallet: the scan set equals the monitored set.
+        let monitored = manager.monitored_script_pubkeys_for(&wallet_id);
+        assert_eq!(manager.scan_script_pubkeys_for(&wallet_id), monitored);
+
+        // Mark a CoinJoin address used with no unspent output — a spent
+        // single-use address. The scan query drops it; the monitored set
+        // keeps it.
+        let info = manager.get_wallet_info_mut(&wallet_id).expect("wallet info");
+        let coinjoin = info.accounts.coinjoin_accounts.get_mut(&0).expect("CoinJoin account 0");
+        let spent_addr = coinjoin.all_addresses().first().cloned().expect("CoinJoin address");
+        assert!(coinjoin.mark_address_used(&spent_addr));
+
+        let monitored = manager.monitored_script_pubkeys_for(&wallet_id);
+        let scan = manager.scan_script_pubkeys_for(&wallet_id);
+        assert!(monitored.contains(&spent_addr.script_pubkey()));
+        assert!(!scan.contains(&spent_addr.script_pubkey()));
+        assert_eq!(scan.len(), monitored.len() - 1);
+
+        // Unknown wallet id yields an empty scan set.
+        assert!(manager.scan_script_pubkeys_for(&[0xff; 32]).is_empty());
+    }
+
     #[tokio::test]
     async fn test_monitor_revision_bumps_and_stability() {
         let mut manager: WalletManager<ManagedWalletInfo> = WalletManager::new(Network::Testnet);

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

@@ -389,6 +389,11 @@ pub struct MockWalletState {
 /// enabling tests that exercise per-wallet attribution paths.
 pub struct MultiMockWallet {
     wallets: std::collections::BTreeMap<WalletId, MockWalletState>,
+    /// Per-wallet override for `scan_script_pubkeys_for`. Wallets absent here
+    /// fall back to the monitored set, mirroring the trait default. Lets tests
+    /// hand the filter scan a pruned query while the monitored set stays full
+    /// (dashpay/rust-dashcore#948).
+    scan_addresses: std::collections::BTreeMap<WalletId, Vec<Address>>,
     event_sender: broadcast::Sender<WalletEvent>,
     /// Track every block processed for assertions.
     processed: Arc<Mutex<Vec<(WalletId, dashcore::BlockHash, u32)>>>,
@@ -405,6 +410,7 @@ impl MultiMockWallet {
         let (event_sender, _) = broadcast::channel(16);
         Self {
             wallets: std::collections::BTreeMap::new(),
+            scan_addresses: std::collections::BTreeMap::new(),
             event_sender,
             processed: Arc::new(Mutex::new(Vec::new())),
         }
@@ -415,6 +421,12 @@ impl MultiMockWallet {
         self.wallets.insert(wallet_id, state);
     }
 
+    /// Override the scan query for one wallet: `scan_script_pubkeys_for`
+    /// returns these addresses' scripts instead of the monitored set.
+    pub fn set_scan_addresses(&mut self, wallet_id: WalletId, addresses: Vec<Address>) {
+        self.scan_addresses.insert(wallet_id, addresses);
+    }
+
     /// Mutable access to a wallet's state, panicking if absent.
     pub fn wallet_mut(&mut self, wallet_id: &WalletId) -> &mut MockWalletState {
         self.wallets.get_mut(wallet_id).expect("wallet present")
@@ -466,6 +478,13 @@ impl WalletInterface for MultiMockWallet {
             .unwrap_or_default()
     }
 
+    fn scan_script_pubkeys_for(&self, wallet_id: &WalletId) -> Vec<ScriptBuf> {
+        match self.scan_addresses.get(wallet_id) {
+            Some(addresses) => addresses.iter().map(|a| a.script_pubkey()).collect(),
+            None => self.monitored_script_pubkeys_for(wallet_id),
+        }
+    }
+
     fn watched_outpoints(&self) -> Vec<OutPoint> {
         Vec::new()
     }

key-wallet-manager/src/wallet_interface.rs

@@ -88,6 +88,21 @@ pub trait WalletInterface: Send + Sync + 'static {
     /// Get cached scriptPubKeys for every address monitored by `wallet_id`.
     fn monitored_script_pubkeys_for(&self, wallet_id: &WalletId) -> Vec<ScriptBuf>;
 
+    /// Get the scriptPubKeys `wallet_id` wants matched during a forward
+    /// compact-filter scan.
+    ///
+    /// Defaults to [`Self::monitored_script_pubkeys_for`]. Implementations may
+    /// return a subset when some monitored scripts can no longer be paid in
+    /// practice — the managed-wallet implementation drops CoinJoin addresses
+    /// whose outputs are all spent, since those are single-use by protocol and
+    /// their monotonic growth dominates per-filter matching cost late in a
+    /// mixing-heavy recovery scan (dashpay/rust-dashcore#948). Block
+    /// processing still checks transactions against the full monitored set, so
+    /// pruning only narrows which blocks the filter scan downloads.
+    fn scan_script_pubkeys_for(&self, wallet_id: &WalletId) -> Vec<ScriptBuf> {
+        self.monitored_script_pubkeys_for(wallet_id)
+    }
+
     /// Get the bare `hash160` compact-filter elements monitored by `wallet_id`
     /// that are not covered by its scriptPubKeys.
     ///