From 6b6c8eb26331d3f5a789f0e8354d05a63b8e383f Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 22 May 2026 01:41:22 -0700 Subject: [PATCH] feat(console): forward Last-Event-ID through SSE proxy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The console SSE proxy (`_proxy_sse`) is the inbound SSE path for multi-node deployments — every browser EventSource that targets a per-node route traverses it. Today's proxy strips client request headers (only `Accept`, `Cache-Control`, and the re-minted auth token make it upstream), so the per-ws / global SSE handlers' `Last-Event-ID` resume (PR-D commit 1) never sees the header in the multi-node shape — every reconnect would be a fresh connect and silently drop events from the disconnect window. Builds the upstream headers dict conditionally: copy `Last-Event-ID` from the incoming request when present, omit otherwise (no fabricated value on fresh connects). Starlette's header dict is case-insensitive so the `request.headers.get("last-event-id")` lookup catches both the spec-recommended capitalization and any intermediary normalisation. The query-param fallback (`?last_event_id=N`) needs no proxy change — `request.url.query` is already forwarded verbatim at the top of the function. Tests in `tests/test_service_auth_boundary.py::TestProxySseLastEventIdForwarding`: - Positive: browser header → upstream header (value preserved). - Negative: browser sends nothing → upstream gets nothing (no fabricated value). --- tests/test_service_auth_boundary.py | 102 ++++++++++++++++++++++++++++ turnstone/console/server.py | 21 +++++- 2 files changed, 122 insertions(+), 1 deletion(-) diff --git a/tests/test_service_auth_boundary.py b/tests/test_service_auth_boundary.py index 6544ebda..7ada00d0 100644 --- a/tests/test_service_auth_boundary.py +++ b/tests/test_service_auth_boundary.py @@ -440,6 +440,108 @@ class TestProxySseNon200LogLevel: assert "\n" not in matches[0].getMessage().split("body=", 1)[-1] +# --------------------------------------------------------------------------- +# _proxy_sse — Last-Event-ID forwarding (PR-D reconnect-with-replay) +# --------------------------------------------------------------------------- + + +class TestProxySseLastEventIdForwarding: + """The console SSE proxy is the inbound SSE path for multi-node + deployments — every browser EventSource traverses it. Without + forwarding ``Last-Event-ID``, the per-ws / global SSE handlers on + the node would treat every reconnect as a fresh connect and silently + drop events emitted during the disconnect window. PR-D's whole + reconnect-with-replay foundation depends on these tests passing.""" + + @pytest.mark.anyio + async def test_forwards_last_event_id_header_to_upstream(self): + """Browser sends ``Last-Event-ID``; upstream node must receive it.""" + from starlette.requests import Request + + from turnstone.console.server import _proxy_sse + + captured_headers: dict[str, str] = {} + + def handler(req: httpx.Request) -> httpx.Response: + # httpx headers are case-insensitive; capture lowercased. + captured_headers.update({k.lower(): v for k, v in req.headers.items()}) + return httpx.Response( + 200, text="data: {}\n\n", headers={"content-type": "text/event-stream"} + ) + + sse_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + proxy_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + scope = { + "type": "http", + "method": "GET", + "path": "/node/n/api/workstreams/ws-1/events", + "headers": [(b"last-event-id", b"42")], + "query_string": b"", + "app": MagicMock( + state=SimpleNamespace(proxy_sse_client=sse_client, proxy_client=proxy_client) + ), + } + + async def _receive(): + return {"type": "http.request", "body": b""} + + request = Request(scope, receive=_receive) + response = await _proxy_sse( + request, "http://node-1:8001", "workstreams/ws-1/events", api_prefix="api" + ) + # Drain so the upstream call actually fires. + async for _ in response.body_iterator: # type: ignore[attr-defined] + pass + + assert captured_headers.get("last-event-id") == "42", ( + f"Last-Event-ID not forwarded to upstream; got headers={captured_headers!r}" + ) + + @pytest.mark.anyio + async def test_omits_last_event_id_when_client_did_not_send_one(self): + """Fresh connect (no header on the browser side) → no header + added on the upstream side either. Guards against + accidentally injecting a stale or fabricated value.""" + from starlette.requests import Request + + from turnstone.console.server import _proxy_sse + + captured_headers: dict[str, str] = {} + + def handler(req: httpx.Request) -> httpx.Response: + captured_headers.update({k.lower(): v for k, v in req.headers.items()}) + return httpx.Response( + 200, text="data: {}\n\n", headers={"content-type": "text/event-stream"} + ) + + sse_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + proxy_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + scope = { + "type": "http", + "method": "GET", + "path": "/node/n/api/workstreams/ws-1/events", + "headers": [], + "query_string": b"", + "app": MagicMock( + state=SimpleNamespace(proxy_sse_client=sse_client, proxy_client=proxy_client) + ), + } + + async def _receive(): + return {"type": "http.request", "body": b""} + + request = Request(scope, receive=_receive) + response = await _proxy_sse( + request, "http://node-1:8001", "workstreams/ws-1/events", api_prefix="api" + ) + async for _ in response.body_iterator: # type: ignore[attr-defined] + pass + + assert "last-event-id" not in captured_headers, ( + f"upstream got an unexpected Last-Event-ID; headers={captured_headers!r}" + ) + + # --------------------------------------------------------------------------- # Gated cluster_events_sse — 503 on scope error (0a) # --------------------------------------------------------------------------- diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 4a12b07d..dc707e0b 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -2866,12 +2866,31 @@ async def _proxy_sse( else: sse_auth = _proxy_auth_headers(request) + # Forward ``Last-Event-ID`` from the browser to the upstream node so + # the per-ws / global SSE handlers can serve the reconnect-with-replay + # buffer slice. Without this, multi-node deployments lose replay + # entirely (the proxy is the only inbound SSE path in that shape); + # the per-ws handler would treat every reconnect as a fresh connect + # and silently drop events emitted during the disconnect window. + # The query-param fallback (``?last_event_id=N``) is already + # forwarded via the existing ``request.url.query`` propagation at + # the top of this function — only the header needs an explicit + # carry-over. Header lookups in Starlette are case-insensitive. + upstream_headers: dict[str, str] = { + **sse_auth, + "Accept": "text/event-stream", + "Cache-Control": "no-store", + } + last_event_id_hdr = request.headers.get("last-event-id") + if last_event_id_hdr is not None: + upstream_headers["Last-Event-ID"] = last_event_id_hdr + async def raw_stream() -> AsyncGenerator[bytes, None]: try: async with sse_client.stream( "GET", target, - headers={**sse_auth, "Accept": "text/event-stream", "Cache-Control": "no-store"}, + headers=upstream_headers, timeout=httpx.Timeout(connect=10, read=None, write=5, pool=None), ) as response: if response.status_code != 200: