Address review comment

Author: bomanaps

libp2p/abc.py

@@ -1581,6 +1581,7 @@ async def close_peer(self, peer_id: ID) -> None:
 
         """
 
+    @abstractmethod
     def get_peer_health_summary(self, peer_id: ID) -> dict[str, Any]:
         """
         Get health summary for a specific peer.
@@ -1599,6 +1600,7 @@ def get_peer_health_summary(self, peer_id: ID) -> dict[str, Any]:
         """
         return {}
 
+    @abstractmethod
     def get_global_health_summary(self) -> dict[str, Any]:
         """
         Get global health summary across all peers.
@@ -1612,6 +1614,7 @@ def get_global_health_summary(self) -> dict[str, Any]:
         """
         return {}
 
+    @abstractmethod
     def export_health_metrics(self, format: str = "json") -> str:
         """
         Export health metrics in specified format.
@@ -1630,6 +1633,7 @@ def export_health_metrics(self, format: str = "json") -> str:
         """
         return "{}" if format == "json" else ""
 
+    @abstractmethod
     async def get_health_monitor_status(self) -> dict[str, Any]:
         """
         Get status information about the health monitoring service.
@@ -1967,6 +1971,7 @@ def get_connection_health(self, peer_id: ID) -> dict[str, Any]:
             Returns empty dict if health monitoring is disabled or peer not found.
 
         """
+        ...
 
     @abstractmethod
     def get_network_health_summary(self) -> dict[str, Any]:
@@ -1980,6 +1985,7 @@ def get_network_health_summary(self) -> dict[str, Any]:
             Returns empty dict if health monitoring is disabled.
 
         """
+        ...
 
     @abstractmethod
     def export_health_metrics(self, format: str = "json") -> str:
@@ -1998,6 +2004,7 @@ def export_health_metrics(self, format: str = "json") -> str:
             Returns empty string or object if health monitoring is disabled.
 
         """
+        ...
 
     @abstractmethod
     async def get_health_monitor_status(self) -> dict[str, Any]:
@@ -2018,6 +2025,7 @@ async def get_health_monitor_status(self) -> dict[str, Any]:
             Returns {"enabled": False} if health monitoring is disabled.
 
         """
+        ...
 
 
 # -------------------------- peer-record interface.py --------------------------

libp2p/network/health/monitor.py

@@ -277,40 +277,25 @@ def _should_replace_connection(self, peer_id: ID, conn: INetConn) -> bool:
     async def _replace_unhealthy_connection(
         self, peer_id: ID, old_conn: INetConn
     ) -> None:
-        """Replace an unhealthy connection with a new one."""
+        """
+        Replace an unhealthy connection with a new one.
+
+        This method establishes a new connection BEFORE closing the old one
+        to avoid dropping below minimum connection threshold.
+        """
         try:
             logger.info(f"Replacing unhealthy connection for peer {peer_id}")
 
-            # Check if we have enough connections remaining
+            # Check current connection count
             current_connections = self.swarm.connections.get(peer_id, [])
-            remaining_after_removal = len(current_connections) - 1
-
-            # Only remove if we have more than the minimum required
-            if remaining_after_removal < self.config.min_connections_per_peer:
-                logger.warning(
-                    f"Not replacing connection to {peer_id}: would go below minimum "
-                    f"({remaining_after_removal} < "
-                    f"{self.config.min_connections_per_peer})"
-                )
-                return
-
-            # Clean up health tracking first
-            self.swarm.cleanup_connection_health(peer_id, old_conn)
-
-            # Remove from active connections
-            if (
-                peer_id in self.swarm.connections
-                and old_conn in self.swarm.connections[peer_id]
-            ):
-                self.swarm.connections[peer_id].remove(old_conn)
+            current_count = len(current_connections)
 
-            # Close the unhealthy connection
-            try:
-                await old_conn.close()
-            except Exception as e:
-                logger.debug(f"Error closing unhealthy connection: {e}")
+            # Strategy: Try to establish new connection first, then close
+            # old one. This prevents us from being stuck with a bad
+            # connection at minimum threshold
 
-            # Try to establish a new connection to maintain connectivity
+            # First, try to establish a new connection
+            new_conn = None
             try:
                 logger.info(f"Attempting to dial replacement connection to {peer_id}")
                 new_conn = await self.swarm.dial_peer_replacement(peer_id)
@@ -322,12 +307,39 @@ async def _replace_unhealthy_connection(
                     logger.warning(
                         f"Failed to establish replacement connection to {peer_id}"
                     )
-
             except Exception as e:
                 logger.error(
                     f"Error establishing replacement connection to {peer_id}: {e}"
                 )
 
+            # If we successfully established a new connection, or if we have enough
+            # connections to safely remove the bad one, proceed with cleanup
+            if new_conn or current_count > self.config.min_connections_per_peer:
+                # Clean up health tracking
+                self.swarm.cleanup_connection_health(peer_id, old_conn)
+
+                # Remove from active connections
+                if (
+                    peer_id in self.swarm.connections
+                    and old_conn in self.swarm.connections[peer_id]
+                ):
+                    self.swarm.connections[peer_id].remove(old_conn)
+
+                # Close the unhealthy connection
+                try:
+                    await old_conn.close()
+                    logger.info(f"Closed unhealthy connection to {peer_id}")
+                except Exception as e:
+                    logger.debug(f"Error closing unhealthy connection: {e}")
+            else:
+                # We couldn't establish a new connection and we're at minimum
+                # Keep the unhealthy connection rather than having no connection
+                logger.warning(
+                    f"Keeping unhealthy connection to {peer_id}: "
+                    f"failed to establish replacement and at minimum threshold "
+                    f"({current_count} connections)"
+                )
+
         except Exception as e:
             logger.error(f"Error replacing connection to {peer_id}: {e}")
 

libp2p/pubsub/pubsub.py

@@ -53,7 +53,6 @@
     StreamClosed,
     StreamEOF,
     StreamError,
-    StreamReset,
 )
 from libp2p.peer.id import (
     ID,

tests/core/transport/test_websocket.py

@@ -5,12 +5,6 @@
 from typing import Any
 
 import pytest
-
-if hasattr(builtins, "ExceptionGroup"):
-    ExceptionGroup = builtins.ExceptionGroup  # type: ignore[misc]
-else:
-    # Fallback for older Python versions
-    ExceptionGroup = Exception  # type: ignore[misc]
 from multiaddr import Multiaddr
 import trio
 
@@ -31,6 +25,10 @@
 
 logger = logging.getLogger(__name__)
 
+# ExceptionGroup type handling for Python 3.11+ compatibility
+# Use getattr to avoid type checker errors on Python 3.10
+ExceptionGroupType: type[BaseException] = getattr(builtins, "ExceptionGroup", Exception)
+
 PLAINTEXT_PROTOCOL_ID = "/plaintext/1.0.0"