fix(server): de-register the global listener when the reconnect window exits early (#881)

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

@@ -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:

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: