Reject LEADER responses with non-zero id and empty address

Upstream raft_leader sets id and address atomically; the C server
substitutes "" for NULL so the only wire-legal shapes are (0, "") and
(nonzero, nonempty). The "non-zero id, empty address" branch was
silently substituting the queried node's own address as the leader,
which trusts a peer that has already violated the wire contract. Raise
ProtocolError instead so find_leader skips the bad peer via its
existing except-tuple and moves on.

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

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -164,15 +164,20 @@ async def _query_leader(
             await protocol.handshake()
             node_id, leader_addr = await protocol.get_leader()
 
-            if not leader_addr and node_id != 0:
-                # Non-zero node_id with empty address: this node is the leader
-                return address
-            elif leader_addr:
-                # Non-empty address: redirect to the reported leader
+            # Upstream dqlite's ``raft_leader`` sets ``id`` and ``address``
+            # atomically: either both are filled in (a leader is known) or
+            # both are zero/NULL. The server substitutes ``""`` for NULL,
+            # so the only wire-legal shapes are ``(0, "")`` and
+            # ``(nonzero, nonempty)``.
+            if node_id != 0 and not leader_addr:
+                raise ProtocolError(
+                    f"server {address} returned node_id={node_id} with empty "
+                    f"leader address; expected both or neither"
+                )
+            if leader_addr:
                 return leader_addr
-            else:
-                # node_id=0 and empty address: no leader known
-                return None
+            # node_id=0 and empty address: no leader known
+            return None
         finally:
             writer.close()
 

tests/test_cluster.py

@@ -539,3 +539,65 @@ async def get_leader(self) -> tuple[int, str]:
 
         assert captured.get("trust_server_heartbeat") is False
 
+
+class TestQueryLeaderRejectsUnreachableCombo:
+    """A (nonzero id, empty address) response is a protocol violation;
+    reject it instead of silently substituting the queried address.
+    """
+
+    async def test_nonzero_id_with_empty_address_raises_protocol_error(self) -> None:
+        from dqliteclient.exceptions import ProtocolError
+
+        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()
+
+        class FakeProto:
+            def __init__(self, *args: object, **kwargs: object) -> None:
+                pass
+
+            async def handshake(self) -> None:
+                pass
+
+            async def get_leader(self) -> tuple[int, str]:
+                return (42, "")
+
+        with (
+            patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)),
+            patch("dqliteclient.cluster.DqliteProtocol", FakeProto),
+            pytest.raises(ProtocolError),
+        ):
+            await client._query_leader("localhost:9001")
+
+    async def test_zero_id_empty_address_returns_none(self) -> None:
+        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()
+
+        class FakeProto:
+            def __init__(self, *args: object, **kwargs: object) -> None:
+                pass
+
+            async def handshake(self) -> None:
+                pass
+
+            async def get_leader(self) -> tuple[int, str]:
+                return (0, "")
+
+        with (
+            patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)),
+            patch("dqliteclient.cluster.DqliteProtocol", FakeProto),
+        ):
+            result = await client._query_leader("localhost:9001")
+
+        assert result is None