feat(sync): handle not connected peers

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<PeerId>,
     /// 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<GetTipResponse, ChainSyncError>>,
     ) -> 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<HeaderId>,
         reply_sender: oneshot::Sender<BoxedStream<Result<SerialisedBlock, ChainSyncError>>>,
     ) -> 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;

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) {

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) {

tests/src/topology/configs/mod.rs

@@ -43,6 +43,14 @@ pub struct GeneralConfig {
 
 #[must_use]
 pub fn create_general_configs(n_nodes: usize) -> Vec<GeneralConfig> {
+    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<GeneralConfig> {
     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<GeneralConfig> {
     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);

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<SwarmConfig>,
+    swarm_configs: &[SwarmConfig],
     network_params: &NetworkParams,
 ) -> Vec<Vec<Multiaddr>> {
     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);
             }
         }
     }