fix: add per-node timeout in find_leader()

Wraps _query_leader() in asyncio.wait_for() so a node that accepts
TCP but hangs during handshake/get_leader doesn't block the entire
leader discovery process indefinitely.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -48,10 +48,15 @@ async def find_leader(self) -> str:
 
         for node in nodes:
             try:
-                leader_address = await self._query_leader(node.address)
+                leader_address = await asyncio.wait_for(
+                    self._query_leader(node.address), timeout=self._timeout
+                )
                 if leader_address:
                     self._leader_address = leader_address
                     return leader_address
+            except TimeoutError:
+                errors.append(f"{node.address}: timed out")
+                continue
             except Exception as e:
                 errors.append(f"{node.address}: {e}")
                 continue

tests/test_cluster.py

@@ -106,6 +106,32 @@ async def test_find_leader_no_leader_known(self) -> None:
         ):
             await client.find_leader()
 
+    async def test_find_leader_node_hangs_after_connect(self) -> None:
+        """A node that accepts TCP but hangs on handshake should be timed out."""
+        import asyncio
+
+        store = MemoryNodeStore(["localhost:9001"])
+        client = ClusterClient(store, timeout=0.1)
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        # Server accepts connection but never sends a response
+        async def hang_forever(*args, **kwargs):
+            await asyncio.sleep(100)
+            return b""
+
+        mock_reader.read.side_effect = hang_forever
+
+        with (
+            patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)),
+            pytest.raises(ClusterError, match="Could not find leader"),
+        ):
+            await client.find_leader()
+
     async def test_update_nodes(self) -> None:
         store = MemoryNodeStore()
         client = ClusterClient(store)