From 986e0fdd85b121d300b97fcab0ed0c4f5c623d3d Mon Sep 17 00:00:00 2001 From: Anthony Ruiz Date: Wed, 23 Sep 2026 10:00:57 +0000 Subject: [PATCH] fix(gateway-watcher): fence stale projections after final subscriber removal --- api/gateway_watcher.py | 66 +++++++------- docs/sse-streams.md | 15 ++++ .../test_gateway_watcher_resubscribe_fresh.py | 90 ++++++++++++++++++- 3 files changed, 139 insertions(+), 32 deletions(-) diff --git a/api/gateway_watcher.py b/api/gateway_watcher.py index b112883d..69b1bbb6 100644 --- a/api/gateway_watcher.py +++ b/api/gateway_watcher.py @@ -203,6 +203,8 @@ class GatewayWatcher: ): self._subscribers: list[queue.Queue] = [] self._sub_lock = threading.Lock() + # Final removal invalidates any projection admitted by an earlier cohort. + self._subscriber_epoch = 0 self._stop_event = threading.Event() # Wakes a poll loop parked because nobody is subscribed (subscribe/stop). self._idle_wake = threading.Event() @@ -286,31 +288,32 @@ class GatewayWatcher: self._idle_wake.set() return q + def _remove_subscriber_locked(self, q: queue.Queue) -> bool: + """Remove a known queue under _sub_lock; fence polls on the last removal.""" + try: + self._subscribers.remove(q) + except ValueError: + return False + if not self._subscribers: + self._subscriber_epoch += 1 + self._last_cheap_fp = '' + self._last_full_projection_at = None + return True + def unsubscribe(self, q: queue.Queue): - """Remove a subscriber queue. - - When the last subscriber leaves, drop the cheap fingerprint and parity - timestamp so the next subscription runs a fresh projection. The poll - loop parks *before* calling ``_poll_once`` once nobody is subscribed, - so the equivalent reset inside ``_poll_once`` is not reached on the - normal disconnect path. - """ + """Remove a subscriber queue, invalidating projection on the last removal.""" with self._sub_lock: - try: - self._subscribers.remove(q) - except ValueError: - return - if not self._subscribers: - self._last_cheap_fp = '' - self._last_full_projection_at = None + self._remove_subscriber_locked(q) - def _notify_subscribers(self, sessions: list): + def _notify_subscribers(self, sessions: list, *, epoch: int | None = None): """Push change event to all subscribers.""" event = { 'type': 'sessions_changed', 'sessions': sessions, } with self._sub_lock: + if epoch is not None and epoch != self._subscriber_epoch: + return # An old poll must not notify a new subscriber cohort. dead = [] for q in self._subscribers: try: @@ -320,10 +323,7 @@ class GatewayWatcher: except Exception: dead.append(q) for q in dead: - try: - self._subscribers.remove(q) - except ValueError: - pass + self._remove_subscriber_locked(q) # Send a None sentinel so the SSE handler unblocks, closes, # and lets the browser's EventSource auto-reconnect. try: @@ -340,9 +340,9 @@ class GatewayWatcher: """ with self._sub_lock: has_subscribers = bool(self._subscribers) + admission_epoch = self._subscriber_epoch if not has_subscribers: - # Force fresh projection on the next subscription, without touching DB. - self._last_full_projection_at = None + # Final removal already invalidated the cache under the same lock. return False db_path = self._state_db_path # A watcher may start before the agent has created state.db. Publishing an @@ -371,14 +371,20 @@ class GatewayWatcher: if sessions is None: return False current_hash = _snapshot_hash(sessions) - if cheap_fp is not None: - self._last_cheap_fp = cheap_fp - self._last_full_projection_at = current_time - - if current_hash != self._last_hash: - self._last_hash = current_hash - self._last_sessions = sessions - self._notify_subscribers(sessions) + with self._sub_lock: + if admission_epoch != self._subscriber_epoch or not self._subscribers: + return False # Never restore a cache invalidated during the DB read. + if cheap_fp is not None: + self._last_cheap_fp = cheap_fp + self._last_full_projection_at = current_time + if current_hash != self._last_hash: + changed = True + self._last_hash = current_hash + self._last_sessions = sessions + else: + changed = False + if changed: + self._notify_subscribers(sessions, epoch=admission_epoch) return True def _poll_loop(self): diff --git a/docs/sse-streams.md b/docs/sse-streams.md index 07fca5b6..12eb3898 100644 --- a/docs/sse-streams.md +++ b/docs/sse-streams.md @@ -56,6 +56,21 @@ infer it from the status code. Clients that only need session updates should use `/api/session/stream` directly; the gateway probe is only relevant for clients that display CLI/TUI/messaging sessions. +## Gateway watcher projection lifecycle + +The optional gateway stream's watcher owns a cheap database fingerprint and a +timestamp for its last full session projection. While clients are subscribed, +an unchanged fingerprint can skip the expensive read until the bounded parity +interval. With no subscribers the poll loop parks instead of reading the DB. +When the final subscriber leaves (including eviction of a full or failed +consumer queue), the watcher invalidates both cache fields. A reconnect must +therefore trigger a fresh projection even when the database fingerprint has not +changed and the parity interval has not elapsed. Final removal also advances a +subscriber epoch: an older in-flight DB read cannot restore the invalidated +cache or publish its result to a later subscriber cohort. The subscriber lock +protects removal, epoch checks and cache commits, but is not held during DB +reads, so disconnect is not blocked by a slow projection. + ## Heartbeats and proxy behavior - All long-lived streams emit SSE keepalive comment lines on the diff --git a/tests/test_gateway_watcher_resubscribe_fresh.py b/tests/test_gateway_watcher_resubscribe_fresh.py index af62a699..01e9baa8 100644 --- a/tests/test_gateway_watcher_resubscribe_fresh.py +++ b/tests/test_gateway_watcher_resubscribe_fresh.py @@ -5,16 +5,19 @@ change (#7694) the poll loop parks *before* calling ``_poll_once`` once the last subscriber leaves, so that reset is not reached on a normal disconnect. A tab that reconnects inside the parity interval then resumes with the stale -cheap fingerprint and skips the projection. The reset now happens in -``unsubscribe()`` when the last subscriber leaves. +cheap fingerprint and skips the projection. Final removal invalidates the +cache, and an epoch fences projections already in flight. """ from __future__ import annotations import importlib import sqlite3 +import threading import time from pathlib import Path +import pytest + def _make_db(tmp_path: Path) -> Path: db = tmp_path / "state.db" @@ -101,6 +104,89 @@ def test_unsubscribe_of_unknown_queue_does_not_reset(tmp_path): assert watcher._last_full_projection_at == 1.0 +def test_slow_consumer_eviction_invalidates_populated_cache(tmp_path): + gw = importlib.import_module("api.gateway_watcher") + watcher = gw.GatewayWatcher(state_db_path=_make_db(tmp_path)) + q = watcher.subscribe() + assert watcher._poll_once(now=1.0) is True + assert watcher._last_cheap_fp + assert watcher._last_full_projection_at == 1.0 + + for _ in range(q.maxsize - q.qsize()): + q.put_nowait({"type": "existing"}) + watcher._notify_subscribers([]) + assert not watcher._has_subscribers() + assert watcher._last_cheap_fp == "" + assert watcher._last_full_projection_at is None + watcher.unsubscribe(q) # The late SSE cleanup cannot undo the reset. + watcher.subscribe() + assert watcher._poll_once(now=2.0) is True + + +@pytest.mark.parametrize("removal", ["unsubscribe", "slow_consumer"]) +@pytest.mark.parametrize("reconnect_while_blocked", [False, True]) +def test_inflight_projection_cannot_restore_cache_after_last_removal( + tmp_path, monkeypatch, removal, reconnect_while_blocked, +): + """The DB read must not serialize final removal or commit across its epoch.""" + gw = importlib.import_module("api.gateway_watcher") + watcher = gw.GatewayWatcher(state_db_path=_make_db(tmp_path)) + q = watcher.subscribe() + entered = threading.Event() + release = threading.Event() + projections = [] + errors = [] + real_projection = gw._get_agent_sessions_from_db + + def blocked_projection(path): + projections.append(path) + if len(projections) == 1: + entered.set() + assert release.wait(5), "test never released the DB read" + return real_projection(path) + + def poll(): + try: + watcher._poll_once(now=1.0) + except BaseException as exc: + errors.append(exc) + + monkeypatch.setattr(gw, "_get_agent_sessions_from_db", blocked_projection) + worker = threading.Thread(target=poll, daemon=True) + worker.start() + new_q = None + try: + assert entered.wait(5), "poll never reached the DB read" + if removal == "unsubscribe": + watcher.unsubscribe(q) + else: + for _ in range(q.maxsize): + q.put_nowait({"type": "existing"}) + watcher._notify_subscribers([]) + # Handler teardown of an evicted queue must be harmless. + watcher.unsubscribe(q) + assert not watcher._has_subscribers() + assert watcher._last_cheap_fp == "" + assert watcher._last_full_projection_at is None + if reconnect_while_blocked: + new_q = watcher.subscribe() + finally: + release.set() + worker.join(timeout=5) + assert not worker.is_alive(), "poll did not finish" + assert not errors, errors + assert watcher._last_cheap_fp == "" + assert watcher._last_full_projection_at is None + if new_q is not None: + assert new_q.empty(), "old projection must not notify a new subscriber" + else: + watcher.subscribe() + assert watcher._poll_once(now=2.0) is True + assert len(projections) == 2 + assert watcher._last_cheap_fp + assert watcher._last_full_projection_at == 2.0 + + def test_poll_loop_reprojects_for_reconnecting_subscriber(tmp_path, monkeypatch): """Through the real loop: disconnect, park, reconnect -> projection runs again.""" gw = importlib.import_module("api.gateway_watcher")