p2p: improve pending_inbound_connections_limit test

Author: ImplOfAnImpl

p2p/src/tests/helpers/test_node.rs

@@ -377,6 +377,8 @@ where
     }
 
     pub async fn wait_for_next_pending_peer_creation_in_backend(&mut self) -> PeerId {
+        log::debug!("Waiting for next pending peer creation in backend");
+
         loop {
             if let BackendNotification::PendingPeerCreated { peer_id } =
                 self.backend_notification_receiver.recv().await.unwrap()
@@ -390,6 +392,8 @@ where
         &mut self,
         expected_peer_id: PeerId,
     ) {
+        log::debug!("Waiting for next pending peer removal in backend");
+
         loop {
             if let BackendNotification::PendingPeerRemoved { peer_id } =
                 self.backend_notification_receiver.recv().await.unwrap()

p2p/src/tests/pending_inbound_connections_limit.rs

@@ -13,17 +13,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use rstest::rstest;
 
 use chainstate::ChainstateConfig;
 use common::primitives::user_agent::mintlayer_core_user_agent;
+use logging::log;
 use networking::{
     test_helpers::{
         TestTransportChannel, TestTransportMaker, TestTransportNoise, TestTransportTcp,
     },
-    transport::{TransportSocket, new_message_stream},
+    transport::{TransportListener as _, TransportSocket, new_message_stream},
 };
 use p2p_test_utils::{SHORT_TIMEOUT, run_with_timeout};
 use randomness::RngExt as _;
@@ -49,17 +50,18 @@ where
     let mut rng = make_seedable_rng(seed);
 
     let max_pending_inbound_connections = 5;
-    let good_peers_count = rng.random_range(1..=3);
+    let good_inbound_peers_count = rng.random_range(1..=3);
 
     let time_getter = BasicTestTimeGetter::new();
     let chain_config = Arc::new(common::chain::config::create_unit_test_config());
 
     let p2p_config = Arc::new(P2pConfig {
         backend_config: BackendConfig {
             max_pending_inbound_connections: max_pending_inbound_connections.into(),
+            // Make sure pending connections don't time out during the test.
+            peer_handshake_timeout: Duration::from_secs(3600).into(),
 
             outbound_connection_timeout: Default::default(),
-            peer_handshake_timeout: Default::default(),
             disconnection_timeout: Default::default(),
             socket_write_timeout: Default::default(),
         },
@@ -101,10 +103,11 @@ where
 
     let peer_time = P2pTimestamp::from_time(time_getter.get_time_getter().get_time());
 
-    let mut pending_peers_readers_writers = Vec::new();
+    let mut pending_inbound_peers_readers_writers = Vec::new();
 
     // A `max_pending_inbound_connections` number of inbound peers connect, but don't initiate
     // the handshake. Each peer is registered as pending.
+    log::debug!("Adding the first batch of inbound peers");
     for _ in 0..max_pending_inbound_connections {
         let stream = transport.connect(test_node.local_address().socket_addr()).await.unwrap();
         let (msg_reader, msg_writer) = new_message_stream::<_, Message>(
@@ -113,13 +116,14 @@ where
         );
         let peer_id = test_node.wait_for_next_pending_peer_creation_in_backend().await;
 
-        pending_peers_readers_writers.push((peer_id, msg_reader, msg_writer));
+        pending_inbound_peers_readers_writers.push((peer_id, msg_reader, msg_writer));
     }
 
     test_node.assert_no_pending_peer_removal_in_backend().await;
 
     // A few more inbound peers attempt to connect. These are not registered as pending and are
     // dropped immediately because the limit for pending peers has been reached.
+    log::debug!("Adding the second batch of inbound peers");
     for _ in 0..3 {
         let stream = transport.connect(test_node.local_address().socket_addr()).await.unwrap();
         let (mut msg_reader, _) = new_message_stream::<_, Message>(
@@ -134,14 +138,35 @@ where
             .unwrap_err();
     }
 
-    let mut good_peers_readers_writers = Vec::new();
+    let mut outbound_peers_data = Vec::new();
+
+    // Sanity check: outbound connections still create pending peers. This will also ensure
+    // that the presence of pending outbound peers has no effect on the pending inbound peer
+    // limiting.
+    log::debug!("Adding outbound peers");
+    for _ in 0..rng.random_range(1..=3) {
+        let mut listener = transport.bind(vec![TTM::make_address()]).await.unwrap();
+        let connect_result_receiver =
+            test_node.start_connecting(listener.local_addresses().unwrap()[0].into());
+        let (stream, _) = listener.accept().await.unwrap();
+        let _peer_id = test_node.wait_for_next_pending_peer_creation_in_backend().await;
+
+        // Avoid dropping the stream and the result receiver to ensure that the peer stays pending.
+        outbound_peers_data.push((stream, connect_result_receiver));
+    }
+
+    test_node.assert_no_pending_peer_removal_in_backend().await;
+
+    let mut good_inbound_peers_readers_writers = Vec::new();
 
-    // A few of the pending peers decide to initiate the handshake, after which they're no longer pending.
-    for _ in 0..good_peers_count {
-        let peer_idx = rng.random_range(0..pending_peers_readers_writers.len());
+    // A few of the pending inbound peers decide to initiate the handshake, after which they're
+    // no longer pending.
+    log::debug!("Good inbound peers initiate handshake");
+    for _ in 0..good_inbound_peers_count {
+        let peer_idx = rng.random_range(0..pending_inbound_peers_readers_writers.len());
 
         let (peer_id, mut msg_reader, mut msg_writer) =
-            pending_peers_readers_writers.remove(peer_idx);
+            pending_inbound_peers_readers_writers.remove(peer_idx);
 
         msg_writer
             .send(Message::Handshake(HandshakeMessage::Hello {
@@ -163,24 +188,26 @@ where
         test_node.wait_for_next_pending_peer_removal_in_backend(peer_id).await;
 
         // Don't drop the readers/writers to make sure the peer continues to exist.
-        good_peers_readers_writers.push((peer_id, msg_reader, msg_writer));
+        good_inbound_peers_readers_writers.push((peer_id, msg_reader, msg_writer));
     }
 
-    // Since `good_peers_count` peers are no longer pending, the node will accept this number
+    // Since `good_inbound_peers_count` peers are no longer pending, the node will accept this number
     // of new peers and they will be registered as pending.
-    for _ in 0..good_peers_count {
+    log::debug!("Adding the third batch of inbound peers");
+    for _ in 0..good_inbound_peers_count {
         let stream = transport.connect(test_node.local_address().socket_addr()).await.unwrap();
         let (msg_reader, msg_writer) = new_message_stream::<_, Message>(
             stream,
             Some(*p2p_config.protocol_config.max_message_size),
         );
         let peer_id = test_node.wait_for_next_pending_peer_creation_in_backend().await;
 
-        pending_peers_readers_writers.push((peer_id, msg_reader, msg_writer));
+        pending_inbound_peers_readers_writers.push((peer_id, msg_reader, msg_writer));
     }
 
     // A few more inbound peers attempt to connect. These are not registered as pending and are
     // dropped immediately because the limit for pending peers has been reached again.
+    log::debug!("Adding the fourth batch of inbound peers");
     for _ in 0..3 {
         let stream = transport.connect(test_node.local_address().socket_addr()).await.unwrap();
         let (mut msg_reader, _) = new_message_stream::<_, Message>(