fix(gateway): a Chronos fire that lands during startup waits for the gateway's adapters instead of running with none
(cherry picked from commit cd1314e0ed830a8b5af689297dde4a434276168e)
This commit is contained in:
@@ -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":
|
||||
|
||||
46
gateway/platforms/api_server_fire_startup.py
Normal file
46
gateway/platforms/api_server_fire_startup.py
Normal file
@@ -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"})
|
||||
114
tests/gateway/test_cron_fire_startup_gate.py
Normal file
114
tests/gateway/test_cron_fire_startup_gate.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user