Rotate away from a connected server that stopped following the chain (B-86, ELC-21, ARC-26)

`server.ping` keeps succeeding on a server that is stalled, and nothing compared its tip
with the others', so it stayed the active server for ever.

- The listener records when the last header was accepted (monotonic clock, a module
  attribute so tests can move it).
- After every successful ping, ElectrumListener._check_not_stalled: once the last accepted
  header is older than 360s the other servers are asked for their tip
  (blockchain.headers.subscribe, ELC-14's timeouts); if any reports a higher height than
  ours the session ends with ServerStalledError and run() rotates to the next server. If
  none does the chain is just quiet and the session goes on.

No existing test changed. ELC-29 and /admin/electrum (the connection state and the age of
the last header, B-92) are untouched.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

tests/unit/test_electrum_listener.py

@@ -1254,3 +1254,138 @@ async def test_the_window_does_not_move_when_the_reorg_cannot_be_recorded(sessio
     await listener._apply_header({"height": 101, "hex": client.chain[101]}, client)
     assert listener.tip_height == 101
     assert len(await _reorg_events(session_factory)) == 1
+
+
+# --- B-86 (ELC-21): a connected server that stopped following the chain is replaced -------
+
+
+class _TipClient:
+    """What a *other* server answers to a headers subscription: its tip."""
+
+    def __init__(self, height: int | None, fail: bool = False):
+        self._height = height
+        self._fail = fail
+
+    async def connect(self):
+        if self._fail:
+            raise OSError("unreachable")
+
+    async def subscribe_headers(self):
+        return {"height": self._height, "hex": "00"}
+
+    async def close(self):
+        pass
+
+
+def _stall_listener(session_factory, monkeypatch, other_heights: dict[str, int | None]):
+    """A listener on `_ENDPOINTS[0]` with a tip at 100 accepted at t=0 and a movable clock;
+    the other two servers answer `other_heights` (None = unreachable)."""
+    from app.electrum import listener as listener_module
+
+    clock = {"now": 0.0}
+    monkeypatch.setattr(listener_module, "_monotonic", lambda: clock["now"])
+
+    def factory(endpoint):
+        height = other_heights[endpoint.host]
+        return _TipClient(height, fail=height is None)
+
+    listener = ElectrumListener(factory, session_factory, _ENDPOINTS)
+    listener._extend_window([(100, _mine_chain(100, 1)[100])])
+    return listener, clock
+
+
+async def test_a_stalled_server_is_replaced_when_another_reports_a_higher_tip(session_factory, monkeypatch):
+    """B-86, ELC-21: the active server keeps answering pings but its last accepted header is
+    older than 360s while another server is already at 101 — the session must end so run()
+    rotates. Nothing used to compare the tips, so a stalled server stayed active for ever."""
+    from app.electrum.listener import ServerStalledError
+
+    listener, clock = _stall_listener(
+        session_factory, monkeypatch, {"first.example": None, "second.example": 101, "third.example": 100}
+    )
+    clock["now"] = 361.0
+    with pytest.raises(ServerStalledError):
+        await listener._check_not_stalled()
+
+
+async def test_a_quiet_chain_does_not_rotate_the_session(session_factory, monkeypatch):
+    """ELC-21: no block for 360s but nobody reports a higher tip — the network produced
+    nothing, and the session goes on."""
+    listener, clock = _stall_listener(
+        session_factory, monkeypatch, {"first.example": None, "second.example": 100, "third.example": 99}
+    )
+    clock["now"] = 10_000.0
+    await listener._check_not_stalled()  # must not raise
+
+
+async def test_the_other_servers_are_not_asked_before_360_seconds(session_factory, monkeypatch):
+    """ELC-21: the check costs a connection to every other server, so it only starts once
+    the last header is older than 360s — exactly 360 is still fine."""
+    listener, clock = _stall_listener(
+        session_factory, monkeypatch, {"first.example": None, "second.example": 500, "third.example": 500}
+    )
+    clock["now"] = 360.0
+    await listener._check_not_stalled()  # must not raise although others are ahead
+
+
+async def test_an_unreachable_other_server_does_not_count_as_ahead(session_factory, monkeypatch):
+    listener, clock = _stall_listener(
+        session_factory, monkeypatch, {"first.example": None, "second.example": None, "third.example": None}
+    )
+    clock["now"] = 1_000.0
+    await listener._check_not_stalled()  # no answers, nothing to rotate to
+
+
+async def test_an_accepted_header_restarts_the_stall_clock(session_factory, monkeypatch):
+    """ELC-21: it is the age of the last *accepted* header that counts."""
+    listener, clock = _stall_listener(
+        session_factory, monkeypatch, {"first.example": None, "second.example": 500, "third.example": 500}
+    )
+    clock["now"] = 400.0
+    listener._extend_window([(101, _mine_chain(101, 1)[101])])
+    clock["now"] = 600.0  # 200s since the header at 101
+    await listener._check_not_stalled()
+
+
+async def test_the_keepalive_ends_the_session_of_a_stalled_server(session_factory, monkeypatch):
+    """B-86: wired into the ping loop — after a successful ping the stall check runs, and its
+    error is what ends the session (`_run_once` then returns and `run` rotates)."""
+    from app.electrum import listener as listener_module
+    from app.electrum.listener import ServerStalledError
+
+    monkeypatch.setattr(listener_module, "_PING_INTERVAL_SECONDS", 0)
+    listener, clock = _stall_listener(
+        session_factory, monkeypatch, {"first.example": None, "second.example": 101, "third.example": 100}
+    )
+    clock["now"] = 400.0
+    pinged = []
+
+    class _Active:
+        async def ping(self):
+            pinged.append(True)
+
+    with pytest.raises(ServerStalledError):
+        await asyncio.wait_for(listener._keepalive(_Active()), timeout=2)
+    assert pinged == [True]
+
+
+async def test_a_run_once_session_ends_when_the_active_server_stalls(session_factory, monkeypatch):
+    """B-86, end to end through _run_once: connected, header accepted at t=0, then the clock
+    passes 360s with another server ahead — the session returns by itself."""
+    from app.electrum import listener as listener_module
+
+    monkeypatch.setattr(listener_module, "_PING_INTERVAL_SECONDS", 0.01)
+    clock = {"now": 0.0}
+    monkeypatch.setattr(listener_module, "_monotonic", lambda: clock["now"])
+    chain = _mine_chain(91, 10)
+    active = _FakeConnectClient({"height": 100, "hex": chain[100]}, chain)
+
+    def factory(endpoint):
+        return active if endpoint == _ENDPOINTS[0] else _TipClient(101)
+
+    listener = ElectrumListener(factory, session_factory, _ENDPOINTS)
+    task = asyncio.create_task(listener._run_once(_ENDPOINTS[0]))
+    await _wait_until(lambda: listener.client is not None)
+    clock["now"] = 361.0
+    assert await asyncio.wait_for(task, timeout=3) is True
+    assert listener.client is None