Skip to content

fix: route getdata requests to the peer that sent the inv #527

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
xdustinface merged 1 commit into from
Mar 17, 2026
Merged

fix: route getdata requests to the peer that sent the inv #527

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions dash-spv/src/network/manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

@ZocoLini ZocoLini Mar 12, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the SendMessage branch we are using the following logic:

let result = match &msg {
                                        // Distribute across peers for parallel sync
                                        NetworkMessage::GetCFHeaders(_)
                                        | NetworkMessage::GetCFilters(_)
                                        | NetworkMessage::GetData(_)
                                        | NetworkMessage::GetMnListD(_)
                                        | NetworkMessage::GetQRInfo(_)
                                        | NetworkMessage::GetHeaders(_)
                                        | NetworkMessage::GetHeaders2(_) => {
                                            this.send_distributed(msg).await
                                        }
                                        _ => {
                                            this.send_to_single_peer(msg).await
                                        }

more specifically, the Inv message triggers the send_to_single_peer path but you are calling send_distributed

@xdustinface xdustinface Mar 13, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah but there is no reason for the other messages to be sent "to a single peer" instead of following the same round robin distribution. There are still few things that im working on cleaning up here. Its okay like this for now.

@QuantumExplorer QuantumExplorer Mar 17, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure this is correct either... if we ask to send to a peer, but the peer doesn't exist, it seems weird to send to the network.

}
};
Comment thread
xdustinface marked this conversation as resolved.
if let Err(e) = result {
log::error!("Request processor: failed to send message to peer {}: {}", peer_address, e);
}
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
None => {
log::info!("Request processor: channel closed");
break;
Expand Down
23 changes: 21 additions & 2 deletions dash-spv/src/network/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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.
Expand All @@ -66,8 +69,24 @@ impl RequestSender {
.map_err(|e| NetworkError::ProtocolError(e.to_string()))
}

pub fn request_inventory(&self, inventory: Vec<Inventory>) -> 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<Inventory>,
peer_address: SocketAddr,
) -> NetworkResult<()> {
self.send_message_to_peer(NetworkMessage::GetData(inventory), peer_address)
}

pub fn request_block_headers(&self, start_hash: BlockHash) -> NetworkResult<()> {
Expand Down
3 changes: 2 additions & 1 deletion dash-spv/src/sync/chainlock/sync_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,8 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> 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 {
Expand Down
4 changes: 3 additions & 1 deletion dash-spv/src/sync/filters/pipeline.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
2 changes: 1 addition & 1 deletion dash-spv/src/sync/instantsend/sync_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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![])
}
Expand Down
Loading