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: