From 821c4f8077b8de8a0302f70af623ac00ab30eb1a Mon Sep 17 00:00:00 2001 From: xdustinface Date: Fri, 3 Apr 2026 11:06:06 +1100 Subject: [PATCH 1/3] feat: rebroadcast unconfirmed self-sent transactions Each self-sent transaction in `recent_sends` tracks when it was last broadcast. Transactions whose last broadcast was more than 10 minutes ago are rebroadcast to all peers on each tick. Add `BroadcastMessage` variant to `NetworkRequest` so the mempool manager can request a broadcast via the request queue. Clean up `recent_sends` when a transaction is confirmed or IS-locked, since there is no need to rebroadcast finalized transactions. --- dash-spv/src/network/manager.rs | 15 ++++ dash-spv/src/network/mod.rs | 9 +++ dash-spv/src/sync/mempool/manager.rs | 92 +++++++++++++++++++++++ dash-spv/src/sync/mempool/sync_manager.rs | 3 + 4 files changed, 119 insertions(+) diff --git a/dash-spv/src/network/manager.rs b/dash-spv/src/network/manager.rs index f1beaabd4..f5927b7b8 100644 --- a/dash-spv/src/network/manager.rs +++ b/dash-spv/src/network/manager.rs @@ -822,6 +822,21 @@ impl PeerNetworkManager { } }); } + Some(NetworkRequest::BroadcastMessage(msg)) => { + tracing::debug!("Request processor: broadcasting {}", msg.cmd()); + let this = this.clone(); + tokio::spawn(async move { + let results = this.broadcast(msg).await; + let failures = results.iter().filter(|r| r.is_err()).count(); + if failures > 0 { + tracing::warn!( + "Request processor: broadcast had {} failures out of {} peers", + failures, + results.len() + ); + } + }); + } None => { tracing::info!("Request processor: channel closed"); break; diff --git a/dash-spv/src/network/mod.rs b/dash-spv/src/network/mod.rs index b4f2b2866..a428ddde1 100644 --- a/dash-spv/src/network/mod.rs +++ b/dash-spv/src/network/mod.rs @@ -47,6 +47,8 @@ pub enum NetworkRequest { SendMessage(NetworkMessage), /// Send a message to a specific peer. SendMessageToPeer(NetworkMessage, SocketAddr), + /// Broadcast a message to all connected peers. + BroadcastMessage(NetworkMessage), } /// Handle for managers to queue outgoing network requests. @@ -81,6 +83,13 @@ impl RequestSender { .map_err(|e| NetworkError::ProtocolError(e.to_string())) } + /// Queue a message to be broadcast to all connected peers. + pub(crate) fn broadcast(&self, msg: NetworkMessage) -> NetworkResult<()> { + self.tx + .send(NetworkRequest::BroadcastMessage(msg)) + .map_err(|e| NetworkError::ProtocolError(e.to_string())) + } + /// Request inventory from a specific peer. pub fn request_inventory( &self, diff --git a/dash-spv/src/sync/mempool/manager.rs b/dash-spv/src/sync/mempool/manager.rs index 333894220..1f83068b7 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -11,6 +11,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use dashcore::ephemerealdata::instant_lock::InstantLock; +use dashcore::network::message::NetworkMessage; use dashcore::network::message_blockdata::Inventory; use dashcore::{Amount, Transaction, Txid}; use rand::seq::IteratorRandom; @@ -42,6 +43,9 @@ const MAX_PENDING_IS_LOCKS: usize = 1000; /// Covers the window where multiple peers respond to the initial `mempool` request. const SEEN_TXID_EXPIRY: Duration = Duration::from_secs(180); +/// Per-transaction interval between rebroadcast attempts (10 minutes). +const REBROADCAST_INTERVAL: Duration = Duration::from_secs(600); + /// Mempool manager that monitors unconfirmed transactions from the P2P network. /// /// Tracks connected peers via a unified map where: @@ -436,6 +440,32 @@ impl MempoolManager { } } + /// Rebroadcast unconfirmed self-sent transactions to all peers. + /// + /// Each transaction in `recent_sends` tracks when it was last broadcast. + /// Transactions whose last broadcast was more than `REBROADCAST_INTERVAL` + /// ago are rebroadcast and their timestamp is reset. + pub(super) async fn rebroadcast_if_due(&mut self, requests: &RequestSender) { + let mut due: Vec = Vec::new(); + for (txid, last_broadcast) in &mut self.recent_sends { + if last_broadcast.elapsed() < REBROADCAST_INTERVAL { + continue; + } + if let Some(unconfirmed) = self.transactions.get(txid) { + due.push(unconfirmed.transaction.clone()); + *last_broadcast = Instant::now(); + } + } + + for tx in &due { + let _ = requests.broadcast(NetworkMessage::Tx(tx.clone())); + } + + if !due.is_empty() { + tracing::info!("Rebroadcast {} unconfirmed transaction(s) to all peers", due.len()); + } + } + fn is_queued(&self, txid: &Txid) -> bool { self.peers.values().filter_map(|v| v.as_ref()).any(|q| q.contains(txid)) } @@ -1603,6 +1633,68 @@ mod tests { ); } + #[tokio::test] + async fn test_rebroadcast_sends_old_recent_sends() { + let (mut manager, requests, mut rx) = create_test_manager(); + + let tx = Transaction { + version: 10, + lock_time: 0, + input: vec![], + output: vec![], + special_transaction_payload: None, + }; + let txid = tx.txid(); + + // Add a transaction and mark its last broadcast as older than the interval + manager.transactions.insert( + txid, + UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, true, Vec::new(), -100_000), + ); + manager + .recent_sends + .insert(txid, Instant::now() - REBROADCAST_INTERVAL - Duration::from_secs(1)); + + manager.rebroadcast_if_due(&requests).await; + + // Should have sent a BroadcastMessage for the transaction + let msg = rx.try_recv().expect("expected a rebroadcast message"); + assert!( + matches!(msg, NetworkRequest::BroadcastMessage(NetworkMessage::Tx(_))), + "expected BroadcastMessage(Tx), got {:?}", + msg + ); + + // Timestamp should be reset, so a second call should not rebroadcast + manager.rebroadcast_if_due(&requests).await; + assert!(rx.try_recv().is_err(), "should not rebroadcast immediately after reset"); + } + + #[tokio::test] + async fn test_rebroadcast_skips_recent_transactions() { + let (mut manager, requests, mut rx) = create_test_manager(); + + let tx = Transaction { + version: 11, + lock_time: 0, + input: vec![], + output: vec![], + special_transaction_payload: None, + }; + let txid = tx.txid(); + + // Add a transaction that was just sent (within the rebroadcast interval) + manager.transactions.insert( + txid, + UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, true, Vec::new(), -50_000), + ); + manager.recent_sends.insert(txid, Instant::now()); + + manager.rebroadcast_if_due(&requests).await; + + assert!(rx.try_recv().is_err(), "recently sent transactions should not be rebroadcast"); + } + #[test] fn test_peer_disconnect_keeps_other_peers_intact() { let (mut manager, _requests, _rx) = create_test_manager(); diff --git a/dash-spv/src/sync/mempool/sync_manager.rs b/dash-spv/src/sync/mempool/sync_manager.rs index e618ee0f9..806905224 100644 --- a/dash-spv/src/sync/mempool/sync_manager.rs +++ b/dash-spv/src/sync/mempool/sync_manager.rs @@ -117,6 +117,9 @@ impl SyncManager for MempoolManager { // Send queued getdata requests now that slots may have freed up self.send_queued(requests).await?; + // Rebroadcast unconfirmed self-sent transactions on a randomized interval + self.rebroadcast_if_due(requests).await; + // Rebuild bloom filter if the wallet's monitored set has changed. // // We poll the revision counter rather than using push-based wallet events From 6c9cc1fcfda6b205fa26c6e8023a8d24163ecaff Mon Sep 17 00:00:00 2001 From: xdustinface Date: Thu, 16 Apr 2026 08:28:47 +1000 Subject: [PATCH 2/3] test(dash-spv): fix Windows `Instant` underflow in rebroadcast test Project `now` forward via a private `rebroadcast_if_due_at` helper instead of subtracting `REBROADCAST_INTERVAL` from `Instant::now()`. On Windows, `Instant` is backed by `QueryPerformanceCounter` measured from boot, so on a freshly provisioned runner with uptime below the interval the subtraction underflows and panics. --- dash-spv/src/sync/mempool/manager.rs | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/dash-spv/src/sync/mempool/manager.rs b/dash-spv/src/sync/mempool/manager.rs index 1f83068b7..32299ec45 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -446,14 +446,21 @@ impl MempoolManager { /// Transactions whose last broadcast was more than `REBROADCAST_INTERVAL` /// ago are rebroadcast and their timestamp is reset. pub(super) async fn rebroadcast_if_due(&mut self, requests: &RequestSender) { + self.rebroadcast_if_due_at(requests, Instant::now()).await + } + + /// `now`-injected variant of [`Self::rebroadcast_if_due`]. Tests project `now` + /// forward instead of subtracting from `Instant::now()`, which underflows on + /// Windows when the QPC-based monotonic clock has a small value at boot. + async fn rebroadcast_if_due_at(&mut self, requests: &RequestSender, now: Instant) { let mut due: Vec = Vec::new(); for (txid, last_broadcast) in &mut self.recent_sends { - if last_broadcast.elapsed() < REBROADCAST_INTERVAL { + if now.saturating_duration_since(*last_broadcast) < REBROADCAST_INTERVAL { continue; } if let Some(unconfirmed) = self.transactions.get(txid) { due.push(unconfirmed.transaction.clone()); - *last_broadcast = Instant::now(); + *last_broadcast = now; } } @@ -1646,16 +1653,16 @@ mod tests { }; let txid = tx.txid(); - // Add a transaction and mark its last broadcast as older than the interval + let t0 = Instant::now(); + let later = t0 + REBROADCAST_INTERVAL + Duration::from_secs(1); + manager.transactions.insert( txid, UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, true, Vec::new(), -100_000), ); - manager - .recent_sends - .insert(txid, Instant::now() - REBROADCAST_INTERVAL - Duration::from_secs(1)); + manager.recent_sends.insert(txid, t0); - manager.rebroadcast_if_due(&requests).await; + manager.rebroadcast_if_due_at(&requests, later).await; // Should have sent a BroadcastMessage for the transaction let msg = rx.try_recv().expect("expected a rebroadcast message"); @@ -1665,8 +1672,9 @@ mod tests { msg ); - // Timestamp should be reset, so a second call should not rebroadcast - manager.rebroadcast_if_due(&requests).await; + // Timestamp should be reset to `later`, so a second call at the same instant + // must not rebroadcast. + manager.rebroadcast_if_due_at(&requests, later).await; assert!(rx.try_recv().is_err(), "should not rebroadcast immediately after reset"); } From 0ca9b398206e6e465fcc753f7f1043552512bce2 Mon Sep 17 00:00:00 2001 From: xdustinface Date: Fri, 17 Apr 2026 07:59:17 +1000 Subject: [PATCH 3/3] refactor: broadcast inline in `rebroadcast_if_due_at` instead of collecting into a `Vec` Addresses ZocoLini review comment on PR #627 https://github.com/dashpay/rust-dashcore/pull/627#discussion_r3096076404 --- dash-spv/src/sync/mempool/manager.rs | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/dash-spv/src/sync/mempool/manager.rs b/dash-spv/src/sync/mempool/manager.rs index 32299ec45..fc8fd23ea 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -453,23 +453,20 @@ impl MempoolManager { /// forward instead of subtracting from `Instant::now()`, which underflows on /// Windows when the QPC-based monotonic clock has a small value at boot. async fn rebroadcast_if_due_at(&mut self, requests: &RequestSender, now: Instant) { - let mut due: Vec = Vec::new(); + let mut count: usize = 0; for (txid, last_broadcast) in &mut self.recent_sends { if now.saturating_duration_since(*last_broadcast) < REBROADCAST_INTERVAL { continue; } if let Some(unconfirmed) = self.transactions.get(txid) { - due.push(unconfirmed.transaction.clone()); + let _ = requests.broadcast(NetworkMessage::Tx(unconfirmed.transaction.clone())); *last_broadcast = now; + count += 1; } } - for tx in &due { - let _ = requests.broadcast(NetworkMessage::Tx(tx.clone())); - } - - if !due.is_empty() { - tracing::info!("Rebroadcast {} unconfirmed transaction(s) to all peers", due.len()); + if count > 0 { + tracing::info!("Rebroadcast {} unconfirmed transaction(s) to all peers", count); } }