Backfill the flowing lanes' activity buffer on relay reconnect

infra/status-relay/relay.py

@@ -10,14 +10,16 @@
 run with nothing but the standard library plus `websockets` installed,
 independent of the CORA deployment's own environment.
 
-Holds the single most-recent snapshot AND a bounded ring of up to
-`_RUN_HISTORY_CACHE_SIZE` pushed run histories, both in process-local
-memory only. It is not a database and is not meant to be one: nothing
-about the beamline persists to disk here, so a relay restart has no
-retention question, just a smaller blast radius than before -- a
-compromise of this box now exposes the last snapshot (~20 KB) AND up to
-20 cached run histories, still never anything not already pushed to it,
-still never anything on disk.
+Holds the single most-recent snapshot, a bounded ring of up to
+`_RUN_HISTORY_CACHE_SIZE` pushed run histories, every enclosure timeline
+ever received, and the last `_ACTIVITY_BUFFER_SECONDS` of activity
+events, all in process-local memory only. It is not a database and is
+not meant to be one: nothing about the beamline persists to disk here,
+so a relay restart has no retention question, just a smaller blast
+radius than before -- a compromise of this box now exposes the last
+snapshot (~20 KB), up to 20 cached run histories, every enclosure's
+permit/lifecycle history, and a few minutes of event metadata, still
+never anything not already pushed to it, still never anything on disk.
 
 Six endpoints, one port, one library (`websockets`' `process_request`
 hook answers plain HTTP so a WebSocket-only library can still serve the
@@ -55,18 +57,21 @@
                               not a human viewer)
   - `WS  /watch`              a browser connects here; sent the current
                               snapshot (or a "no producer yet" state), the
-                              run-history index, and every cached enclosure
-                              timeline immediately on connect, then every
-                              subsequent snapshot, run-history index
-                              update, enclosure-timeline update, and
-                              producer connect / disconnect transition,
-                              live. Enclosure timelines are pure
-                              pass-through PLUS a replay cache (unlike
-                              run history's cache-or-ask-the-producer
-                              shape): at pilot scale there are only a
-                              handful of enclosures, so the whole set
-                              fits in memory with no eviction and no
-                              on-demand request path is needed at all
+                              run-history index, every cached enclosure
+                              timeline, and a backfill of the last
+                              `_ACTIVITY_BUFFER_SECONDS` of activity events
+                              immediately on connect, then every subsequent
+                              snapshot, run-history index update,
+                              enclosure-timeline update, and activity
+                              event, live, plus producer connect /
+                              disconnect transitions. Enclosure timelines
+                              and activity are pure pass-through PLUS a
+                              replay cache (unlike run history's
+                              cache-or-ask-the-producer shape): at pilot
+                              scale there are only a handful of enclosures
+                              and a bounded few minutes of activity, so
+                              both fit in memory with no on-demand request
+                              path needed at all
 
 Run: `STATUS_RELAY_TOKEN=<token> STATUS_RELAY_VIEWER_USER=<user>
 STATUS_RELAY_VIEWER_PASSWORD=<password> python relay.py [--host 0.0.0.0]
@@ -82,6 +87,7 @@
 import os
 import sys
 from collections import OrderedDict
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 from uuid import UUID, uuid4
@@ -125,6 +131,17 @@
 Exceeding `open_timeout` aborts the TCP handshake with no HTTP response
 at all, turning a legitimate 504 into an unexplained `Failed to fetch`."""
 
+_ACTIVITY_BUFFER_SECONDS = 15 * 60
+"""How long this relay backfills a freshly-connecting watcher with recent
+`"activity"` events, mirroring `page.html`'s own `FLOWING_WINDOW_MS`: a
+separate literal, not a shared constant, since this relay imports nothing
+from `cora` and `page.html` is served verbatim with no build step either
+(see the module docstring). Keeping the two in sync matters only in the
+direction that this value should be >= the browser's own window --
+buffering less would leave a visible gap on reconnect, buffering more is
+harmless since the browser prunes anything older than its own window on
+receipt (`pruneFlowingBuffer`)."""
+
 _PAGE_PATH = Path(__file__).parent / "page.html"
 _SCRUBBER_JS_PATH = Path(__file__).parent / "scrubber.js"
 
@@ -160,6 +177,16 @@
 reaching any of them needs no producer round trip the way reaching any
 RUN does -- see `cora.api._status_push._EnclosureTimelineTail`'s own
 docstring for why."""
+_activity_buffer: list[dict[str, Any]] = []
+"""Individual events (not whole `"activity"` messages) from the last
+`_ACTIVITY_BUFFER_SECONDS`, flattened across however many producer
+messages they arrived in, pruned by `occurred_at` on every new arrival
+and again right before a replay. Exists purely so a watcher that connects
+mid-quiet-period still gets caught up: before this, flowing mode was a
+pure pass-through with no way for a fresh connection (or a browser
+refresh, or a resumed SSH tunnel) to see anything that happened before it
+attached, unlike the snapshot and enclosure-timeline rings which already
+replay on connect."""
 
 
 def _require_token() -> str:
@@ -239,6 +266,52 @@ def _store_enclosure_timeline(message: dict[str, Any]) -> None:
     _enclosure_timelines[enclosure_id] = message
 
 
+def _prune_activity_buffer() -> None:
+    cutoff = datetime.now(UTC).timestamp() - _ACTIVITY_BUFFER_SECONDS
+    global _activity_buffer  # noqa: PLW0603
+    _activity_buffer = [event for event in _activity_buffer if _event_epoch(event) >= cutoff]
+
+
+def _event_epoch(event: dict[str, Any]) -> float:
+    """`occurred_at` as a Unix timestamp, or `-inf` for a malformed/missing
+    one so it prunes out on the next pass rather than wedging the buffer
+    open forever."""
+    occurred_at = event.get("occurred_at")
+    if not isinstance(occurred_at, str):
+        return float("-inf")
+    try:
+        return datetime.fromisoformat(occurred_at).timestamp()
+    except ValueError:
+        return float("-inf")
+
+
+def _store_activity_events(message: dict[str, Any]) -> None:
+    events = message.get("events")
+    if not isinstance(events, list):
+        _log.warning("producer.malformed_activity")
+        return
+    _activity_buffer.extend(events)
+    _prune_activity_buffer()
+
+
+def _activity_replay_message() -> dict[str, Any] | None:
+    """One synthetic `"activity"` message backfilling a freshly-connecting
+    watcher, in the exact shape `build_activity_message` already produces
+    (`page.html`'s `handleActivity` just concatenates `events`, so it needs
+    no replay-specific handling). `None` when the buffer is empty, mirroring
+    the producer's own "never sent when rows is empty" convention."""
+    _prune_activity_buffer()
+    if not _activity_buffer:
+        return None
+    return {
+        "kind": "activity",
+        "schema_version": 1,
+        "producer_id": _producer_id,
+        "generated_at": datetime.now(UTC).isoformat(),
+        "events": list(_activity_buffer),
+    }
+
+
 def _fail_all_pending(reason: str) -> None:
     """Resolve every in-flight `run_history_request` with an exception
     immediately, rather than letting each one sit until its own
@@ -300,11 +373,12 @@ async def _handle_producer(ws: ServerConnection) -> None:
                 if _watchers:
                     websockets.broadcast(_watchers, message)
             elif kind == "activity":
-                # No relay-side cache, unlike snapshot/run_history: flowing
-                # mode's rolling window lives in each browser, not here, so
-                # this is a pure pass-through. A watcher that connects mid-
-                # flow simply starts receiving from that point on, same as
-                # it already does for the live tables.
+                # Buffered (`_store_activity_events`) AND passed through
+                # live: a watcher connecting mid-flow gets the buffer as a
+                # replay in `_handle_watcher`, then this same broadcast
+                # keeps it current, same as it already does for the live
+                # tables.
+                _store_activity_events(payload)
                 if _watchers:
                     websockets.broadcast(_watchers, message)
             elif kind == "run_history_response":
@@ -344,6 +418,9 @@ async def _handle_watcher(ws: ServerConnection) -> None:
         await ws.send(json.dumps(_run_history_index()))
         for message in _enclosure_timelines.values():
             await ws.send(json.dumps(message))
+        activity_replay = _activity_replay_message()
+        if activity_replay is not None:
+            await ws.send(json.dumps(activity_replay))
         async for _ in ws:
             pass  # watchers never send anything meaningful; drain and ignore
     finally:

infra/status-relay/test_relay.py

@@ -26,6 +26,7 @@
 import sys
 import urllib.error
 import urllib.request
+from datetime import UTC, datetime
 from typing import TYPE_CHECKING
 from uuid import uuid4
 
@@ -104,6 +105,7 @@ async def __aenter__(self) -> _RelayHarness:
         relay._watchers.clear()
         relay._run_histories.clear()
         relay._enclosure_timelines.clear()
+        relay._activity_buffer.clear()
 
         check_viewer_auth = relay.basic_auth(
             realm="cora-status", credentials=(_VIEWER_USER, _VIEWER_PASSWORD)
@@ -377,10 +379,10 @@ async def test_enclosure_timeline_from_producer_is_broadcast_to_a_connected_watc
 
 async def test_enclosure_timeline_is_replayed_to_a_watcher_that_connects_later() -> None:
     """The behavior that makes REWIND reach an enclosure without any
-    on-demand request path: unlike `activity`, a pushed timeline is
-    cached (`relay._enclosure_timelines`) and replayed on every new
-    `/watch` connection, not just broadcast to whoever happened to
-    already be listening."""
+    on-demand request path: a pushed timeline is cached
+    (`relay._enclosure_timelines`, no eviction, unlike `activity`'s
+    bounded time window) and replayed on every new `/watch` connection,
+    not just broadcast to whoever happened to already be listening."""
     async with _RelayHarness() as harness:
         enclosure_id = str(uuid4())
         async with harness.connect_producer() as producer:
@@ -414,6 +416,96 @@ async def test_malformed_enclosure_timeline_without_enclosure_id_is_dropped() ->
         assert relay._enclosure_timelines == {}
 
 
+def _recent_iso() -> str:
+    """A timestamp inside `relay._ACTIVITY_BUFFER_SECONDS` of real now,
+    since pruning compares `occurred_at` against the wall clock at test
+    run time, not a fixed date."""
+    return datetime.now(UTC).isoformat()
+
+
+def _sample_activity_message(*, occurred_at: str, stream_type: str = "Run") -> dict:
+    return {
+        "kind": "activity",
+        "schema_version": 1,
+        "producer_id": "p1",
+        "generated_at": "2026-08-30T00:05:00+00:00",
+        "events": [
+            {
+                "stream_type": stream_type,
+                "stream_id": str(uuid4()),
+                "event_type": "RunStarted",
+                "occurred_at": occurred_at,
+                "recorded_at": occurred_at,
+            }
+        ],
+    }
+
+
+async def test_activity_from_producer_is_broadcast_to_a_connected_watcher() -> None:
+    async with _RelayHarness() as harness, harness.connect_watcher() as watcher:
+        async with harness.connect_producer() as producer:
+            message = _sample_activity_message(occurred_at=_recent_iso())
+            await producer.send(json.dumps(message))
+            received = await _recv_until_kind(watcher, "activity")
+            assert received == message, received
+
+
+async def test_activity_is_replayed_to_a_watcher_that_connects_later() -> None:
+    """The fix for the flowing lanes going empty on a refresh or a
+    resumed SSH tunnel: before this, `activity` was a pure pass-through
+    with nothing for a fresh `/watch` connection to catch up on. Now the
+    last `relay._ACTIVITY_BUFFER_SECONDS` of events are buffered
+    (`relay._activity_buffer`) and replayed as one synthetic `activity`
+    message on connect, in the same shape `page.html`'s `handleActivity`
+    already knows how to consume."""
+    async with _RelayHarness() as harness:
+        async with harness.connect_producer() as producer:
+            message = _sample_activity_message(occurred_at=_recent_iso())
+            await producer.send(json.dumps(message))
+            # Give the relay's event loop a turn to process the frame
+            # before a fresh watcher connects and expects to see it.
+            await asyncio.sleep(0.1)
+
+            async with harness.connect_watcher() as watcher:
+                received = await _recv_until_kind(watcher, "activity")
+                assert received["events"] == message["events"], received
+
+
+async def test_activity_older_than_the_buffer_window_is_not_replayed() -> None:
+    async with _RelayHarness() as harness:
+        async with harness.connect_producer() as producer:
+            stale = _sample_activity_message(occurred_at="2020-01-01T00:00:00+00:00")
+            await producer.send(json.dumps(stale))
+            await asyncio.sleep(0.1)
+
+            async with harness.connect_watcher() as watcher:
+                got_activity = False
+                try:
+                    await _recv_until_kind(watcher, "activity", max_frames=3, timeout=0.5)
+                    got_activity = True
+                except (TimeoutError, AssertionError):
+                    pass
+                assert not got_activity, "a stale event outside the buffer window was replayed"
+
+
+async def test_malformed_activity_without_events_list_is_dropped() -> None:
+    async with _RelayHarness() as harness, harness.connect_producer() as producer:
+        await producer.send(
+            json.dumps(
+                {
+                    "kind": "activity",
+                    "schema_version": 1,
+                    "producer_id": "p1",
+                    "generated_at": "2026-08-30T00:00:00+00:00",
+                }
+            )
+        )
+        # No reply is expected either way; give the relay a turn to process
+        # (and, if it were going to, mis-store) the frame.
+        await asyncio.sleep(0.1)
+        assert relay._activity_buffer == []
+
+
 TESTS: list[Callable[[], Awaitable[None]]] = [
     obj
     for name, obj in list(globals().items())