Merge pull request #347 from mintlayer/p2p-test-conn-improvement

Add timeout for connection establishment in P2P tests

Author: altonen

p2p/p2p-test-utils/src/lib.rs

@@ -32,9 +32,12 @@ use common::{
 };
 use crypto::random::SliceRandom;
 use libp2p::Multiaddr;
-use std::net::SocketAddr;
-use std::sync::Arc;
-use tokio::net::{TcpListener, TcpStream};
+use p2p::net::{types::ConnectivityEvent, ConnectivityService, NetworkingService};
+use std::{net::SocketAddr, sync::Arc, time::Duration};
+use tokio::{
+    net::{TcpListener, TcpStream},
+    time::timeout,
+};
 
 pub fn make_libp2p_addr() -> Multiaddr {
     "/ip6/::1/tcp/0".parse().unwrap()
@@ -67,6 +70,37 @@ pub async fn get_tcp_socket() -> TcpStream {
     TcpStream::connect(addr).await.unwrap()
 }
 
+pub async fn connect_services<T>(
+    conn1: &mut T::ConnectivityHandle,
+    conn2: &mut T::ConnectivityHandle,
+) where
+    T: NetworkingService + std::fmt::Debug,
+    T::ConnectivityHandle: ConnectivityService<T>,
+{
+    let addr = timeout(Duration::from_secs(5), conn2.local_addr())
+        .await
+        .expect("local address fetch not to timeout")
+        .unwrap()
+        .unwrap();
+    conn1.connect(addr).await.expect("dial to succeed");
+
+    match timeout(Duration::from_secs(5), conn2.poll_next()).await {
+        Ok(event) => match event.unwrap() {
+            ConnectivityEvent::InboundAccepted { .. } => {}
+            event => panic!("expected `InboundAccepted`, got {event:?}"),
+        },
+        Err(_err) => panic!("did not receive `InboundAccepted` in time"),
+    }
+
+    match timeout(Duration::from_secs(5), conn1.poll_next()).await {
+        Ok(event) => match event.unwrap() {
+            ConnectivityEvent::OutboundAccepted { .. } => {}
+            event => panic!("expected `OutboundAccepted`, got {event:?}"),
+        },
+        Err(_err) => panic!("did not receive `OutboundAccepted` in time"),
+    }
+}
+
 pub type ChainstateHandle = subsystem::Handle<Box<dyn ChainstateInterface + 'static>>;
 
 #[derive(Debug, Clone, PartialEq, Eq)]

p2p/src/net/libp2p/behaviour.rs

@@ -61,12 +61,12 @@ use std::{
     poll_method = "poll"
 )]
 pub struct Libp2pBehaviour {
+    pub connmgr: connectivity::ConnectionManager,
+    pub identify: identify::Identify,
+    pub discovery: discovery::DiscoveryManager,
     pub gossipsub: Gossipsub,
     pub ping: ping::Behaviour,
-    pub identify: identify::Identify,
     pub sync: RequestResponse<SyncingCodec>,
-    pub connmgr: connectivity::ConnectionManager,
-    pub discovery: discovery::DiscoveryManager,
 
     #[behaviour(ignore)]
     pub events: VecDeque<Libp2pBehaviourEvent>,

p2p/src/net/libp2p/tests/mod.rs

@@ -38,7 +38,7 @@ use std::{
     sync::Arc,
     time::Duration,
 };
-use tokio::sync::mpsc;
+use tokio::{sync::mpsc, time::timeout};
 
 #[cfg(test)]
 mod frontend;
@@ -247,7 +247,11 @@ pub async fn make_libp2p_with_ping(
 
 async fn get_address<T: NetworkBehaviour>(swarm: &mut Swarm<T>) -> Multiaddr {
     loop {
-        if let SwarmEvent::NewListenAddr { address, .. } = swarm.select_next_some().await {
+        if let SwarmEvent::NewListenAddr { address, .. } =
+            timeout(Duration::from_secs(5), swarm.select_next_some())
+                .await
+                .expect("event to be received")
+        {
             return address;
         }
     }
@@ -262,7 +266,7 @@ where
     let addr = get_address::<A>(swarm1).await;
 
     for _ in 0..3 {
-        swarm2.dial(addr.clone()).expect("swarm dial failed");
+        swarm2.dial(addr.clone()).expect("dial to succeed");
 
         loop {
             tokio::select! {

p2p/src/swarm/tests/ban.rs

@@ -16,7 +16,7 @@
 use crate::{
     error::{P2pError, PeerError},
     net::{self, libp2p::Libp2pService, mock::MockService, ConnectivityService, NetworkingService},
-    swarm::tests::make_peer_manager,
+    swarm::tests::{connect_services, make_peer_manager},
 };
 use common::chain::config;
 use libp2p::Multiaddr;
@@ -26,7 +26,7 @@ use std::sync::Arc;
 // ban peer whose connected to us
 async fn ban_connected_peer<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -35,13 +35,8 @@ where
     let mut swarm1 = make_peer_manager::<T>(addr1, Arc::clone(&config)).await;
     let mut swarm2 = make_peer_manager::<T>(addr2, config).await;
 
-    let addr = swarm2.handle.local_addr().await.unwrap().unwrap();
-    let (_conn1_res, conn2_res) =
-        tokio::join!(swarm1.handle.connect(addr), swarm2.handle.poll_next(),);
-
-    if let Ok(net::types::ConnectivityEvent::InboundAccepted { peer_info, address }) = conn2_res {
-        swarm2.accept_inbound_connection(address, peer_info).await.unwrap();
-    }
+    let (address, peer_info) = connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
+    swarm2.accept_inbound_connection(address, peer_info).await.unwrap();
 
     let peer_id = *swarm1.handle_mut().peer_id();
     assert_eq!(swarm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
@@ -74,13 +69,8 @@ where
     let mut swarm1 = make_peer_manager::<T>(addr1, Arc::clone(&config)).await;
     let mut swarm2 = make_peer_manager::<T>(addr2, config).await;
 
-    let addr = swarm2.handle.local_addr().await.unwrap().unwrap();
-    let (_conn1_res, conn2_res) =
-        tokio::join!(swarm1.handle.connect(addr), swarm2.handle.poll_next(),);
-
-    if let Ok(net::types::ConnectivityEvent::InboundAccepted { peer_info, address }) = conn2_res {
-        swarm2.accept_inbound_connection(address, peer_info).await.unwrap();
-    }
+    let (address, peer_info) = connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
+    swarm2.accept_inbound_connection(address, peer_info).await.unwrap();
 
     let peer_id = *swarm1.handle_mut().peer_id();
     assert_eq!(swarm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
@@ -116,7 +106,7 @@ async fn banned_peer_attempts_to_connect_mock() {
 // attempt to connect to banned peer
 async fn connect_to_banned_peer<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -125,13 +115,8 @@ where
     let mut swarm1 = make_peer_manager::<T>(addr1, Arc::clone(&config)).await;
     let mut swarm2 = make_peer_manager::<T>(addr2, config).await;
 
-    let addr = swarm2.handle.local_addr().await.unwrap().unwrap();
-    let (_conn1_res, conn2_res) =
-        tokio::join!(swarm1.handle.connect(addr), swarm2.handle.poll_next(),);
-
-    if let Ok(net::types::ConnectivityEvent::InboundAccepted { peer_info, address }) = conn2_res {
-        swarm2.accept_inbound_connection(address, peer_info).await.unwrap();
-    }
+    let (address, peer_info) = connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
+    swarm2.accept_inbound_connection(address, peer_info).await.unwrap();
 
     let peer_id = *swarm1.handle_mut().peer_id();
     assert_eq!(swarm2.adjust_peer_score(peer_id, 1000).await, Ok(()));
@@ -391,10 +376,7 @@ where
     )
     .await;
 
-    let addr = swarm2.handle.local_addr().await.unwrap().unwrap();
-    let (_conn1_res, conn2_res) =
-        tokio::join!(swarm1.handle.connect(addr), swarm2.handle.poll_next());
-    let _conn2_res: net::types::ConnectivityEvent<T> = conn2_res.unwrap();
+    connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
     let swarm1_id = *swarm1.handle.peer_id();
 
     // run the first peer manager in the background and poll events from the peer manager

p2p/src/swarm/tests/connections.rs

@@ -21,7 +21,10 @@ use crate::{
         mock::{types::MockPeerId, MockService},
         ConnectivityService, NetworkingService,
     },
-    swarm::{self, tests::make_peer_manager},
+    swarm::{
+        self,
+        tests::{connect_services, make_peer_manager},
+    },
 };
 use common::chain::config;
 use libp2p::{Multiaddr, PeerId};
@@ -31,7 +34,7 @@ use std::{net::SocketAddr, sync::Arc};
 // try to connect to an address that no one listening on and verify it fails
 async fn test_swarm_connect<T: NetworkingService>(bind_addr: T::Address, remote_addr: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -73,7 +76,7 @@ async fn test_swarm_connect_libp2p() {
 // is below the desired threshold and there are idle peers in the peerdb
 async fn test_auto_connect<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -118,7 +121,7 @@ async fn test_auto_connect_mock() {
 
 async fn connect_outbound_same_network<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -127,14 +130,7 @@ where
     let mut swarm1 = make_peer_manager::<T>(addr1, Arc::clone(&config)).await;
     let mut swarm2 = make_peer_manager::<T>(addr2, config).await;
 
-    let addr = swarm2.handle.local_addr().await.unwrap().unwrap();
-    let (_conn1_res, _conn2_res) =
-        tokio::join!(swarm1.handle.connect(addr), swarm2.handle.poll_next());
-
-    assert!(std::matches!(
-        swarm1.handle.poll_next().await,
-        Ok(net::types::ConnectivityEvent::OutboundAccepted { .. })
-    ));
+    connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
 }
 
 #[tokio::test]
@@ -196,7 +192,7 @@ async fn test_validate_supported_protocols() {
 
 async fn connect_outbound_different_network<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -209,17 +205,8 @@ where
     )
     .await;
 
-    let addr = swarm2.handle.local_addr().await.unwrap().unwrap();
-    tokio::spawn(async move { swarm2.handle.poll_next().await.unwrap() });
-    swarm1.handle.connect(addr).await.unwrap();
-
-    if let Ok(net::types::ConnectivityEvent::OutboundAccepted {
-        peer_info,
-        address: _,
-    }) = swarm1.handle.poll_next().await
-    {
-        assert_ne!(peer_info.magic_bytes, *config.magic_bytes());
-    }
+    let (_address, peer_info) = connect_services::<T>(&mut swarm2.handle, &mut swarm1.handle).await;
+    assert_ne!(peer_info.magic_bytes, *config.magic_bytes());
 }
 
 #[tokio::test]
@@ -235,7 +222,7 @@ async fn connect_outbound_different_network_mock() {
 
 async fn connect_inbound_same_network<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -244,19 +231,11 @@ where
     let mut swarm1 = make_peer_manager::<T>(addr1, Arc::clone(&config)).await;
     let mut swarm2 = make_peer_manager::<T>(addr2, config).await;
 
-    let (_conn1_res, conn2_res) = tokio::join!(
-        swarm1.handle.connect(swarm2.handle.local_addr().await.unwrap().unwrap()),
-        swarm2.handle.poll_next()
+    let (address, peer_info) = connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
+    assert_eq!(
+        swarm2.accept_inbound_connection(address, peer_info).await,
+        Ok(())
     );
-    let conn2_res: net::types::ConnectivityEvent<T> = conn2_res.unwrap();
-    if let net::types::ConnectivityEvent::InboundAccepted { peer_info, address } = conn2_res {
-        assert_eq!(
-            swarm2.accept_inbound_connection(address, peer_info).await,
-            Ok(())
-        );
-    } else {
-        panic!("invalid event received");
-    }
 }
 
 #[tokio::test]
@@ -272,7 +251,7 @@ async fn connect_inbound_same_network_mock() {
 
 async fn connect_inbound_different_network<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -284,23 +263,15 @@ where
     )
     .await;
 
-    let (_conn1_res, conn2_res) = tokio::join!(
-        swarm1.handle.connect(swarm2.handle.local_addr().await.unwrap().unwrap()),
-        swarm2.handle.poll_next()
+    let (address, peer_info) = connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
+
+    assert_eq!(
+        swarm2.accept_inbound_connection(address, peer_info).await,
+        Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork(
+            [1, 2, 3, 4],
+            *config::create_mainnet().magic_bytes(),
+        )))
     );
-    let conn2_res: net::types::ConnectivityEvent<T> = conn2_res.unwrap();
-
-    if let net::types::ConnectivityEvent::InboundAccepted { peer_info, address } = conn2_res {
-        assert_eq!(
-            swarm2.accept_inbound_connection(address, peer_info).await,
-            Err(P2pError::ProtocolError(ProtocolError::DifferentNetwork(
-                [1, 2, 3, 4],
-                *config::create_mainnet().magic_bytes(),
-            )))
-        );
-    } else {
-        panic!("invalid event received");
-    }
 }
 
 #[tokio::test]
@@ -316,28 +287,16 @@ async fn connect_inbound_different_network_mock() {
 
 async fn remote_closes_connection<T>(addr1: T::Address, addr2: T::Address)
 where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
 {
     let mut swarm1 = make_peer_manager::<T>(addr1, Arc::new(config::create_mainnet())).await;
     let mut swarm2 = make_peer_manager::<T>(addr2, Arc::new(config::create_mainnet())).await;
 
-    let (_conn1_res, conn2_res) = tokio::join!(
-        swarm1.handle.connect(swarm2.handle.local_addr().await.unwrap().unwrap()),
-        swarm2.handle.poll_next()
-    );
-    let conn2_res: net::types::ConnectivityEvent<T> = conn2_res.unwrap();
-
-    assert!(std::matches!(
-        conn2_res,
-        net::types::ConnectivityEvent::InboundAccepted { .. }
-    ));
-    assert!(std::matches!(
-        swarm1.handle.poll_next().await,
-        Ok(net::types::ConnectivityEvent::OutboundAccepted { .. })
-    ));
+    let (_address, _peer_info) =
+        connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
 
     assert_eq!(
         swarm2.handle.disconnect(*swarm1.handle.peer_id()).await,
@@ -366,7 +325,7 @@ async fn inbound_connection_too_many_peers<T>(
     default_addr: T::Address,
     peers: Vec<net::types::PeerInfo<T>>,
 ) where
-    T: NetworkingService + 'static,
+    T: NetworkingService + 'static + std::fmt::Debug,
     T::ConnectivityHandle: ConnectivityService<T>,
     <T as net::NetworkingService>::Address: std::str::FromStr,
     <<T as net::NetworkingService>::Address as std::str::FromStr>::Err: std::fmt::Debug,
@@ -383,10 +342,8 @@ async fn inbound_connection_too_many_peers<T>(
         swarm::MAX_ACTIVE_CONNECTIONS
     );
 
-    let addr = swarm2.handle.local_addr().await.unwrap().unwrap();
-    let (_conn1_res, conn2_res) =
-        tokio::join!(swarm1.handle.connect(addr), swarm2.handle.poll_next());
-    let _conn2_res: net::types::ConnectivityEvent<T> = conn2_res.unwrap();
+    let (_address, _peer_info) =
+        connect_services::<T>(&mut swarm1.handle, &mut swarm2.handle).await;
     let swarm1_id = *swarm1.handle.peer_id();
 
     // run the first peer manager in the background and poll events from the peer manager