From 4232136d26c464072b52b486b64fc47df496842f Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Wed, 22 Jul 2026 22:10:46 -0700 Subject: [PATCH] fix(server): de-register the global listener when the reconnect window exits early (#881) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-1's lock-scope fix moved the snapshot build after listener registration but left it unguarded: a raising _build_node_snapshot (storage reads, per-ws locks) propagated before the generator — whose finally owns de-registration — ever existed, stranding a dead 1000-slot queue in the fan-out list forever (the fan-out thread never removes listeners; pre-branch the append was the LAST locked statement precisely so a raising build could not strand it). The whole post-registration window (build, log, response construction) now runs under a guard that de-registers on ANY exit and re-raises; _deregister is shared with the generator's finally so the discipline has one owner. BaseException because the window must stay guarded even if a future edit introduces an await (today it is await-free, so a cancel cannot land inside it). Tests (round-2 review): the leak path is pinned (raising build → exception propagates AND the listener list is empty); the registration-before-build + lock-released ordering is pinned by a probe builder asserting both at build time; the caught-up-cursor test is rebuilt around a sentinel live event so it asserts the no-envelope shape positively instead of truncating the drain at the retry frame. --- tests/test_global_sse_boot_epoch.py | 104 +++++++++++++++++++++++++--- turnstone/server.py | 99 +++++++++++++++++--------- 2 files changed, 161 insertions(+), 42 deletions(-) diff --git a/tests/test_global_sse_boot_epoch.py b/tests/test_global_sse_boot_epoch.py index 9053886b..6e3e184d 100644 --- a/tests/test_global_sse_boot_epoch.py +++ b/tests/test_global_sse_boot_epoch.py @@ -31,6 +31,7 @@ import collections import json import queue import threading +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace as SimpleNS from typing import Any @@ -204,14 +205,42 @@ def test_same_epoch_cursor_at_head_is_caught_up_replay_ok( monkeypatch: pytest.MonkeyPatch, ) -> None: """A cursor at the newest id is the normal caught-up reconnect: - nothing to replay, no snapshot, and crucially no false truncated.""" - yields = _drain( - monkeypatch, - buffered=[_ev(4), _ev(5)], - headers={"Last-Event-ID": f"{EPOCH}-5"}, - max_yields=1, - ) - assert _types(yields) == [] + nothing to replay, no snapshot, and crucially no false truncated. + + Asserted POSITIVELY: a sentinel live event is queued to the + registered listener before draining, so the drain runs through the + live loop and the sentinel must be the FIRST data frame — a spurious + envelope or snapshot would land ahead of it. (A drain truncated at + the always-first ``retry`` frame asserts nothing — round-2 review.) + """ + monkeypatch.setattr(server_mod, "_build_node_snapshot", lambda _s: dict(_SNAPSHOT_STUB)) + app_state = _make_app_state(buffered=[_ev(4), _ev(5)]) + executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="boot-epoch-test") + app_state.sse_executor = executor + req = _fake_request(app_state, headers={"Last-Event-ID": f"{EPOCH}-5"}) + + async def _run() -> list[dict[str, Any]]: + async with asyncio.timeout(10): + resp: Any = await server_mod.global_events_sse(req) + # Registered by handler-await time; the live sentinel the + # drain below must reach as its first data frame. + app_state.global_listeners[0].put_nowait( + {"type": "ws_state", "ws_id": "sentinel", "_event_id": 6} + ) + out: list[dict[str, Any]] = [] + async for chunk in resp.body_iterator: + out.append(chunk) + if len(out) >= 2: + break + await resp.body_iterator.aclose() + return out + + try: + yields = asyncio.run(_run()) + finally: + executor.shutdown(wait=True) + frames = _data_frames(yields) + assert [f.get("ws_id") for f in frames] == ["sentinel"] def test_same_epoch_ring_miss_is_truncated_with_honest_counts( @@ -377,6 +406,65 @@ def test_epoch_is_hex_and_dashless_by_construction() -> None: assert "app.state.global_boot_epoch = secrets.token_hex(" in src +def test_snapshot_builds_after_registration_outside_the_lock( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Pins the branch's concurrency contract AT THE BUILD SITE: when + ``_build_node_snapshot`` runs, (1) the listener is already + REGISTERED — registration-before-build is the no-loss ordering + (every event from registration on is queued, so the newer snapshot + plus the queued deltas covers everything), and (2) + ``global_listeners_lock`` is RELEASED — the O(workstreams) build + must not stall the fan-out thread. A refactor "harmonizing" this + lane with the per-ws build-under-lock shape, or reordering + registration after the build, fails here and nowhere else.""" + app_state = _make_app_state(buffered=[_ev(1)]) + probed: dict[str, Any] = {} + + def _probe(_s: Any) -> dict[str, Any]: + probed["registered"] = len(app_state.global_listeners) + acquired = app_state.global_listeners_lock.acquire(blocking=False) + probed["lock_free"] = acquired + if acquired: + app_state.global_listeners_lock.release() + return dict(_SNAPSHOT_STUB) + + monkeypatch.setattr(server_mod, "_build_node_snapshot", _probe) + req = _fake_request(app_state, headers={"Last-Event-ID": "deadbeef-1"}) + + async def _run() -> None: + resp: Any = await server_mod.global_events_sse(req) + await resp.body_iterator.aclose() + + asyncio.run(_run()) + assert probed == {"registered": 1, "lock_free": True} + + +def test_raising_snapshot_build_deregisters_the_listener( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The registration-window guard (#881 round-2 review): registration + precedes the build, so a raising ``_build_node_snapshot`` fires + before the generator — whose ``finally`` is the normal owner of + de-registration — ever exists. The window guard must remove the + queue, or it sits dead in the fan-out list forever (the fan-out + thread never removes listeners).""" + app_state = _make_app_state(buffered=[_ev(1)]) + + def _boom(_s: Any) -> dict[str, Any]: + raise RuntimeError("storage hiccup") + + monkeypatch.setattr(server_mod, "_build_node_snapshot", _boom) + req = _fake_request(app_state, headers={"Last-Event-ID": "deadbeef-1"}) + + async def _run() -> None: + await server_mod.global_events_sse(req) + + with pytest.raises(RuntimeError, match="storage hiccup"): + asyncio.run(_run()) + assert app_state.global_listeners == [] + + def test_listener_registered_exactly_once_per_connect( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/turnstone/server.py b/turnstone/server.py index 827dd729..1bb51baf 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -1160,37 +1160,17 @@ async def global_events_sse(request: Request) -> Response: replay_events = [ev for eid, ev in buffered if eid > last_event_id] listeners.append(client_queue) - if replay_status != "replay_ok": - # Snapshot built AFTER the lock: ``_build_node_snapshot`` is an - # O(workstreams) walk taking each ws's ``_ws_lock``, and under - # ``global_listeners_lock`` it would stall the fanout thread's - # per-event stamping for EVERY listener — N-fold during a - # restart herd of stale-cursor reconnects, exactly when the - # reborn node is emitting its re-open events. Registration - # already happened under the lock above, so every event from - # that instant on is queued to this listener: nothing is lost, - # and deltas stamped while the snapshot builds re-apply - # idempotently over the (newer) snapshot at the consumer - # (state-of-world contract — see the endpoint docstring). - # ``replay_ok`` is by design exactly the no-snapshot shape - # (truncated always carries the snapshot floor, fresh always - # snapshots), so this predicate and the generator's emission - # branch stay one rule. - snapshot = _build_node_snapshot(request.app.state) + def _deregister() -> None: + """Remove this connection's queue from the fan-out list. - if replay_status == "truncated": - # Envelope chokepoint log, analog of ``ws.events.replay_truncated`` - # on the per-ws lane — one line per gap detection, the operator - # signal that clients are being told to resync (a burst of - # ``reason=boot_epoch`` right after startup is the restart herd - # healing; sustained ``reason=ring_evicted`` means the ring cap - # is being outrun). - log.info( - "global.events.replay_truncated", - reason=truncated_reason, - lost_count=lost_count if truncated_reason == "ring_evicted" else None, - cursor=(last_event_id_raw or "")[:64], - ) + Idempotent (``in`` check under the lock). Two owners share it: + the generator's ``finally`` on every stream exit, and the + registration-window guard below for exits that fire before the + generator exists. + """ + with listeners_lock: + if client_queue in listeners: + listeners.remove(client_queue) async def event_generator() -> AsyncGenerator[dict[str, Any], None]: _metrics.record_sse_connect() @@ -1262,11 +1242,62 @@ async def global_events_sse(request: Request) -> Response: pass # poll timeout, retry finally: _metrics.record_sse_disconnect() - with listeners_lock: - if client_queue in listeners: - listeners.remove(client_queue) + _deregister() - return EventSourceResponse(event_generator(), ping=5) + # Registration-window guard (#881 round-2 review): from the append + # above until the return below, THIS frame owns the listener — the + # generator's ``finally`` (the normal owner) cannot run before the + # generator exists, so any exit in this window must de-register or + # the queue leaks into the fan-out loop forever (the fan-out thread + # never removes dead listeners; pre-#881 the append was the LAST + # statement of the locked block precisely so a raising snapshot + # build could not strand it). Everything that can raise lives + # inside the try — the snapshot build (storage reads + per-ws + # locks), the log call, and response construction; the two ``def``s + # above are pure bindings and cannot. ``BaseException``: the + # window is await-free today, so a cancellation cannot land inside + # it, but the guard must not silently narrow if a future edit + # introduces an await. Ownership passes to the generator's + # ``finally`` once the return succeeds. + try: + if replay_status != "replay_ok": + # Snapshot built AFTER the lock: ``_build_node_snapshot`` + # is an O(workstreams) walk taking each ws's ``_ws_lock``, + # and under ``global_listeners_lock`` it would stall the + # fanout thread's per-event stamping for EVERY listener — + # N-fold during a restart herd of stale-cursor reconnects, + # exactly when the reborn node is emitting its re-open + # events. Registration already happened under the lock + # above, so every event from that instant on is queued to + # this listener: nothing is lost, and deltas stamped while + # the snapshot builds re-apply idempotently over the + # (newer) snapshot at the consumer (state-of-world + # contract — see the endpoint docstring). ``replay_ok`` + # is by design exactly the no-snapshot shape (truncated + # always carries the snapshot floor, fresh always + # snapshots), so this predicate and the generator's + # emission branch stay one rule. + snapshot = _build_node_snapshot(request.app.state) + + if replay_status == "truncated": + # Envelope chokepoint log, analog of + # ``ws.events.replay_truncated`` on the per-ws lane — one + # line per gap detection, the operator signal that clients + # are being told to resync (a burst of ``reason=boot_epoch`` + # right after startup is the restart herd healing; sustained + # ``reason=ring_evicted`` means the ring cap is being + # outrun). + log.info( + "global.events.replay_truncated", + reason=truncated_reason, + lost_count=lost_count if truncated_reason == "ring_evicted" else None, + cursor=(last_event_id_raw or "")[:64], + ) + + return EventSourceResponse(event_generator(), ping=5) + except BaseException: + _deregister() + raise async def dashboard(request: Request) -> JSONResponse: