fix: narrow find_leader exception catch and chain cause

find_leader previously caught bare Exception, stringified any error (even
TypeError/KeyError programming bugs) into the errors list, and raised
ClusterError with no __cause__. ClusterError is in retry_with_backoff's
retryable set, so a programming bug got retried 5× before propagating as
a generic error with no traceback.

Narrow the catch to the real transport-level errors
(DqliteConnectionError, ProtocolError, OperationalError, OSError) so
programming bugs propagate untouched, and chain the final ClusterError's
__cause__ to the last caught exception so logs surface the underlying
reason.

Also wrap wire-library protocol errors in DqliteProtocol reads so they
surface as the client's ProtocolError rather than raw DecodeError/etc.,
which keeps the narrow catch honest.

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

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -3,7 +3,12 @@
 import asyncio
 
 from dqliteclient.connection import DqliteConnection, _parse_address
-from dqliteclient.exceptions import ClusterError, DqliteConnectionError, OperationalError
+from dqliteclient.exceptions import (
+    ClusterError,
+    DqliteConnectionError,
+    OperationalError,
+    ProtocolError,
+)
 from dqliteclient.node_store import MemoryNodeStore, NodeInfo, NodeStore
 from dqliteclient.protocol import DqliteProtocol
 from dqliteclient.retry import retry_with_backoff
@@ -46,6 +51,7 @@ async def find_leader(self) -> str:
             raise ClusterError("No nodes configured")
 
         errors: list[str] = []
+        last_exc: BaseException | None = None
 
         for node in nodes:
             try:
@@ -54,14 +60,19 @@ async def find_leader(self) -> str:
                 )
                 if leader_address:
                     return leader_address
-            except TimeoutError:
+            except TimeoutError as e:
                 errors.append(f"{node.address}: timed out")
+                last_exc = e
                 continue
-            except Exception as e:
+            except (DqliteConnectionError, ProtocolError, OperationalError, OSError) as e:
+                # Narrow the catch so programming bugs (TypeError, KeyError,
+                # etc.) propagate directly instead of being stringified into
+                # a retryable ClusterError.
                 errors.append(f"{node.address}: {e}")
+                last_exc = e
                 continue
 
-        raise ClusterError(f"Could not find leader. Errors: {'; '.join(errors)}")
+        raise ClusterError(f"Could not find leader. Errors: {'; '.join(errors)}") from last_exc
 
     async def _query_leader(self, address: str) -> str | None:
         """Query a node for the current leader."""

src/dqliteclient/protocol.py

@@ -7,6 +7,7 @@
 
 from dqliteclient.exceptions import DqliteConnectionError, OperationalError, ProtocolError
 from dqlitewire import MessageDecoder, MessageEncoder
+from dqlitewire.exceptions import ProtocolError as _WireProtocolError
 from dqlitewire.messages import (
     ClientRequest,
     DbResponse,
@@ -298,12 +299,15 @@ async def _read_continuation(self, deadline: float | None = None) -> RowsRespons
         """
         if deadline is None:
             deadline = self._operation_deadline()
-        while True:
-            result = self._decoder.decode_continuation()
-            if result is not None:
-                return result
-            data = await self._read_data(deadline=deadline)
-            self._decoder.feed(data)
+        try:
+            while True:
+                result = self._decoder.decode_continuation()
+                if result is not None:
+                    return result
+                data = await self._read_data(deadline=deadline)
+                self._decoder.feed(data)
+        except _WireProtocolError as e:
+            raise ProtocolError(f"Wire decode failed: {e}") from e
 
     async def _read_response(self, deadline: float | None = None) -> Message:
         """Read and decode the next response message.
@@ -315,11 +319,15 @@ async def _read_response(self, deadline: float | None = None) -> Message:
         """
         if deadline is None:
             deadline = self._operation_deadline()
-        while not self._decoder.has_message():
-            data = await self._read_data(deadline=deadline)
-            self._decoder.feed(data)
+        try:
+            while not self._decoder.has_message():
+                data = await self._read_data(deadline=deadline)
+                self._decoder.feed(data)
+
+            message = self._decoder.decode()
+        except _WireProtocolError as e:
+            raise ProtocolError(f"Wire decode failed: {e}") from e
 
-        message = self._decoder.decode()
         if message is None:
             raise ProtocolError("Failed to decode message")
 

tests/test_cluster.py

@@ -162,6 +162,8 @@ async def test_query_leader_closes_writer_on_handshake_error(self) -> None:
 
     async def test_query_leader_closes_writer_on_protocol_init_error(self) -> None:
         """Writer must be closed even if DqliteProtocol construction fails."""
+        from dqliteclient.exceptions import DqliteConnectionError
+
         store = MemoryNodeStore(["localhost:9001"])
         client = ClusterClient(store, timeout=1.0)
 
@@ -175,7 +177,7 @@ async def test_query_leader_closes_writer_on_protocol_init_error(self) -> None:
             patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)),
             patch(
                 "dqliteclient.cluster.DqliteProtocol",
-                side_effect=RuntimeError("init failed"),
+                side_effect=DqliteConnectionError("init failed"),
             ),
             pytest.raises(ClusterError),
         ):
@@ -223,6 +225,45 @@ async def hang_forever():
 
         mock_writer.close.assert_called()
 
+    async def test_find_leader_propagates_programming_bugs(self) -> None:
+        """Programming bugs (TypeError etc.) must propagate, not be swallowed
+        into a generic ClusterError. ClusterError is retryable upstream, so
+        stringifying a bug here amplifies it N*retries times.
+        """
+        store = MemoryNodeStore(["localhost:9001", "localhost:9002"])
+        client = ClusterClient(store, timeout=0.5)
+
+        async def buggy_query(_address: str) -> str | None:
+            raise TypeError("programmer mistake")
+
+        with (
+            patch.object(client, "_query_leader", side_effect=buggy_query),
+            pytest.raises(TypeError, match="programmer mistake"),
+        ):
+            await client.find_leader()
+
+    async def test_find_leader_transport_error_chains_cause(self) -> None:
+        """When every node yields a transport error, the final ClusterError
+        must have __cause__ set so logs show the underlying reason, not just
+        the generic message.
+        """
+        from dqliteclient.exceptions import DqliteConnectionError
+
+        store = MemoryNodeStore(["localhost:9001"])
+        client = ClusterClient(store, timeout=0.5)
+
+        boom = DqliteConnectionError("handshake failed")
+
+        async def failing_query(_address: str) -> str | None:
+            raise boom
+
+        with (
+            patch.object(client, "_query_leader", side_effect=failing_query),
+            pytest.raises(ClusterError) as exc_info,
+        ):
+            await client.find_leader()
+        assert exc_info.value.__cause__ is boom
+
     async def test_update_nodes(self) -> None:
         store = MemoryNodeStore()
         client = ClusterClient(store)