From 51e424d523e10c552ff476a2de7204d3a49a960e Mon Sep 17 00:00:00 2001 From: Andrus Salumets Date: Tue, 26 Aug 2025 14:36:43 +0200 Subject: [PATCH] feat(sync): handle not connected peers --- .../cryptarchia-sync/src/libp2p/behaviour.rs | 60 +------------------ tests/src/tests/cryptarchia/bootstrap.rs | 11 +++- tests/src/tests/cryptarchia/orphan.rs | 10 +++- tests/src/topology/configs/mod.rs | 10 +++- tests/src/topology/configs/network.rs | 36 +++++++---- 5 files changed, 53 insertions(+), 74 deletions(-) diff --git a/consensus/cryptarchia-sync/src/libp2p/behaviour.rs b/consensus/cryptarchia-sync/src/libp2p/behaviour.rs index b7dc5a635..39e850cc6 100644 --- a/consensus/cryptarchia-sync/src/libp2p/behaviour.rs +++ b/consensus/cryptarchia-sync/src/libp2p/behaviour.rs @@ -8,8 +8,8 @@ use libp2p::{ core::{transport::PortUse, Endpoint}, futures::stream::FuturesUnordered, swarm::{ - behaviour::ConnectionEstablished, ConnectionClosed, ConnectionDenied, ConnectionHandler, - ConnectionId, FromSwarm, NetworkBehaviour, THandlerInEvent, ToSwarm, + ConnectionDenied, ConnectionHandler, ConnectionId, FromSwarm, NetworkBehaviour, + THandlerInEvent, ToSwarm, }, Multiaddr, PeerId, Stream as Libp2pStream, Stream, StreamProtocol, }; @@ -22,7 +22,7 @@ use crate::{ config::Config, libp2p::{ downloader::Downloader, - errors::{ChainSyncError, ChainSyncErrorKind}, + errors::ChainSyncError, provider::{Provider, ReceivingRequestStream, MAX_ADDITIONAL_BLOCKS}, }, messages::{DownloadBlocksRequest, GetTipResponse, RequestMessage, SerialisedBlock}, @@ -124,8 +124,6 @@ pub struct Behaviour { control: Control, /// A handle to listen to incoming stream requests. incoming_streams: IncomingStreams, - /// List of connected peers. - connected_peers: HashSet, /// Futures for reading incoming requests. This is common to both tip and /// block because initially we don't know which request we receive over /// stream. After reading the request, we use dedicated `FuturesUnordered`, @@ -169,7 +167,6 @@ impl Behaviour { stream_behaviour, control, incoming_streams, - connected_peers: HashSet::new(), receiving_block_responses: FuturesUnordered::new(), sending_block_responses: FuturesUnordered::new(), receiving_requests: FuturesUnordered::new(), @@ -183,26 +180,11 @@ impl Behaviour { } } - fn add_peer(&mut self, peer: PeerId) { - self.connected_peers.insert(peer); - } - - fn remove_peer(&mut self, peer: &PeerId) { - self.connected_peers.remove(peer); - } - pub fn request_tip( &self, peer_id: PeerId, reply_sender: oneshot::Sender>, ) -> Result<(), ChainSyncError> { - if !self.connected_peers.contains(&peer_id) { - return Err(ChainSyncError { - peer: peer_id, - kind: ChainSyncErrorKind::RequestTipError("Peer is not connected".to_owned()), - }); - } - let mut control = self.control.clone(); self.sending_tip_requests.push( @@ -224,17 +206,7 @@ impl Behaviour { additional_blocks: HashSet, reply_sender: oneshot::Sender>>, ) -> Result<(), ChainSyncError> { - if !self.connected_peers.contains(&peer_id) { - return Err(ChainSyncError { - peer: peer_id, - kind: ChainSyncErrorKind::RequestBlocksDownloadError( - "Peer is neither connected nor known".to_owned(), - ), - }); - } - let control = self.control.clone(); - let request = DownloadBlocksRequest::new( target_block, local_tip, @@ -430,15 +402,6 @@ impl NetworkBehaviour for Behaviour { } fn on_swarm_event(&mut self, event: FromSwarm) { - match event { - FromSwarm::ConnectionEstablished(ConnectionEstablished { peer_id, .. }) => { - self.add_peer(peer_id); - } - FromSwarm::ConnectionClosed(ConnectionClosed { peer_id, .. }) => { - self.remove_peer(&peer_id); - } - _ => {} - } self.stream_behaviour.on_swarm_event(event); } @@ -591,23 +554,6 @@ mod tests { assert_eq!(errors.len(), 0); } - #[tokio::test] - async fn test_download_with_no_peers() { - let (response_tx, _response_rx) = oneshot::channel(); - let err = Behaviour::new(Config::default()) - .start_blocks_download( - PeerId::random(), - HeaderId::from([0; 32]), - HeaderId::from([0; 32]), - HeaderId::from([0; 32]), - HashSet::new(), - response_tx, - ) - .unwrap_err(); - - matches!(err.kind, ChainSyncErrorKind::RequestBlocksDownloadError(_)); - } - #[tokio::test] async fn test_reject_excess_download_requests() { let (mut downloader_swarm, provider_peer_id) = start_provider_and_downloader(1).await; diff --git a/tests/src/tests/cryptarchia/bootstrap.rs b/tests/src/tests/cryptarchia/bootstrap.rs index 582d2db8c..28070f55e 100644 --- a/tests/src/tests/cryptarchia/bootstrap.rs +++ b/tests/src/tests/cryptarchia/bootstrap.rs @@ -9,7 +9,11 @@ use tests::{ common::sync::{wait_for_validators_mode, wait_for_validators_mode_and_height}, nodes::validator::{create_validator_config, Validator}, secret_key_to_peer_id, - topology::configs::{create_general_configs, GeneralConfig}, + topology::configs::{ + create_general_configs_with_network, + network::{Libp2pNetworkLayout, NetworkParams}, + GeneralConfig, + }, }; #[tokio::test] @@ -17,7 +21,10 @@ async fn test_ibd_behind_nodes() { let n_validators = 4; let n_initial_validators = 2; - let general_configs = create_general_configs(n_validators); + let network_params = NetworkParams { + libp2p_network_layout: Libp2pNetworkLayout::Full, + }; + let general_configs = create_general_configs_with_network(n_validators, &network_params); let mut initial_validators = vec![]; for config in general_configs.iter().take(n_initial_validators) { diff --git a/tests/src/tests/cryptarchia/orphan.rs b/tests/src/tests/cryptarchia/orphan.rs index 55f4a5f69..b3486fb62 100644 --- a/tests/src/tests/cryptarchia/orphan.rs +++ b/tests/src/tests/cryptarchia/orphan.rs @@ -5,7 +5,10 @@ use tests::{ adjust_timeout, common::sync::wait_for_validators_mode_and_height, nodes::validator::{create_validator_config, Validator}, - topology::configs::create_general_configs, + topology::configs::{ + create_general_configs_with_network, + network::{Libp2pNetworkLayout, NetworkParams}, + }, }; #[tokio::test] @@ -13,7 +16,10 @@ async fn test_orphan_handling() { let n_validators = 3; let min_height = 5; - let general_configs = create_general_configs(n_validators); + let network_params = NetworkParams { + libp2p_network_layout: Libp2pNetworkLayout::Full, + }; + let general_configs = create_general_configs_with_network(n_validators, &network_params); let mut validators = vec![]; for config in general_configs.iter().take(2) { diff --git a/tests/src/topology/configs/mod.rs b/tests/src/topology/configs/mod.rs index ff0214ef9..135995828 100644 --- a/tests/src/topology/configs/mod.rs +++ b/tests/src/topology/configs/mod.rs @@ -43,6 +43,14 @@ pub struct GeneralConfig { #[must_use] pub fn create_general_configs(n_nodes: usize) -> Vec { + create_general_configs_with_network(n_nodes, &NetworkParams::default()) +} + +#[must_use] +pub fn create_general_configs_with_network( + n_nodes: usize, + network_params: &NetworkParams, +) -> Vec { let mut ids = vec![[0; 32]; n_nodes]; let mut ports = vec![]; @@ -54,7 +62,7 @@ pub fn create_general_configs(n_nodes: usize) -> Vec { let consensus_params = ConsensusParams::default_for_participants(n_nodes); let consensus_configs = consensus::create_consensus_configs(&ids, &consensus_params); let bootstrap_config = bootstrap::create_bootstrap_configs(&ids, Duration::from_secs(20)); - let network_configs = network::create_network_configs(&ids, &NetworkParams::default()); + let network_configs = network::create_network_configs(&ids, network_params); let da_configs = da::create_da_configs(&ids, &DaParams::default(), &ports); let api_configs = api::create_api_configs(&ids); let blend_configs = blend::create_blend_configs(&ids); diff --git a/tests/src/topology/configs/network.rs b/tests/src/topology/configs/network.rs index 799a3662a..ea36bd3ac 100644 --- a/tests/src/topology/configs/network.rs +++ b/tests/src/topology/configs/network.rs @@ -9,6 +9,7 @@ pub enum Libp2pNetworkLayout { #[default] Star, Chain, + Full, } #[derive(Default)] @@ -45,7 +46,7 @@ pub fn create_network_configs( }) .collect(); - let all_initial_peers = initial_peers_by_network_layout(swarm_configs.clone(), network_params); + let all_initial_peers = initial_peers_by_network_layout(&swarm_configs, network_params); swarm_configs .iter() @@ -58,29 +59,40 @@ pub fn create_network_configs( } fn initial_peers_by_network_layout( - mut swarm_configs: Vec, + swarm_configs: &[SwarmConfig], network_params: &NetworkParams, ) -> Vec> { let mut all_initial_peers = vec![]; - let first_swarm = swarm_configs.remove(0); - let first_addr = node_address_from_port(first_swarm.port); match network_params.libp2p_network_layout { Libp2pNetworkLayout::Star => { - let other_initial_peers = vec![first_addr]; - all_initial_peers.push(vec![]); // First node has no initial peers. + // First node is the hub - has no initial peers + all_initial_peers.push(vec![]); + let first_addr = node_address_from_port(swarm_configs[0].port); - for _ in swarm_configs { - all_initial_peers.push(other_initial_peers.clone()); + // All other nodes connect to the first node + for _ in 1..swarm_configs.len() { + all_initial_peers.push(vec![first_addr.clone()]); } } Libp2pNetworkLayout::Chain => { - let mut prev_addr = first_addr; - all_initial_peers.push(vec![]); // First node has no initial peers. + // First node has no initial peers + all_initial_peers.push(vec![]); - for swarm in swarm_configs { + // Each subsequent node connects to the previous one + for i in 1..swarm_configs.len() { + let prev_addr = node_address_from_port(swarm_configs[i - 1].port); all_initial_peers.push(vec![prev_addr]); - prev_addr = node_address_from_port(swarm.port); + } + } + Libp2pNetworkLayout::Full => { + // Each node connects to all previous nodes, unidirectional connections + for i in 0..swarm_configs.len() { + let mut peers = vec![]; + for swarm_config in swarm_configs.iter().take(i) { + peers.push(node_address_from_port(swarm_config.port)); + } + all_initial_peers.push(peers); } } }