diff --git a/dash-spv/Cargo.toml b/dash-spv/Cargo.toml index 08064df27..c05f46999 100644 --- a/dash-spv/Cargo.toml +++ b/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" } diff --git a/key-wallet-ffi/Cargo.toml b/key-wallet-ffi/Cargo.toml index f6b293056..cfbe497d7 100644 --- a/key-wallet-ffi/Cargo.toml +++ b/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" diff --git a/key-wallet/Cargo.toml b/key-wallet/Cargo.toml index 6fc6a1a9f..ed2f485bd 100644 --- a/key-wallet/Cargo.toml +++ b/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"] } diff --git a/key-wallet/src/lib.rs b/key-wallet/src/lib.rs index b08372884..26344ea77 100644 --- a/key-wallet/src/lib.rs +++ b/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; diff --git a/key-wallet/src/manager/matching.rs b/key-wallet/src/manager/matching.rs index 3a3b008c1..92200eac9 100644 --- a/key-wallet/src/manager/matching.rs +++ b/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> = 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)] diff --git a/key-wallet/src/test_utils/mock_wallet.rs b/key-wallet/src/test_utils/mock_wallet.rs new file mode 100644 index 000000000..856dc31e0 --- /dev/null +++ b/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)>>>; + +pub struct MockWallet { + processed_blocks: Arc>>, + processed_transactions: Arc>>, + // Map txid -> (net_amount, addresses) + effects: TransactionEffectsMap, + synced_height: CoreBlockHeight, + event_sender: broadcast::Sender, + /// When true, process_mempool_transaction returns is_relevant=true. + mempool_relevant: bool, + /// Addresses returned by monitored_addresses. + addresses: Vec
, + /// Outpoints returned by watched_outpoints. + outpoints: Vec, + /// New addresses returned by process_mempool_transaction. + mempool_new_addresses: Vec
, + /// Recorded status change notifications for test assertions. + status_changes: Arc>>, + /// 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
) { + self.addresses = addresses; + self.monitor_revision += 1; + } + + /// Set the outpoints returned by watched_outpoints. + pub fn set_outpoints(&mut self, outpoints: Vec) { + 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
) { + self.mempool_new_addresses = addresses; + } + + pub fn status_changes(&self) -> Arc>> { + self.status_changes.clone() + } + + pub async fn set_effect(&self, txid: dashcore::Txid, net: i64, addresses: Vec) { + let mut map = self.effects.lock().await; + map.insert(txid, (net, addresses)); + } + + pub fn processed_blocks(&self) -> Arc>> { + self.processed_blocks.clone() + } + + pub fn processed_transactions(&self) -> Arc>> { + 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::::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)> { + let map = self.effects.lock().await; + map.get(&tx.txid()).cloned() + } + + fn monitored_addresses(&self) -> Vec
{ + self.addresses.clone() + } + + fn watched_outpoints(&self) -> Vec { + 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 { + 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, +} + +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
{ + Vec::new() + } + + fn watched_outpoints(&self) -> Vec { + 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 { + self.event_sender.subscribe() + } + + async fn describe(&self) -> String { + "NonMatchingWallet (test implementation)".to_string() + } +} diff --git a/key-wallet/src/test_utils/mod.rs b/key-wallet/src/test_utils/mod.rs index ba8709556..bb5390835 100644 --- a/key-wallet/src/test_utils/mod.rs +++ b/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; diff --git a/key-wallet/src/test_utils/wallet.rs b/key-wallet/src/test_utils/wallet.rs index 5cf6ded55..139bca640 100644 --- a/key-wallet/src/test_utils/wallet.rs +++ b/key-wallet/src/test_utils/wallet.rs @@ -13,251 +13,6 @@ impl ManagedWalletInfo { } } -use crate::manager::MempoolTransactionResult; -use crate::manager::{BlockProcessingResult, WalletEvent, WalletInterface}; -use dashcore::address::NetworkUnchecked; -use dashcore::prelude::CoreBlockHeight; -use dashcore::{Block, OutPoint}; -use std::str::FromStr; -use std::{collections::BTreeMap, sync::Arc}; -use tokio::sync::{broadcast, Mutex}; - -// Type alias for transaction effects map -type TransactionEffectsMap = Arc)>>>; - -pub struct MockWallet { - processed_blocks: Arc>>, - processed_transactions: Arc>>, - // Map txid -> (net_amount, addresses) - effects: TransactionEffectsMap, - synced_height: CoreBlockHeight, - event_sender: broadcast::Sender, - /// When true, process_mempool_transaction returns is_relevant=true. - mempool_relevant: bool, - /// Addresses returned by monitored_addresses. - addresses: Vec
, - /// Outpoints returned by watched_outpoints. - outpoints: Vec, - /// New addresses returned by process_mempool_transaction. - mempool_new_addresses: Vec
, - /// Recorded status change notifications for test assertions. - status_changes: Arc>>, - /// 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
) { - self.addresses = addresses; - self.monitor_revision += 1; - } - - /// Set the outpoints returned by watched_outpoints. - pub fn set_outpoints(&mut self, outpoints: Vec) { - 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
) { - self.mempool_new_addresses = addresses; - } - - pub fn status_changes(&self) -> Arc>> { - self.status_changes.clone() - } - - pub async fn set_effect(&self, txid: dashcore::Txid, net: i64, addresses: Vec) { - let mut map = self.effects.lock().await; - map.insert(txid, (net, addresses)); - } - - pub fn processed_blocks(&self) -> Arc>> { - self.processed_blocks.clone() - } - - pub fn processed_transactions(&self) -> Arc>> { - 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::::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)> { - let map = self.effects.lock().await; - map.get(&tx.txid()).cloned() - } - - fn monitored_addresses(&self) -> Vec
{ - self.addresses.clone() - } - - fn watched_outpoints(&self) -> Vec { - 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 { - 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, -} - -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
{ - Vec::new() - } - - fn watched_outpoints(&self) -> Vec { - 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 { - self.event_sender.subscribe() - } - - async fn describe(&self) -> String { - "NonMatchingWallet (test implementation)".to_string() - } -} - /// Pre-built wallet context for transaction checking tests. /// /// Provides a testnet wallet with a default BIP44 account, a pre-derived