From 7b304f9f2f7bfdd72f4c82de70735a1fa7d3296a Mon Sep 17 00:00:00 2001 From: xdustinface Date: Thu, 12 Mar 2026 17:55:06 +0700 Subject: [PATCH] fix: route `getdata` requests to the peer that sent the `inv` We currently just randomly distribute `getdata` requests which leads to issues where we request inventory `x`, which we received from peer `A`, from peer `B` which doesn't have it (yet). This PR: - adds `SendMessageToPeer` variant to `NetworkRequest` and adjusts `request_inventory` in `RequestSender` to accept a `peer`. The request processing falls back to distributed send if the target peer disconnects. - adjusts the `instantsend` and `chainlock` `inv`->`getdata` flows. --- dash-spv/src/network/manager.rs | 30 +++++++++++++++++++ dash-spv/src/network/mod.rs | 23 ++++++++++++-- dash-spv/src/sync/chainlock/sync_manager.rs | 3 +- dash-spv/src/sync/filters/pipeline.rs | 4 ++- dash-spv/src/sync/instantsend/sync_manager.rs | 2 +- 5 files changed, 57 insertions(+), 5 deletions(-) diff --git a/dash-spv/src/network/manager.rs b/dash-spv/src/network/manager.rs index 95337fd5e..db294190e 100644 --- a/dash-spv/src/network/manager.rs +++ b/dash-spv/src/network/manager.rs @@ -790,6 +790,36 @@ impl PeerNetworkManager { } }); } + Some(NetworkRequest::SendMessageToPeer(msg, peer_address)) => { + log::debug!("Request processor: sending {} to peer {}", msg.cmd(), peer_address); + let this = this.clone(); + tokio::spawn(async move { + let fallback_msg = msg.clone(); + let result = match this.pool.get_peer(&peer_address).await { + Some(peer) => match this.send_message_to_peer(&peer_address, &peer, msg).await { + Ok(()) => Ok(()), + Err(err) => { + log::warn!( + "Target peer {} send failed ({}), falling back to distributed send", + peer_address, + err + ); + this.send_distributed(fallback_msg).await + } + }, + None => { + log::warn!( + "Target peer {} disconnected, falling back to distributed send", + peer_address + ); + this.send_distributed(fallback_msg).await + } + }; + if let Err(e) = result { + log::error!("Request processor: failed to send message to peer {}: {}", peer_address, e); + } + }); + } None => { log::info!("Request processor: channel closed"); break; diff --git a/dash-spv/src/network/mod.rs b/dash-spv/src/network/mod.rs index 18ed9710b..35695cdaa 100644 --- a/dash-spv/src/network/mod.rs +++ b/dash-spv/src/network/mod.rs @@ -34,6 +34,7 @@ pub use manager::PeerNetworkManager; pub use message_dispatcher::{Message, MessageDispatcher}; pub use message_type::MessageType; pub use peer::Peer; +use std::net::SocketAddr; use tokio::sync::mpsc::UnboundedReceiver; const FILTER_TYPE_DEFAULT: u8 = 0; @@ -43,6 +44,8 @@ const FILTER_TYPE_DEFAULT: u8 = 0; pub enum NetworkRequest { /// Send a message to the network. SendMessage(NetworkMessage), + /// Send a message to a specific peer. + SendMessageToPeer(NetworkMessage, SocketAddr), } /// Handle for managers to queue outgoing network requests. @@ -66,8 +69,24 @@ impl RequestSender { .map_err(|e| NetworkError::ProtocolError(e.to_string())) } - pub fn request_inventory(&self, inventory: Vec) -> NetworkResult<()> { - self.send_message(NetworkMessage::GetData(inventory)) + /// Queue a message to be sent to a specific peer. + fn send_message_to_peer( + &self, + msg: NetworkMessage, + peer_address: SocketAddr, + ) -> NetworkResult<()> { + self.tx + .send(NetworkRequest::SendMessageToPeer(msg, peer_address)) + .map_err(|e| NetworkError::ProtocolError(e.to_string())) + } + + /// Request inventory from a specific peer. + pub fn request_inventory( + &self, + inventory: Vec, + peer_address: SocketAddr, + ) -> NetworkResult<()> { + self.send_message_to_peer(NetworkMessage::GetData(inventory), peer_address) } pub fn request_block_headers(&self, start_hash: BlockHash) -> NetworkResult<()> { diff --git a/dash-spv/src/sync/chainlock/sync_manager.rs b/dash-spv/src/sync/chainlock/sync_manager.rs index 7fc277cf2..5cdc320ea 100644 --- a/dash-spv/src/sync/chainlock/sync_manager.rs +++ b/dash-spv/src/sync/chainlock/sync_manager.rs @@ -58,7 +58,8 @@ impl SyncManager for ChainLockManager "Received {} ChainLock announcements, requesting via getdata", chainlocks_to_request.len() ); - requests.request_inventory(chainlocks_to_request.clone())?; + requests + .request_inventory(chainlocks_to_request.clone(), msg.peer_address())?; for item in &chainlocks_to_request { if let Inventory::ChainLock(hash) = item { diff --git a/dash-spv/src/sync/filters/pipeline.rs b/dash-spv/src/sync/filters/pipeline.rs index 6906afef1..b52a6ace3 100644 --- a/dash-spv/src/sync/filters/pipeline.rs +++ b/dash-spv/src/sync/filters/pipeline.rs @@ -785,7 +785,9 @@ mod tests { // Verify message was sent let request = rx.try_recv().unwrap(); - let NetworkRequest::SendMessage(msg) = request; + let NetworkRequest::SendMessage(msg) = request else { + panic!("Expected SendMessage variant"); + }; if let NetworkMessage::GetCFilters(gcf) = msg { assert_eq!(gcf.start_height, 0); assert_eq!(gcf.filter_type, 0); diff --git a/dash-spv/src/sync/instantsend/sync_manager.rs b/dash-spv/src/sync/instantsend/sync_manager.rs index b9f58b77b..fdac17b77 100644 --- a/dash-spv/src/sync/instantsend/sync_manager.rs +++ b/dash-spv/src/sync/instantsend/sync_manager.rs @@ -49,7 +49,7 @@ impl SyncManager for InstantSendManager { "Received {} InstantSendLock announcements, requesting via getdata", islocks_to_request.len() ); - requests.request_inventory(islocks_to_request)?; + requests.request_inventory(islocks_to_request, msg.peer_address())?; } Ok(vec![]) }