fix: run connection waiting concurrently with event listener (#2218)

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: nacho.d.g <iduartgomez@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 4 people

apps/freenet-ping/Cargo.lock

@@ -1,4 +1,4 @@
-use std::{collections::HashSet, convert::Infallible, sync::Arc, time::Duration};
+use std::{convert::Infallible, sync::Arc, time::Duration};
 
 use futures::{future::BoxFuture, FutureExt};
 use tokio::task::JoinHandle;
@@ -21,12 +21,9 @@ use crate::{
         self, ContractHandler, ContractHandlerChannel, ExecutorToEventLoopChannel,
         NetworkEventListenerHalve, WaitingResolution,
     },
-    message::{NetMessage, NetMessageV1, NodeEvent},
+    message::NodeEvent,
     node::NodeConfig,
-    operations::{
-        connect::{self, ConnectOp},
-        OpEnum,
-    },
+    operations::connect,
 };
 
 use super::OpManager;
@@ -49,10 +46,14 @@ pub(crate) struct NodeP2P {
 }
 
 impl NodeP2P {
-    /// Aggressively establish connections during startup to avoid on-demand delays
-    async fn aggressive_initial_connections(&self) {
-        let min_connections = self.op_manager.ring.connection_manager.min_connections;
-
+    /// Aggressively wait for connections during startup to avoid on-demand delays.
+    /// This is an associated function that can be spawned as a task to run concurrently
+    /// with the event listener. Without the event listener running, connection
+    /// handshakes won't be processed.
+    async fn aggressive_initial_connections_impl(
+        op_manager: &Arc<OpManager>,
+        min_connections: usize,
+    ) {
         tracing::info!(
             "Starting aggressive connection acquisition phase (target: {} connections)",
             min_connections
@@ -63,14 +64,13 @@ impl NodeP2P {
         let start = std::time::Instant::now();
         let max_duration = Duration::from_secs(10);
         let mut last_connection_count = 0;
-        let mut stable_rounds = 0;
 
         while start.elapsed() < max_duration {
             // Cooperative yielding for CI environments with limited CPU cores
-            // Research shows CI (2 cores) needs explicit yields to prevent task starvation
+            // This is critical - the event listener needs CPU time to process handshakes
             tokio::task::yield_now().await;
 
-            let current_connections = self.op_manager.ring.open_connections();
+            let current_connections = op_manager.ring.open_connections();
 
             // If we've reached our target, we're done
             if current_connections >= min_connections {
@@ -82,50 +82,34 @@ impl NodeP2P {
                 break;
             }
 
-            // If connection count is stable for 3 rounds, actively trigger more connections
-            if current_connections == last_connection_count {
-                stable_rounds += 1;
-                if stable_rounds >= 3 && current_connections > 0 {
-                    tracing::info!(
-                        "Connection count stable at {}, triggering active peer discovery",
-                        current_connections
-                    );
-
-                    // Trigger the connection maintenance task to actively look for more peers
-                    // In small networks, we want to be more aggressive
-                    for _ in 0..3 {
-                        // Yield before each connection attempt to prevent blocking other tasks
-                        tokio::task::yield_now().await;
-
-                        if let Err(e) = self.trigger_connection_maintenance().await {
-                            tracing::warn!("Failed to trigger connection maintenance: {}", e);
-                        }
-                        tokio::time::sleep(Duration::from_millis(100)).await;
-                    }
-                    stable_rounds = 0;
-                }
-            } else {
-                stable_rounds = 0;
+            // Log progress when connection count changes
+            if current_connections != last_connection_count {
+                tracing::info!(
+                    "Connection progress: {}/{} (elapsed: {}s)",
+                    current_connections,
+                    min_connections,
+                    start.elapsed().as_secs()
+                );
                 last_connection_count = current_connections;
+            } else {
+                tracing::debug!(
+                    "Current connections: {}/{}, waiting for more peers (elapsed: {}s)",
+                    current_connections,
+                    min_connections,
+                    start.elapsed().as_secs()
+                );
             }
 
-            tracing::debug!(
-                "Current connections: {}/{}, waiting for more peers (elapsed: {}s)",
-                current_connections,
-                min_connections,
-                start.elapsed().as_secs()
-            );
-
-            // Check more frequently at the beginning
+            // Check more frequently at the beginning to detect quick connections
             let sleep_duration = if start.elapsed() < Duration::from_secs(3) {
-                Duration::from_millis(500)
+                Duration::from_millis(250)
             } else {
-                Duration::from_secs(1)
+                Duration::from_millis(500)
             };
             tokio::time::sleep(sleep_duration).await;
         }
 
-        let final_connections = self.op_manager.ring.open_connections();
+        let final_connections = op_manager.ring.open_connections();
         tracing::info!(
             "Aggressive connection phase complete. Final connections: {}/{} (took {}s)",
             final_connections,
@@ -134,56 +118,6 @@ impl NodeP2P {
         );
     }
 
-    /// Trigger the connection maintenance task to actively look for more peers
-    async fn trigger_connection_maintenance(&self) -> anyhow::Result<()> {
-        let ideal_location = Location::random();
-
-        // Find a connected peer to query
-        let query_target = {
-            let router = self.op_manager.ring.router.read();
-            self.op_manager.ring.connection_manager.routing(
-                ideal_location,
-                None,
-                &HashSet::<std::net::SocketAddr>::new(),
-                &router,
-            )
-        };
-
-        if let Some(query_target) = query_target {
-            let joiner = self.op_manager.ring.connection_manager.own_location();
-            let ttl = self
-                .op_manager
-                .ring
-                .max_hops_to_live
-                .max(1)
-                .min(u8::MAX as usize) as u8;
-            let target_connections = self.op_manager.ring.connection_manager.min_connections;
-
-            let (tx, op, msg) = ConnectOp::initiate_join_request(
-                joiner,
-                query_target.clone(),
-                ideal_location,
-                ttl,
-                target_connections,
-                self.op_manager.connect_forward_estimator.clone(),
-            );
-
-            tracing::debug!(
-                %tx,
-                query_peer = %query_target,
-                %ideal_location,
-                "Triggering connection maintenance connect request"
-            );
-            self.op_manager
-                .notify_op_change(
-                    NetMessage::V1(NetMessageV1::Connect(msg)),
-                    OpEnum::Connect(Box::new(op)),
-                )
-                .await?;
-        }
-
-        Ok(())
-    }
     pub(super) async fn run_node(mut self) -> anyhow::Result<Infallible> {
         if self.should_try_connect {
             let join_handle = connect::initial_join_procedure(
@@ -193,11 +127,26 @@ impl NodeP2P {
             .await?;
             self.initial_join_task = Some(join_handle);
 
-            // After connecting to gateways, aggressively try to reach min_connections
-            // This is important for fast startup and avoiding on-demand connection delays
-            self.aggressive_initial_connections().await;
+            // Note: We don't run aggressive_initial_connections here because
+            // the event listener hasn't started yet. The connect requests from
+            // initial_join_procedure are queued but won't be processed until
+            // the event listener runs. Instead, we'll run the aggressive
+            // connection phase concurrently with the event listener below.
         }
 
+        // Spawn aggressive connection task to run concurrently with event listener.
+        // This is needed because connection handshakes are processed by the event
+        // listener, so we can't block waiting for connections before it starts.
+        let aggressive_conn_task = if self.should_try_connect {
+            let op_manager = self.op_manager.clone();
+            let min_connections = op_manager.ring.connection_manager.min_connections;
+            Some(tokio::spawn(async move {
+                Self::aggressive_initial_connections_impl(&op_manager, min_connections).await;
+            }))
+        } else {
+            None
+        };
+
         let f = self.conn_manager.run_event_listener(
             self.op_manager.clone(),
             self.client_wait_for_transaction,
@@ -226,6 +175,9 @@ impl NodeP2P {
         if let Some(handle) = join_task {
             handle.abort();
         }
+        if let Some(handle) = aggressive_conn_task {
+            handle.abort();
+        }
 
         result
     }