From 830305555dce4bd84583a4e14ef01e50cbf19688 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 22 May 2026 19:27:18 -0700 Subject: [PATCH] fix(sse): address PR #561 review round 2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 5 follow-up comments from Copilot, all valid: 1. **CRITICAL — snap_seq race with split writer** (concurrency, 001). Round-1's fix lifted snapshot capture into register_listener_with_replay under nested locks, but the WRITER side (on_content_token / on_reasoning_token) still released _ws_lock before calling _enqueue (which bumps _event_id under _listeners_lock). A reader could interleave between writer's release and writer's _enqueue: capture inflight WITH the new text, read STALE _event_id, return snap_seq < new_event_id. The new event's live emit then has _seq > snap_seq, slips past the dedup filter, and double-renders text the snapshot already contained. Fix: move self._enqueue(...) INSIDE the with self._ws_lock: block in both token writers. The inflight mutation and the _event_id advancement are now atomic against any snapshot reader. Lock order _ws_lock (outer) → _listeners_lock (inner via _enqueue) matches the snapshot helpers, so no deadlock. Fan-out's put_nowait calls happen under _ws_lock for token writers — microsecond cost per listener, acceptable for the correctness guarantee. 2. **NIT — stale comment ref to buffered[-1]._event_id** (docs, 002). The comment referenced a local var (buffered) that lives in register_listener_with_replay, not in the events handler. Reworded to describe the cutoff in terms of the last replayed event id and the atomic-against-writers registration. 3. **MODERATE — 401 branch leaves reconnect loop** (bug, 003). The coord's onerror schedules a 5 s CLOSED-state recovery timer unconditionally. In the 401-expired-session branch we close evtSource and showLogin — but the timer still fires 5 s later, observes !evtSource, and calls scheduleReconnect(), which opens a new EventSource that 401s again → infinite reconnect loop while the login overlay is up. Fix: cancel reconnectTimer in the 401 branch. 4. **MODERATE — race test was vacuous** (test_coverage, 004). The previous regression test drained the listener queue after register_listener_with_replay returned, but the helper doesn't backfill buffered events into the queue, so the loop was almost always a no-op and the assertion never executed. Rewrote with a monkey-patched _enqueue that sleeps 50 ms before bumping _event_id — widens the race window deterministically. Verified: the test FAILS on pre-fix code (snap.content has marker but snap.seq=0 < final_event_id=1) and PASSES on post-fix code (writer holds _ws_lock through _enqueue, so the reader blocks until writer fully done). Also pinned the no-backfill contract so a future change adding listener-queue backfill remembers to keep snap_seq the high-water mark. 5. **NIT — except Exception too broad in test** (best_practices, 005). Tightened except Exception: to except queue.Empty: so unexpected exceptions aren't silently swallowed in the drain loop. Tests: - 86 tests in test_sse_reconnect_replay.py + test_session_ui_base.py pass (existing 84 + 2 new race regressions). - Full non-live suite: 6347 passed, 15 skipped, no regressions. - Ruff + mypy clean on changed .py files; JS parses. --- tests/test_sse_reconnect_replay.py | 139 +++++++++++++++--- .../console/static/coordinator/coordinator.js | 14 +- turnstone/core/session_routes.py | 9 +- turnstone/core/session_ui_base.py | 36 +++-- 4 files changed, 161 insertions(+), 37 deletions(-) diff --git a/tests/test_sse_reconnect_replay.py b/tests/test_sse_reconnect_replay.py index 3a54b3d6..f48f45ba 100644 --- a/tests/test_sse_reconnect_replay.py +++ b/tests/test_sse_reconnect_replay.py @@ -341,36 +341,129 @@ def test_truncated_path_snapshot_captures_real_snap_seq() -> None: assert "t9" in snap["content"] -def test_truncated_path_filters_already_in_snapshot_tokens() -> None: - """End-to-end: a token landing in the listener queue between - ``register_listener_with_replay`` returning and the live drain's - first read must be filtered by the handler's - ``_seq <= snap_seq`` dedup, because its text is ALSO in the - snapshot we just emitted. Without the snap_seq fix this test - would observe a duplicate live token after the snapshot.""" - import collections +def test_snap_seq_high_water_mark_holds_under_writer_race() -> None: + """Regression for PR #561 review comment 1. + + The invariant: every token whose text appears in + ``snapshot["content"]`` (or ``"reasoning"``) must have its + ``_event_id`` <= ``snapshot["seq"]``. Equivalently, any token + that fires AFTER the snapshot was captured must have + ``_event_id > snap_seq``. Otherwise the events handler's + ``_seq <= snap_seq`` live-drain filter would let the new token + through AND its text would already be in the snapshot text → + double-render. + + The pre-fix race: ``on_content_token`` took ``_ws_lock``, + appended to inflight, released ``_ws_lock``, then called + ``_enqueue`` (which bumps ``_event_id``). A snapshot reader + interleaving between the release and the ``_enqueue`` would + capture inflight (with the new text) and read a STALE + ``_event_id``. Snap_seq below new event's id → filter slips → + double-render. + + The race window in plain Python is narrow (a few bytecodes + between lock release and the ``_enqueue`` call), so a pure + barrier-based race rarely hits it. This test injects a + deterministic sleep into ``_enqueue`` via monkey-patch to + widen the window enough to be reliably observed under the + pre-fix code path — AND to be reliably AVOIDED under the + post-fix code path (because the post-fix + ``on_content_token`` calls ``_enqueue`` while still holding + ``_ws_lock``, so the snapshot reader can't acquire + ``_ws_lock`` until the writer is fully done). + """ + import queue + import threading + import time ui = _make_ui() - ui._event_buffer = collections.deque(maxlen=3) - for i in range(10): - ui.on_content_token(f"t{i}") - lq, _, status, _, _, snap = ui.register_listener_with_replay(1) - assert status == "truncated" - captured_seq = snap["seq"] - # Simulate the race: at this point the listener queue already - # holds the buffered token events. Drain them and confirm - # every one has ``_seq <= snap_seq`` — i.e. the handler's filter - # would correctly drop them. + marker = "RACE-MARKER" + original_enqueue = ui._enqueue + + # Widen the race window: sleep just BEFORE the original + # ``_enqueue`` runs (which is where ``_event_id`` would advance). + # Post-fix this sleep happens while the writer still holds + # ``_ws_lock`` — readers block. Pre-fix the writer has + # released ``_ws_lock`` before reaching this monkey-patch, so + # the reader gets a clean window to capture an inconsistent + # ``(inflight, _event_id)`` pair. + def slow_enqueue(data: dict[str, Any]) -> None: + time.sleep(0.05) # 50 ms — orders of magnitude wider than the GIL switch interval + return original_enqueue(data) + + ui._enqueue = slow_enqueue # type: ignore[method-assign] + + snap_box: dict[str, Any] = {} + writer_done = threading.Event() + + def _writer() -> None: + ui.on_content_token(marker) + writer_done.set() + + def _reader() -> None: + # Give the writer time to enter ``on_content_token`` and + # (pre-fix) release ``_ws_lock`` before the snapshot. 50 ms + # is conservative; 5 ms would also work in practice. + time.sleep(0.025) + _, _, _, _, _, snap = ui.register_listener_with_replay(0) + snap_box["snap"] = snap + snap_box["event_id_at_snapshot_return"] = ui._event_id + + wt = threading.Thread(target=_writer) + rt = threading.Thread(target=_reader) + wt.start() + rt.start() + wt.join(timeout=5) + rt.join(timeout=5) + assert writer_done.is_set(), "writer thread did not complete" + + snap = snap_box["snap"] + final_event_id = ui._event_id + + # Core invariant: if the snapshot's content includes the marker + # text, snap.seq must be >= the writer's final _event_id. + # Pre-fix this fails (snap.seq=0 while final_event_id=1 and + # snap.content="RACE-MARKER"); post-fix the reader can't acquire + # ``_ws_lock`` until the writer completes, so snap is either + # (content="", seq=0) — reader won first — or + # (content="RACE-MARKER", seq=1) — writer won first. + assert marker in snap["content"] or snap["content"] == "", ( + f"unexpected snap content: {snap['content']!r}" + ) + if marker in snap["content"]: + assert snap["seq"] >= final_event_id, ( + f"snap captured '{marker}' but snap.seq={snap['seq']} < " + f"final _event_id={final_event_id}; the live emission of " + f"this token would slip past the events handler's " + f"_seq <= snap_seq filter and double-render text the " + f"snapshot already contained. Pre-fix race window " + f"opened by ``_enqueue`` running outside ``_ws_lock``." + ) + + # Sanity: also exercise the post-truncated drain shape so the + # test file pins both the contract AND the no-backfill behaviour + # (a future change that adds backfill into the listener queue + # must keep the dedup invariant above true). + ui2 = _make_ui() + for j in range(5): + ui2.on_content_token(f"x{j}") + lq, _, status, _, _, snap2 = ui2.register_listener_with_replay(0) + captured_seq = snap2["seq"] + drained = 0 while True: try: ev = lq.get_nowait() - except Exception: + except queue.Empty: break + drained += 1 if ev.get("type") == "content": - assert ev["_seq"] <= captured_seq, ( - f"token _seq={ev['_seq']} > snap_seq={captured_seq}; " - "would slip past the live-drain dedup and double-render" - ) + assert ev["_seq"] <= captured_seq, f"token _seq={ev['_seq']} > snap_seq={captured_seq}" + assert drained == 0, ( + f"register_listener_with_replay backfilled {drained} events " + f"into the listener queue; if intentional, the dedup " + f"invariant above must still hold and this assertion should " + f"be updated." + ) # --------------------------------------------------------------------------- diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index fb463107..549ba9cf 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -2011,6 +2011,17 @@ /* noop */ } evtSource = null; + // Cancel the pending CLOSED-state recovery timer (set + // below). Without this, 5 s later the timer would + // observe ``!evtSource`` and call ``scheduleReconnect``, + // which would open a new EventSource that gets 401 again + // → infinite reconnect loop while the login overlay is + // up. The login flow re-arms ``connectSSE`` after a + // successful sign-in via its own callback path. + if (reconnectTimer) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } showLogin("Session expired. Please sign in to reconnect."); } }, @@ -2027,7 +2038,8 @@ // lastEventId via the URL query param, so replay still // works across the manual reconnect). Cancel/replace the // existing timer so successive onerror fires don't pile up - // multiple checks for the same source. + // multiple checks for the same source. The 401 branch above + // ALSO cancels this timer when it fires — see comment there. if (reconnectTimer) clearTimeout(reconnectTimer); reconnectTimer = setTimeout(function () { reconnectTimer = null; diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 7ac9991e..68820579 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -1514,10 +1514,11 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler: # synthetic snapshot/state_change/history emission is # skipped by the events handler. No live-dedup # filtering required because the buffered events - # themselves are the cutoff — anything past - # ``buffered[-1]._event_id`` is genuinely new live - # traffic that lands in the listener queue after the - # buffer snapshot. + # themselves are the cutoff — anything past the last + # replayed event id is genuinely new live traffic + # that lands in the listener queue after the buffer + # snapshot was taken (atomic-against-writers under + # the registration's nested locks). in_progress_snap = {"content": "", "reasoning": "", "seq": 0} # Per-kind executor for the blocking ``client_queue.get`` diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index c8682889..fbcbed95 100644 --- a/turnstone/core/session_ui_base.py +++ b/turnstone/core/session_ui_base.py @@ -1737,12 +1737,25 @@ class SessionUIBase: snapshot text up to the cap and then live tokens past it, with a visual gap equal to the past-cap chunk. No silent drop of subsequent tokens. + + **Lock coupling**: ``_enqueue`` is called WHILE still + holding ``_ws_lock`` so the inflight append AND the + ``_event_id`` advancement happen atomically against a + snapshot reader. Without this coupling a reader could + capture the inflight (with the new text) and read + ``_event_id`` BEFORE the writer's ``_enqueue`` bumped it, + producing a ``snap_seq`` lower than the new event's + ``_event_id``. The new event would then slip past the + ``_seq <= snap_seq`` live-drain dedup and double-render + the text the snapshot already contained. Acquisition + order ``_ws_lock`` (outer) → ``_listeners_lock`` (inner via + ``_enqueue``) matches the snapshot helpers, so no deadlock. """ with self._ws_lock: if self._ws_inflight_reasoning_size < _MAX_TURN_CONTENT_CHARS: self._ws_inflight_reasoning.append(text) self._ws_inflight_reasoning_size += len(text) - self._enqueue({"type": "reasoning", "text": text}) + self._enqueue({"type": "reasoning", "text": text}) def on_content_token(self, text: str) -> None: """Append to both turn-content buffers (capped) + enqueue. @@ -1758,18 +1771,23 @@ class SessionUIBase: is stamped by :meth:`_enqueue` against the per-ws ``_event_id`` counter, which advances on EVERY emit regardless of cap state — see :meth:`on_reasoning_token` for - the full rationale. + the full rationale, including why ``_enqueue`` runs while + still holding ``_ws_lock`` (the lock coupling that makes + ``snap_seq`` a true high-water mark for the snapshot text). - The cap-check + append + size-update run under ``_ws_lock`` - so a concurrent + The cap-check + append + size-update + enqueue all run under + ``_ws_lock`` so a concurrent :meth:`snapshot_and_consume_state_payload` IDLE/ERROR drain or a concurrent :meth:`register_listener_with_in_progress_snapshot` - can't see a torn list mid-append. In production this is - single-writer-per-ws (the worker thread) but the snapshot + / :meth:`register_listener_with_replay` sees a consistent + ``(inflight_content, _event_id)`` pair. In production this + is single-writer-per-ws (the worker thread) but the snapshot reader runs from coord's adapter via ``mgr.set_state``; without the lock the writer's append could land in an - orphaned list reference the snapshot just swapped out. Lock - hold is microseconds. + orphaned list reference the snapshot just swapped out, AND + the inflight/counter pair could de-sync. Lock hold is + microseconds (the fan-out's ``put_nowait`` calls are O(N + listeners) but each is a single non-blocking enqueue). """ with self._ws_lock: if self._ws_turn_content_size < _MAX_TURN_CONTENT_CHARS: @@ -1778,7 +1796,7 @@ class SessionUIBase: if self._ws_inflight_content_size < _MAX_TURN_CONTENT_CHARS: self._ws_inflight_content.append(text) self._ws_inflight_content_size += len(text) - self._enqueue({"type": "content", "text": text}) + self._enqueue({"type": "content", "text": text}) def on_stream_end(self) -> None: with self._ws_lock: