Chain find_leader per-node failures via BaseExceptionGroup when more than one node fails

``ClusterClient._find_leader_impl`` overwrote ``last_exc`` on
every iteration; the final ``raise ClusterError(...) from
last_exc`` chained only the LAST iteration's exception even
though the message listed ALL N nodes' failures. Code branching
on ``e.__cause__`` type (e.g. routing security alerts for
``ProtocolError``-from-malformed-redirect) saw a non-
deterministic decision based on iteration ordering — depending
on the random shuffle, a security-relevant ``ProtocolError`` on
node 1 was overwritten by a benign ``TimeoutError`` on node 3.

Collect every per-node exception into a list and chain via
``BaseExceptionGroup`` when more than one node contributed.
Single-exception case keeps the narrow chain so existing
callers continue to work. No-exception case (every node
returned no-leader-known) raises with no chain — the message
itself is the diagnostic.

Mirrors the discipline already applied to
``ConnectionPool.initialize`` partial-failure handling.

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

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -327,7 +327,15 @@ async def _find_leader_impl(self, *, trust_server_heartbeat: bool) -> str:
         nodes.sort(key=lambda n: 0 if n.role == NodeRole.VOTER else 1)
 
         errors: list[str] = []
-        last_exc: BaseException | None = None
+        # Collect every per-node BaseException so the final
+        # ``ClusterError`` can chain them all via
+        # ``BaseExceptionGroup``. Previously only the LAST iteration's
+        # exception was preserved on ``__cause__`` — code that
+        # branches on the cause class (e.g. routing security alerts
+        # for ``ProtocolError``-from-malformed-redirect) saw a non-
+        # deterministic decision based on iteration ordering. Mirrors
+        # the discipline already applied to ``ConnectionPool.initialize``.
+        per_node_excs: list[BaseException] = []
         total_nodes = len(nodes)
 
         for idx, node in enumerate(nodes):
@@ -386,7 +394,7 @@ async def _find_leader_impl(self, *, trust_server_heartbeat: bool) -> str:
                     total_nodes,
                 )
                 errors.append(f"{_safe_addr}: timed out")
-                last_exc = e
+                per_node_excs.append(e)
                 continue
             except (DqliteConnectionError, ProtocolError, OperationalError, OSError) as e:
                 # Narrow the catch so programming bugs (TypeError, KeyError,
@@ -402,7 +410,7 @@ async def _find_leader_impl(self, *, trust_server_heartbeat: bool) -> str:
                     total_nodes,
                 )
                 errors.append(f"{_sanitize_display_text(node.address)}: {_truncate_error(str(e))}")
-                last_exc = e
+                per_node_excs.append(e)
                 continue
 
         joined = "; ".join(errors)
@@ -411,7 +419,21 @@ async def _find_leader_impl(self, *, trust_server_heartbeat: bool) -> str:
             joined = (
                 joined[:_MAX_AGGREGATE_ERROR_PAYLOAD] + f"... [aggregate truncated, {kept} chars]"
             )
-        raise ClusterError(f"Could not find leader. Errors: {joined}") from last_exc
+        # Chain via ``BaseExceptionGroup`` when more than one node
+        # contributed a real exception (the no-leader-known arm
+        # produces no exception, only an entry in ``errors``). Single-
+        # exception case keeps the narrow chain so existing callers
+        # that branch on ``e.__cause__`` type continue to work.
+        # No-exception case (every node returned no-leader-known)
+        # raises with no chain — the message itself is the
+        # diagnostic. Mirrors ``ConnectionPool.initialize``'s discipline.
+        if len(per_node_excs) > 1:
+            raise ClusterError(f"Could not find leader. Errors: {joined}") from BaseExceptionGroup(
+                "find_leader: per-node failures", per_node_excs
+            )
+        if per_node_excs:
+            raise ClusterError(f"Could not find leader. Errors: {joined}") from per_node_excs[0]
+        raise ClusterError(f"Could not find leader. Errors: {joined}")
 
     async def _query_leader(
         self, address: str, *, trust_server_heartbeat: bool = False

tests/test_find_leader_aggregate_exception_group.py

@@ -0,0 +1,80 @@
+"""Pin: ``ClusterClient._find_leader_impl`` chains per-node
+failures via ``BaseExceptionGroup`` when more than one node
+contributed a real exception.
+
+Pre-fix, only the LAST iteration's exception was preserved on
+``__cause__`` — code that branches on the cause class (e.g.
+routing security alerts for ``ProtocolError``-from-malformed-
+redirect) saw a non-deterministic decision based on iteration
+ordering.
+
+Mirrors ``ConnectionPool.initialize``'s
+``BaseExceptionGroup``-on-multi-failure discipline.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from dqliteclient.cluster import ClusterClient
+from dqliteclient.exceptions import (
+    ClusterError,
+    DqliteConnectionError,
+    ProtocolError,
+)
+from dqliteclient.node_store import MemoryNodeStore
+
+
+@pytest.mark.asyncio
+async def test_find_leader_aggregate_chains_via_exception_group_when_multiple_failures() -> None:
+    store = MemoryNodeStore(["node-a:9001", "node-b:9001", "node-c:9001"])
+    cluster = ClusterClient(store, timeout=0.5)
+
+    # node-a → ProtocolError (security-relevant)
+    # node-b → DqliteConnectionError (transport)
+    # node-c → TimeoutError
+    async def _query_leader_per_node(address: str, **_kw: object) -> None:
+        if address == "node-a:9001":
+            raise ProtocolError("malformed redirect from node-a")
+        if address == "node-b:9001":
+            raise DqliteConnectionError("connection refused on node-b")
+        raise TimeoutError("node-c timed out")
+
+    cluster._query_leader = AsyncMock(side_effect=_query_leader_per_node)
+
+    with pytest.raises(ClusterError) as exc_info:
+        await cluster.find_leader()
+
+    cause = exc_info.value.__cause__
+    # Multi-failure case: cause must be BaseExceptionGroup so callers
+    # branching on .split(ProtocolError) recover the security-relevant
+    # exception regardless of iteration order.
+    assert isinstance(cause, BaseExceptionGroup), (
+        f"expected BaseExceptionGroup chain on multi-failure, got {type(cause).__name__}"
+    )
+    matched, _ = cause.split(ProtocolError)
+    assert matched is not None and len(matched.exceptions) == 1
+    matched_transport, _ = cause.split(DqliteConnectionError)
+    assert matched_transport is not None and len(matched_transport.exceptions) == 1
+
+
+@pytest.mark.asyncio
+async def test_find_leader_aggregate_keeps_narrow_chain_on_single_failure() -> None:
+    """Backward-compat: single-failure case keeps the narrow chain so
+    callers that branch on ``e.__cause__`` type continue to work."""
+    store = MemoryNodeStore(["node-a:9001"])
+    cluster = ClusterClient(store, timeout=0.5)
+
+    async def _query_leader_fails(address: str, **_kw: object) -> None:
+        raise ProtocolError("malformed redirect")
+
+    cluster._query_leader = AsyncMock(side_effect=_query_leader_fails)
+
+    with pytest.raises(ClusterError) as exc_info:
+        await cluster.find_leader()
+
+    cause = exc_info.value.__cause__
+    assert isinstance(cause, ProtocolError)
+    assert "malformed redirect" in str(cause)