fixes

Author: ChrisPaprocki

modules/peer_network_interface/config.default.toml

@@ -38,3 +38,7 @@ peer-sharing-enabled = true
 churn-interval-secs = 600
 # Timeout in seconds for a full peer-sharing exchange
 peer-sharing-timeout-secs = 10
+# Timeout in seconds for TCP connect to a peer
+connect-timeout-secs = 15
+# Accept IPv6 peer addresses from peer-sharing (set true on dual-stack hosts)
+ipv6-enabled = false

modules/peer_network_interface/src/block_flow_consensus_scenarios_tests.rs

@@ -113,6 +113,8 @@ async fn make_harness() -> TestHarness {
         peer_sharing_enabled: false,
         churn_interval_secs: 600,
         peer_sharing_timeout_secs: 10,
+        connect_timeout_secs: 15,
+        ipv6_enabled: false,
     };
 
     let flow = BlockFlowHandler::new(

modules/peer_network_interface/src/configuration.rs

@@ -38,6 +38,10 @@ pub struct InterfaceConfig {
     pub churn_interval_secs: u64,
     #[serde(default = "default_peer_sharing_timeout_secs")]
     pub peer_sharing_timeout_secs: u64,
+    #[serde(default = "default_connect_timeout_secs")]
+    pub connect_timeout_secs: u64,
+    #[serde(default = "default_ipv6_enabled")]
+    pub ipv6_enabled: bool,
 }
 
 fn default_consensus_topic() -> String {
@@ -68,6 +72,14 @@ fn default_peer_sharing_timeout_secs() -> u64 {
     10
 }
 
+fn default_connect_timeout_secs() -> u64 {
+    15
+}
+
+fn default_ipv6_enabled() -> bool {
+    false
+}
+
 impl InterfaceConfig {
     pub fn try_load(config: &Config) -> Result<Self> {
         let full_config = Config::builder()

modules/peer_network_interface/src/connection.rs

@@ -25,11 +25,18 @@ pub struct PeerConnection {
 }
 
 impl PeerConnection {
-    pub fn new(address: String, magic: u32, sender: PeerMessageSender, delay: Duration) -> Self {
+    pub fn new(
+        address: String,
+        magic: u32,
+        sender: PeerMessageSender,
+        delay: Duration,
+        connect_timeout: Duration,
+    ) -> Self {
         let worker = PeerConnectionWorker {
             address: address.clone(),
             magic,
             sender,
+            connect_timeout,
         };
         let (chainsync_tx, chainsync_rx) = mpsc::unbounded_channel();
         let (blockfetch_tx, blockfetch_rx) = mpsc::unbounded_channel();
@@ -100,6 +107,7 @@ struct PeerConnectionWorker {
     address: String,
     magic: u32,
     sender: PeerMessageSender,
+    connect_timeout: Duration,
 }
 
 impl PeerConnectionWorker {
@@ -119,7 +127,13 @@ impl PeerConnectionWorker {
         chainsync: mpsc::UnboundedReceiver<ChainsyncCommand>,
         blockfetch: mpsc::UnboundedReceiver<BlockfetchCommand>,
     ) -> Result<()> {
-        let client = PeerClient::connect(self.address.clone(), self.magic.into()).await?;
+        let timeout_dur = self.connect_timeout;
+        let client = tokio::time::timeout(
+            timeout_dur,
+            PeerClient::connect(self.address.clone(), self.magic.into()),
+        )
+        .await
+        .map_err(|_| anyhow::anyhow!("connect timeout after {}s", timeout_dur.as_secs()))??;
         select! {
             res = self.run_chainsync(client.chainsync, chainsync) => res,
             res = self.run_blockfetch(client.blockfetch, blockfetch) => res,

modules/peer_network_interface/src/network.rs

@@ -94,6 +94,8 @@ pub struct NetworkManager {
     /// Addresses from the static `node_addresses` config. Configured peers are always
     /// retried on disconnect and are never blacklisted via `mark_failed`.
     configured_addrs: HashSet<String>,
+    connect_timeout: Duration,
+    ipv6_enabled: bool,
 }
 
 impl NetworkManager {
@@ -110,6 +112,8 @@ impl NetworkManager {
         peer_sharing_enabled: bool,
         churn_interval_secs: u64,
         peer_sharing_timeout_secs: u64,
+        connect_timeout_secs: u64,
+        ipv6_enabled: bool,
     ) -> Self {
         let peer_manager = if peer_sharing_enabled {
             Some(PeerManager::new(PeerManagerConfig {
@@ -140,6 +144,8 @@ impl NetworkManager {
             min_hot_peers,
             cold_origin: HashSet::new(),
             configured_addrs,
+            connect_timeout: Duration::from_secs(connect_timeout_secs),
+            ipv6_enabled,
         };
 
         if peer_sharing_enabled {
@@ -217,16 +223,17 @@ impl NetworkManager {
                 let hot: HashSet<String> =
                     self.peers.values().map(|p| p.conn.address.clone()).collect();
                 if let Some(ref mut pm) = self.peer_manager {
-                    let count = addresses.len();
+                    let received = addresses.len();
                     let queried_peer = self
                         .peers
                         .get(&from_peer)
                         .map(|p| p.conn.address.as_str())
                         .unwrap_or("unknown");
-                    pm.add_discovered(addresses, &hot);
+                    let added = pm.add_discovered(addresses, &hot);
                     info!(
                         queried_peer,
-                        discovered = count,
+                        received,
+                        added,
                         cold_count = pm.cold_count(),
                         "peer-sharing discovery batch complete"
                     );
@@ -329,6 +336,7 @@ impl NetworkManager {
         let amount = pm.config().target_peer_count.min(255) as u8;
         let timeout = Duration::from_secs(pm.config().peer_sharing_timeout_secs);
         let sender = self.events_sender.clone();
+        let ipv6 = self.ipv6_enabled;
 
         info!(
             peer = %address,
@@ -339,7 +347,7 @@ impl NetworkManager {
         );
 
         tokio::spawn(async move {
-            match request_peers(&address, magic, amount, timeout).await {
+            match request_peers(&address, magic, amount, timeout, ipv6).await {
                 Ok(addrs) => {
                     info!(
                         peer = %address,
@@ -448,7 +456,13 @@ impl NetworkManager {
             sink: self.events_sender.clone(),
             id,
         };
-        let conn = PeerConnection::new(address, self.network_magic, sender, delay);
+        let conn = PeerConnection::new(
+            address,
+            self.network_magic,
+            sender,
+            delay,
+            self.connect_timeout,
+        );
         let peer = PeerData::new(conn);
         let points = self.flow_handler.handle_new_connection(id, self.sync_point.as_ref());
         peer.find_intersect(points);
@@ -545,7 +559,10 @@ impl NetworkManager {
             // Ghost disconnect: peer was already removed (e.g. churn dropped the PeerData,
             // which caused the worker to exit and emit a Disconnected event).  Safe to
             // ignore — flow_handler was already called in on_churn.
-            debug!(peer_id = id.0, "ignoring ghost disconnect for already-removed peer");
+            debug!(
+                peer_id = id.0,
+                "ignoring ghost disconnect for already-removed peer"
+            );
             return;
         };
         warn!(address = %peer.conn.address, "disconnected from peer");
@@ -601,7 +618,8 @@ impl NetworkManager {
         let needs_promotion = self.peers.len() < self.min_hot_peers;
         let promoted = needs_promotion && self.try_promote_cold_peer();
         if promoted {
-            let hot: HashSet<String> = self.peers.values().map(|p| p.conn.address.clone()).collect();
+            let hot: HashSet<String> =
+                self.peers.values().map(|p| p.conn.address.clone()).collect();
             if let Some(ref mut pm) = self.peer_manager {
                 pm.demote_to_cold(address, &hot);
             }
@@ -746,6 +764,8 @@ mod tests {
             peer_sharing_enabled: false,
             churn_interval_secs: 600,
             peer_sharing_timeout_secs: 10,
+            connect_timeout_secs: 15,
+            ipv6_enabled: false,
         };
 
         let flow_handler = BlockFlowHandler::new(
@@ -769,6 +789,8 @@ mod tests {
             false,
             600,
             10,
+            15,
+            false,
         )
     }
 
@@ -783,6 +805,7 @@ mod tests {
             0,
             sender,
             Duration::from_secs(3600),
+            Duration::from_secs(15),
         );
         manager.peers.insert(peer, PeerData::new(conn));
     }
@@ -823,6 +846,8 @@ mod tests {
             peer_sharing_enabled: true,
             churn_interval_secs: 600,
             peer_sharing_timeout_secs: 10,
+            connect_timeout_secs: 15,
+            ipv6_enabled: false,
         };
 
         let flow_handler = BlockFlowHandler::new(
@@ -846,6 +871,8 @@ mod tests {
             cfg.peer_sharing_enabled,
             cfg.churn_interval_secs,
             cfg.peer_sharing_timeout_secs,
+            cfg.connect_timeout_secs,
+            cfg.ipv6_enabled,
         );
 
         // Seed a cold peer manually
@@ -874,15 +901,25 @@ mod tests {
             pm.contains_cold("hot.peer.example.com:3001"),
             "disconnected hot peer should be returned to cold to stay in rotation"
         );
-        assert_eq!(pm.cold_count(), 1, "net cold count: promoted one, returned one");
+        assert_eq!(
+            pm.cold_count(),
+            1,
+            "net cold count: promoted one, returned one"
+        );
     }
 
     fn add_test_peer_with_address(manager: &mut NetworkManager, peer: PeerId, address: &str) {
         let sender = PeerMessageSender {
             sink: manager.events_sender.clone(),
             id: peer,
         };
-        let conn = PeerConnection::new(address.to_string(), 0, sender, Duration::from_secs(3600));
+        let conn = PeerConnection::new(
+            address.to_string(),
+            0,
+            sender,
+            Duration::from_secs(3600),
+            Duration::from_secs(15),
+        );
         manager.peers.insert(peer, PeerData::new(conn));
     }
 
@@ -908,6 +945,8 @@ mod tests {
             peer_sharing_enabled: false, // disabled
             churn_interval_secs: 600,
             peer_sharing_timeout_secs: 10,
+            connect_timeout_secs: 15,
+            ipv6_enabled: false,
         };
 
         let flow_handler = BlockFlowHandler::new(
@@ -931,6 +970,8 @@ mod tests {
             cfg.peer_sharing_enabled,
             cfg.churn_interval_secs,
             cfg.peer_sharing_timeout_secs,
+            cfg.connect_timeout_secs,
+            cfg.ipv6_enabled,
         );
 
         assert!(
@@ -966,6 +1007,8 @@ mod tests {
             peer_sharing_enabled: true,
             churn_interval_secs: 600,
             peer_sharing_timeout_secs: 10,
+            connect_timeout_secs: 15,
+            ipv6_enabled: false,
         };
 
         let flow_handler = BlockFlowHandler::new(
@@ -989,6 +1032,8 @@ mod tests {
             cfg.peer_sharing_enabled,
             cfg.churn_interval_secs,
             cfg.peer_sharing_timeout_secs,
+            cfg.connect_timeout_secs,
+            cfg.ipv6_enabled,
         );
 
         let addresses = vec![
@@ -1033,6 +1078,8 @@ mod tests {
             peer_sharing_enabled: true,
             churn_interval_secs: 600,
             peer_sharing_timeout_secs: 10,
+            connect_timeout_secs: 15,
+            ipv6_enabled: false,
         };
 
         let flow_handler = BlockFlowHandler::new(
@@ -1056,6 +1103,8 @@ mod tests {
             cfg.peer_sharing_enabled,
             cfg.churn_interval_secs,
             cfg.peer_sharing_timeout_secs,
+            cfg.connect_timeout_secs,
+            cfg.ipv6_enabled,
         );
 
         // Add 4 hot peers
@@ -1088,6 +1137,8 @@ mod tests {
             peer_sharing_enabled: true,
             churn_interval_secs: 600,
             peer_sharing_timeout_secs: 10,
+            connect_timeout_secs: 15,
+            ipv6_enabled: false,
         };
 
         let flow_handler = BlockFlowHandler::new(
@@ -1111,6 +1162,8 @@ mod tests {
             cfg.peer_sharing_enabled,
             cfg.churn_interval_secs,
             cfg.peer_sharing_timeout_secs,
+            cfg.connect_timeout_secs,
+            cfg.ipv6_enabled,
         );
 
         // Add exactly min_hot_peers = 3 peers

modules/peer_network_interface/src/peer_manager.rs

@@ -105,17 +105,21 @@ impl PeerManager {
     /// # TODO(ledger-peers): `seed_from_ledger(addresses, hot)` follows the same pattern —
     /// add a sibling method that accepts relay addresses from `SPOStateMessage` and applies
     /// the same deduplication and cap enforcement.
-    pub fn add_discovered(&mut self, addresses: Vec<String>, hot: &HashSet<String>) {
+    pub fn add_discovered(&mut self, addresses: Vec<String>, hot: &HashSet<String>) -> usize {
         let cold_cap = self.config.target_peer_count * 4;
+        let mut added = 0usize;
         for addr in addresses {
             if self.is_known(&addr, hot) {
                 continue;
             }
             if self.cold_peers.len() >= cold_cap {
                 self.evict_random_cold();
             }
-            self.cold_peers.insert(addr);
+            if self.cold_peers.insert(addr) {
+                added += 1;
+            }
         }
+        added
     }
 
     /// Remove and return a randomly selected cold peer address for promotion.

modules/peer_network_interface/src/peer_network_interface.rs

@@ -155,6 +155,8 @@ impl PeerNetworkInterface {
                 cfg.peer_sharing_enabled,
                 cfg.churn_interval_secs,
                 cfg.peer_sharing_timeout_secs,
+                cfg.connect_timeout_secs,
+                cfg.ipv6_enabled,
             );
 
             match sync_point {