diff --git a/tests/test_global_sse_boot_epoch.py b/tests/test_global_sse_boot_epoch.py new file mode 100644 index 00000000..e35d4aa9 --- /dev/null +++ b/tests/test_global_sse_boot_epoch.py @@ -0,0 +1,391 @@ +"""Boot-epoch staleness signal on the node-global SSE stream (#881). + +The global ring buffer and its event counter are process-local: unlike the +per-ws lane (whose counter ``_seed_event_id_from_storage`` seeds from +``MAX(conversations.event_id)``), ``global_event_id_holder`` reboots at 0 +with the process. A bare integer cursor therefore cannot prove which boot +minted it — after a restart it is first "ahead" of the reborn ring (empty +replay slice) and then aliases into the new id space as the counter +re-grows, both of which used to draw ``replay_ok`` and silently skip the +restart boundary. + +The fix stamps every global SSE ``id:`` as ``"{boot_epoch}-{counter}"`` +and treats any cursor that does not carry the live epoch as stale, +answering ``replay_truncated`` (``reason="boot_epoch"``) plus the +``node_snapshot`` recovery floor. These tests walk the reconnect matrix +at the :func:`turnstone.server.global_events_sse` boundary: + + cursor ∈ {absent, same-epoch, stale-epoch, legacy bare-int, garbage, + negative, forged-against-empty-ring} + × ring ∈ {empty, covers cursor, evicted past cursor} + +The browser half (app.js capture/presentation) is pinned in +``test_app_js.py``; the end-to-end restart loop is +``scripts/recovery_e2e.py --scenario roster-restart``. +""" + +from __future__ import annotations + +import asyncio +import collections +import json +import queue +import threading +from types import SimpleNamespace as SimpleNS +from typing import Any + +import pytest +from starlette.requests import Request + +import turnstone.server as server_mod + +EPOCH = "0badf00d" # test-pinned boot epoch (hex, like secrets.token_hex(4)) + + +def _make_app_state( + *, + buffered: list[tuple[int, dict[str, Any]]] | None = None, + epoch: str = EPOCH, +) -> SimpleNS: + """Minimal ``app.state`` for the global SSE handler. + + Real lock / deque / list so registration and slicing run the + production code paths; only the snapshot builder is stubbed (its + composition has its own coverage in ``test_console.py``). + """ + buf: collections.deque[tuple[int, dict[str, Any]]] = collections.deque(maxlen=50) + for item in buffered or []: + buf.append(item) + return SimpleNS( + node_id="node-under-test", + global_listeners=[], + global_listeners_lock=threading.Lock(), + global_event_buffer=buf, + global_boot_epoch=epoch, + sse_executor=None, # live loop is never entered by these tests + ) + + +def _fake_request( + app_state: SimpleNS, + *, + headers: dict[str, str] | None = None, + query: dict[str, str] | None = None, +) -> Request: + """ASGI-scope-honest request carrying a service-scoped principal.""" + header_list = [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()] + query_string = "&".join(f"{k}={v}" for k, v in query.items()).encode() if query else b"" + scope = { + "type": "http", + "method": "GET", + "headers": header_list, + "path": "/v1/api/events/global", + "raw_path": b"/v1/api/events/global", + "query_string": query_string, + "app": SimpleNS(state=app_state), + "state": {"auth_result": SimpleNS(scopes=["service"])}, + } + + async def _recv() -> dict[str, Any]: # noqa: RUF029 — async signature required + return {"type": "http.disconnect"} + + return Request(scope, receive=_recv) + + +_SNAPSHOT_STUB = {"type": "node_snapshot", "node_id": "node-under-test", "workstreams": []} + + +def _drain( + monkeypatch: pytest.MonkeyPatch, + *, + buffered: list[tuple[int, dict[str, Any]]] | None = None, + headers: dict[str, str] | None = None, + query: dict[str, str] | None = None, + max_yields: int = 8, + epoch: str = EPOCH, +) -> list[dict[str, Any]]: + """Run the handler and collect its pre-live yields as raw dicts. + + ``max_yields`` must not exceed the pre-live yield count for the + branch under test + 1 — the live loop blocks on an executor draw, + so every test enumerates its expected frames and stops short. + """ + monkeypatch.setattr(server_mod, "_build_node_snapshot", lambda _s: dict(_SNAPSHOT_STUB)) + app_state = _make_app_state(buffered=buffered, epoch=epoch) + req = _fake_request(app_state, headers=headers, query=query) + + async def _run() -> list[dict[str, Any]]: + # Guard against a miscounted ``max_yields`` reaching the live + # loop (which blocks on an executor draw and would hang the + # test + leak the draw thread) — fail fast and visibly instead. + async with asyncio.timeout(10): + # ``Any``: the endpoint is annotated ``-> Response``; the SSE + # subtype's ``body_iterator`` is what the drain consumes. + resp: Any = await server_mod.global_events_sse(req) + out: list[dict[str, Any]] = [] + async for chunk in resp.body_iterator: + out.append(chunk) + if len(out) >= max_yields: + break + await resp.body_iterator.aclose() + return out + + return asyncio.run(_run()) + + +def _data_frames(yields: list[dict[str, Any]]) -> list[dict[str, Any]]: + """JSON-decoded ``data:`` payloads, in yield order.""" + return [json.loads(y["data"]) for y in yields if "data" in y] + + +def _types(yields: list[dict[str, Any]]) -> list[str]: + return [d.get("type", "") for d in _data_frames(yields)] + + +def _ev(eid: int, **extra: Any) -> tuple[int, dict[str, Any]]: + """A buffered ring entry the way ``_global_fanout_thread`` stores it: + the event dict carries ``_event_id`` and the tuple repeats the id.""" + return eid, {"type": "ws_state", "ws_id": f"ws-{eid}", "_event_id": eid, **extra} + + +# --------------------------------------------------------------------------- +# Fresh connect (no cursor) +# --------------------------------------------------------------------------- + + +def test_fresh_connect_gets_snapshot_no_envelope(monkeypatch: pytest.MonkeyPatch) -> None: + yields = _drain(monkeypatch, buffered=[_ev(1)], max_yields=2) + assert "retry" in yields[0] + assert _types(yields) == ["node_snapshot"] + + +# --------------------------------------------------------------------------- +# Same-epoch cursors — ring logic must behave exactly as before +# --------------------------------------------------------------------------- + + +def test_same_epoch_cursor_replays_slice_without_snapshot( + monkeypatch: pytest.MonkeyPatch, +) -> None: + yields = _drain( + monkeypatch, + buffered=[_ev(1), _ev(2), _ev(3)], + headers={"Last-Event-ID": f"{EPOCH}-1"}, + max_yields=3, + ) + frames = _data_frames(yields) + assert [f["ws_id"] for f in frames] == ["ws-2", "ws-3"] + assert "node_snapshot" not in _types(yields) + assert "replay_truncated" not in _types(yields) + + +def test_replayed_and_live_ids_are_epoch_tagged(monkeypatch: pytest.MonkeyPatch) -> None: + """Every ``id:`` on the wire carries the live epoch — the browser + echoes it verbatim on native reconnect, which is what lets the + server prove cursor provenance without any client cooperation.""" + yields = _drain( + monkeypatch, + buffered=[_ev(1), _ev(2)], + headers={"Last-Event-ID": f"{EPOCH}-0"}, + max_yields=3, + ) + ids = [y["id"] for y in yields if "id" in y] + assert ids == [f"{EPOCH}-1", f"{EPOCH}-2"] + # And the internal ``_event_id`` never leaks onto the wire. + assert all("_event_id" not in f for f in _data_frames(yields)) + + +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) == [] + + +def test_same_epoch_ring_miss_is_truncated_with_honest_counts( + monkeypatch: pytest.MonkeyPatch, +) -> None: + yields = _drain( + monkeypatch, + buffered=[_ev(10), _ev(11)], + headers={"Last-Event-ID": f"{EPOCH}-3"}, + max_yields=3, + ) + envelope, snapshot = _data_frames(yields) + assert envelope["type"] == "replay_truncated" + assert envelope["reason"] == "ring_evicted" + assert envelope["lost_count"] == 6 # ids 4..9 died with the ring + assert envelope["earliest_available_id"] == 10 + assert snapshot["type"] == "node_snapshot" + + +# --------------------------------------------------------------------------- +# Stale / foreign cursors — the #881 class +# --------------------------------------------------------------------------- + + +def _assert_boot_epoch_truncated(yields: list[dict[str, Any]]) -> None: + """Envelope (reason=boot_epoch, no invented counts) then snapshot, + and no replay slice leaked around them.""" + frames = _data_frames(yields) + assert [f["type"] for f in frames] == ["replay_truncated", "node_snapshot"] + envelope = frames[0] + assert envelope["reason"] == "boot_epoch" + assert "lost_count" not in envelope + assert "earliest_available_id" not in envelope + + +def test_prior_boot_cursor_against_regrown_ring_is_truncated( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """THE restart-aliasing case: the reborn counter has re-grown past + the stale cursor, so pre-#881 the ring sliced from it as if the ids + were contiguous across the boot — silently skipping the restart + boundary. The epoch mismatch must win over the plausible slice.""" + yields = _drain( + monkeypatch, + buffered=[_ev(1), _ev(2), _ev(3)], + headers={"Last-Event-ID": "deadbeef-2"}, + max_yields=3, + ) + _assert_boot_epoch_truncated(yields) + + +def test_legacy_bare_int_cursor_on_empty_ring_is_truncated( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The original #881 lie: empty reborn ring + pre-restart cursor + used to answer ``replay_ok`` with nothing. A bare-int cursor (no + epoch half — pre-#881 client or prior-boot native echo) must draw + the truncated floor instead of the silent gap.""" + yields = _drain( + monkeypatch, + buffered=[], + headers={"Last-Event-ID": "42"}, + max_yields=3, + ) + _assert_boot_epoch_truncated(yields) + + +def test_legacy_bare_int_cursor_on_live_ring_is_truncated( + monkeypatch: pytest.MonkeyPatch, +) -> None: + yields = _drain( + monkeypatch, + buffered=[_ev(1), _ev(2)], + headers={"Last-Event-ID": "1"}, + max_yields=3, + ) + _assert_boot_epoch_truncated(yields) + + +def test_same_epoch_cursor_against_empty_ring_fails_safe( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Same-epoch ids imply a non-empty ring (single append site, first + id 1, never cleared) — a same-epoch cursor over an empty ring is + forged or a bug, and must fail to the snapshot floor rather than + resurrect the silent-gap shape.""" + yields = _drain( + monkeypatch, + buffered=[], + headers={"Last-Event-ID": f"{EPOCH}-5"}, + max_yields=3, + ) + _assert_boot_epoch_truncated(yields) + + +@pytest.mark.parametrize( + "cursor", + [ + "not-a-cursor", # wrong epoch, non-numeric counter + f"{EPOCH}-xyz", # live epoch, garbage counter + f"{EPOCH}--5", # live epoch, negative (forged) counter + "-", # empty epoch, empty counter + ], +) +def test_unusable_cursor_shapes_all_draw_the_truncated_floor( + monkeypatch: pytest.MonkeyPatch, cursor: str +) -> None: + """A present-but-unusable cursor must NOT fall back to ``fresh``: + fresh is the no-loss shape, and these callers provably lost events. + (Pre-#881 the unparseable arm silently became fresh — the semantic + shift is deliberate and this test pins it.)""" + yields = _drain( + monkeypatch, + buffered=[_ev(1)], + headers={"Last-Event-ID": cursor}, + max_yields=3, + ) + _assert_boot_epoch_truncated(yields) + + +# --------------------------------------------------------------------------- +# Transport details +# --------------------------------------------------------------------------- + + +def test_query_param_fallback_matches_header_semantics( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Manual browser reconnects can't set headers on ``new + EventSource(url)`` — the ``?last_event_id=`` fallback must run the + same epoch logic (app.js presents its stored cursor this way).""" + yields = _drain( + monkeypatch, + buffered=[_ev(1), _ev(2)], + query={"last_event_id": f"{EPOCH}-1"}, + max_yields=2, + ) + frames = _data_frames(yields) + assert [f["ws_id"] for f in frames] == ["ws-2"] + + yields = _drain( + monkeypatch, + buffered=[_ev(1), _ev(2)], + query={"last_event_id": "deadbeef-1"}, + max_yields=3, + ) + _assert_boot_epoch_truncated(yields) + + +def test_epoch_is_hex_and_dashless_by_construction() -> None: + """The parse splits on the FIRST ``-``; the epoch half must never + contain one. ``secrets.token_hex`` guarantees pure hex — this pin + exists so a future 'readable epoch' refactor (timestamps, uuids + with dashes) fails here instead of corrupting cursor parsing.""" + import secrets + + for _ in range(64): + assert "-" not in secrets.token_hex(4) + # And the production init uses token_hex — source-level pin. + import inspect + + src = inspect.getsource(server_mod) + assert "app.state.global_boot_epoch = secrets.token_hex(" in src + + +def test_listener_registered_exactly_once_per_connect( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Registration is atomic with the replay decision for every branch + (stale cursors included) — the truncated floor must not skip or + double the listener append.""" + monkeypatch.setattr(server_mod, "_build_node_snapshot", lambda _s: dict(_SNAPSHOT_STUB)) + app_state = _make_app_state(buffered=[_ev(1)]) + req = _fake_request(app_state, headers={"Last-Event-ID": "deadbeef-1"}) + + async def _run() -> None: + resp: Any = await server_mod.global_events_sse(req) + assert len(app_state.global_listeners) == 1 + assert isinstance(app_state.global_listeners[0], queue.Queue) + await resp.body_iterator.aclose() + + asyncio.run(_run()) diff --git a/turnstone/api/server_spec.py b/turnstone/api/server_spec.py index 8df956ae..71dc53a0 100644 --- a/turnstone/api/server_spec.py +++ b/turnstone/api/server_spec.py @@ -186,7 +186,13 @@ SERVER_ENDPOINTS: list[EndpointSpec] = [ "Emits a node_snapshot event on connect (workstreams, health, aggregate), " "followed by real-time delta events (ws_state, ws_activity, ws_created, " "ws_closed, ws_rename, health_changed, aggregate). " - "Pass ?expected_node_id=X for identity verification (returns 409 on mismatch).", + "Pass ?expected_node_id=X for identity verification (returns 409 on mismatch). " + "Every event's SSE id is an opaque '{boot_epoch}-{counter}' string; presenting " + "it on reconnect (Last-Event-ID header or ?last_event_id=) replays missed " + "events, or emits a replay_truncated event (reason: ring_evicted with " + "lost_count + earliest_available_id, or boot_epoch when the cursor predates " + "this server process) followed by a fresh node_snapshot. Treat the id as " + "opaque — its format may change.", tags=["Streaming"], ), EndpointSpec( diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index 7a3330b2..1720d337 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -277,7 +277,20 @@ class ClusterCollector: await task async def _node_sse_task(self, node_id: str, stop_event: asyncio.Event) -> None: - """Persistent SSE connection to a single server node.""" + """Persistent SSE connection to a single server node. + + DELIBERATELY cursorless (#881 audit ruling): every (re)connect is + fresh — no ``Last-Event-ID``, so the node answers with a full + ``node_snapshot`` and the collector rebuilds wholesale. That is + this consumer's recovery model (idempotent state-of-world, not + append-only history), it side-steps cursor staleness across node + restarts entirely, and it means the epoch-tagged ids and the + ``replay_truncated`` envelope on this stream can never fire here. + If a cursor is ever adopted, present the SSE ``id:`` VERBATIM + (opaque ``"{boot_epoch}-{counter}"`` — never parse it) and handle + ``replay_truncated`` explicitly; today an unknown event type + no-ops in ``_apply_delta`` by design. + """ backoff = 1.0 while not stop_event.is_set() and self._running: url = self._get_node_url(node_id) diff --git a/turnstone/server.py b/turnstone/server.py index b4594bf4..e0f52af4 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -22,6 +22,7 @@ import os import queue import random import re +import secrets import sys import textwrap import threading @@ -1009,6 +1010,19 @@ async def global_events_sse(request: Request) -> Response: On connect, emits a ``node_snapshot`` event with the full node state (workstreams, health, aggregate) followed by real-time delta events. The snapshot and listener registration are atomic — no events are lost. + + Resume contract (#881): every event's SSE ``id:`` is + ``"{boot_epoch}-{counter}"``, where ``boot_epoch`` is a per-process + nonce and ``counter`` is the process-local monotonic event id. A + reconnect presenting a cursor (``Last-Event-ID`` header or + ``?last_event_id=``) whose epoch matches gets the ring replay + (``replay_ok`` past the counter, or a ``replay_truncated`` envelope + with ``reason="ring_evicted"`` when the ring has moved on); a cursor + from any other epoch — a prior boot, another node, a pre-#881 + bare-int client — gets ``replay_truncated`` with + ``reason="boot_epoch"`` followed by a fresh ``node_snapshot``, since + the events it missed died with the process that minted it. Clients + treat the cursor as an opaque string; only this handler parses it. """ # -- Service-scope gate --------------------------------------------------- # The global stream carries cluster-wide workstream inventory across @@ -1036,14 +1050,44 @@ async def global_events_sse(request: Request) -> Response: # Native EventSource sets the header on auto-reconnect; the # manual-reconnect path (which can't set custom headers on # ``new EventSource(url)``) uses the query-param fallback. + # + # Global ids are ``"{boot_epoch}-{counter}"`` (#881): the epoch half + # proves the cursor was minted by THIS process. The counter reboots + # at 0 with the process, so without the epoch a pre-restart cursor is + # first invisibly "ahead" of the reborn ring (empty replay slice) and + # then, as the counter re-grows past it, aliases into the new id + # space — both draw ``replay_ok`` and silently skip the restart + # boundary. Parse outcomes: + # - no cursor → ``last_event_id = None`` (fresh); + # - epoch matches → ``last_event_id = counter`` (ring logic); + # - anything else → ``cursor_stale = True`` (a cursor was + # presented but is unusable: prior-boot epoch, another node's + # epoch, a bare-int cursor from a pre-#881 client, or garbage) + # → the ``replay_truncated`` + node_snapshot floor below. A + # present-but-unusable cursor must NOT fall back to ``fresh``: + # fresh is the no-loss shape, and this caller has provably + # lost the events between its cursor and this boot. + boot_epoch: str = request.app.state.global_boot_epoch last_event_id_raw = request.headers.get("Last-Event-ID") or request.query_params.get( "last_event_id" ) - last_event_id: int | None - try: - last_event_id = int(last_event_id_raw) if last_event_id_raw else None - except (TypeError, ValueError): - last_event_id = None + last_event_id: int | None = None + cursor_stale = False + if last_event_id_raw: + cursor_epoch, sep, counter_raw = last_event_id_raw.partition("-") + if sep and cursor_epoch == boot_epoch: + try: + last_event_id = int(counter_raw) + except ValueError: + cursor_stale = True + else: + if last_event_id < 0: + # Our counters start at 1; a negative counter can + # only be forged. Same floor as garbage. + last_event_id = None + cursor_stale = True + else: + cursor_stale = True # -- Atomic snapshot / replay-slice + listener registration --------------- client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=1000) @@ -1053,40 +1097,54 @@ async def global_events_sse(request: Request) -> Response: request.app.state.global_event_buffer ) - # Three replay shapes, matching :func:`make_events_handler`: + # Four replay shapes (the first three matching :func:`make_events_handler`): # - ``last_event_id is None`` → ``"fresh"``: emit node_snapshot # then live. # - ``last_event_id`` + buffer covers gap → ``"replay_ok"``: # emit buffered events past the id, SKIP node_snapshot, then # live. - # - ``last_event_id`` + buffer too short → ``"truncated"``: emit - # a ``replay_truncated`` envelope then fall through to - # ``"fresh"`` (node_snapshot is the recovery floor). + # - ``last_event_id`` + buffer too short → ``"truncated"`` + # (``reason="ring_evicted"``, honest ``lost_count``): emit a + # ``replay_truncated`` envelope then fall through to ``"fresh"`` + # (node_snapshot is the recovery floor). + # - ``cursor_stale`` (#881) → ``"truncated"`` + # (``reason="boot_epoch"``): the cursor is from another boot — + # the prior ring died with its process, so what was lost is + # unknowable; the envelope omits ``lost_count`` / + # ``earliest_available_id`` rather than guess. The per-ws + # stream never needs this arm: its counter is storage-seeded + # across restarts, so plain counter comparison detects staleness + # there (see ``register_listener_with_replay``); the global + # counter is process-local and reboots at 0, hence the epoch. replay_status: str + truncated_reason = "ring_evicted" replay_events: list[dict[str, Any]] = [] lost_count = 0 earliest_available_id = 0 snapshot: dict[str, Any] | None = None with listeners_lock: - if last_event_id is None: + if cursor_stale: + replay_status = "truncated" + truncated_reason = "boot_epoch" + snapshot = _build_node_snapshot(request.app.state) + elif last_event_id is None: replay_status = "fresh" snapshot = _build_node_snapshot(request.app.state) else: buffered = list(event_buffer) if not buffered: - # KNOWN GAP (#881): an empty ring after a node restart - # reports ``replay_ok`` even when the client's cursor is - # stale, silently skipping the events lost with the old - # process. The per-ws stream fixes this by deriving - # staleness from its storage-seeded ``_event_id`` counter - # (see ``register_listener_with_replay``); the global - # counter (``global_event_id_holder``) resets to 0 at - # boot, so the same rule cannot apply — a boot-epoch - # staleness signal is the fix direction tracked in #881. - # Tolerable meanwhile: consumers are idempotent roster - # state refreshed periodically, not append-only history. - replay_status = "replay_ok" + # Same-epoch cursor against an empty ring is impossible + # by construction (ids are only minted by the fanout + # thread's single append site, first id 1, and the ring + # is never cleared — an id implies a non-empty ring), so + # a cursor landing here is forged or a bug. Fail SAFE + # to the truncated floor: the snapshot rebuild is + # idempotent, whereas trusting the cursor would replay + # nothing and ghost the roster (#881's original lie). + replay_status = "truncated" + truncated_reason = "boot_epoch" + snapshot = _build_node_snapshot(request.app.state) else: earliest_available_id = buffered[0][0] if last_event_id < earliest_available_id - 1: @@ -1098,16 +1156,40 @@ 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 == "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], + ) + async def event_generator() -> AsyncGenerator[dict[str, Any], None]: _metrics.record_sse_connect() def _format_event(event: dict[str, Any]) -> dict[str, str]: - """Strip ``_event_id`` from the wire dict, attach SSE ``id:``.""" + """Strip ``_event_id`` from the wire dict, attach SSE ``id:``. + + The id is ``"{boot_epoch}-{counter}"`` — GLOBAL-LANE ONLY + (#881). Do not propagate the epoch tag to the per-ws + formatter in ``make_events_handler``: per-ws ids must stay + bare ints — their counter is storage-seeded across restarts + (epoch unnecessary), clients seed cursors from ``/history``'s + integer ``cursor``, and coordinator.js's counter-reset + detector does ``Number(event.lastEventId)`` comparisons that + an epoch prefix would turn into ``NaN`` and silently kill. + """ ev_copy = dict(event) eid = ev_copy.pop("_event_id", None) out: dict[str, str] = {"data": json.dumps(ev_copy)} if eid is not None: - out["id"] = str(eid) + out["id"] = f"{boot_epoch}-{eid}" return out try: @@ -1117,15 +1199,23 @@ async def global_events_sse(request: Request) -> Response: yield {"retry": random.randint(2500, 4500)} if replay_status == "truncated": - yield { - "data": json.dumps( - { - "type": "replay_truncated", - "lost_count": lost_count, - "earliest_available_id": earliest_available_id, - } - ) + # ``reason`` distinguishes an in-epoch ring overrun + # (``ring_evicted`` — ``lost_count`` / + # ``earliest_available_id`` are honest) from a + # cross-boot cursor (``boot_epoch`` — the prior ring + # died with its process, so the loss is unknowable and + # the numeric fields are OMITTED rather than invented). + # No first-party consumer reads the numeric fields + # (app.js reads only ``type`` and resyncs), so the + # conditional shape is wire-safe. + envelope: dict[str, Any] = { + "type": "replay_truncated", + "reason": truncated_reason, } + if truncated_reason == "ring_evicted": + envelope["lost_count"] = lost_count + envelope["earliest_available_id"] = earliest_available_id + yield {"data": json.dumps(envelope)} if replay_status == "replay_ok": for ev in replay_events: yield _format_event(ev) @@ -4972,6 +5062,18 @@ def create_app( # Single-element list as a mutable int holder so the fanout # thread can ``counter_holder[0] += 1`` under the lock. app.state.global_event_id_holder = [0] + # Per-process boot epoch for the global SSE lane (#881). The + # counter above reboots at 0 with the process (unlike the per-ws + # counters, which ``_seed_event_id_from_storage`` seeds from + # ``MAX(conversations.event_id)``), so a bare integer cursor from a + # prior boot is indistinguishable from — and eventually aliases + # into — the new id space. Every global SSE ``id:`` is therefore + # stamped ``"{epoch}-{counter}"``; a reconnect cursor whose epoch + # differs from the live process is stale by construction and draws + # the ``replay_truncated`` + node_snapshot recovery floor in + # :func:`global_events_sse`. Hex nonce (never contains ``-``), so + # ``partition("-")`` splits the id unambiguously. + app.state.global_boot_epoch = secrets.token_hex(4) app.state.skip_permissions = skip_permissions app.state.jwt_secret = jwt_secret app.state.auth_storage = auth_storage