diff --git a/CHANGELOG.md b/CHANGELOG.md index 60c1b39..8222fde 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] ### Fixed +- **A saturated replica no longer reads as down to a gateway.** The standalone server's request limiter no longer sheds `GET /v1/models` or `/-/healthz` (Ray Serve's health path, a 404 here), which a gateway's readiness probe asks, and `/v1/models` is answered on the event loop rather than behind fits and predicts in the threadpool. - A predict reported `cache_outcome: hit` when the table was restored from disk while its rows were being read, as on a replica after a restart (its schema not yet recorded); the restore is now counted wherever it happens. ### Changed - `cache_outcome` is required in the predict response's schema (the server always sent it; generated clients now see it as always present). The coalescing batcher closes a batch and releases everyone in it even if its window's wait fails. diff --git a/src/tabctx/serve/service.py b/src/tabctx/serve/service.py index 808ec09..7f37f4a 100644 --- a/src/tabctx/serve/service.py +++ b/src/tabctx/serve/service.py @@ -1104,10 +1104,12 @@ def livez(self): @fastapi_app.get( "/v1/models", summary="List the models served", tags=["deployment"] ) - def models(self): + async def models(self): """Which models this deployment serves: pass one of these ids as `model` in fit requests. The first is the default.""" + # async: answered on the event loop, not from the threadpool that + # fits and predicts hold, so a readiness probe never waits on them. return { "data": [ { diff --git a/src/tabctx/serve/standalone.py b/src/tabctx/serve/standalone.py index aeb359b..d3ce0f2 100644 --- a/src/tabctx/serve/standalone.py +++ b/src/tabctx/serve/standalone.py @@ -12,8 +12,10 @@ - Request shedding (the deployment's max_ongoing_requests / max_queued_requests): the limiter below, configured by TABCTX_MAX_ONGOING_REQUESTS and TABCTX_MAX_QUEUED_REQUESTS (defaults - 8 / 16, matching serve/app.py). Health routes bypass it -- a saturated - service must still answer its probes. + 8 / 16, matching serve/app.py). Health routes, the model list and + Ray's /-/healthz path (a 404 here) bypass it -- a saturated service + must still answer its probes, and a gateway judges readiness from the + list after asking /-/healthz. - Multi-replica session affinity: nothing replaces it. Run ONE replica, or front several with a router that pins the session-affinity header (serve/affinity.py) to a replica. @@ -43,7 +45,11 @@ DEFAULT_MAX_QUEUED = 16 # Never shed probes: a platform restarts a replica whose liveness probe # times out, and a loaded replica is exactly when that must not happen. -_UNSHED_PATHS = frozenset({"/livez", "/readyz", "/healthz"}) +# A gateway's readiness probe is one too (e.g. aieng-proxy-gateway's, +# behind the inference site's status page): it asks Ray Serve's health +# path, /-/healthz (a 404 here: not Ray), then reports a model ready only +# while /v1/models names it. Neither reads engine state. +_UNSHED_PATHS = frozenset({"/livez", "/readyz", "/healthz", "/v1/models", "/-/healthz"}) def _positive_int_env(name: str, default: int) -> int: diff --git a/tests/unit/test_standalone_service.py b/tests/unit/test_standalone_service.py index 1366df7..796b8f8 100644 --- a/tests/unit/test_standalone_service.py +++ b/tests/unit/test_standalone_service.py @@ -10,9 +10,11 @@ import asyncio import subprocess import sys +import threading import time import urllib.parse +import anyio.to_thread import httpx import pytest from fastapi.responses import JSONResponse @@ -22,6 +24,7 @@ from tabctx.serve.factory import BACKEND_ENV_VAR from tabctx.serve.service import ROUTE_SPECS, TabctxService, fastapi_app from tabctx.serve.standalone import ( + _UNSHED_PATHS, MAX_ONGOING_ENV_VAR, MAX_QUEUED_ENV_VAR, _RequestLimiter, @@ -232,6 +235,22 @@ def test_every_registered_route_is_rebound(self, app): assert want <= got # the made app adds only its own docs routes +async def _until(condition, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while not condition(): + assert time.monotonic() < deadline, "timed out" + await asyncio.sleep(0.005) + + +def _limiter_of(app) -> _RequestLimiter: + """The app's limiter instance (built with its middleware stack on the + first request).""" + layer = app.middleware_stack + while not isinstance(layer, _RequestLimiter): + layer = layer.app + return layer + + class TestRequestLimiter: def test_sheds_beyond_the_queue_and_never_sheds_probes(self): async def scenario(): @@ -239,7 +258,7 @@ async def scenario(): started = asyncio.Event() async def app(scope, receive, send): - if scope["path"] == "/livez": + if scope["path"] in _UNSHED_PATHS: await JSONResponse({"status": "alive"})(scope, receive, send) return started.set() @@ -264,8 +283,8 @@ async def app(scope, receive, send): assert shed.status_code == 503 assert "retry" in shed.json()["detail"] - probe = await client.get("/livez") # bypasses the limiter - assert probe.status_code == 200 + for probe in _UNSHED_PATHS: # bypass the limiter + assert (await client.get(probe)).status_code == 200, probe release.set() assert (await first).status_code == 200 @@ -273,6 +292,65 @@ async def app(scope, receive, send): asyncio.run(scenario()) + def test_unshed_paths_are_the_probe_routes(self): + # Each is a GET route the server answers, except Ray's /-/healthz, + # which it must answer (404) just as fast. + routes = {s.path for s in ROUTE_SPECS if s.methods == ("GET",)} + assert _UNSHED_PATHS - routes == {"/-/healthz"} + + def test_a_gateway_probe_answers_a_saturated_server(self, monkeypatch): + # The real app, its one slot and one queue place taken, and the + # threadpool's one thread held by that slot's work: the probe + # aieng-proxy-gateway runs (/-/healthz must 404, then /v1/models + # must name the model) still answers at once, while work is shed. + monkeypatch.setenv(BACKEND_ENV_VAR, "fake") + monkeypatch.delenv(REQUIRE_TENANT_ENV_VAR, raising=False) + monkeypatch.setenv(MAX_ONGOING_ENV_VAR, "1") + monkeypatch.setenv(MAX_QUEUED_ENV_VAR, "1") + service = TabctxService() + held = threading.Event() + release = threading.Event() + limits = service.limits + + def slow_limits(): # the handler's signature: FastAPI reads it + held.set() + release.wait(10) + return limits() + + service.limits = slow_limits + app = make_app(service) + + async def scenario(): + anyio.to_thread.current_default_thread_limiter().total_tokens = 1 + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, base_url="http://test" + ) as client: + first = asyncio.create_task(client.get("/v1/tabctx/limits")) + await _until(held.is_set) # holds the one slot + limiter = _limiter_of(app) + second = asyncio.create_task(client.get("/v1/tabctx/limits")) + await _until(lambda: limiter._queued == 1) # the queue place + assert (await client.get("/v1/tabctx/datasets")).status_code == 503 + + async def probe() -> httpx.Response: + assert (await client.get("/-/healthz")).status_code == 404 + return await client.get("/v1/models") + + models = await asyncio.wait_for(probe(), timeout=1.0) + assert models.status_code == 200 + assert [m["id"] for m in models.json()["data"]] == ["fake"] + + release.set() + assert (await first).status_code == 200 + assert (await second).status_code == 200 + + try: + asyncio.run(scenario()) + finally: + release.set() + service.close() + def test_rejects_bad_limit_configuration(self, monkeypatch): monkeypatch.setenv(BACKEND_ENV_VAR, "fake") monkeypatch.setenv(MAX_ONGOING_ENV_VAR, "0")