diff --git a/api/main.py b/api/main.py index 76221ae..3fc8ada 100644 --- a/api/main.py +++ b/api/main.py @@ -107,6 +107,49 @@ def _catalyst_engine(store, catalyst: dict) -> dict: "dilutive": dilutive, "vetoes_today": vetoes} +class _ChangeSignal: + """One process-wide event-change signal shared by ALL SSE clients. + + Before this, every SSE client ran ``COUNT(*)`` on a FRESH EventStore + connection every ~1s (re-applying the 473-line schema unless + ``PG_SKIP_SCHEMA=1``). Now a single long-lived connection polls + ``max(timestamp)`` (an index-tail seek) at most once per second GLOBALLY, + behind a lock — N clients cost one query/sec, not N. ``reader`` is injectable + so tests can count underlying queries without a DB. + """ + + def __init__(self, db_path: str | None = None, *, reader=None, min_interval: float = 1.0): + self._db_path = db_path + self._reader = reader + self._min_interval = float(min_interval) + self._lock = threading.Lock() + self._store = None + self._stamp = None + self._checked_at = 0.0 + self.query_count = 0 # test hook: number of underlying reads performed + + def _read(self): + self.query_count += 1 + if self._reader is not None: + return self._reader() + if self._store is None: + self._store = EventStore(self._db_path) # long-lived; schema skipped via PG_SKIP_SCHEMA + return self._store.max_timestamp() + + def current(self): + """Latest change token (max event timestamp), refreshed at most once per + ``min_interval`` seconds across all callers.""" + now = time.monotonic() + with self._lock: + if self._checked_at == 0.0 or (now - self._checked_at) >= self._min_interval: + try: + self._stamp = self._read() + except Exception: # noqa: BLE001 — a transient read must not kill the stream + pass + self._checked_at = now + return self._stamp + + class DashboardState: """Shared state: event store path + optional live execution service.""" @@ -125,6 +168,16 @@ class DashboardState: self.research_con = research_con self.watch_inject = watch_inject self._lock = threading.Lock() + # lazily-built, process-wide SSE change signal (shared by all clients) + self._change_signal: _ChangeSignal | None = None + + def change_stamp(self): + """SSE change token from the shared once-per-second signal (see _ChangeSignal).""" + if self._change_signal is None: + with self._lock: + if self._change_signal is None: + self._change_signal = _ChangeSignal(self.event_db_path) + return self._change_signal.current() def open_store(self) -> EventStore: return EventStore(self.event_db_path) @@ -350,14 +403,15 @@ class DashboardHandler(BaseHTTPRequestHandler): self.end_headers() shutdown = getattr(type(self), "shutdown_event", None) - last_count = -1 + _sentinel = object() + last_stamp = _sentinel last_heartbeat = 0.0 try: while shutdown is None or not shutdown.is_set(): - count = self.state.event_count() + stamp = self.state.change_stamp() now = time.monotonic() - if count != last_count: - last_count = count + if stamp != last_stamp: # last_stamp starts as a unique sentinel + last_stamp = stamp payload = _dump(self.state.snapshots()).decode() self.wfile.write(f"event: snapshot\ndata: {payload}\n\n".encode()) self.wfile.flush() diff --git a/storage/event_store.py b/storage/event_store.py index 844e1c5..6f6e823 100644 --- a/storage/event_store.py +++ b/storage/event_store.py @@ -200,6 +200,16 @@ class EventStore: row = cursor.fetchone() return int(row[0]) if row else 0 + def max_timestamp(self): + """Newest events.timestamp (None on an empty store). The SSE change + signal polls this instead of COUNT(*): idx_events_timestamp makes it an + index-tail seek, whereas COUNT(*) scans the whole table every tick.""" + with self._lock: + cursor = self.con.cursor() + cursor.execute("SELECT max(timestamp) FROM events") + row = cursor.fetchone() + return row[0] if row else None + def get_symbol_timeline(self, symbol: str) -> list[dict]: """Get full timeline for a symbol. diff --git a/tests/unit/api/test_sse_change_signal.py b/tests/unit/api/test_sse_change_signal.py new file mode 100644 index 0000000..5f26136 --- /dev/null +++ b/tests/unit/api/test_sse_change_signal.py @@ -0,0 +1,51 @@ +"""The shared SSE change signal must run ONE underlying query per second across +ALL clients, not one per client per tick (plan 007 step 2).""" + +import time + +from api.main import _ChangeSignal + + +def test_shared_signal_query_budget(): + calls = {"n": 0} + + def reader(): + calls["n"] += 1 + return calls["n"] # a monotonic stamp + + sig = _ChangeSignal(reader=reader, min_interval=1.0) + + # Two "clients" hammering current() 10 times total inside the 1s window must + # collapse to a single underlying read (the first one). + for _ in range(5): + sig.current() # client A + sig.current() # client B + assert calls["n"] == 1, f"expected 1 read in the window, got {calls['n']}" + assert sig.query_count == 1 + + +def test_signal_refreshes_after_interval(): + calls = {"n": 0} + + def reader(): + calls["n"] += 1 + return calls["n"] + + sig = _ChangeSignal(reader=reader, min_interval=0.05) + first = sig.current() + time.sleep(0.06) + second = sig.current() + assert first == 1 and second == 2 # re-read once the interval elapsed + + +def test_signal_survives_reader_error(): + def boom(): + raise RuntimeError("db blip") + + sig = _ChangeSignal(reader=boom, min_interval=1.0) + assert sig.current() is None # swallowed; stream keeps running + + +if __name__ == "__main__": # pragma: no cover + import pytest + pytest.main([__file__, "-q"])