Merge pull request #1069 from sumanjeet0012/fix/pubsub-protocol-negotiation-crash-910

Add exception handling to prevent pubsub service crashes.

Author: seetadev

libp2p/pubsub/pubsub.py

@@ -481,6 +481,17 @@ async def _handle_new_peer(self, peer_id: ID) -> None:
 
         logger.debug("added new peer %s", peer_id)
 
+    async def _handle_new_peer_safe(self, peer_id: ID) -> None:
+        """
+        Safely handle new peer with exception handling.
+        This wrapper ensures that any exceptions during peer negotiation
+        don't crash the entire pubsub service.
+        """
+        try:
+            await self._handle_new_peer(peer_id)
+        except Exception as error:
+            logger.info(f"Protocol negotiation failed for peer {peer_id}: {error}")
+
     def _handle_dead_peer(self, peer_id: ID) -> None:
         if peer_id not in self.peers:
             return
@@ -503,8 +514,8 @@ async def handle_peer_queue(self) -> None:
         async with self.peer_receive_channel:
             self.event_handle_peer_queue_started.set()
             async for peer_id in self.peer_receive_channel:
-                # Add Peer
-                self.manager.run_task(self._handle_new_peer, peer_id)
+                # Add Peer - wrap in exception handler to prevent service crash
+                self.manager.run_task(self._handle_new_peer_safe, peer_id)
 
     async def handle_dead_peer_queue(self) -> None:
         """

newsfragments/910.bugfix.rst

@@ -0,0 +1 @@
+Fixed pubsub service crashes when protocol negotiation fails by adding proper exception handling.

tests/core/pubsub/test_pubsub.py

@@ -1277,3 +1277,29 @@ def custom_msg_id_constructor(msg):
             expected_custom_id = msg.data + msg.from_id
 
             assert custom_msg_id == expected_custom_id
+
+
+@pytest.mark.trio
+async def test_handle_peer_queue_exception_handling():
+    """Test that _handle_new_peer_safe gracefully handles exceptions."""
+    async with PubsubFactory.create_batch_with_floodsub(1) as pubsubs_fsub:
+        pubsub = pubsubs_fsub[0]
+
+        original_handle_new_peer = pubsub._handle_new_peer
+
+        async def mock_handle_new_peer(peer_id):
+            raise Exception("Protocol negotiation failed")
+
+        pubsub._handle_new_peer = mock_handle_new_peer
+
+        test_peer = IDFactory()
+
+        # Directly call the safe wrapper that's used by handle_peer_queue
+        await pubsub._handle_new_peer_safe(test_peer)
+
+        # The key test: service should still be running despite the exception
+        assert pubsub.manager.is_running, (
+            "Pubsub service should continue running even when peer negotiation fails"
+        )
+
+        pubsub._handle_new_peer = original_handle_new_peer