From 83c560f626f0dd60c10f2a3047571047fa8284a6 Mon Sep 17 00:00:00 2001 From: Alex Date: Wed, 8 Jul 2026 00:23:32 +0100 Subject: [PATCH] feat: mini replay improvements --- application/api/events/routes.py | 87 +++++++++++------- application/asgi.py | 3 +- application/core/settings.py | 13 +++ tests/api/test_events_routes.py | 149 +++++++++++++++++++++++++------ tests/test_asgi.py | 12 +++ 5 files changed, 207 insertions(+), 57 deletions(-) diff --git a/application/api/events/routes.py b/application/api/events/routes.py index 9080a58a..d1097157 100644 --- a/application/api/events/routes.py +++ b/application/api/events/routes.py @@ -103,33 +103,18 @@ def _allow_replay( ) -> bool: """Per-user sliding-window snapshot-replay budget. - Fails open on Redis errors or when the budget is disabled. Empty-backlog - no-cursor connects skip INCR so dev double-mounts don't trip 429. + Fails open on Redis errors or when the budget is disabled. No-cursor + connects never consume budget: fresh sessions start live and do no + snapshot work, so counting them only starved real reconnects (a burst + of fresh tabs could 429 a user's cursor-bearing reconnect). """ budget = int(settings.EVENTS_REPLAY_BUDGET_REQUESTS_PER_WINDOW) if budget <= 0: return True if redis_client is None: return True - - # Cheap pre-check: only INCR when we might actually replay. XLEN - # is one Redis op; the alternative (INCR every connect) is two - # ops AND wrongly counts no-op probes. The check is conservative: - # if ``last_event_id`` is set we always INCR, even if the cursor - # has already overtaken the latest entry — that case is rare and - # short-lived, and probing further would mean a redundant XRANGE. if last_event_id is None: - try: - if int(redis_client.xlen(stream_key(user_id))) == 0: - return True - except Exception: - # XLEN probe failed; fall through to the INCR path so a - # transient Redis hiccup can't bypass the budget. - logger.debug( - "XLEN probe failed for replay budget check user=%s; " - "proceeding to INCR", - user_id, - ) + return True window = max(1, int(settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS)) key = replay_budget_key(user_id) @@ -154,7 +139,7 @@ def _normalize_last_event_id(raw: Optional[str]) -> Optional[str]: Returns the value unchanged when it parses as a Redis Streams id, otherwise ``None`` — callers treat ``None`` as "client has nothing" - and replay from the start of the retained window. Invalid ids would + and start the stream live (no snapshot). Invalid ids would otherwise pass straight to XRANGE and surface as a quiet replay failure plus broken truncation detection. """ @@ -166,6 +151,21 @@ def _normalize_last_event_id(raw: Optional[str]) -> Optional[str]: return raw +def _replay_floor_id() -> Optional[str]: + """Oldest stream id the snapshot replay may reach, or ``None``. + + Streams ids are millisecond timestamps, so an age ceiling maps + directly to an id floor. MAXLEN caps the stream by count, not time — + for a low-traffic user 1000 entries can span weeks, and shipping + that on reconnect helps no one. + """ + max_age_hours = int(settings.EVENTS_REPLAY_MAX_AGE_HOURS) + if max_age_hours <= 0: + return None + floor_ms = int(time.time() * 1000) - max_age_hours * 3600 * 1000 + return f"{floor_ms}-0" + + def _replay_backlog( redis_client, user_id: str, last_event_id: Optional[str], max_count: int ) -> Iterator[tuple[str, str]]: @@ -175,8 +175,16 @@ def _replay_backlog( Parse failures are skipped; the Streams id is injected into the envelope so replay matches live-tail shape. """ - # Exclusive start: '(' skips the already-delivered entry. - start = f"({last_event_id}" if last_event_id else "-" + floor = _replay_floor_id() + if last_event_id is None: + start = floor or "-" + elif floor and stream_id_compare(last_event_id, floor) < 0: + # Cursor is past the age ceiling: clamp to the floor (inclusive). + # The caller emits ``backlog.truncated`` for this case. + start = floor + else: + # Exclusive start: '(' skips the already-delivered entry. + start = f"({last_event_id}" try: entries = redis_client.xrange( stream_key(user_id), min=start, max="+", count=max_count @@ -385,15 +393,34 @@ def stream_events() -> Response: try: if redis_client is None: return + if last_event_id is None: + # Fresh session: start live. A no-cursor connect has + # no state to catch up on — replaying the whole + # retained window shipped weeks-old entries on every + # tab-open (MAXLEN caps by count, not age). Clients + # that had state present a cursor; the malformed- + # cursor case already got its truncation notice + # before the subscribe loop. + return oldest = _oldest_retained_id(redis_client, user_id) - if ( - last_event_id - and oldest - and stream_id_compare(last_event_id, oldest) < 0 + floor = _replay_floor_id() + # The snapshot can't reach past whichever is newer: + # the MAXLEN'd window edge or the age floor. + effective_oldest = oldest + if floor and ( + effective_oldest is None + or stream_id_compare(floor, effective_oldest) > 0 ): - # The Last-Event-ID has slid off the MAXLEN window. - # Tell the client so it can fetch full state. - replay_lines.append(_truncation_notice_line(oldest)) + effective_oldest = floor + if ( + effective_oldest + and stream_id_compare(last_event_id, effective_oldest) < 0 + ): + # The Last-Event-ID has slid off the replayable + # window. Tell the client so it can fetch full state. + replay_lines.append( + _truncation_notice_line(effective_oldest) + ) replay_cap = int(settings.EVENTS_REPLAY_MAX_PER_REQUEST) for entry_id, sse_line in _replay_backlog( redis_client, user_id, last_event_id, replay_cap diff --git a/application/asgi.py b/application/asgi.py index 1ffb1d6d..787f94ca 100644 --- a/application/asgi.py +++ b/application/asgi.py @@ -10,9 +10,10 @@ from starlette.routing import Mount from application.api.async_sse import async_sse_routes from application.app import app as flask_app +from application.core.settings import settings from application.mcp_server import mcp -_WSGI_THREADPOOL = 32 +_WSGI_THREADPOOL = int(settings.WSGI_THREADPOOL_WORKERS) mcp_app = mcp.http_app(path="/") diff --git a/application/core/settings.py b/application/core/settings.py index a5d5b8ec..52caed41 100644 --- a/application/core/settings.py +++ b/application/core/settings.py @@ -272,6 +272,13 @@ class Settings(BaseSettings): # --graceful-timeout). Keep below the gunicorn --timeout (180) watchdog. # Used by gunicorn_worker.BoundedDrainUvicornWorker. GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS: int = 30 + # a2wsgi thread-pool size per worker process — the hard ceiling on + # concurrent Flask requests (SSE holders + in-flight chats included). + # Threads parked on idle SSE reads are nearly free; exhaustion is + # not (every Flask request on that worker queues behind LLM + # generations). Sized so saturation needs hundreds of concurrent + # holders per worker, not 32. + WSGI_THREADPOOL_WORKERS: int = 96 # SSE keepalive comment cadence. Must sit under Cloudflare's 100s idle # close and iOS Safari's ~60s — 15s gives generous headroom. SSE_KEEPALIVE_SECONDS: int = 15 @@ -288,6 +295,12 @@ class Settings(BaseSettings): # connection cap above and the windowed budget below, total # enumeration throughput is bounded. EVENTS_REPLAY_MAX_PER_REQUEST: int = 200 + # Age ceiling on snapshot replay. MAXLEN caps the stream by COUNT, + # so a low-traffic user's window can reach weeks back — replaying + # month-old ingest ticks on reconnect helps no one. Entries older + # than this never replay (the client gets ``backlog.truncated`` and + # refetches state instead). 0 disables the floor. + EVENTS_REPLAY_MAX_AGE_HOURS: int = 48 # Sliding-window cap on snapshot replays per user. Once the budget # is exhausted the route returns HTTP 429 with the cursor pinned; # the client backs off and retries after the window rolls over. diff --git a/tests/api/test_events_routes.py b/tests/api/test_events_routes.py index 629acd39..8eef96c3 100644 --- a/tests/api/test_events_routes.py +++ b/tests/api/test_events_routes.py @@ -450,11 +450,10 @@ class TestReplayRateLimit: mock_settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS = 60 assert _allow_replay(None, "alice", "1735682400000-0") is True - def test_allow_replay_skips_incr_when_no_cursor_and_empty_backlog(self): - """Fresh client with no cursor and an empty user stream cannot - do snapshot work — INCR'ing the counter would needlessly - burn budget. Catches the React-StrictMode dev-burst case where - double-mounted components would otherwise 429 in 5 connects. + def test_allow_replay_never_consumes_budget_without_cursor(self): + """No cursor ⇒ no replay work ⇒ no budget. Fresh sessions start + live and must never 429 on the replay budget, no matter how many + tabs open at once (30 fresh connects/min used to exhaust it). """ from application.api.events.routes import _allow_replay @@ -462,31 +461,13 @@ class TestReplayRateLimit: mock_settings.EVENTS_REPLAY_BUDGET_REQUESTS_PER_WINDOW = 3 mock_settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS = 60 redis = MagicMock() - redis.xlen.return_value = 0 + redis.xlen.return_value = 42 - # 5 connects in a row, all with no cursor — none consume - # budget because the backlog is empty. - for _ in range(5): + for _ in range(50): assert _allow_replay(redis, "alice", None) is True - redis.xlen.assert_called() redis.incr.assert_not_called() - - def test_allow_replay_incrs_when_no_cursor_but_backlog_present(self): - """A no-cursor connect against a non-empty backlog *will* do - snapshot work, so it consumes budget normally. - """ - from application.api.events.routes import _allow_replay - - with patch("application.api.events.routes.settings") as mock_settings: - mock_settings.EVENTS_REPLAY_BUDGET_REQUESTS_PER_WINDOW = 5 - mock_settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS = 60 - redis = MagicMock() - redis.xlen.return_value = 42 - redis.incr.return_value = 1 - - assert _allow_replay(redis, "alice", None) is True - redis.incr.assert_called_once() + redis.xlen.assert_not_called() def test_allow_replay_passes_until_budget_exhausted(self): from application.api.events.routes import _allow_replay @@ -647,3 +628,119 @@ class TestFormatHelpers: from application.api.events.routes import _normalize_last_event_id assert _normalize_last_event_id(candidate) == expected + + +# ── replay policy: fresh sessions start live, snapshots are age-capped ── + + +class TestReplayPolicy: + def test_replay_floor_id_disabled_when_setting_zero(self): + from application.api.events.routes import _replay_floor_id + + with patch("application.api.events.routes.settings") as mock_settings: + mock_settings.EVENTS_REPLAY_MAX_AGE_HOURS = 0 + assert _replay_floor_id() is None + + def test_replay_floor_id_is_ms_stream_id(self): + from application.api.events.routes import _replay_floor_id + + with patch("application.api.events.routes.settings") as mock_settings: + mock_settings.EVENTS_REPLAY_MAX_AGE_HOURS = 48 + with patch( + "application.api.events.routes.time.time", + return_value=1_800_000_000.0, + ): + floor = _replay_floor_id() + assert floor == f"{(1_800_000_000 - 48 * 3600) * 1000}-0" + + def test_replay_backlog_uses_cursor_when_newer_than_floor(self): + from application.api.events.routes import _replay_backlog + + redis = MagicMock() + redis.xrange.return_value = [] + with patch( + "application.api.events.routes._replay_floor_id", + return_value="1000-0", + ): + list(_replay_backlog(redis, "alice", "2000-0", 200)) + assert redis.xrange.call_args.kwargs["min"] == "(2000-0" + + def test_replay_backlog_clamps_start_to_age_floor(self): + from application.api.events.routes import _replay_backlog + + redis = MagicMock() + redis.xrange.return_value = [] + with patch( + "application.api.events.routes._replay_floor_id", + return_value="5000-0", + ): + list(_replay_backlog(redis, "alice", "2000-0", 200)) + # Cursor is older than the floor: replay starts at the floor + # (inclusive), not the cursor — entries past their shelf life + # are never shipped. + assert redis.xrange.call_args.kwargs["min"] == "5000-0" + + def test_no_cursor_connect_never_replays(self): + """A fresh session (no Last-Event-ID) starts live: no XRANGE, no + weeks-old backlog on every tab-open.""" + from application.api.events import routes as events_module + + app = _make_app() + redis_client = MagicMock() + redis_client.incr.return_value = 1 + + def _fake_subscribe(self, on_subscribe=None, poll_timeout=1.0): + if on_subscribe is not None: + on_subscribe() + yield None + + with patch.object( + events_module, "get_redis_instance", return_value=redis_client + ), patch.object( + events_module.Topic, "subscribe", _fake_subscribe, create=False + ): + with app.test_client() as c: + r = c.get("/api/events", headers={"X-Test-Sub": "alice"}) + body = _drain_until( + r, lambda b: b": connected" in b, max_chunks=40 + ) + assert b": connected" in body + redis_client.xrange.assert_not_called() + + def test_cursor_older_than_floor_emits_truncation_notice(self): + """An age-clamped snapshot has a gap the client can't see from + entry ids alone — it must get the truncation notice so it + refetches full state instead of trusting a partial replay.""" + from application.api.events import routes as events_module + + app = _make_app() + redis_client = MagicMock() + redis_client.incr.return_value = 1 + # No retained-oldest available: the floor alone must trigger it. + redis_client.xinfo_stream.side_effect = Exception("nope") + redis_client.xrange.return_value = [] + + def _fake_subscribe(self, on_subscribe=None, poll_timeout=1.0): + if on_subscribe is not None: + on_subscribe() + yield None + + with patch.object( + events_module, "get_redis_instance", return_value=redis_client + ), patch.object( + events_module.Topic, "subscribe", _fake_subscribe, create=False + ), patch.object( + events_module, "_replay_floor_id", return_value="9999999999999-0" + ): + with app.test_client() as c: + r = c.get( + "/api/events", + headers={ + "X-Test-Sub": "alice", + "Last-Event-ID": "1735682400000-0", + }, + ) + body = _drain_until( + r, lambda b: b"backlog.truncated" in b, max_chunks=40 + ) + assert b"backlog.truncated" in body diff --git a/tests/test_asgi.py b/tests/test_asgi.py index 556a47a6..da516fa2 100644 --- a/tests/test_asgi.py +++ b/tests/test_asgi.py @@ -156,3 +156,15 @@ def test_cors_preflight_on_mcp_route(): assert r.status_code in (200, 204) assert r.headers.get("access-control-allow-origin") == "*" assert "Mcp-Session-Id" in r.headers.get("access-control-allow-headers", "") + + +@pytest.mark.unit +def test_wsgi_threadpool_sized_from_settings(): + """The Flask thread pool is the app's request-capacity ceiling — + it must be operator-tunable (prod incident 2026-07-05→07: 32 slots + exhausted by SSE holders starved all other requests).""" + from application import asgi + from application.core.settings import settings + + assert asgi._WSGI_THREADPOOL == int(settings.WSGI_THREADPOOL_WORKERS) + assert asgi._WSGI_THREADPOOL >= 64