fix(gateway): a fire waiting on startup is refused at once when the gateway starts draining

A gateway that is stopped during boot never reaches _running, so a fire parked in the startup
wait held its drain reservation for the full budget before refusing, and stalled the shutdown
drain with it. The wait now exits on _draining / _external_drain_active with the same retryable
503 it returns on timeout.

gateway/platforms/api_server_fire_startup.py

@@ -32,7 +32,9 @@ async def refuse_until_started(runner: Any, job_id: str, *, received_at: float)
     loop = asyncio.get_running_loop()
     deadline = received_at + FIRE_STARTUP_WAIT_SECONDS
     while runner is not None and not getattr(runner, "_running", True):
-        if loop.time() >= deadline:
+        # A gateway stopping mid-boot never reaches _running: refuse now, not after the full wait.
+        draining = getattr(runner, "_draining", False) or getattr(runner, "_external_drain_active", False)
+        if draining or loop.time() >= deadline:
             return web.json_response(
                 {"error": "gateway unreachable; retry", "job_id": job_id}, status=503, headers={"Retry-After": "60"})
         await asyncio.sleep(_POLL_SECONDS)

tests/gateway/test_cron_fire_startup_gate.py

@@ -106,6 +106,24 @@ async def test_fire_is_retryable_and_never_claimed_when_startup_does_not_finish(
     assert provider.claimed == [] and provider.fired == []  # nothing durably claimed for a fire we refused
 
 
+@pytest.mark.asyncio
+async def test_gateway_that_starts_draining_mid_boot_is_refused_without_the_full_wait(
+        adapter, provider, monkeypatch):
+    monkeypatch.setattr(api_server_fire_startup, "FIRE_STARTUP_WAIT_SECONDS", 30.0)
+    runner = SimpleNamespace(_draining=False, _external_drain_active=False, _running=False, adapters={})
+
+    async def _stop_during_startup():
+        await asyncio.sleep(0.3)
+        runner._draining = True
+
+    stopper = asyncio.ensure_future(_stop_during_startup())
+    resp = await asyncio.wait_for(_post_fire(adapter, runner), timeout=10.0)
+    await stopper
+
+    assert resp.status == 503 and resp.headers["Retry-After"] == "60"
+    assert provider.claimed == [] and provider.fired == []
+
+
 @pytest.mark.asyncio
 async def test_started_gateway_is_accepted_immediately(adapter, provider):
     runner = SimpleNamespace(_draining=False, _external_drain_active=False, _running=True, adapters={"relay": object()})