From f19f0e6cf82c17810c6711a0ef4f2aed9b777b90 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:01:29 +0000 Subject: [PATCH 1/4] fix(backend): the events long-poll observes events from any writer poll_events only woke on the waiter that server.api.events.emit sets, so a client parked on wait=20 slept for the whole wait while the orchestrator's server.orchestrator.events.emit appended events straight to state storage. Wait in slices of HSS_INGEST_EVENTS_POLL_MS and re-read the store between slices; an in-process emit still wakes the poll the moment it sets the waiter, and wait=0 keeps returning immediately. --- backend/server/api/events.py | 10 +- .../tests/test_events_orchestrator_wake.py | 140 ++++++++++++++++++ 2 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 backend/server/api/tests/test_events_orchestrator_wake.py diff --git a/backend/server/api/events.py b/backend/server/api/events.py index 8926556c8..5c1c45d94 100644 --- a/backend/server/api/events.py +++ b/backend/server/api/events.py @@ -38,6 +38,14 @@ async def poll_events(deps: Deps, record: CaptureRecord, after: int, wait_s: flo try: deadline = time.monotonic() + wait_s waiter = _waiter(deps, record.capture_id) + # Only same-process server.api.events.emit sets the wake event. The + # orchestrator's server.orchestrator.events.emit writes straight to state + # storage, and a separate worker or Modal deployment shares no process with + # this loop, so its writers can never set the waiter. Wait in slices of + # HSS_INGEST_EVENTS_POLL_MS and re-read the store between slices, so the + # poll observes events from any writer while an in-process emit still + # wakes it the moment the waiter is set. + slice_s = deps.settings.HSS_INGEST_EVENTS_POLL_MS / 1000 while True: rows = deps.state.events_after(record.capture_id, after, limit=100) timed_out = time.monotonic() >= deadline @@ -47,7 +55,7 @@ async def poll_events(deps: Deps, record: CaptureRecord, after: int, wait_s: flo if remaining <= 0: break try: - await asyncio.wait_for(waiter.wait(), timeout=remaining) + await asyncio.wait_for(waiter.wait(), timeout=min(slice_s, remaining)) except TimeoutError: pass waiter.clear() diff --git a/backend/server/api/tests/test_events_orchestrator_wake.py b/backend/server/api/tests/test_events_orchestrator_wake.py new file mode 100644 index 000000000..c5d024a07 --- /dev/null +++ b/backend/server/api/tests/test_events_orchestrator_wake.py @@ -0,0 +1,140 @@ +"""The events long-poll must observe events the orchestrator writes to state. + +PR #286's device run held the app's second events poll for its full 20,022 ms +although the orchestrator had finished within a second: poll_events only wakes +on deps.wake, which server.api.events.emit sets, while the orchestrator's +server.orchestrator.events.emit writes straight to state storage and never +touches the waiter. These tests drive the real app over uvicorn on loopback +and use only the orchestrator path to append the event a waiter is waiting on. +""" + +from __future__ import annotations + +import threading +import time + +import httpx +import pytest +import uvicorn + +from server.api.app import create_app +from server.common.config import Settings +from server.contracts.api import FailedData +from server.orchestrator import events as orchestrator_events + +TOKEN = "d" * 40 + + +def _settings(tmp_path) -> Settings: + return Settings( + STORAGE_BACKEND="volume", + STATE_BACKEND="memory", + VOLUME_ROOT=str(tmp_path / "vol"), + CAPTURE_API_PUBLIC_URL="http://testserver", + CAPTURE_API_TOKENS=f"device:{TOKEN}", + UPLOAD_URL_HMAC_SECRET="h" * 32, + HSS_INGEST_HOOKS="null", + HSS_AUTH_MODE="static", + GIT_SHA="0123456789abcdef", + VISION_EXTERNAL_CALLS="on", + SMP_SERVICE_NAME="capture-api", + ) + + +@pytest.fixture +def server(tmp_path): + """The real app served by uvicorn on loopback; yields (base_url, deps).""" + + app = create_app(settings=_settings(tmp_path)) + config = uvicorn.Config(app, host="127.0.0.1", port=0, log_level="warning") + server = uvicorn.Server(config) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + deadline = time.monotonic() + 10 + while not server.started: + if time.monotonic() > deadline or not thread.is_alive(): + pytest.fail("uvicorn never started") + time.sleep(0.01) + yield f"http://127.0.0.1:{server.servers[0].sockets[0].getsockname()[1]}", app.state.deps + server.should_exit = True + thread.join(timeout=10) + + +def _create_capture(base_url: str) -> str: + body = { + "packetId": "packet-orch-wake-0001", + "formatVersion": "0.4", + "tier": "photo_only", + "device": {"model": "iPhone", "systemVersion": "18", "appVersion": "1"}, + "flow": "photo", + } + response = httpx.post( + f"{base_url}/v1/captures", + json=body, + headers={"Authorization": f"Bearer {TOKEN}", "Idempotency-Key": "packet-orch-wake-0001"}, + timeout=10, + ) + assert response.status_code == 201, response.text + return response.json()["captureId"] + + +def test_long_poll_returns_within_one_poll_interval_of_an_orchestrator_event(server): + """wait=20 returns in under 2 s once the orchestrator appends an event.""" + + base_url, deps = server + capture_id = _create_capture(base_url) + + result: dict = {} + + def poll(): + started = time.perf_counter() + response = httpx.get( + f"{base_url}/v1/captures/{capture_id}/events", + params={"after": 0, "wait": 20}, + headers={"Authorization": f"Bearer {TOKEN}"}, + timeout=30, + ) + result["elapsed"] = time.perf_counter() - started + result["response"] = response + + client = threading.Thread(target=poll) + client.start() + try: + time.sleep(0.3) # the poll is parked in poll_events before the event lands + record = orchestrator_events.emit( + deps.state, + capture_id, + "failed", + FailedData(code="processing_failed", message="orchestrator blew up", next="manual_review"), + run_id="run-orch-wake-1", + ) + finally: + client.join(timeout=30) + assert not client.is_alive(), "the long-poll client never returned" + + assert result["response"].status_code == 200, result["response"].text + body = result["response"].json() + assert [event["type"] for event in body["events"]] == ["failed"] + assert body["events"][0]["seq"] == record.seq + assert body["next"] == record.seq + assert result["elapsed"] < 2.0, ( + f"the poll held for {result['elapsed']:.3f} s although the orchestrator event landed within one poll interval" + ) + + +def test_wait_zero_returns_immediately_when_the_feed_is_empty(server): + """wait=0 stays an immediate empty-page read, with no parked wait.""" + + base_url, _ = server + capture_id = _create_capture(base_url) + started = time.perf_counter() + response = httpx.get( + f"{base_url}/v1/captures/{capture_id}/events", + params={"after": 0, "wait": 0}, + headers={"Authorization": f"Bearer {TOKEN}"}, + timeout=10, + ) + elapsed = time.perf_counter() - started + assert response.status_code == 200 + assert response.json() == {"status": "uploading", "next": 0, "events": []} + assert elapsed < 1.0, f"wait=0 took {elapsed:.3f} s" From cb4949c6a89826f67efef316d4dd95bfd05c9ca0 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:58:18 +0000 Subject: [PATCH 2/4] fix(backend): the events long-poll reads the store off the event loop With ModalDictState every events_after and get_capture is a synchronous remote read; poll_events called them directly on the API's event loop, so each slice of a quiet 20 s poll parked the whole app for the length of its read, roughly 40 times per poll. Run every store read in poll_events through asyncio.to_thread so waiting clients cannot delay other requests on the same app. --- backend/server/api/events.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/backend/server/api/events.py b/backend/server/api/events.py index 5c1c45d94..3e7a4f53b 100644 --- a/backend/server/api/events.py +++ b/backend/server/api/events.py @@ -44,10 +44,13 @@ async def poll_events(deps: Deps, record: CaptureRecord, after: int, wait_s: flo # this loop, so its writers can never set the waiter. Wait in slices of # HSS_INGEST_EVENTS_POLL_MS and re-read the store between slices, so the # poll observes events from any writer while an in-process emit still - # wakes it the moment the waiter is set. + # wakes it the moment the waiter is set. With ModalDictState the store + # reads are synchronous remote calls, so every read runs in a worker + # thread (asyncio.to_thread): on the event loop, each slice would park + # the whole app for the length of its read. slice_s = deps.settings.HSS_INGEST_EVENTS_POLL_MS / 1000 while True: - rows = deps.state.events_after(record.capture_id, after, limit=100) + rows = await asyncio.to_thread(deps.state.events_after, record.capture_id, after, limit=100) timed_out = time.monotonic() >= deadline if rows or wait_s == 0 or timed_out: break @@ -59,8 +62,8 @@ async def poll_events(deps: Deps, record: CaptureRecord, after: int, wait_s: flo except TimeoutError: pass waiter.clear() - rows = deps.state.events_after(record.capture_id, after, limit=100) - fresh = deps.state.get_capture(record.capture_id) or record + rows = await asyncio.to_thread(deps.state.events_after, record.capture_id, after, limit=100) + fresh = await asyncio.to_thread(deps.state.get_capture, record.capture_id) or record events = [Event(seq=row.seq, type=row.type, at=row.at, data=row.data) for row in rows] nxt = events[-1].seq if events else after return EventsResponse(status=fresh.status, next=nxt, events=events) From 1719a87e99837d71e4cef9c12e2ddb7031a86493 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:58:26 +0000 Subject: [PATCH 3/4] fix(backend): the ingest config refuses a non-positive events poll interval HSS_INGEST_EVENTS_POLL_MS=0 (or negative) makes every wait_for slice time out immediately, so a parked events poll spins on the state store for its whole wait. Reject the value where the setting is declared instead of silently clamping it. --- backend/server/common/config.py | 11 +++++++++ .../common/tests/test_settings_poll_ms.py | 24 +++++++++++++++++++ 2 files changed, 35 insertions(+) create mode 100644 backend/server/common/tests/test_settings_poll_ms.py diff --git a/backend/server/common/config.py b/backend/server/common/config.py index 07d7b492d..41e81d201 100644 --- a/backend/server/common/config.py +++ b/backend/server/common/config.py @@ -98,6 +98,17 @@ def unknown_environment_is_prod(cls, value: object) -> object: HSS_AUTH_AUDIENCE: str = "capture-api" HSS_API_MIN_CONTAINERS: int = Field(default=1) + @field_validator("HSS_INGEST_EVENTS_POLL_MS") + @classmethod + def poll_interval_must_be_positive(cls, value: int) -> int: + """A zero or negative slice spins the events long-poll on the store.""" + if value <= 0: + raise ValueError( + "HSS_INGEST_EVENTS_POLL_MS must be a positive number of milliseconds; " + "the events long-poll waits one slice between store reads" + ) + return value + @model_validator(mode="after") def dev_deploy_auth_defaults_to_none(self) -> Settings: """Dev deploy is auth-off unless HSS_AUTH_MODE is configured (SD-86).""" diff --git a/backend/server/common/tests/test_settings_poll_ms.py b/backend/server/common/tests/test_settings_poll_ms.py new file mode 100644 index 000000000..20ebc0d02 --- /dev/null +++ b/backend/server/common/tests/test_settings_poll_ms.py @@ -0,0 +1,24 @@ +"""HSS_INGEST_EVENTS_POLL_MS must stay positive. + +poll_events slices the events long-poll's wait with this setting. A zero or +negative value turns every asyncio.wait_for slice into an immediate timeout, +so a parked poll spins on the state store for its whole wait instead of +waiting one slice between reads (Codex P2 and CodeRabbit minor, PR #290). +The setting is rejected where it is declared, never silently clamped. +""" + +import pytest +from pydantic import ValidationError + +from server.common.config import Settings + + +@pytest.mark.parametrize("value", [0, -1, -500]) +def test_non_positive_poll_ms_is_rejected(value: int) -> None: + with pytest.raises(ValidationError, match="HSS_INGEST_EVENTS_POLL_MS"): + Settings(HSS_INGEST_EVENTS_POLL_MS=value) + + +def test_positive_poll_ms_is_accepted() -> None: + settings = Settings(HSS_INGEST_EVENTS_POLL_MS=250) + assert settings.HSS_INGEST_EVENTS_POLL_MS == 250 From f6cc69089276a80147315b83009759645791e128 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:58:26 +0000 Subject: [PATCH 4/4] test(backend): the orchestrator-wake test proves the poll is parked before the event lands time.sleep(0.3) did not establish that the client had reached the empty-feed wait, so if HTTP setup ran slow the orchestrator event landed early and even a poll that never re-read the store returned it on the first read. Wrap the state's events_after to signal when a read has come back empty, wait on that signal, and only then emit; a new test pins that a slow store read no longer stalls a concurrent request on the same app. --- .../tests/test_events_orchestrator_wake.py | 81 ++++++++++++++++++- 1 file changed, 77 insertions(+), 4 deletions(-) diff --git a/backend/server/api/tests/test_events_orchestrator_wake.py b/backend/server/api/tests/test_events_orchestrator_wake.py index c5d024a07..69a5cee97 100644 --- a/backend/server/api/tests/test_events_orchestrator_wake.py +++ b/backend/server/api/tests/test_events_orchestrator_wake.py @@ -60,9 +60,9 @@ def server(tmp_path): thread.join(timeout=10) -def _create_capture(base_url: str) -> str: +def _create_capture(base_url: str, packet_id: str = "packet-orch-wake-0001") -> str: body = { - "packetId": "packet-orch-wake-0001", + "packetId": packet_id, "formatVersion": "0.4", "tier": "photo_only", "device": {"model": "iPhone", "systemVersion": "18", "appVersion": "1"}, @@ -71,7 +71,7 @@ def _create_capture(base_url: str) -> str: response = httpx.post( f"{base_url}/v1/captures", json=body, - headers={"Authorization": f"Bearer {TOKEN}", "Idempotency-Key": "packet-orch-wake-0001"}, + headers={"Authorization": f"Bearer {TOKEN}", "Idempotency-Key": packet_id}, timeout=10, ) assert response.status_code == 201, response.text @@ -97,10 +97,26 @@ def poll(): result["elapsed"] = time.perf_counter() - started result["response"] = response + # The orchestrator may only emit once the server has read an empty feed, + # so the event cannot land before the poll parks: an emit that arrives + # earlier is picked up by the first store read of any implementation, and + # the test would pass without exercising the wait at all. Wrap the state's + # events_after to signal when a read has come back empty. + empty_feed = threading.Event() + original_events_after = deps.state.events_after + + def spy_events_after(capture_id: str, seq: int, limit: int = 100): + rows = original_events_after(capture_id, seq, limit=limit) + if not rows: + empty_feed.set() + return rows + + deps.state.events_after = spy_events_after + client = threading.Thread(target=poll) client.start() try: - time.sleep(0.3) # the poll is parked in poll_events before the event lands + assert empty_feed.wait(timeout=10), "the poll never reached the empty-feed wait" record = orchestrator_events.emit( deps.state, capture_id, @@ -122,6 +138,63 @@ def poll(): ) +def test_a_slow_state_read_does_not_stall_other_requests_on_the_same_app(server): + """poll_events reads the store off the event loop, so one slow read cannot park the app. + + With ModalDictState every events_after is a synchronous remote read. Reading + it on the event loop blocks every other request for the length of the read, + roughly 40 times per quiet 20 s poll (Greptile P2, PR #290). The parked + poll's first read gates on a threading.Event here; a wait=0 request for a + second capture must return while that gate is still shut. + """ + + base_url, deps = server + parked_capture = _create_capture(base_url, "packet-orch-wake-0002") + probe_capture = _create_capture(base_url, "packet-orch-wake-0003") + + entered = threading.Event() + release = threading.Event() + original_events_after = deps.state.events_after + + def gated_events_after(capture_id: str, seq: int, limit: int = 100): + if capture_id == parked_capture: + entered.set() + release.wait(timeout=4.0) + return original_events_after(capture_id, seq, limit=limit) + + deps.state.events_after = gated_events_after + + def poll(): + httpx.get( + f"{base_url}/v1/captures/{parked_capture}/events", + params={"after": 0, "wait": 3}, + headers={"Authorization": f"Bearer {TOKEN}"}, + timeout=30, + ) + + poller = threading.Thread(target=poll) + poller.start() + try: + assert entered.wait(timeout=10), "the long-poll never reached its first store read" + started = time.perf_counter() + response = httpx.get( + f"{base_url}/v1/captures/{probe_capture}/events", + params={"after": 0, "wait": 0}, + headers={"Authorization": f"Bearer {TOKEN}"}, + timeout=30, + ) + probe_elapsed = time.perf_counter() - started + finally: + release.set() + poller.join(timeout=30) + + assert response.status_code == 200, response.text + assert probe_elapsed < 1.5, ( + f"a wait=0 request took {probe_elapsed:.3f} s while another poll's store read was in flight: " + "poll_events is reading the state store on the event loop" + ) + + def test_wait_zero_returns_immediately_when_the_feed_is_empty(server): """wait=0 stays an immediate empty-page read, with no parked wait."""