diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index d4905b704..58c37bc93 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -124,11 +124,9 @@ impl // Build mempool manager if tracking is enabled if config.enable_mempool_tracking { - let mempool_state = Arc::new(RwLock::new(Default::default())); let initial_revision = wallet.read().await.monitor_revision(); managers.mempool = Some(MempoolManager::new( wallet.clone(), - mempool_state, config.mempool_strategy, config.max_mempool_transactions, initial_revision, diff --git a/dash-spv/src/sync/mempool/manager.rs b/dash-spv/src/sync/mempool/manager.rs index 2d55fab6b..333894220 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -23,7 +23,7 @@ use crate::error::SyncResult; use crate::network::RequestSender; use crate::sync::mempool::MempoolProgress; use crate::sync::SyncEvent; -use crate::types::{MempoolState, UnconfirmedTransaction}; +use crate::types::UnconfirmedTransaction; use key_wallet_manager::WalletInterface; /// Timeout for pruning expired mempool transactions (24 hours). @@ -50,7 +50,8 @@ const SEEN_TXID_EXPIRY: Duration = Duration::from_secs(180); pub(crate) struct MempoolManager { pub(super) progress: MempoolProgress, pub(super) wallet: Arc>, - pub(super) mempool_state: Arc>, + pub(super) transactions: HashMap, + pub(super) recent_sends: HashMap, strategy: MempoolStrategy, max_transactions: usize, /// Txids we have requested via getdata but not yet received, with request time. @@ -69,11 +70,10 @@ pub(crate) struct MempoolManager { } impl MempoolManager { - /// Creates a new mempool manager with the given wallet, shared mempool state, + /// Creates a new mempool manager with the given wallet, /// bloom filter strategy, and transaction capacity limit. pub(crate) fn new( wallet: Arc>, - mempool_state: Arc>, strategy: MempoolStrategy, max_transactions: usize, initial_monitor_revision: u64, @@ -81,7 +81,8 @@ impl MempoolManager { Self { progress: MempoolProgress::default(), wallet, - mempool_state, + transactions: HashMap::new(), + recent_sends: HashMap::new(), strategy, max_transactions, pending_requests: HashMap::new(), @@ -197,8 +198,7 @@ impl MempoolManager { peer: SocketAddr, requests: &RequestSender, ) -> SyncResult> { - let mempool_full = - self.mempool_state.read().await.transactions.len() >= self.max_transactions; + let mempool_full = self.transactions.len() >= self.max_transactions; if mempool_full { return Ok(vec![]); } @@ -214,7 +214,7 @@ impl MempoolManager { if self.seen_txids.get(txid).is_some_and(|t| t.elapsed() < SEEN_TXID_EXPIRY) || self.pending_requests.contains_key(txid) || self.is_queued(txid) - || self.mempool_state.read().await.transactions.contains_key(txid) + || self.transactions.contains_key(txid) { continue; } @@ -268,7 +268,7 @@ impl MempoolManager { break; }; if self.pending_requests.contains_key(&txid) - || self.mempool_state.read().await.transactions.contains_key(&txid) + || self.transactions.contains_key(&txid) { continue; } @@ -309,10 +309,10 @@ impl MempoolManager { let is_local = peer.ip().is_unspecified(); // Skip if already tracked (e.g., locally broadcast then received from a peer) - if self.mempool_state.read().await.transactions.contains_key(&txid) { + if self.transactions.contains_key(&txid) { self.seen_txids.insert(txid, Instant::now()); if is_local { - self.mempool_state.write().await.record_send(txid); + self.recent_sends.insert(txid, Instant::now()); } return Ok(vec![]); } @@ -346,36 +346,30 @@ impl MempoolManager { result.addresses, result.net_amount, ); - { - let mut state = self.mempool_state.write().await; - state.add_transaction(unconfirmed_tx); - if is_local { - state.record_send(txid); - } - self.progress.set_tracked(state.transactions.len() as u32); + self.transactions.insert(txid, unconfirmed_tx); + if is_local { + self.recent_sends.insert(txid, Instant::now()); } + self.progress.set_tracked(self.transactions.len() as u32); Ok(vec![]) } /// Remove transactions from the mempool that have been confirmed in a block. - pub(super) async fn remove_confirmed(&mut self, txids: &[Txid]) { + pub(super) fn remove_confirmed(&mut self, txids: &[Txid]) { self.seen_txids.retain(|_, t| t.elapsed() < SEEN_TXID_EXPIRY); let mut removed = Vec::new(); - { - let mut state = self.mempool_state.write().await; - for txid in txids { - if state.remove_transaction(txid).is_some() { - state.recent_sends.remove(txid); - removed.push(*txid); - } - } - if !removed.is_empty() { - self.progress.add_removed(removed.len() as u32); - self.progress.set_tracked(state.transactions.len() as u32); - tracing::debug!("Removed {} confirmed transactions from mempool", removed.len()); + for txid in txids { + if self.transactions.remove(txid).is_some() { + self.recent_sends.remove(txid); + removed.push(*txid); } } + if !removed.is_empty() { + self.progress.add_removed(removed.len() as u32); + self.progress.set_tracked(self.transactions.len() as u32); + tracing::debug!("Removed {} confirmed transactions from mempool", removed.len()); + } } /// Mark a mempool transaction as InstantSend-locked and notify the wallet. @@ -384,10 +378,9 @@ impl MempoolManager { /// can be applied when the transaction is later received via `handle_tx`. pub(super) async fn process_instant_send(&mut self, instant_lock: InstantLock) { let txid = instant_lock.txid; - let mut state = self.mempool_state.write().await; - let instant_lock_opt = if let Some(tx) = state.transactions.get_mut(&txid) { + let instant_lock_opt = if let Some(tx) = self.transactions.get_mut(&txid) { tx.is_instant_send = true; - state.recent_sends.remove(&txid); + self.recent_sends.remove(&txid); tracing::debug!("Marked mempool tx {} as InstantSend-locked", txid); Some(instant_lock) } else if self.pending_is_locks.len() < MAX_PENDING_IS_LOCKS { @@ -402,7 +395,6 @@ impl MempoolManager { ); None }; - drop(state); if let Some(lock) = instant_lock_opt { let mut wallet = self.wallet.write().await; wallet.process_instant_send_lock(lock); @@ -410,14 +402,27 @@ impl MempoolManager { } /// Prune transactions and pending IS locks older than `timeout`. - pub(super) async fn prune_expired(&mut self, timeout: Duration) { - let mut state = self.mempool_state.write().await; - let pruned = state.prune_expired(timeout); - if !pruned.is_empty() { - self.progress.add_removed(pruned.len() as u32); - self.progress.set_tracked(state.transactions.len() as u32); - tracing::debug!("Pruned {} expired mempool transactions", pruned.len()); - for txid in &pruned { + pub(super) fn prune_expired(&mut self, timeout: Duration) { + let mut expired_txids = Vec::new(); + self.transactions.retain(|txid, tx| { + if tx.is_expired(timeout) { + expired_txids.push(*txid); + false + } else { + true + } + }); + + // Prune old recent sends + if let Some(cutoff) = Instant::now().checked_sub(timeout) { + self.recent_sends.retain(|_, &mut timestamp| timestamp > cutoff); + } + + if !expired_txids.is_empty() { + self.progress.add_removed(expired_txids.len() as u32); + self.progress.set_tracked(self.transactions.len() as u32); + tracing::debug!("Pruned {} expired mempool transactions", expired_txids.len()); + for txid in &expired_txids { self.pending_is_locks.remove(txid); } } @@ -548,12 +553,10 @@ mod tests { fn create_test_manager( ) -> (MempoolManager, RequestSender, mpsc::UnboundedReceiver) { let wallet = Arc::new(RwLock::new(MockWallet::new())); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::FetchAll, 1000, 0); + let mut manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 1000, 0); manager.progress.set_state(SyncState::Synced); (manager, requests, rx) @@ -562,12 +565,10 @@ mod tests { fn create_bloom_manager( ) -> (MempoolManager, RequestSender, mpsc::UnboundedReceiver) { let wallet = Arc::new(RwLock::new(MockWallet::new())); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::BloomFilter, 1000, 0); + let manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); (manager, requests, rx) } @@ -628,13 +629,11 @@ mod tests { #[tokio::test] async fn test_handle_inv_capacity_limit() { let wallet = Arc::new(RwLock::new(MockWallet::new())); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, _rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); let mut manager = MempoolManager::new( wallet, - mempool_state.clone(), MempoolStrategy::FetchAll, 2, // Very small capacity 0, @@ -643,25 +642,19 @@ mod tests { manager.peers.insert(peer, Some(VecDeque::new())); // Fill mempool to capacity - { - let mut state = mempool_state.write().await; - for i in 0..2u32 { - let tx = Transaction { - version: 1, - lock_time: i, - input: vec![], - output: vec![], - special_transaction_payload: None, - }; - state.add_transaction(UnconfirmedTransaction::new( - tx, - Amount::from_sat(0), - false, - false, - Vec::new(), - 0, - )); - } + for i in 0..2u32 { + let tx = Transaction { + version: 1, + lock_time: i, + input: vec![], + output: vec![], + special_transaction_payload: None, + }; + let txid = tx.txid(); + manager.transactions.insert( + txid, + UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0), + ); } // New transactions should be filtered out @@ -675,12 +668,10 @@ mod tests { #[tokio::test] async fn test_handle_inv_pending_requests_limit() { let wallet = Arc::new(RwLock::new(MockWallet::new())); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, _rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::FetchAll, 2, 0); + let mut manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 2, 0); manager.progress.set_state(SyncState::Synced); let peer = test_socket_address(1); manager.peers.insert(peer, Some(VecDeque::new())); @@ -701,12 +692,10 @@ mod tests { #[test] fn test_prune_pending_requests_timeout() { let wallet = Arc::new(RwLock::new(MockWallet::new())); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, _rx) = mpsc::unbounded_channel::(); let _requests = RequestSender::new(tx); - let mut manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::FetchAll, 1000, 0); + let mut manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 1000, 0); let fresh_txid = Txid::from_byte_array([1; 32]); let stale_txid = Txid::from_byte_array([2; 32]); @@ -740,9 +729,8 @@ mod tests { assert!(events.is_empty()); assert_eq!(manager.progress.received(), 1); - // Irrelevant tx should not be stored in mempool state - let state = manager.mempool_state.read().await; - assert!(!state.transactions.contains_key(&txid)); + // Irrelevant tx should not be stored + assert!(!manager.transactions.contains_key(&txid)); assert_eq!(manager.progress.relevant(), 0); } @@ -763,8 +751,8 @@ mod tests { assert_eq!(manager.pending_requests.len(), 1); } - #[tokio::test] - async fn test_prune_expired() { + #[test] + fn test_prune_expired() { let (mut manager, _requests, _rx) = create_test_manager(); let fresh_tx = Transaction { @@ -786,35 +774,26 @@ mod tests { let expired_txid = expired_tx.txid(); let test_timeout = Duration::from_secs(2); - { - let mut state = manager.mempool_state.write().await; - state.add_transaction(UnconfirmedTransaction::new( - fresh_tx, - Amount::from_sat(0), - false, - false, - Vec::new(), - 0, - )); - let mut expired_utx = UnconfirmedTransaction::new( - expired_tx, - Amount::from_sat(0), - false, - false, - Vec::new(), - 0, - ); - expired_utx.first_seen = Instant::now() - test_timeout - Duration::from_secs(1); - state.add_transaction(expired_utx); - } + manager.transactions.insert( + fresh_txid, + UnconfirmedTransaction::new(fresh_tx, Amount::from_sat(0), false, false, Vec::new(), 0), + ); + let mut expired_utx = UnconfirmedTransaction::new( + expired_tx, + Amount::from_sat(0), + false, + false, + Vec::new(), + 0, + ); + expired_utx.first_seen = Instant::now() - test_timeout - Duration::from_secs(1); + manager.transactions.insert(expired_txid, expired_utx); - manager.prune_expired(test_timeout).await; + manager.prune_expired(test_timeout); - let state = manager.mempool_state.read().await; - assert_eq!(state.transactions.len(), 1); - assert!(state.transactions.contains_key(&fresh_txid)); - assert!(!state.transactions.contains_key(&expired_txid)); - drop(state); + assert_eq!(manager.transactions.len(), 1); + assert!(manager.transactions.contains_key(&fresh_txid)); + assert!(!manager.transactions.contains_key(&expired_txid)); assert_eq!(manager.progress.removed(), 1); } @@ -826,17 +805,10 @@ mod tests { let mut mock = MockWallet::new(); mock.set_mempool_relevant(true); let wallet = Arc::new(RwLock::new(mock)); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, _rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = MempoolManager::new( - wallet.clone(), - mempool_state, - MempoolStrategy::BloomFilter, - 1000, - 0, - ); + let manager = MempoolManager::new(wallet.clone(), MempoolStrategy::BloomFilter, 1000, 0); (manager, requests, wallet) } @@ -857,13 +829,11 @@ mod tests { let events = manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); assert!(events.is_empty()); - // Verify transaction was stored in mempool state - let state = manager.mempool_state.read().await; - assert!(state.transactions.contains_key(&txid)); + // Verify transaction was stored + assert!(manager.transactions.contains_key(&txid)); assert_eq!(manager.progress.received(), 1); assert_eq!(manager.progress.relevant(), 1); assert_eq!(manager.progress.tracked(), 1); - drop(state); // Processing the same transaction again should be a no-op (dedup guard) let tx2 = Transaction { @@ -876,8 +846,7 @@ mod tests { let events = manager.handle_tx(tx2, test_socket_address(1)).await.unwrap(); assert!(events.is_empty()); - let state = manager.mempool_state.read().await; - assert_eq!(state.transactions.len(), 1); + assert_eq!(manager.transactions.len(), 1); // Progress counters should not have incremented assert_eq!(manager.progress.received(), 1); assert_eq!(manager.progress.relevant(), 1); @@ -900,10 +869,9 @@ mod tests { let local_addr = SocketAddr::from(([0, 0, 0, 0], 0)); manager.handle_tx(tx, local_addr).await.unwrap(); - let state = manager.mempool_state.read().await; - assert!(state.transactions.contains_key(&txid)); + assert!(manager.transactions.contains_key(&txid)); assert!( - state.is_recent_send(&txid, Duration::from_secs(10)), + manager.recent_sends.contains_key(&txid), "locally dispatched transaction should be recorded as a recent send" ); } @@ -923,10 +891,9 @@ mod tests { manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); - let state = manager.mempool_state.read().await; - assert!(state.transactions.contains_key(&txid)); + assert!(manager.transactions.contains_key(&txid)); assert!( - !state.is_recent_send(&txid, Duration::from_secs(10)), + !manager.recent_sends.contains_key(&txid), "peer-received transaction should not be recorded as a recent send" ); } @@ -953,8 +920,7 @@ mod tests { assert!(!manager.pending_requests.contains_key(&txid)); // Since the manager uses BloomFilter strategy (relevant mock), tx should be stored - let state = manager.mempool_state.read().await; - assert!(state.transactions.contains_key(&txid)); + assert!(manager.transactions.contains_key(&txid)); } fn create_bloom_manager_with_addresses( @@ -963,12 +929,10 @@ mod tests { let mut mock = MockWallet::new(); mock.set_addresses(addresses); let wallet = Arc::new(RwLock::new(mock)); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::BloomFilter, 1000, 0); + let manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); (manager, requests, rx) } @@ -1012,29 +976,20 @@ mod tests { special_transaction_payload: None, }; let txid = tx.txid(); - { - let mut state = manager.mempool_state.write().await; - state.add_transaction(UnconfirmedTransaction::new( - tx, - Amount::from_sat(0), - false, - false, - Vec::new(), - 0, - )); - state.record_send(txid); - } + manager.transactions.insert( + txid, + UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0), + ); + manager.recent_sends.insert(txid, Instant::now()); manager.process_instant_send(dummy_instant_lock(txid)).await; - // Verify mempool state reflects IS flag and recent_sends is cleaned up - let state = manager.mempool_state.read().await; - assert!(state.transactions.get(&txid).unwrap().is_instant_send); + // Verify IS flag and recent_sends cleanup + assert!(manager.transactions.get(&txid).unwrap().is_instant_send); assert!( - !state.recent_sends.contains_key(&txid), + !manager.recent_sends.contains_key(&txid), "IS-locked transaction should be removed from recent_sends" ); - drop(state); let wallet = manager.wallet.read().await; let status_changes = wallet.status_changes(); @@ -1140,18 +1095,11 @@ mod tests { // Enqueue the txid on an activated peer manager.peers.insert(peer, Some(VecDeque::from([txid]))); - // Simulate the transaction arriving in mempool_state before send - { - let mut state = manager.mempool_state.write().await; - state.add_transaction(UnconfirmedTransaction::new( - tx, - Amount::from_sat(0), - false, - false, - Vec::new(), - 0, - )); - } + // Simulate the transaction arriving before send + manager.transactions.insert( + txid, + UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0), + ); manager.send_queued(&requests).await.unwrap(); // Txid should have been skipped, not added to pending @@ -1227,9 +1175,7 @@ mod tests { assert!(manager.pending_is_locks.is_empty()); // Transaction stored with IS flag set - let state = manager.mempool_state.read().await; - assert!(state.transactions.get(&txid).unwrap().is_instant_send); - drop(state); + assert!(manager.transactions.get(&txid).unwrap().is_instant_send); // Wallet received the IS lock payload with the correct cyclehash let w = wallet.read().await; @@ -1266,9 +1212,8 @@ mod tests { // Pending IS lock cleaned up (no leak) assert!(manager.pending_is_locks.is_empty()); - // Irrelevant tx should not be stored in mempool state - let state = manager.mempool_state.read().await; - assert!(!state.transactions.contains_key(&txid)); + // Irrelevant tx should not be stored + assert!(!manager.transactions.contains_key(&txid)); } #[tokio::test] @@ -1291,8 +1236,8 @@ mod tests { assert_eq!(manager.pending_is_locks.len(), MAX_PENDING_IS_LOCKS); } - #[tokio::test] - async fn test_prune_expired_removes_is_lock_for_expired_tx() { + #[test] + fn test_prune_expired_removes_is_lock_for_expired_tx() { let (mut manager, _requests, _rx) = create_test_manager(); let tx = Transaction { @@ -1307,13 +1252,10 @@ mod tests { let test_timeout = Duration::from_secs(2); // Add the tx with a timestamp in the past so it expires - { - let mut state = manager.mempool_state.write().await; - let mut utx = - UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0); - utx.first_seen = Instant::now() - test_timeout - Duration::from_secs(1); - state.add_transaction(utx); - } + let mut utx = + UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0); + utx.first_seen = Instant::now() - test_timeout - Duration::from_secs(1); + manager.transactions.insert(txid, utx); // Also store a pending IS lock for this txid and an unrelated one let unrelated_txid = Txid::from_byte_array([0xdd; 32]); @@ -1322,7 +1264,7 @@ mod tests { .pending_is_locks .insert(unrelated_txid, (dummy_instant_lock(unrelated_txid), Instant::now())); - manager.prune_expired(test_timeout).await; + manager.prune_expired(test_timeout); // The expired tx's IS lock should be removed assert!( @@ -1336,8 +1278,8 @@ mod tests { ); } - #[tokio::test] - async fn test_prune_expired_removes_stale_pending_is_locks() { + #[test] + fn test_prune_expired_removes_stale_pending_is_locks() { let (mut manager, _requests, _rx) = create_test_manager(); let test_timeout = Duration::from_secs(2); @@ -1358,7 +1300,7 @@ mod tests { .pending_is_locks .insert(fresh_txid, (dummy_instant_lock(fresh_txid), Instant::now())); - manager.prune_expired(test_timeout).await; + manager.prune_expired(test_timeout); assert!( !manager.pending_is_locks.contains_key(&stale_txid), @@ -1407,12 +1349,10 @@ mod tests { let mut mock = MockWallet::new(); mock.set_addresses(vec![addr]); let wallet = Arc::new(RwLock::new(mock)); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::BloomFilter, 1000, 0); + let mut manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); // Drop receiver so send_filter_load fails drop(rx); @@ -1445,8 +1385,7 @@ mod tests { manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); - let state = manager.mempool_state.read().await; - let stored = state.transactions.get(&txid).unwrap(); + let stored = manager.transactions.get(&txid).unwrap(); assert_eq!(stored.net_amount, 50000); assert!(!stored.is_outgoing); assert!(!stored.is_instant_send); @@ -1474,8 +1413,7 @@ mod tests { manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); - let state = manager.mempool_state.read().await; - let stored = state.transactions.get(&txid).unwrap(); + let stored = manager.transactions.get(&txid).unwrap(); assert_eq!(stored.net_amount, -30000); assert!(stored.is_outgoing); assert!(!stored.is_instant_send); @@ -1558,62 +1496,52 @@ mod tests { assert!(manager.peers.is_empty()); } - #[tokio::test] - async fn test_remove_confirmed_removes_txids() { + #[test] + fn test_remove_confirmed_removes_txids() { let (mut manager, _requests, _rx) = create_test_manager(); let mut txids = Vec::new(); - { - let mut state = manager.mempool_state.write().await; - for i in 0..3u32 { - let tx = Transaction { - version: 1, - lock_time: i, - input: vec![], - output: vec![], - special_transaction_payload: None, - }; - let txid = tx.txid(); - txids.push(txid); - state.add_transaction(UnconfirmedTransaction::new( - tx, - Amount::from_sat(0), - false, - false, - Vec::new(), - 0, - )); - } - assert_eq!(state.transactions.len(), 3); - // Mark two as recent sends - state.record_send(txids[0]); - state.record_send(txids[1]); + for i in 0..3u32 { + let tx = Transaction { + version: 1, + lock_time: i, + input: vec![], + output: vec![], + special_transaction_payload: None, + }; + let txid = tx.txid(); + txids.push(txid); + manager.transactions.insert( + txid, + UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0), + ); } + assert_eq!(manager.transactions.len(), 3); + // Mark two as recent sends + manager.recent_sends.insert(txids[0], Instant::now()); + manager.recent_sends.insert(txids[1], Instant::now()); // Remove 2 of the 3 transactions - manager.remove_confirmed(&txids[..2]).await; + manager.remove_confirmed(&txids[..2]); - let state = manager.mempool_state.read().await; - assert_eq!(state.transactions.len(), 1); - assert!(state.transactions.contains_key(&txids[2])); - assert!(!state.recent_sends.contains_key(&txids[0])); - assert!(!state.recent_sends.contains_key(&txids[1])); - drop(state); + assert_eq!(manager.transactions.len(), 1); + assert!(manager.transactions.contains_key(&txids[2])); + assert!(!manager.recent_sends.contains_key(&txids[0])); + assert!(!manager.recent_sends.contains_key(&txids[1])); assert_eq!(manager.progress.removed(), 2); assert_eq!(manager.progress.tracked(), 1); } - #[tokio::test] - async fn test_remove_confirmed_unknown_txids_noop() { + #[test] + fn test_remove_confirmed_unknown_txids_noop() { let (mut manager, _requests, _rx) = create_test_manager(); let unknown = vec![Txid::from_byte_array([0xaa; 32]), Txid::from_byte_array([0xbb; 32])]; - manager.remove_confirmed(&unknown).await; + manager.remove_confirmed(&unknown); - let state = manager.mempool_state.read().await; - assert!(state.transactions.is_empty()); + assert!(manager.transactions.is_empty()); assert_eq!(manager.progress.removed(), 0); } diff --git a/dash-spv/src/sync/mempool/sync_manager.rs b/dash-spv/src/sync/mempool/sync_manager.rs index 61a1f324a..e618ee0f9 100644 --- a/dash-spv/src/sync/mempool/sync_manager.rs +++ b/dash-spv/src/sync/mempool/sync_manager.rs @@ -88,7 +88,7 @@ impl SyncManager for MempoolManager { // Remove confirmed transactions from mempool. // Bloom filter rebuild is handled by the tick's revision check. if !confirmed_txids.is_empty() { - self.remove_confirmed(confirmed_txids).await; + self.remove_confirmed(confirmed_txids); } Ok(vec![]) } @@ -109,7 +109,7 @@ impl SyncManager for MempoolManager { } // Prune expired transactions periodically - self.prune_expired(MEMPOOL_TX_EXPIRY).await; + self.prune_expired(MEMPOOL_TX_EXPIRY); // Prune pending requests that never received a response self.prune_pending_requests(); @@ -189,7 +189,6 @@ mod tests { use crate::client::config::MempoolStrategy; use crate::network::NetworkRequest; use crate::test_utils::test_socket_address; - use crate::types::MempoolState; use dashcore::hashes::Hash; use key_wallet_manager::test_utils::MockWallet; use std::sync::Arc; @@ -198,12 +197,10 @@ mod tests { fn create_test_manager( ) -> (MempoolManager, RequestSender, mpsc::UnboundedReceiver) { let wallet = Arc::new(RwLock::new(MockWallet::new())); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::FetchAll, 1000, 0); + let manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 1000, 0); (manager, requests, rx) } @@ -360,29 +357,29 @@ mod tests { }; manager.handle_sync_event(&sync, &requests).await.unwrap(); - // Add transactions to mempool state + // Add transactions to mempool let mut txids = Vec::new(); - { - let mut state = manager.mempool_state.write().await; - for i in 0..2u32 { - let tx = dashcore::Transaction { - version: 1, - lock_time: i, - input: vec![], - output: vec![], - special_transaction_payload: None, - }; - let txid = tx.txid(); - txids.push(txid); - state.add_transaction(crate::types::UnconfirmedTransaction::new( + for i in 0..2u32 { + let tx = dashcore::Transaction { + version: 1, + lock_time: i, + input: vec![], + output: vec![], + special_transaction_payload: None, + }; + let txid = tx.txid(); + txids.push(txid); + manager.transactions.insert( + txid, + crate::types::UnconfirmedTransaction::new( tx, dashcore::Amount::from_sat(0), false, false, Vec::new(), 0, - )); - } + ), + ); } let event = SyncEvent::BlockProcessed { @@ -394,8 +391,7 @@ mod tests { let events = manager.handle_sync_event(&event, &requests).await.unwrap(); assert!(events.is_empty()); - let state = manager.mempool_state.read().await; - assert!(state.transactions.is_empty()); + assert!(manager.transactions.is_empty()); } #[tokio::test] @@ -419,17 +415,17 @@ mod tests { special_transaction_payload: None, }; let txid = tx.txid(); - { - let mut state = manager.mempool_state.write().await; - state.add_transaction(crate::types::UnconfirmedTransaction::new( + manager.transactions.insert( + txid, + crate::types::UnconfirmedTransaction::new( tx, dashcore::Amount::from_sat(0), false, false, Vec::new(), 0, - )); - } + ), + ); // Fire InstantLockReceived with a lock whose txid matches let mut is_lock = dashcore::InstantLock::dummy(0..1); @@ -442,8 +438,7 @@ mod tests { let events = manager.handle_sync_event(&event, &requests).await.unwrap(); assert!(events.is_empty()); - let state = manager.mempool_state.read().await; - assert!(state.transactions.get(&txid).unwrap().is_instant_send); + assert!(manager.transactions.get(&txid).unwrap().is_instant_send); } #[tokio::test] @@ -553,12 +548,10 @@ mod tests { let addr = dashcore::Address::from_script(&script, dashcore::Network::Testnet).unwrap(); mock.set_addresses(vec![addr]); let wallet = Arc::new(RwLock::new(mock)); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, mut rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = - MempoolManager::new(wallet, mempool_state, MempoolStrategy::BloomFilter, 1000, 0); + let mut manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); let peer = test_socket_address(1); manager.handle_peer_connected(peer); @@ -623,13 +616,11 @@ mod tests { mock.set_addresses(vec![addr.clone()]); let initial_revision = mock.monitor_revision(); let wallet = Arc::new(RwLock::new(mock)); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, mut rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); let mut manager = MempoolManager::new( wallet.clone(), - mempool_state, MempoolStrategy::BloomFilter, 1000, initial_revision, @@ -685,12 +676,10 @@ mod tests { #[tokio::test] async fn test_tick_skips_rebuild_for_fetch_all_strategy() { let wallet = Arc::new(RwLock::new(MockWallet::new())); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, mut rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = - MempoolManager::new(wallet.clone(), mempool_state, MempoolStrategy::FetchAll, 1000, 0); + let mut manager = MempoolManager::new(wallet.clone(), MempoolStrategy::FetchAll, 1000, 0); let peer = test_socket_address(1); manager.handle_peer_connected(peer); @@ -736,13 +725,11 @@ mod tests { mock.set_addresses(vec![addr]); let initial_revision = mock.monitor_revision(); let wallet = Arc::new(RwLock::new(mock)); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx, mut rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); let mut manager = MempoolManager::new( wallet.clone(), - mempool_state, MempoolStrategy::BloomFilter, 1000, initial_revision, @@ -787,13 +774,11 @@ mod tests { mock.set_addresses(vec![addr]); let initial_revision = mock.monitor_revision(); let wallet = Arc::new(RwLock::new(mock)); - let mempool_state = Arc::new(RwLock::new(MempoolState::default())); let (tx_chan, mut rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx_chan); let mut manager = MempoolManager::new( wallet.clone(), - mempool_state, MempoolStrategy::BloomFilter, 1000, initial_revision, diff --git a/dash-spv/src/types.rs b/dash-spv/src/types.rs index 8f632482d..dc21b4732 100644 --- a/dash-spv/src/types.rs +++ b/dash-spv/src/types.rs @@ -247,85 +247,3 @@ impl UnconfirmedTransaction { self.fee.to_sat() as f64 / self.size as f64 } } - -/// Mempool state tracking. -#[derive(Debug, Clone, Default)] -pub struct MempoolState { - /// Currently tracked unconfirmed transactions. - pub transactions: std::collections::HashMap, - /// Recent sends (txid -> timestamp) for Selective strategy. - pub recent_sends: std::collections::HashMap, - /// Total pending balance change. - pub pending_balance: i64, - /// Total pending InstantSend balance. - pub pending_instant_balance: i64, -} - -impl MempoolState { - /// Add a transaction to mempool. - pub fn add_transaction(&mut self, tx: UnconfirmedTransaction) { - if tx.is_instant_send { - self.pending_instant_balance += tx.net_amount; - } else { - self.pending_balance += tx.net_amount; - } - - let txid = tx.txid(); - self.transactions.insert(txid, tx); - } - - /// Remove a transaction from mempool. - pub fn remove_transaction(&mut self, txid: &Txid) -> Option { - if let Some(tx) = self.transactions.remove(txid) { - if tx.is_instant_send { - self.pending_instant_balance -= tx.net_amount; - } else { - self.pending_balance -= tx.net_amount; - } - Some(tx) - } else { - None - } - } - - /// Prune expired transactions. - pub fn prune_expired(&mut self, timeout: Duration) -> Vec { - let mut expired = Vec::new(); - - self.transactions.retain(|txid, tx| { - if tx.is_expired(timeout) { - expired.push(*txid); - if tx.is_instant_send { - self.pending_instant_balance -= tx.net_amount; - } else { - self.pending_balance -= tx.net_amount; - } - false - } else { - true - } - }); - - // Also prune old recent sends - if let Some(cutoff) = Instant::now().checked_sub(timeout) { - self.recent_sends.retain(|_, &mut timestamp| timestamp > cutoff); - } - - expired - } - - /// Record a recent send. - pub fn record_send(&mut self, txid: Txid) { - self.recent_sends.insert(txid, Instant::now()); - } - - /// Check if a transaction was recently sent. - pub fn is_recent_send(&self, txid: &Txid, window: Duration) -> bool { - self.recent_sends.get(txid).map(|×tamp| timestamp.elapsed() < window).unwrap_or(false) - } - - /// Get total pending balance (regular + InstantSend). - pub fn total_pending_balance(&self) -> i64 { - self.pending_balance + self.pending_instant_balance - } -}