refactor: gate manager module behind feature flag

dash-spv/Cargo.toml

@@ -12,7 +12,7 @@ rust-version = "1.89"
 # Core Dash libraries
 dashcore = { path = "../dash", features = ["serde", "core-block-hash-use-x11", "message_verification", "bls", "quorum_validation"] }
 dashcore_hashes = { path = "../hashes" }
-key-wallet = { path = "../key-wallet" }
+key-wallet = { path = "../key-wallet", features = ["parallel-filters"] }
 
 # BLS signatures
 blsful = { git = "https://github.com/dashpay/agora-blsful", rev = "0c34a7a488a0bd1c9a9a2196e793b303ad35c900" }

key-wallet-ffi/Cargo.toml

@@ -20,7 +20,7 @@ eddsa = ["dashcore/eddsa", "key-wallet/eddsa"]
 bls = ["dashcore/bls", "key-wallet/bls"]
 
 [dependencies]
-key-wallet = { path = "../key-wallet", default-features = false, features = ["std"] }
+key-wallet = { path = "../key-wallet", default-features = false, features = ["std", "manager"] }
 dashcore = { path = "../dash" }
 secp256k1 = { version = "0.30.0", features = ["global-context"] }
 tokio = { version = "1.32", features = ["rt-multi-thread", "sync"] }
@@ -31,6 +31,6 @@ hex = "0.4"
 cbindgen = "0.29"
 
 [dev-dependencies]
-key-wallet = { path = "../key-wallet", default-features = false, features = ["std", "test-utils"] }
+key-wallet = { path = "../key-wallet", default-features = false, features = ["std", "manager", "test-utils"] }
 tempfile = "3.0"
 hex = "0.4"

key-wallet/Cargo.toml

@@ -9,8 +9,10 @@ readme = "README.md"
 license = "CC0-1.0"
 
 [features]
-default = ["std"]
+default = ["std", "manager"]
 std = ["secp256k1/std", "bip39/std", "getrandom", "rand"]
+manager = ["dep:tokio"]
+parallel-filters = ["manager", "dep:rayon"]
 serde = ["dep:serde", "dep:serde_json", "dashcore_hashes/serde", "secp256k1/serde", "dashcore/serde"]
 bincode = ["serde", "dep:bincode", "dep:bincode_derive", "dashcore_hashes/bincode", "dashcore/bincode"]
 bip38 = ["scrypt", "aes", "bs58", "rand"]
@@ -43,8 +45,8 @@ hex = { version = "0.4"}
 zeroize = { version = "1.8", features = ["derive"] }
 tracing = "0.1"
 async-trait = "0.1"
-tokio = { version = "1", features = ["macros", "rt", "sync"] }
-rayon = { version = "1.11" }
+tokio = { version = "1", features = ["macros", "rt", "sync"], optional = true }
+rayon = { version = "1.11", optional = true }
 
 [dev-dependencies]
 dashcore = { path="../dash", features = ["test-utils"] }

key-wallet/src/lib.rs

@@ -43,6 +43,7 @@ pub mod dip9;
 pub mod error;
 pub mod gap_limit;
 pub mod managed_account;
+#[cfg(feature = "manager")]
 pub mod manager;
 pub mod mnemonic;
 pub mod psbt;

key-wallet/src/manager/matching.rs

@@ -2,6 +2,7 @@ use alloc::vec::Vec;
 use dashcore::bip158::BlockFilter;
 use dashcore::prelude::CoreBlockHeight;
 use dashcore::{Address, BlockHash};
+#[cfg(feature = "parallel-filters")]
 use rayon::prelude::{IntoParallelIterator, ParallelIterator};
 use std::collections::{BTreeSet, HashMap};
 
@@ -34,15 +35,22 @@ pub fn check_compact_filters_for_addresses(
     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.clone())
-        })
-        .collect()
+    let match_filter = |(key, filter): (&FilterMatchKey, &BlockFilter)| {
+        filter
+            .match_any(key.hash(), script_pubkey_bytes.iter().map(|v| v.as_slice()))
+            .unwrap_or(false)
+            .then_some(key.clone())
+    };
+
+    #[cfg(feature = "parallel-filters")]
+    {
+        input.into_par_iter().filter_map(match_filter).collect()
+    }
+
+    #[cfg(not(feature = "parallel-filters"))]
+    {
+        input.iter().filter_map(match_filter).collect()
+    }
 }
 
 #[cfg(test)]

key-wallet/src/test_utils/mock_wallet.rs

@@ -0,0 +1,246 @@
+use crate::manager::MempoolTransactionResult;
+use crate::manager::{BlockProcessingResult, WalletEvent, WalletInterface};
+use crate::transaction_checking::TransactionContext;
+use dashcore::address::NetworkUnchecked;
+use dashcore::prelude::CoreBlockHeight;
+use dashcore::{Address, Block, OutPoint, Transaction, Txid};
+use std::collections::BTreeMap;
+use std::str::FromStr;
+use std::sync::Arc;
+use tokio::sync::{broadcast, Mutex};
+
+// Type alias for transaction effects map
+type TransactionEffectsMap = Arc<Mutex<BTreeMap<Txid, (i64, Vec<String>)>>>;
+
+pub struct MockWallet {
+    processed_blocks: Arc<Mutex<Vec<(dashcore::BlockHash, u32)>>>,
+    processed_transactions: Arc<Mutex<Vec<dashcore::Txid>>>,
+    // Map txid -> (net_amount, addresses)
+    effects: TransactionEffectsMap,
+    synced_height: CoreBlockHeight,
+    event_sender: broadcast::Sender<WalletEvent>,
+    /// When true, process_mempool_transaction returns is_relevant=true.
+    mempool_relevant: bool,
+    /// Addresses returned by monitored_addresses.
+    addresses: Vec<Address>,
+    /// Outpoints returned by watched_outpoints.
+    outpoints: Vec<OutPoint>,
+    /// New addresses returned by process_mempool_transaction.
+    mempool_new_addresses: Vec<Address>,
+    /// Recorded status change notifications for test assertions.
+    status_changes: Arc<Mutex<Vec<(Txid, TransactionContext)>>>,
+    /// Monitor revision counter for staleness detection.
+    monitor_revision: u64,
+}
+
+impl Default for MockWallet {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl MockWallet {
+    pub fn new() -> Self {
+        let (event_sender, _) = broadcast::channel(16);
+        Self {
+            processed_blocks: Arc::new(Mutex::new(Vec::new())),
+            processed_transactions: Arc::new(Mutex::new(Vec::new())),
+            effects: Arc::new(Mutex::new(BTreeMap::new())),
+            synced_height: 0,
+            event_sender,
+            mempool_relevant: false,
+            addresses: Vec::new(),
+            outpoints: Vec::new(),
+            mempool_new_addresses: Vec::new(),
+            status_changes: Arc::new(Mutex::new(Vec::new())),
+            monitor_revision: 0,
+        }
+    }
+
+    /// Configure whether mempool transactions are reported as relevant.
+    pub fn set_mempool_relevant(&mut self, relevant: bool) {
+        self.mempool_relevant = relevant;
+    }
+
+    /// Set the addresses returned by monitored_addresses.
+    pub fn set_addresses(&mut self, addresses: Vec<Address>) {
+        self.addresses = addresses;
+        self.monitor_revision += 1;
+    }
+
+    /// Set the outpoints returned by watched_outpoints.
+    pub fn set_outpoints(&mut self, outpoints: Vec<OutPoint>) {
+        self.outpoints = outpoints;
+        self.monitor_revision += 1;
+    }
+
+    /// Set new addresses returned by process_mempool_transaction.
+    pub fn set_mempool_new_addresses(&mut self, addresses: Vec<Address>) {
+        self.mempool_new_addresses = addresses;
+    }
+
+    pub fn status_changes(&self) -> Arc<Mutex<Vec<(Txid, TransactionContext)>>> {
+        self.status_changes.clone()
+    }
+
+    pub async fn set_effect(&self, txid: dashcore::Txid, net: i64, addresses: Vec<String>) {
+        let mut map = self.effects.lock().await;
+        map.insert(txid, (net, addresses));
+    }
+
+    pub fn processed_blocks(&self) -> Arc<Mutex<Vec<(dashcore::BlockHash, u32)>>> {
+        self.processed_blocks.clone()
+    }
+
+    pub fn processed_transactions(&self) -> Arc<Mutex<Vec<dashcore::Txid>>> {
+        self.processed_transactions.clone()
+    }
+}
+
+#[async_trait::async_trait]
+impl WalletInterface for MockWallet {
+    async fn process_block(&mut self, block: &Block, height: u32) -> BlockProcessingResult {
+        let mut processed = self.processed_blocks.lock().await;
+        processed.push((block.block_hash(), height));
+
+        BlockProcessingResult {
+            new_txids: block.txdata.iter().map(|tx| tx.txid()).collect(),
+            existing_txids: Vec::new(),
+            new_addresses: Vec::new(),
+        }
+    }
+
+    async fn process_mempool_transaction(
+        &mut self,
+        tx: &Transaction,
+        _is_instant_send: bool,
+    ) -> MempoolTransactionResult {
+        let mut processed = self.processed_transactions.lock().await;
+        processed.push(tx.txid());
+
+        if !self.mempool_relevant {
+            return MempoolTransactionResult::default();
+        }
+
+        let effects = self.effects.lock().await;
+        let (net_amount, addresses) = if let Some((net, addr_strs)) = effects.get(&tx.txid()) {
+            let addrs = addr_strs
+                .iter()
+                .filter_map(|s| {
+                    Address::<NetworkUnchecked>::from_str(s).ok().map(|a| a.assume_checked())
+                })
+                .collect();
+            (*net, addrs)
+        } else {
+            (0, Vec::new())
+        };
+
+        MempoolTransactionResult {
+            is_relevant: true,
+            net_amount,
+            is_outgoing: net_amount < 0,
+            addresses,
+            new_addresses: self.mempool_new_addresses.clone(),
+        }
+    }
+
+    async fn describe(&self) -> String {
+        "MockWallet (test implementation)".to_string()
+    }
+
+    async fn transaction_effect(&self, tx: &Transaction) -> Option<(i64, Vec<String>)> {
+        let map = self.effects.lock().await;
+        map.get(&tx.txid()).cloned()
+    }
+
+    fn monitored_addresses(&self) -> Vec<Address> {
+        self.addresses.clone()
+    }
+
+    fn watched_outpoints(&self) -> Vec<OutPoint> {
+        self.outpoints.clone()
+    }
+
+    fn synced_height(&self) -> CoreBlockHeight {
+        self.synced_height
+    }
+
+    fn update_synced_height(&mut self, height: CoreBlockHeight) {
+        self.synced_height = height;
+    }
+
+    fn monitor_revision(&self) -> u64 {
+        self.monitor_revision
+    }
+
+    fn subscribe_events(&self) -> broadcast::Receiver<WalletEvent> {
+        self.event_sender.subscribe()
+    }
+
+    fn process_instant_send_lock(&mut self, txid: Txid) {
+        let mut changes =
+            self.status_changes.try_lock().expect("status_changes lock contention in test helper");
+        changes.push((txid, TransactionContext::InstantSend));
+    }
+}
+
+/// Mock wallet that returns false for filter checks
+pub struct NonMatchingMockWallet {
+    synced_height: CoreBlockHeight,
+    event_sender: broadcast::Sender<WalletEvent>,
+}
+
+impl Default for NonMatchingMockWallet {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl NonMatchingMockWallet {
+    pub fn new() -> Self {
+        let (event_sender, _) = broadcast::channel(16);
+        Self {
+            synced_height: 0,
+            event_sender,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl WalletInterface for NonMatchingMockWallet {
+    async fn process_block(&mut self, _block: &Block, _height: u32) -> BlockProcessingResult {
+        BlockProcessingResult::default()
+    }
+
+    async fn process_mempool_transaction(
+        &mut self,
+        _tx: &Transaction,
+        _is_instant_send: bool,
+    ) -> MempoolTransactionResult {
+        MempoolTransactionResult::default()
+    }
+
+    fn monitored_addresses(&self) -> Vec<Address> {
+        Vec::new()
+    }
+
+    fn watched_outpoints(&self) -> Vec<OutPoint> {
+        Vec::new()
+    }
+
+    fn synced_height(&self) -> CoreBlockHeight {
+        self.synced_height
+    }
+
+    fn update_synced_height(&mut self, height: CoreBlockHeight) {
+        self.synced_height = height;
+    }
+
+    fn subscribe_events(&self) -> broadcast::Receiver<WalletEvent> {
+        self.event_sender.subscribe()
+    }
+
+    async fn describe(&self) -> String {
+        "NonMatchingWallet (test implementation)".to_string()
+    }
+}

key-wallet/src/test_utils/mod.rs

@@ -1,7 +1,11 @@
 mod account;
+#[cfg(feature = "manager")]
+mod mock_wallet;
 mod utxo;
 mod wallet;
 
-pub use wallet::MockWallet;
-pub use wallet::NonMatchingMockWallet;
+#[cfg(feature = "manager")]
+pub use mock_wallet::MockWallet;
+#[cfg(feature = "manager")]
+pub use mock_wallet::NonMatchingMockWallet;
 pub use wallet::TestWalletContext;