address review comment

Author: bomanaps

libp2p/network/health/monitor.py

@@ -49,7 +49,7 @@ async def run(self) -> None:
         logger.info("Starting ConnectionHealthMonitor service")
 
         # Only run if health monitoring is enabled
-        if not self._is_health_monitoring_enabled():
+        if not self._is_health_monitoring_enabled:
             logger.debug("Health monitoring disabled, skipping monitor service")
             return
 
@@ -128,16 +128,29 @@ async def _check_connection_health(self, peer_id: ID, conn: INetConn) -> None:
         try:
             # Skip checks during connection warmup window
             warmup = getattr(self.config, "health_warmup_window", 0.0)
-            if warmup and hasattr(conn, "established_at"):
-                # Fallback to event_started if established_at not present
-                try:
+            if warmup:
+                # Check if we have health data with established_at timestamp
+                if self._has_health_data(peer_id, conn):
                     import time
 
-                    established_at = getattr(conn, "established_at")
-                    if established_at and (time.time() - established_at) < warmup:
+                    health = self.swarm.health_data[peer_id][conn]
+                    if (
+                        health.established_at
+                        and (time.time() - health.established_at) < warmup
+                    ):
+                        logger.debug(
+                            f"Skipping health check for {peer_id} during warmup window"
+                        )
                         return
-                except Exception:
-                    pass
+                else:
+                    # If no health data yet, this is likely a new connection
+                    # Initialize health tracking and skip the first check
+                    self.swarm.initialize_connection_health(peer_id, conn)
+                    logger.debug(
+                        f"Skipping health check for {peer_id} - "
+                        f"initializing health data"
+                    )
+                    return
 
             # Ensure health tracking is initialized
             if not self._has_health_data(peer_id, conn):
@@ -299,24 +312,15 @@ async def _replace_unhealthy_connection(
 
             # Try to establish a new connection to maintain connectivity
             try:
-                # Get peer info for dialing
-                peer_info = self.swarm.peerstore.peer_info(peer_id)
-                if peer_info and peer_info.addrs:
-                    logger.info(f"Attempting to dial new connection to {peer_id}")
-                    new_conn = await self.swarm.dial_peer(peer_id)
-                    if new_conn:
-                        logger.info(
-                            f"Successfully established replacement connection to "
-                            f"{peer_id}"
-                        )
-                    else:
-                        logger.warning(
-                            f"Failed to establish replacement connection to {peer_id}"
-                        )
+                logger.info(f"Attempting to dial replacement connection to {peer_id}")
+                new_conn = await self.swarm.dial_peer_replacement(peer_id)
+                if new_conn:
+                    logger.info(
+                        f"Successfully established replacement connection to {peer_id}"
+                    )
                 else:
                     logger.warning(
-                        f"No addresses available for {peer_id}, "
-                        f"cannot establish replacement"
+                        f"Failed to establish replacement connection to {peer_id}"
                     )
 
             except Exception as e:
@@ -327,9 +331,10 @@ async def _replace_unhealthy_connection(
         except Exception as e:
             logger.error(f"Error replacing connection to {peer_id}: {e}")
 
+    @property
     def _is_health_monitoring_enabled(self) -> bool:
         """Check if health monitoring is enabled."""
-        return self.swarm._is_health_monitoring_enabled()
+        return self.swarm._is_health_monitoring_enabled
 
     def _has_health_data(self, peer_id: ID, conn: INetConn) -> bool:
         """Check if health data exists for a connection."""
@@ -341,7 +346,7 @@ def _has_health_data(self, peer_id: ID, conn: INetConn) -> bool:
 
     async def get_monitoring_status(self) -> HealthMonitorStatus:
         """Get current monitoring status and statistics."""
-        if not self._is_health_monitoring_enabled():
+        if not self._is_health_monitoring_enabled:
             return HealthMonitorStatus(enabled=False)
 
         total_connections = sum(len(conns) for conns in self.swarm.connections.values())

libp2p/network/swarm.py

@@ -159,7 +159,7 @@ async def run(self) -> None:
                 self.transport.set_swarm(self)
 
             # Start health monitoring service if enabled
-            if self._is_health_monitoring_enabled():
+            if self._is_health_monitoring_enabled:
                 from libp2p.network.health.monitor import ConnectionHealthMonitor
 
                 self._health_monitor = ConnectionHealthMonitor(self)
@@ -414,6 +414,51 @@ async def dial_addr(self, addr: Multiaddr, peer_id: ID) -> INetConn:
         """
         return await self._dial_with_retry(addr, peer_id)
 
+    async def dial_peer_replacement(self, peer_id: ID) -> INetConn | None:
+        """
+        Create a new connection to peer_id for replacement purposes.
+        This bypasses the existing connection check and always creates a new connection.
+
+        :param peer_id: peer ID to dial
+        :raises SwarmException: raised when an error occurs
+        :return: new network connection or None if no addresses available
+        """
+        logger.debug("attempting to dial replacement connection to peer %s", peer_id)
+
+        try:
+            # Get peer info from peer store
+            addrs = self.peerstore.addrs(peer_id)
+        except PeerStoreError:
+            logger.warning(f"No known addresses to peer {peer_id} for replacement")
+            return None
+
+        if not addrs:
+            logger.warning(f"No addresses available for {peer_id} for replacement")
+            return None
+
+        exceptions: list[SwarmException] = []
+
+        # Try all known addresses with retry logic
+        for multiaddr in addrs:
+            try:
+                connection = await self._dial_with_retry(multiaddr, peer_id)
+                logger.info(
+                    f"Successfully established replacement connection to {peer_id}"
+                )
+                return connection
+            except SwarmException as e:
+                exceptions.append(e)
+                logger.debug(
+                    "encountered swarm exception when trying to connect to %s, "
+                    "trying next address...",
+                    multiaddr,
+                    exc_info=e,
+                )
+
+        # All addresses failed
+        logger.warning(f"Failed to establish replacement connection to {peer_id}")
+        return None
+
     async def new_stream(self, peer_id: ID) -> INetStream:
         """
         Enhanced: Create a new stream with load balancing across multiple connections.
@@ -839,6 +884,7 @@ async def notify_all(self, notifier: Callable[[INotifee], Awaitable[None]]) -> N
 
     # Health monitoring methods (conditional on health monitoring being enabled)
 
+    @property
     def _is_health_monitoring_enabled(self) -> bool:
         """Check if health monitoring is enabled."""
         return (
@@ -849,7 +895,7 @@ def _is_health_monitoring_enabled(self) -> bool:
 
     def initialize_connection_health(self, peer_id: ID, connection: INetConn) -> None:
         """Initialize health tracking for a new connection."""
-        if not self._is_health_monitoring_enabled():
+        if not self._is_health_monitoring_enabled:
             return
 
         from libp2p.network.health.data_structures import (
@@ -871,7 +917,7 @@ def initialize_connection_health(self, peer_id: ID, connection: INetConn) -> Non
 
     def cleanup_connection_health(self, peer_id: ID, connection: INetConn) -> None:
         """Clean up health tracking for a closed connection."""
-        if not self._is_health_monitoring_enabled():
+        if not self._is_health_monitoring_enabled:
             return
 
         if peer_id in self.health_data and connection in self.health_data[peer_id]:
@@ -885,7 +931,7 @@ def record_connection_event(
     ) -> None:
         """Record a connection lifecycle event."""
         if (
-            self._is_health_monitoring_enabled()
+            self._is_health_monitoring_enabled
             and peer_id in self.health_data
             and connection in self.health_data[peer_id]
         ):
@@ -896,15 +942,15 @@ def record_connection_error(
     ) -> None:
         """Record a connection error."""
         if (
-            self._is_health_monitoring_enabled()
+            self._is_health_monitoring_enabled
             and peer_id in self.health_data
             and connection in self.health_data[peer_id]
         ):
             self.health_data[peer_id][connection].add_error(error)
 
     def get_peer_health_summary(self, peer_id: ID) -> dict[str, Any]:
         """Get health summary for a specific peer."""
-        if not self._is_health_monitoring_enabled():
+        if not self._is_health_monitoring_enabled:
             return {}
 
         if peer_id not in self.health_data:
@@ -942,7 +988,7 @@ def get_peer_health_summary(self, peer_id: ID) -> dict[str, Any]:
 
     def get_global_health_summary(self) -> dict[str, Any]:
         """Get global health summary across all peers."""
-        if not self._is_health_monitoring_enabled():
+        if not self._is_health_monitoring_enabled:
             return {}
 
         all_peers = list(self.health_data.keys())
@@ -975,7 +1021,7 @@ def get_global_health_summary(self) -> dict[str, Any]:
 
     def export_health_metrics(self, format: str = "json") -> str:
         """Export health metrics in various formats."""
-        if not self._is_health_monitoring_enabled():
+        if not self._is_health_monitoring_enabled:
             return "{}" if format == "json" else ""
 
         summary = self.get_global_health_summary()
@@ -1018,7 +1064,7 @@ def _format_prometheus_metrics(self, summary: dict[str, Any]) -> str:
 
     async def get_health_monitor_status(self) -> dict[str, Any]:
         """Get status information about the health monitoring service."""
-        if not self._is_health_monitoring_enabled() or self._health_monitor is None:
+        if not self._is_health_monitoring_enabled or self._health_monitor is None:
             return {"enabled": False}
 
         status = await self._health_monitor.get_monitoring_status()