Propagate dial-side OSError from _query_leader for per-node attribution

Pre-fix the function swallowed pre-handshake OSError as None,
losing per-node attribution: a node that consistently refused
connection contributed to the aggregate ClusterError as
"<addr>: no leader known" rather than
"<addr>: ConnectionRefusedError". Operators reading the aggregate
could not distinguish "node unreachable" from "node up, no leader
elected" — different remediation.

Now propagate OSError (and TimeoutError, ConnectionRefusedError,
etc.) to the caller. _probe_one already wraps the call in a
try/except that classifies transport errors per node into the
aggregate. Mirrors go-dqlite's connector.go shape where every
failure class is attributed.

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

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -924,19 +924,33 @@ async def _probe_one(idx: int, node: _StoreNodeInfo) -> _LeaderHit | _ProbeMiss:
     async def _query_leader(
         self, address: str, *, trust_server_heartbeat: bool = False
     ) -> str | None:
-        """Query a node for the current leader."""
-        try:
-            reader, writer = await asyncio.wait_for(
-                open_connection(address, dial_func=self._dial_func),
-                timeout=self._dial_timeout,
-            )
-        except OSError:
-            # OSError subsumes TimeoutError, BrokenPipeError,
-            # ConnectionError, ConnectionRefusedError, and the rest
-            # of the stdlib transport-error shapes. Any one of those
-            # here means the node is unreachable; surface "unknown
-            # leader" to the caller so it can try another node.
-            return None
+        """Query a node for the current leader.
+
+        Raises ``OSError`` (or a subclass) if the dial itself fails;
+        ``DqliteConnectionError`` / ``ProtocolError`` /
+        ``OperationalError`` for handshake-or-later failures. The
+        caller (``_probe_one``) attributes each failure class to the
+        per-node aggregate ``ClusterError``. Returns ``None`` when
+        the node is reachable but reports no leader (the
+        "unknown leader" branch the wire-level ``LeaderResponse``
+        handles).
+
+        Pre-fix this swallowed pre-handshake ``OSError`` as ``None``,
+        losing per-node attribution: a node that consistently
+        refused connection contributed to the aggregate as
+        ``"<addr>: no leader known"`` rather than
+        ``"<addr>: ConnectionRefusedError"``. Operators reading the
+        aggregate cannot remediate without distinguishing
+        "node unreachable" from "node up, no leader elected".
+        """
+        # Let OSError (subsumes TimeoutError, ConnectionRefused,
+        # BrokenPipe, etc.) propagate to the caller. ``_probe_one``
+        # already wraps this call in a try/except that classifies
+        # transport errors per node into the aggregate ClusterError.
+        reader, writer = await asyncio.wait_for(
+            open_connection(address, dial_func=self._dial_func),
+            timeout=self._dial_timeout,
+        )
 
         try:
             protocol = DqliteProtocol(

tests/test_dial_attempt_timeout_split.py

@@ -141,15 +141,16 @@ async def slow_open(*args: object, **kwargs: object) -> tuple[object, object]:
         "dqliteclient._dial.open_connection_with_keepalive",
         side_effect=slow_open,
     ):
-        # ``_query_leader`` returns ``None`` on OSError-family
-        # (which subsumes TimeoutError on cancellation). The test
-        # passes if it returns within the dial_timeout budget, NOT
-        # after the full ``timeout=10.0``.
+        # ``_query_leader`` raises TimeoutError when the dial budget
+        # is exhausted (the per-node attribution is the caller's
+        # job — see ``_probe_one``). The test passes if the raise
+        # arrives within the dial_timeout budget, NOT after the
+        # full ``timeout=10.0``.
         start = asyncio.get_running_loop().time()
-        result = await cluster._query_leader("localhost:9001")
+        with pytest.raises(TimeoutError):
+            await cluster._query_leader("localhost:9001")
         elapsed = asyncio.get_running_loop().time() - start
 
-    assert result is None
     assert elapsed < 1.5, (
         f"expected query_leader to give up within ~dial_timeout (0.5s); "
         f"actually took {elapsed:.3f}s"

tests/test_query_leader_dial_error_attribution.py

@@ -0,0 +1,88 @@
+"""``_query_leader`` propagates pre-handshake transport errors
+(OSError, TimeoutError, ConnectionRefusedError) to the caller so
+``_probe_one`` can attribute them to the per-node aggregate
+``ClusterError`` with a specific exception class.
+
+Pre-fix the function swallowed OSError as ``None``, conflating
+"node unreachable" with "node up, no leader elected" — operators
+reading the aggregate could not distinguish the two and could not
+remediate appropriately.
+"""
+
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from dqliteclient.cluster import ClusterClient
+from dqliteclient.exceptions import ClusterError
+from dqliteclient.node_store import MemoryNodeStore
+
+
+@pytest.mark.asyncio
+async def test_query_leader_propagates_connection_refused() -> None:
+    cluster = ClusterClient(MemoryNodeStore(["unreachable:9001"]), timeout=2.0)
+
+    async def refused_open(*_args: object, **_kwargs: object):
+        raise ConnectionRefusedError("nothing listening")
+
+    with (
+        patch(
+            "dqliteclient._dial.open_connection_with_keepalive",
+            side_effect=refused_open,
+        ),
+        pytest.raises(ConnectionRefusedError),
+    ):
+        await cluster._query_leader("unreachable:9001")
+
+
+@pytest.mark.asyncio
+async def test_query_leader_propagates_timeout() -> None:
+    cluster = ClusterClient(
+        MemoryNodeStore(["slow:9001"]),
+        timeout=2.0,
+        dial_timeout=0.05,
+    )
+
+    async def slow_open(*_args: object, **_kwargs: object):
+        await asyncio.sleep(1.0)
+
+    with (
+        patch(
+            "dqliteclient._dial.open_connection_with_keepalive",
+            side_effect=slow_open,
+        ),
+        pytest.raises(TimeoutError),
+    ):
+        await cluster._query_leader("slow:9001")
+
+
+@pytest.mark.asyncio
+async def test_find_leader_aggregate_attributes_dial_error() -> None:
+    """End-to-end: a node that refuses connection appears in the
+    aggregate ClusterError with its specific exception class, not
+    a vague "no leader known"."""
+    cluster = ClusterClient(
+        MemoryNodeStore(["unreachable:9001"]),
+        timeout=2.0,
+        dial_timeout=0.5,
+        attempt_timeout=1.0,
+    )
+
+    async def refused_open(*_args: object, **_kwargs: object):
+        raise ConnectionRefusedError("nothing listening")
+
+    with (
+        patch(
+            "dqliteclient._dial.open_connection_with_keepalive",
+            side_effect=refused_open,
+        ),
+        pytest.raises(ClusterError) as excinfo,
+    ):
+        await cluster.find_leader()
+
+    # The aggregate error message must mention the node and the
+    # specific transport error rather than just "no leader known".
+    msg = str(excinfo.value)
+    assert "unreachable:9001" in msg
+    assert "ConnectionRefusedError" in msg or "refused" in msg.lower() or "nothing listening" in msg