feat: implement parallel filter matching

Author: xdustinface

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

@@ -8,16 +8,16 @@
 //! - Efficient filter matching using BIP158 algorithms
 //! - Block download coordination for matches
 
+use crate::error::{SyncError, SyncResult};
+use crate::network::NetworkManager;
+use crate::storage::StorageManager;
 use dashcore::{
     bip158::{BlockFilterReader, Error as Bip158Error},
     network::message::NetworkMessage,
     network::message_blockdata::Inventory,
     BlockHash, ScriptBuf,
 };
-
-use crate::error::{SyncError, SyncResult};
-use crate::network::NetworkManager;
-use crate::storage::StorageManager;
+use key_wallet_manager::wallet_manager::{FilterMatchInput, FilterMatchOutput};
 
 impl<S: StorageManager, N: NetworkManager> super::manager::FilterSyncManager<S, N> {
     pub async fn check_filter_for_matches<
@@ -41,6 +41,16 @@ impl<S: StorageManager, N: NetworkManager> super::manager::FilterSyncManager<S,
         }
     }
 
+    pub async fn check_filters_for_matches<
+        W: key_wallet_manager::wallet_interface::WalletInterface,
+    >(
+        &self,
+        input_map: FilterMatchInput,
+        wallet: &W,
+    ) -> FilterMatchOutput {
+        wallet.check_compact_filters(input_map).await
+    }
+
     /// Check if filter matches any of the provided scripts using BIP158 GCS filter.
     #[allow(dead_code)]
     fn filter_matches_scripts(

key-wallet-manager/Cargo.toml

@@ -25,6 +25,7 @@ serde = { version = "1.0", default-features = false, features = ["derive"], opti
 async-trait = "0.1"
 bincode = { version = "=2.0.0-rc.3", optional = true }
 zeroize = { version = "1.8", features = ["derive"] }
+rayon = "1.11"
 tokio = { version = "1.32", features = ["full"] }
 
 [dev-dependencies]

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

@@ -3,6 +3,7 @@ use std::{collections::BTreeMap, sync::Arc};
 use dashcore::{Block, Transaction, Txid};
 use tokio::sync::Mutex;
 
+use crate::wallet_manager::{FilterMatchInput, FilterMatchOutput};
 use crate::{wallet_interface::WalletInterface, BlockProcessingResult};
 
 // Type alias for transaction effects map
@@ -74,6 +75,10 @@ impl WalletInterface for MockWallet {
         let map = self.effects.lock().await;
         map.get(&tx.txid()).cloned()
     }
+
+    async fn check_compact_filters(&self, _input: FilterMatchInput) -> FilterMatchOutput {
+        FilterMatchOutput::default()
+    }
 }
 
 /// Mock wallet that returns false for filter checks
@@ -103,6 +108,10 @@ impl WalletInterface for NonMatchingMockWallet {
         false
     }
 
+    async fn check_compact_filters(&self, _input: FilterMatchInput) -> FilterMatchOutput {
+        FilterMatchOutput::default()
+    }
+
     async fn describe(&self) -> String {
         "NonMatchingWallet (test implementation)".to_string()
     }

key-wallet-manager/src/wallet_interface.rs

@@ -2,6 +2,7 @@
 //!
 //! This module defines the trait that SPV clients use to interact with wallets.
 
+use crate::wallet_manager::{FilterMatchInput, FilterMatchOutput};
 use alloc::string::String;
 use alloc::vec::Vec;
 use async_trait::async_trait;
@@ -41,6 +42,10 @@ pub trait WalletInterface: Send + Sync + 'static {
         block_hash: &dashcore::BlockHash,
     ) -> bool;
 
+    /// Check compact filters against watched addresses in batch
+    /// Returns map of filter keys to match results
+    async fn check_compact_filters(&self, input: FilterMatchInput) -> FilterMatchOutput;
+
     /// Return the wallet's per-transaction net change and involved addresses if known.
     /// Returns (net_amount, addresses) where net_amount is received - sent in satoshis.
     /// If the wallet has no record for the transaction, returns None.

key-wallet-manager/src/wallet_manager/matching.rs

@@ -0,0 +1,169 @@
+use alloc::vec::Vec;
+use dashcore::bip158::BlockFilter;
+use dashcore::prelude::CoreBlockHeight;
+use dashcore::{Address, BlockHash};
+use rayon::prelude::{IntoParallelIterator, ParallelIterator};
+use std::collections::{BTreeSet, HashMap};
+
+pub type FilterMatchInput = HashMap<FilterMatchKey, BlockFilter>;
+pub type FilterMatchOutput = BTreeSet<FilterMatchKey>;
+
+#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
+pub struct FilterMatchKey {
+    height: CoreBlockHeight,
+    hash: BlockHash,
+}
+
+impl FilterMatchKey {
+    pub fn new(height: CoreBlockHeight, hash: BlockHash) -> Self {
+        Self {
+            height,
+            hash,
+        }
+    }
+    pub fn height(&self) -> CoreBlockHeight {
+        self.height
+    }
+    pub fn hash(&self) -> &BlockHash {
+        &self.hash
+    }
+}
+
+/// Check compact filters for addresses and return the keys that matched.
+pub fn check_compact_filters_for_addresses(
+    input: FilterMatchInput,
+    addresses: Vec<Address>,
+) -> FilterMatchOutput {
+    let script_pubkey_bytes: Vec<Vec<u8>> =
+        addresses.iter().map(|address| address.script_pubkey().to_bytes()).collect();
+
+    input
+        .into_par_iter()
+        .filter_map(|(key, filter)| {
+            filter
+                .match_any(key.hash(), script_pubkey_bytes.iter().map(|v| v.as_slice()))
+                .unwrap_or(false)
+                .then_some(key)
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::Network;
+    use dashcore::{Block, Transaction};
+
+    #[test]
+    fn test_empty_input_returns_empty() {
+        let result = check_compact_filters_for_addresses(FilterMatchInput::new(), vec![]);
+        assert!(result.is_empty());
+    }
+
+    #[test]
+    fn test_empty_addresses_returns_empty() {
+        let address = Address::dummy(Network::Regtest, 1);
+        let tx = Transaction::dummy(&address, 0..0, &[1]);
+        let block = Block::dummy(100, vec![tx]);
+        let filter = BlockFilter::dummy(&block);
+        let key = FilterMatchKey::new(100, block.block_hash());
+
+        let mut input = FilterMatchInput::new();
+        input.insert(key.clone(), filter);
+
+        let output = check_compact_filters_for_addresses(input, vec![]);
+        assert!(!output.contains(&key));
+    }
+
+    #[test]
+    fn test_matching_filter() {
+        let address = Address::dummy(Network::Regtest, 1);
+        let tx = Transaction::dummy(&address, 0..0, &[1]);
+        let block = Block::dummy(100, vec![tx]);
+        let filter = BlockFilter::dummy(&block);
+        let key = FilterMatchKey::new(100, block.block_hash());
+
+        let mut input = FilterMatchInput::new();
+        input.insert(key.clone(), filter);
+
+        let output = check_compact_filters_for_addresses(input, vec![address]);
+        assert!(output.contains(&key));
+    }
+
+    #[test]
+    fn test_non_matching_filter() {
+        let address = Address::dummy(Network::Regtest, 1);
+        let address_other = Address::dummy(Network::Regtest, 2);
+
+        let tx = Transaction::dummy(&address_other, 0..0, &[1]);
+        let block = Block::dummy(100, vec![tx]);
+        let filter = BlockFilter::dummy(&block);
+        let key = FilterMatchKey::new(100, block.block_hash());
+
+        let mut input = FilterMatchInput::new();
+        input.insert(key.clone(), filter);
+
+        let output = check_compact_filters_for_addresses(input, vec![address]);
+        assert!(!output.contains(&key));
+    }
+
+    #[test]
+    fn test_batch_mixed_results() {
+        let unrelated_address = Address::dummy(Network::Regtest, 0);
+        let address_1 = Address::dummy(Network::Regtest, 1);
+        let address_2 = Address::dummy(Network::Regtest, 2);
+
+        let tx_1 = Transaction::dummy(&address_1, 0..0, &[1]);
+        let block_1 = Block::dummy(100, vec![tx_1]);
+        let filter_1 = BlockFilter::dummy(&block_1);
+        let key_1 = FilterMatchKey::new(100, block_1.block_hash());
+
+        let tx_2 = Transaction::dummy(&address_2, 0..0, &[2]);
+        let block_2 = Block::dummy(100, vec![tx_2]);
+        let filter_2 = BlockFilter::dummy(&block_2);
+        let key_2 = FilterMatchKey::new(200, block_2.block_hash());
+
+        let tx_3 = Transaction::dummy(&unrelated_address, 0..0, &[10]);
+        let block_3 = Block::dummy(100, vec![tx_3]);
+        let filter_3 = BlockFilter::dummy(&block_3);
+        let key_3 = FilterMatchKey::new(300, block_3.block_hash());
+
+        let mut input = FilterMatchInput::new();
+        input.insert(key_1.clone(), filter_1);
+        input.insert(key_2.clone(), filter_2);
+        input.insert(key_3.clone(), filter_3);
+
+        let output = check_compact_filters_for_addresses(input, vec![address_1, address_2]);
+        assert_eq!(output.len(), 2);
+        assert!(output.contains(&key_1));
+        assert!(output.contains(&key_2));
+        assert!(!output.contains(&key_3));
+    }
+
+    #[test]
+    fn test_output_sorted_by_height() {
+        let address = Address::dummy(Network::Regtest, 1);
+
+        // Create blocks at different heights (inserted in non-sorted order)
+        let heights = [500, 100, 300, 200, 400];
+        let mut input = FilterMatchInput::new();
+
+        for (i, &height) in heights.iter().enumerate() {
+            let tx = Transaction::dummy(&address, 0..0, &[i as u64]);
+            let block = Block::dummy(height, vec![tx]);
+            let filter = BlockFilter::dummy(&block);
+            let key = FilterMatchKey::new(height, block.block_hash());
+            input.insert(key, filter);
+        }
+
+        let output = check_compact_filters_for_addresses(input, vec![address]);
+
+        // Verify output is sorted by height (ascending)
+        let heights_out: Vec<CoreBlockHeight> = output.iter().map(|k| k.height()).collect();
+        let mut sorted_heights = heights_out.clone();
+        sorted_heights.sort();
+
+        assert_eq!(heights_out, sorted_heights);
+        assert_eq!(heights_out, vec![100, 200, 300, 400, 500]);
+    }
+}

key-wallet-manager/src/wallet_manager/mod.rs

@@ -4,9 +4,11 @@
 //! each of which can have multiple accounts. This follows the architecture
 //! pattern where a manager oversees multiple distinct wallets.
 
+mod matching;
 mod process_block;
 mod transaction_building;
 
+pub use crate::wallet_manager::matching::{FilterMatchInput, FilterMatchKey, FilterMatchOutput};
 use alloc::collections::BTreeMap;
 use alloc::string::String;
 use alloc::vec::Vec;

key-wallet-manager/src/wallet_manager/process_block.rs

@@ -1,4 +1,7 @@
 use crate::wallet_interface::{BlockProcessingResult, WalletInterface};
+use crate::wallet_manager::matching::{
+    check_compact_filters_for_addresses, FilterMatchInput, FilterMatchOutput,
+};
 use crate::WalletManager;
 use alloc::string::String;
 use alloc::vec::Vec;
@@ -80,6 +83,10 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
         hit
     }
 
+    async fn check_compact_filters(&self, input: FilterMatchInput) -> FilterMatchOutput {
+        check_compact_filters_for_addresses(input, self.monitored_addresses())
+    }
+
     async fn transaction_effect(&self, tx: &Transaction) -> Option<(i64, Vec<String>)> {
         // Aggregate across all managed wallets. If any wallet considers it relevant,
         // compute net = total_received - total_sent and collect involved addresses.