fix: close writer directly in _query_leader() to prevent socket leak

The try/finally in _query_leader() closed the protocol object, but if
DqliteProtocol construction failed (or any error occurred between
open_connection and the try block), the writer/socket was leaked.

Widen the try/finally to cover protocol construction and close the
writer directly instead of through the protocol object.

Fixes #102

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

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -75,9 +75,8 @@ async def _query_leader(self, address: str) -> str | None:
         except (TimeoutError, OSError):
             return None
 
-        protocol = DqliteProtocol(reader, writer, timeout=self._timeout)
-
         try:
+            protocol = DqliteProtocol(reader, writer, timeout=self._timeout)
             await protocol.handshake()
             node_id, leader_addr = await protocol.get_leader()
 
@@ -91,8 +90,8 @@ async def _query_leader(self, address: str) -> str | None:
                 # node_id=0 and empty address: no leader known
                 return None
         finally:
-            protocol.close()
-            await protocol.wait_closed()
+            writer.close()
+            await writer.wait_closed()
 
     async def connect(self, database: str = "default") -> DqliteConnection:
         """Connect to the cluster leader.

tests/test_cluster.py

@@ -137,6 +137,53 @@ async def hang_forever(*args, **kwargs):
         ):
             await client.find_leader()
 
+    async def test_query_leader_closes_writer_on_handshake_error(self) -> None:
+        """Writer must be closed even if handshake raises an unexpected error."""
+        store = MemoryNodeStore(["localhost:9001"])
+        client = ClusterClient(store, timeout=1.0)
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        # Handshake data that triggers a protocol error
+        mock_reader.read.side_effect = [b"\x00" * 64]
+
+        with (
+            patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)),
+            pytest.raises(ClusterError),
+        ):
+            await client.find_leader()
+
+        # The writer must have been closed to avoid socket leak
+        mock_writer.close.assert_called()
+
+    async def test_query_leader_closes_writer_on_protocol_init_error(self) -> None:
+        """Writer must be closed even if DqliteProtocol construction fails."""
+        store = MemoryNodeStore(["localhost:9001"])
+        client = ClusterClient(store, timeout=1.0)
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        with (
+            patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)),
+            patch(
+                "dqliteclient.cluster.DqliteProtocol",
+                side_effect=RuntimeError("init failed"),
+            ),
+            pytest.raises(ClusterError),
+        ):
+            await client.find_leader()
+
+        # Even though DqliteProtocol() failed, the writer must be closed
+        mock_writer.close.assert_called()
+
     async def test_update_nodes(self) -> None:
         store = MemoryNodeStore()
         client = ClusterClient(store)