fix: allow parallel connection attempts (max 3 concurrent)

Previously the connection maintenance loop only allowed one connection
attempt at a time. If a connection took 60 seconds to complete, all other
attempts were blocked. This caused star topology formation in small
networks since peers couldn't form connections fast enough.

Now we allow up to MAX_CONCURRENT_CONNECTIONS (3) parallel connection
attempts by using live_tx_tracker.active_transaction_count() instead
of tracking a single live_tx.

Fixes the serialized connection acquisition from #2173.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: sanity

crates/core/src/ring/live_tx.rs

@@ -47,11 +47,15 @@ impl LiveTransactionTracker {
         self.tx_per_peer.contains_key(&peer_addr)
     }
 
-    pub(crate) fn still_alive(&self, tx: &Transaction) -> bool {
-        self.tx_per_peer.iter().any(|e| e.value().contains(tx))
-    }
-
     pub(crate) fn len(&self) -> usize {
         self.tx_per_peer.len()
     }
+
+    /// Returns the total number of active transactions across all peers.
+    pub(crate) fn active_transaction_count(&self) -> usize {
+        self.tx_per_peer
+            .iter()
+            .map(|entry| entry.value().len())
+            .sum()
+    }
 }

crates/core/src/ring/mod.rs

@@ -383,6 +383,11 @@ impl Ring {
 
         const REGENERATE_DENSITY_MAP_INTERVAL: Duration = Duration::from_secs(60);
 
+        /// Maximum number of concurrent connection acquisition attempts.
+        /// Allows parallel connection attempts to speed up network formation
+        /// instead of serial blocking on a single connection at a time.
+        const MAX_CONCURRENT_CONNECTIONS: usize = 3;
+
         let mut check_interval = tokio::time::interval(CHECK_TICK_DURATION);
         check_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
         let mut refresh_density_map = tokio::time::interval(REGENERATE_DENSITY_MAP_INTERVAL);
@@ -393,7 +398,6 @@ impl Ring {
         tokio::time::sleep(Duration::from_secs(2)).await;
         tracing::info!("Connection maintenance task: initial sleep completed");
 
-        let mut live_tx = None;
         let mut pending_conn_adds = BTreeSet::new();
         let mut this_peer = None;
         loop {
@@ -416,20 +420,17 @@ impl Ring {
             let mut skip_list = HashSet::new();
             skip_list.insert(this_peer);
 
-            // if there are no open connections, we need to acquire more
-            if let Some(tx) = &live_tx {
-                if !live_tx_tracker.still_alive(tx) {
-                    let _ = live_tx.take();
-                }
-            }
-
+            // Acquire new connections up to MAX_CONCURRENT_CONNECTIONS limit
+            let active_count = live_tx_tracker.active_transaction_count();
             if let Some(ideal_location) = pending_conn_adds.pop_first() {
-                if live_tx.is_none() {
+                if active_count < MAX_CONCURRENT_CONNECTIONS {
                     tracing::info!(
+                        active_connections = active_count,
+                        max_concurrent = MAX_CONCURRENT_CONNECTIONS,
                         "Attempting to acquire new connection for location: {:?}",
                         ideal_location
                     );
-                    live_tx = self
+                    let tx = self
                         .acquire_new(
                             ideal_location,
                             &skip_list,
@@ -445,18 +446,23 @@ impl Ring {
                             );
                             error
                         })?;
-                    if live_tx.is_none() {
+                    if tx.is_none() {
                         let conns = self.connection_manager.connection_count();
                         tracing::warn!(
                             "acquire_new returned None - likely no peers to query through (connections: {})",
                             conns
                         );
                     } else {
-                        tracing::info!("Successfully initiated connection acquisition");
+                        tracing::info!(
+                            active_connections = active_count + 1,
+                            "Successfully initiated connection acquisition"
+                        );
                     }
                 } else {
                     tracing::debug!(
-                        "Skipping connection attempt - live transaction still active, re-queuing location {}",
+                        active_connections = active_count,
+                        max_concurrent = MAX_CONCURRENT_CONNECTIONS,
+                        "At max concurrent connections, re-queuing location {}",
                         ideal_location
                     );
                     pending_conn_adds.insert(ideal_location);