diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 530f80c39a..568103f37f 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -4031,13 +4031,13 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): from cron.scheduler_provider import provider_supports_split_fire, resolve_cron_scheduler provider = resolve_cron_scheduler() loop = asyncio.get_running_loop() - # Live adapters (parity with the built-in ticker): E2EE / relay-fronted platforms - # have no native credential, so without them delivery fails. runner = self.gateway_runner or request.app.get("gateway_runner") if runner is None: with suppress(Exception): from gateway.run import _gateway_runner_ref runner = _gateway_runner_ref() + from gateway.platforms.api_server_fire_startup import refuse_until_started # a mid-boot fire waits for adapters + if (refusal := await refuse_until_started(runner, job_id)) is not None: return refusal adapters = getattr(runner, "adapters", None) or None def _detach_fire(fire_fn, *fire_args) -> "web.Response": diff --git a/gateway/platforms/api_server_fire_startup.py b/gateway/platforms/api_server_fire_startup.py new file mode 100644 index 0000000000..39f3c5bdaf --- /dev/null +++ b/gateway/platforms/api_server_fire_startup.py @@ -0,0 +1,46 @@ +"""Startup gate for the Chronos fire webhook (``POST /api/cron/fire``). + +A backend that stops the guest on sleep wakes the agent FOR the fire, so the fire reaches the webhook in +the first second of a cold boot. The api_server adapter listens before the gateway has published any +adapter into ``runner.adapters`` (an adapter is registered only after its ``connect()`` returns), so a +fire taken then ran its job with no live adapters and a relay-fronted platform failed with "has no live +gateway transport" (the result was lost). Wait for the gateway to finish starting instead. +""" + +from __future__ import annotations + +import asyncio +from typing import Any, Optional + +from aiohttp import web + +# How long a fire that lands during startup waits before it is refused as retryable. A cold boot +# publishes its adapters within seconds; the bound keeps a gateway that never finishes starting from +# holding the request open (NAS's callback timeout is 30s). +FIRE_STARTUP_WAIT_SECONDS = 10.0 +_POLL_SECONDS = 0.1 + + +async def await_gateway_started(runner: Any, timeout: Optional[float] = None) -> bool: + """True once ``runner`` has finished starting (``_running``: every platform connect attempted and its + adapters published), or when there is no runner to ask (a self-hosted api_server, a test double). + False when it is still starting after ``timeout`` seconds; the caller answers a retryable 503.""" + if runner is None or getattr(runner, "_running", True): + return True + loop = asyncio.get_running_loop() + deadline = loop.time() + (FIRE_STARTUP_WAIT_SECONDS if timeout is None else timeout) + while not runner._running: + if loop.time() >= deadline: + return False + await asyncio.sleep(_POLL_SECONDS) + return True + + +async def refuse_until_started(runner: Any, job_id: str) -> Optional["web.Response"]: + """Wait out a cold boot so the fire's job gets the live adapters (E2EE and relay-fronted platforms have + no native credential, so without them delivery fails). None when the gateway is up; otherwise the + retryable 503, worded like the dashboard's own "gateway unreachable" so NAS classifies it the same.""" + if await await_gateway_started(runner): + return None + return web.json_response( + {"error": "gateway unreachable; retry", "job_id": job_id}, status=503, headers={"Retry-After": "2"}) diff --git a/tests/gateway/test_cron_fire_startup_gate.py b/tests/gateway/test_cron_fire_startup_gate.py new file mode 100644 index 0000000000..1d109ae488 --- /dev/null +++ b/tests/gateway/test_cron_fire_startup_gate.py @@ -0,0 +1,114 @@ +"""A Chronos fire that lands while the gateway is still starting waits for its adapters (cold boot race). + +A backend that stops the guest on sleep (Azure) wakes the agent FOR the fire, so the fire reaches +``POST /api/cron/fire`` in the first second of a cold boot. The api_server adapter listens before the +gateway has published any adapter into ``runner.adapters`` (``_publish_primary_adapter`` registers an +adapter only after its ``connect()`` returns), so the handler's snapshot was an empty dict, which +``or None`` turned into ``None``. The job then ran with no adapters and a relay-fronted platform failed +with "has no live gateway transport", losing the result. The handler must wait for the gateway to finish +starting (``runner._running``) and then hand over the LIVE adapters. +""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from aiohttp import web +from aiohttp.test_utils import TestClient, TestServer + +from gateway.config import PlatformConfig +from gateway.platforms import api_server_fire_startup +from gateway.platforms.api_server import APIServerAdapter, cors_middleware + + +@pytest.fixture +def adapter(): + return APIServerAdapter(PlatformConfig(enabled=True, extra={"key": "sk-secret"})) + + +class _SpyProvider: + def __init__(self): + self.claimed = [] + self.fired = [] + + def claim_fire(self, job_id): + self.claimed.append(job_id) + return {"id": job_id, "execution_id": f"exec-{job_id}"} + + def fire_claimed(self, job, *, adapters=None, loop=None): + self.fired.append((job["id"], adapters)) + return True + + +@pytest.fixture +def provider(monkeypatch): + spy = _SpyProvider() + monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy) + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: (lambda **kw: {"purpose": "cron_fire"}), + ) + return spy + + +async def _post_fire(adapter, runner): + app = web.Application(middlewares=[cors_middleware]) + app["api_server_adapter"] = adapter + app.router.add_post("/api/cron/fire", adapter._handle_cron_fire) + with patch("gateway.run._gateway_runner_ref", lambda: runner): + async with TestClient(TestServer(app)) as cli: + return await cli.post( + "/api/cron/fire", headers={"Authorization": "Bearer good"}, json={"job_id": "job-1"}) + + +async def _wait_for(predicate, timeout=2.0): + for _ in range(int(timeout / 0.01)): + if predicate(): + return True + await asyncio.sleep(0.01) + return False + + +@pytest.mark.asyncio +async def test_fire_during_startup_waits_and_receives_the_live_adapters(adapter, provider): + """The cold-boot shape: api_server already accepting, adapters not yet published.""" + relay = object() + runner = SimpleNamespace(_draining=False, _external_drain_active=False, _running=False, adapters={}) + + async def _finish_startup(): + await asyncio.sleep(0.3) + runner.adapters["relay"] = relay + runner._running = True + + finisher = asyncio.ensure_future(_finish_startup()) + resp = await _post_fire(adapter, runner) + await finisher + + assert resp.status == 202 + assert await _wait_for(lambda: provider.fired) + _job_id, adapters = provider.fired[0] + assert adapters is runner.adapters and adapters.get("relay") is relay + + +@pytest.mark.asyncio +async def test_fire_is_retryable_and_never_claimed_when_startup_does_not_finish(adapter, provider, monkeypatch): + monkeypatch.setattr(api_server_fire_startup, "FIRE_STARTUP_WAIT_SECONDS", 0.2) + runner = SimpleNamespace(_draining=False, _external_drain_active=False, _running=False, adapters={}) + + resp = await _post_fire(adapter, runner) + + assert resp.status == 503 + assert provider.claimed == [] and provider.fired == [] # nothing durably claimed for a fire we refused + + +@pytest.mark.asyncio +async def test_started_gateway_is_not_delayed(adapter, provider): + runner = SimpleNamespace(_draining=False, _external_drain_active=False, _running=True, adapters={"relay": object()}) + + loop = asyncio.get_running_loop() + started = loop.time() + resp = await _post_fire(adapter, runner) + + assert resp.status == 202 + assert loop.time() - started < 1.0