feat: implement parallel filter matching

dash-spv/src/client/block_processor_test.rs

@@ -7,7 +7,7 @@ mod tests {
     use crate::storage::DiskStorageManager;
     use crate::types::{SpvEvent, SpvStats};
     use dashcore::{blockdata::constants::genesis_block, Block, Network, Transaction};
-
+    use key_wallet_manager::wallet_manager::matching::{FilterMatchInput, FilterMatchOutput};
     use std::sync::Arc;
     use tokio::sync::{mpsc, oneshot, Mutex, RwLock};
 
@@ -62,6 +62,10 @@ mod tests {
             true
         }
 
+        async fn check_compact_filters(&self, input: FilterMatchInput) -> FilterMatchOutput {
+            input.keys().cloned().collect()
+        }
+
         async fn describe(&self) -> String {
             "MockWallet (test implementation)".to_string()
         }
@@ -263,6 +267,10 @@ mod tests {
                 false
             }
 
+            async fn check_compact_filters(&self, _input: FilterMatchInput) -> FilterMatchOutput {
+                FilterMatchOutput::new()
+            }
+
             async fn describe(&self) -> String {
                 "NonMatchingWallet (test implementation)".to_string()
             }

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::matching::{FilterMatchInput, FilterMatchOutput};
 
 impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync + 'static>
     super::manager::FilterSyncManager<S, N>
@@ -43,6 +43,16 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
         }
     }
 
+    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

@@ -24,10 +24,12 @@ 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"
 
 [dev-dependencies]
 hex = "0.4"
 serde_json = "1.0"
+dashcore-test-utils = { path = "../test-utils" }
 tokio = { version = "1.32", features = ["full"] }
 
 [lints.rust]

key-wallet-manager/src/wallet_interface.rs

@@ -2,11 +2,28 @@
 //!
 //! This module defines the trait that SPV clients use to interact with wallets.
 
+use crate::wallet_manager::matching::{FilterMatchInput, FilterMatchOutput};
 use alloc::string::String;
+use alloc::vec::Vec;
 use async_trait::async_trait;
 use dashcore::bip158::BlockFilter;
 use dashcore::prelude::CoreBlockHeight;
-use dashcore::{Block, Transaction, Txid};
+use dashcore::{Block, BlockHash, Transaction, Txid};
+
+#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
+pub struct FilterMatchKey {
+    pub block_height: CoreBlockHeight,
+    pub block_hash: BlockHash,
+}
+
+impl FilterMatchKey {
+    pub fn new(height: CoreBlockHeight, hash: BlockHash) -> Self {
+        Self {
+            block_height: height,
+            block_hash: hash,
+        }
+    }
+}
 
 /// Trait for wallet implementations to receive SPV events
 #[async_trait]
@@ -26,6 +43,10 @@ pub trait WalletInterface: Send + Sync {
         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,123 @@
+use crate::wallet_interface::FilterMatchKey;
+use alloc::vec::Vec;
+use dashcore::bip158::BlockFilter;
+use dashcore::Address;
+use rayon::prelude::{IntoParallelIterator, ParallelIterator};
+use std::collections::{BTreeSet, HashMap};
+
+pub type FilterMatchInput = HashMap<FilterMatchKey, BlockFilter>;
+pub type FilterMatchOutput = BTreeSet<FilterMatchKey>;
+
+/// 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.block_hash, script_pubkey_bytes.iter().map(|v| v.as_slice()))
+                .unwrap_or(false)
+                .then_some(key)
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use dashcore::blockdata::script::ScriptBuf;
+    use dashcore_test_utils::{
+        create_filter_for_block, create_test_block, create_test_transaction_to_script, test_address,
+    };
+
+    #[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 tx = create_test_transaction_to_script(ScriptBuf::new());
+        let block = create_test_block(100, vec![tx]);
+        let filter = create_filter_for_block(&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 = test_address(0);
+
+        let tx = create_test_transaction_to_script(address.script_pubkey());
+        let block = create_test_block(100, vec![tx]);
+        let filter = create_filter_for_block(&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 = test_address(0);
+        let other_address = test_address(1);
+
+        let tx = create_test_transaction_to_script(other_address.script_pubkey());
+        let block = create_test_block(100, vec![tx]);
+        let filter = create_filter_for_block(&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 address1 = test_address(0);
+        let address2 = test_address(1);
+        let unrelated_address = test_address(2);
+
+        let tx1 = create_test_transaction_to_script(address1.script_pubkey());
+        let block1 = create_test_block(100, vec![tx1]);
+        let filter1 = create_filter_for_block(&block1);
+        let key1 = FilterMatchKey::new(100, block1.block_hash());
+
+        let tx2 = create_test_transaction_to_script(address2.script_pubkey());
+        let block2 = create_test_block(200, vec![tx2]);
+        let filter2 = create_filter_for_block(&block2);
+        let key2 = FilterMatchKey::new(200, block2.block_hash());
+
+        let tx3 = create_test_transaction_to_script(unrelated_address.script_pubkey());
+        let block3 = create_test_block(300, vec![tx3]);
+        let filter3 = create_filter_for_block(&block3);
+        let key3 = FilterMatchKey::new(300, block3.block_hash());
+
+        let mut input = FilterMatchInput::new();
+        input.insert(key1.clone(), filter1);
+        input.insert(key2.clone(), filter2);
+        input.insert(key3.clone(), filter3);
+
+        let output = check_compact_filters_for_addresses(input, vec![address1, address2]);
+        assert_eq!(output.len(), 2);
+        assert!(output.contains(&key1));
+        assert!(output.contains(&key2));
+        assert!(!output.contains(&key3));
+    }
+}

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

@@ -4,6 +4,7 @@
 //! each of which can have multiple accounts. This follows the architecture
 //! pattern where a manager oversees multiple distinct wallets.
 
+pub mod matching;
 mod process_block;
 mod transaction_building;
 

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

@@ -1,4 +1,7 @@
 use crate::wallet_interface::WalletInterface;
+use crate::wallet_manager::matching::{
+    check_compact_filters_for_addresses, FilterMatchInput, FilterMatchOutput,
+};
 use crate::WalletManager;
 use alloc::string::String;
 use alloc::vec::Vec;
@@ -78,6 +81,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.