diff --git a/README.md b/README.md index 720e2ac..79abe20 100644 --- a/README.md +++ b/README.md @@ -198,8 +198,9 @@ audit-closure sprint: (ADR 0011: Order 360, stuck-orders worklist, exception inbox), and the three-node demo topology (ADR 0012) implemented and deployed to Hugging Face Spaces (the `center` hub and the `spb` edge answer live; `ekb` and - the standalone demo Space are paused — the free tier keeps three - `cpu-basic` Spaces awake) — plus the G2 audit closure (spec/seed + the standalone demo Space are paused — the free tier caps how many + `cpu-basic` Spaces one account runs at once, and other projects hold the + rest) — plus the G2 audit closure (spec/seed consistency, journal-scan hardening, live evidence re-captures). The tagged line and `main` are in sync as of `v2.0.0`. See the diff --git a/docs/dv2-multi-branch/RELEASE_STATUS.md b/docs/dv2-multi-branch/RELEASE_STATUS.md index 277202e..f6340ec 100644 --- a/docs/dv2-multi-branch/RELEASE_STATUS.md +++ b/docs/dv2-multi-branch/RELEASE_STATUS.md @@ -53,10 +53,16 @@ Scorecard channel (5.8 → 7.0). ## Live demo surfaces Four Docker Spaces exist under `liovina`, all built from the same image. -The free tier keeps **three** `cpu-basic` Spaces awake at a time, and the -account runs other projects, so the three-node topology is deployed but not -fully awake. A paused Space answers `503` until it is restarted; nothing -about the deployment is missing, only the compute quota. +The free tier caps how many `cpu-basic` Spaces one account may run at once; +that cap is currently reached by **four** running Spaces on this account, two of +which serve other projects. `POST .../restart` on a paused Space is refused +outright (`403`, "you've reached your cpu-basic quota limit"), and a second +account is not a way around it — creating an additional free Docker Space is +refused too (`POST /api/repos/create` → `402`; only static Spaces are free). + +So the three-node topology is deployed but not fully awake. A paused Space +answers `503` until a running one is paused to make room; nothing about the +deployment is missing, only the concurrent-compute quota. | Space | Role | Runtime stage | `/v1/health` | |-------|------|---------------|--------------| @@ -65,7 +71,8 @@ about the deployment is missing, only the compute quota. | [`agentflow-edge-ekb`](https://liovina-agentflow-edge-ekb.hf.space) | edge branch `ekb` | PAUSED | `503` | | [`agentflow-demo`](https://liovina-agentflow-demo.hf.space) | standalone demo | PAUSED | `503` | -Probed 2026-07-09 (`GET /v1/health` + `GET /api/spaces/liovina/{name}`). +Probed 2026-07-09 (`GET /v1/health` + `GET /api/spaces/liovina/{name}`; the two +quota errors above were reproduced the same day). The cross-node evidence ("Verify live" in `deploy/hf-space/three-node/DEPLOY.md`) was captured on 2026-07-06 while `ekb` was awake. diff --git a/docs/perf/usage-write-bifurcation-2026-07-09.md b/docs/perf/usage-write-bifurcation-2026-07-09.md new file mode 100644 index 0000000..925fb26 --- /dev/null +++ b/docs/perf/usage-write-bifurcation-2026-07-09.md @@ -0,0 +1,124 @@ +# The `Load Test` was not bimodal because of the runner + +**Date:** 2026-07-09 · **Status:** root cause found and fixed · **Supersedes:** the +"CI runner speed" reading of finding N1 in `plan_07_07_26.md`. + +## What we thought + +`Load Test` runs split into two clusters — roughly 44 rps at p99 ~150 ms, or +roughly 29 rps at p99 ~1500 ms against thresholds of 900–1200 ms. Runs of both +kinds existed before and after the B1 usage-write fix, so the split was read as +runner speed and filed as "not code". The proposed remedies were all variations +on making the gate tolerate it: re-run on red, take the best of two runs, or +normalise the threshold against measured rps. + +## Why that reading does not survive arithmetic + +Across the red/green runs, rps varies by 1.5× while p99 varies by 10×. A +machine that is 1.5× slower does not produce a 10× tail; it produces a 1.5× +tail. The nonlinearity means something in the system amplifies a small change +in service time, which is the signature of a queue, not of a slow CPU. + +The decisive clue is in the spread. If the red branch were "an unlucky runner", +red runs would scatter. They do not: + +| branch | runs | aggregate rps | +|--------|------|---------------| +| red | 3 | 29.4 · 29.1 · 28.9 — spread 1.7% | +| green | 9 | 37.0 … 46.2 — spread 25% | + +Three independent red runs land within 1.7% of each other. That is an +attractor, not bad luck. + +## The mechanism + +Every authenticated request wrote its own `api_usage` row before the response +was produced (`AuthMiddleware` → `run_in_threadpool(record_usage)` → +`EmbeddedControlPlaneStore.record_api_usage`). DuckDB serializes writers and +commits per row, so each request queued behind one fsync. + +The load client is closed-loop: 15 users, `wait_time = between(0.1, 0.5)`. Let +`s` be the usage-write service time. + +- **Unsaturated branch.** The writer keeps up, so rps is bounded by think time: + `15 / (0.3 + L) ≈ 48.9` at small `L`. Observed green maximum: 46.2 rps. +- **Saturated branch.** Requests queue at the serialized writer, so + `L ≈ C·s` where the server-side concurrency is `C = rps · L`. Substituting + gives `rps = 1/s` — **independent of the number of users**. That is why the + red runs agree with each other: they are all sitting on `1/s`. + +At 29 rps the implied `s` is ≈ 34 ms, a plausible fsync on a slow shared disk. +The bifurcation point is `s ≈ 1/48.9 ≈ 20 ms`: below it the run is think-time +bound and fast, above it the run collapses onto `1/s`. The runner's disk merely +decides which side of 20 ms the commit lands on. **The cap is ours.** + +## Evidence + +**1. The writer serializes (isolated, `duckdb 1.5.1`, one connection, cursors +per call, the shipped retry loop).** p50 grows linearly with concurrency and +throughput saturates — the shape of a single-server queue. The retry/backoff +path was never entered, so it is not the amplifier: + +| threads | 1 | 2 | 4 | 8 | 15 | 30 | +|---------|---|---|---|---|----|----| +| p50 (ms) | 2.6 | 4.1 | 9.0 | 18.0 | 34.8 | 62.7 | +| inserts/s | 148 | 244 | 275 | 326 | 325 | 374 | + +**2. The collapse reproduces on the shipped code.** Serving the real API with +the CI load profile and `sleep(s)` injected into `record_api_usage`: + +| `s` | rps | p50 (ms) | p99 (ms) | +|-----|-----|----------|----------| +| 0 ms | 43.9 | 27 | 100 | +| 10 ms | 37.8 | 76 | 250 | +| 25 ms | 36.6 | 82 | 320 | +| **34 ms** | **31.4** | **160** | 430 | +| 60 ms | 8.3 | 110 | 21000 | + +At `s = 34 ms` the harness lands on 31.4 rps / p50 160 ms; CI's red branch is +29.1 rps / p50 161 ms. + +**3. Taking the write off the request path removes the sensitivity.** Same +injected `s`, rows enqueued to a background writer: + +| `s` | in-path | off-path | +|-----|---------|----------| +| 34 ms | 31.4 rps · p99 430 | 37.9 rps · p99 340 | +| 60 ms | 8.3 rps · p99 21000 · **FAIL** | 37.2 rps · p99 320 · PASS | + +Off-path throughput is flat in `s`. In-path throughput is `1/s`. + +## The fix + +`src/serving/api/auth/usage_writer.py`. The request enqueues a `UsageRow` and +returns; one background thread drains the queue and writes. + +Batching is not an optimisation here, it is part of the fix. A per-row +background writer would still commit at `1/s` rows per second — below the +request rate the API can otherwise serve — so the ceiling would simply move +from request latency into a queue that silently overflows. `record_api_usage_batch` +puts one commit under N rows, lifting the accounting ceiling to `N/s`. + +## What this costs + +Durability moves from "committed before the response" to "committed shortly +after". A crash loses at most the queued rows. `api_usage` backs one admin read +(`GET /v1/admin/usage`) — it is not billing and not rate limiting — and rows +were already droppable when the store exhausted its retries. Reads that must +see their own writes call `flush_usage`; the API lifespan closes the writer on +shutdown. A full queue sheds rows into `agentflow_usage_rows_dropped_total` +rather than stalling the request it was counting. + +## What this does not claim + +The CI disk is still variable, and a heavily degraded runner can still miss a +p99 threshold. What is gone is the *amplifier*: a 1.5× slower commit now costs +a 1.5× slower commit, not a collapse of the whole API onto `1/s`. Finding N1 +therefore needs no threshold normalisation — the gate was reporting a real +defect, and the honest response was to fix it rather than widen the gate. + +## Beyond CI + +This was never only a CI property. Any deployment on the embedded control plane +served at most `1 / commit_latency` authenticated requests per second, and every +request paid the queueing delay. The Load Test was the only place it was visible. diff --git a/src/serving/api/auth/key_rotation.py b/src/serving/api/auth/key_rotation.py index 4f6a852..66a65e5 100644 --- a/src/serving/api/auth/key_rotation.py +++ b/src/serving/api/auth/key_rotation.py @@ -196,6 +196,9 @@ def shutdown(self) -> None: def old_key_usage_by_key_id(self) -> dict[str, int]: # ADR 0010 slice 4: routed through the ControlPlaneStore port — # was a direct connect_duckdb(self._manager.db_path) query. + # api_usage rows are written off the request path, so every reader of + # the table drains the writer first or it counts a stale total. + self._manager.flush_usage() return self._manager.store.get_old_key_usage_by_key_id() def old_key_usage_last_hour(self, key_id: str) -> int: @@ -204,6 +207,7 @@ def old_key_usage_last_hour(self, key_id: str) -> int: def _usage_by_key(self) -> dict[tuple[str, str], int]: # ADR 0010 slice 4: routed through the ControlPlaneStore port — # was a direct connect_duckdb(self._manager.db_path) query. + self._manager.flush_usage() return self._manager.store.get_usage_by_key() def write_config(self, config: ApiKeysConfig) -> None: diff --git a/src/serving/api/auth/manager.py b/src/serving/api/auth/manager.py index a142143..5c7c815 100644 --- a/src/serving/api/auth/manager.py +++ b/src/serving/api/auth/manager.py @@ -191,8 +191,12 @@ def __init__( if rate_limiter is None and resolved_redis_url is None: self.rate_limiter._redis = None from .key_rotation import KeyRotator + from .usage_writer import UsageWriter self._key_rotator: KeyRotator = KeyRotator(self) + # Constructed eagerly, but its thread starts on the first submitted row + # — most AuthManagers (tests, CLI) never record a request. + self._usage_writer = UsageWriter(self.store, self.audit_publisher) def load(self) -> None: with self._config_lock: @@ -444,16 +448,42 @@ def is_entity_allowed(self, tenant_key: TenantKey, entity_type: str) -> bool: return entity_type in tenant_key.allowed_entity_types def record_usage(self, tenant_key: TenantKey, endpoint: str) -> None: + """Write the row synchronously and durably. Kept for callers that want + the row on disk when this returns; the request path uses + ``submit_usage`` instead.""" from .usage_table import record_usage record_usage(self, tenant_key, endpoint) + def submit_usage(self, tenant_key: TenantKey, endpoint: str) -> bool: + """Hand the row to the off-path writer. Never blocks, never raises.""" + from src.serving.control_plane.store import UsageRow + + return self._usage_writer.submit( + UsageRow( + tenant=tenant_key.tenant, + key_name=tenant_key.name, + endpoint=endpoint, + key_id=tenant_key.key_id, + key_slot=tenant_key.matched_slot, + ) + ) + + def flush_usage(self, timeout: float = 5.0) -> bool: + """Block until queued usage rows are written — read-your-writes.""" + return self._usage_writer.flush(timeout) + + def close_usage_writer(self, timeout: float = 5.0) -> None: + self._usage_writer.close(timeout) + def list_keys_with_usage(self) -> list[dict]: + # KeyRotator._usage_by_key flushes — every api_usage reader does. return self._key_rotator.list_keys_with_usage() def usage_by_tenant(self) -> list[dict]: from .usage_table import usage_by_tenant + self.flush_usage() return usage_by_tenant(self) def create_key(self, payload: KeyCreateRequest) -> TenantKey: diff --git a/src/serving/api/auth/middleware.py b/src/serving/api/auth/middleware.py index f8757b3..5616902 100644 --- a/src/serving/api/auth/middleware.py +++ b/src/serving/api/auth/middleware.py @@ -9,10 +9,9 @@ import structlog from fastapi import Header, HTTPException, Request, Response from fastapi.responses import JSONResponse -from starlette.concurrency import run_in_threadpool from src.constants import DEFAULT_RATE_LIMIT_WINDOW_SECONDS, FAILED_AUTH_WINDOW_SECONDS -from src.serving.api.metrics import AUTH_FAILURES, USAGE_RECORD_FAILURES +from src.serving.api.metrics import AUTH_FAILURES from src.serving.api.security import redact_sensitive_headers from .manager import _CURRENT_TENANT_ID, TenantKey, get_auth_manager @@ -107,28 +106,25 @@ async def __call__( ) manager.clear_failed_auth(client_ip) - # record_usage opens a DuckDB connection, writes, and retries with a - # blocking sleep; running it inline froze the event loop on every - # authenticated request. Offload to a worker thread. (audit_28_06_26.md #13) + # Usage accounting is a side-channel: it may not fail the request it is + # counting, and it may not pace it either. Writing the row here — even + # offloaded to a worker thread — put a serialized DuckDB commit on the + # critical path of every authenticated request, capping the API at + # `1 / commit_latency` rps and tipping the CI load test into a + # saturated equilibrium whenever the runner's disk was slow + # (docs/perf/usage-write-bifurcation-2026-07-09.md). # - # Usage accounting is a side-channel. The store deliberately raises on - # exhausted retries (`ControlPlaneStore.record_api_usage`) so that - # `record_usage` skips its audit publish — but that exception used to - # escape here and turn an otherwise-successful request into a 500 - # (seen under load, 2026-07-09). Count the dropped row and serve the - # request; the counter is the thing to alert on, not the client. - try: - await run_in_threadpool(manager.record_usage, tenant_key, path) - except Exception: + # Hand the row to the writer thread and move on. A full queue sheds the + # row and counts it; a failed write counts it too. Both counters are + # what to alert on, never the client. + if not manager.submit_usage(tenant_key, path): from src.serving.api import auth as auth_package - USAGE_RECORD_FAILURES.inc() auth_package.logger.warning( "api_usage_record_skipped", tenant=tenant_key.tenant, key_name=tenant_key.name, path=path, - exc_info=True, ) is_allowed, remaining, reset_at = await manager.check_rate_limit(tenant_key) rate_limit_headers = { diff --git a/src/serving/api/auth/usage_writer.py b/src/serving/api/auth/usage_writer.py new file mode 100644 index 0000000..b094b6a --- /dev/null +++ b/src/serving/api/auth/usage_writer.py @@ -0,0 +1,224 @@ +"""Off-request-path writer for ``api_usage`` rows. + +Why this exists +--------------- + +Every authenticated request used to write its own ``api_usage`` row before the +response was produced. The embedded store serializes writers and commits each +row, so that write put an fsync on the critical path of every request and +capped the whole API at ``1 / commit_latency`` requests per second. + +That cap is why the ``Load Test`` workflow read as bimodal. The load client is +closed-loop (15 users, 0.1-0.5 s think time), so it has two equilibria: an +unsaturated one bounded by think time (~46 rps, p50 7 ms), and a saturated one +pinned at ``rps = 1 / commit_latency``. A runner whose disk pushed the commit +past ~20 ms tipped the run onto the saturated branch, where three separate red +runs landed within 1.7% of each other (29.4 / 29.1 / 28.9 rps) while green runs +spread across 37-46. Measured, not inferred: +``docs/perf/usage-write-bifurcation-2026-07-09.md``. + +The contract this keeps +----------------------- + +Accounting is a side-channel. It already could not fail a request (the store +raises on exhausted retries and the caller counts the drop). It must not +*pace* one either. So the request enqueues a row and returns; a single writer +thread drains the queue in batches, one commit per batch. + +What that trades +---------------- + +Durability moves from "the row is committed before the response" to "the row is +committed shortly after". A crash loses at most the queued rows. ``api_usage`` +feeds one admin read (``GET /v1/admin/usage``); it is not billing and not rate +limiting, and rows were already droppable on exhausted retries. Reads that must +see their own writes call ``flush``; the API lifespan closes the writer on +shutdown. + +Backpressure is bounded, never blocking: a full queue drops the row and counts +it in ``agentflow_usage_rows_dropped_total``. Dropping a counter row is +strictly better than stalling the request it was counting -- that stall is the +bug this module exists to remove. +""" + +from __future__ import annotations + +import queue +import threading +from typing import TYPE_CHECKING, Protocol + +import structlog + +from src.serving.api.metrics import USAGE_RECORD_FAILURES, USAGE_ROWS_DROPPED +from src.serving.control_plane.store import UsageRow + +if TYPE_CHECKING: + from src.serving.control_plane.store import ControlPlaneStore + +logger = structlog.get_logger(__name__) + +DEFAULT_MAX_QUEUE = 10_000 +DEFAULT_MAX_BATCH = 256 +_SHUTDOWN = object() + + +class AuditPublisher(Protocol): + def publish(self, payload: dict) -> object: ... + + +class UsageWriter: + """Drains queued ``api_usage`` rows on one background thread. + + The thread starts on the first ``submit`` so that constructing an + ``AuthManager`` — which tests do constantly — costs no thread. + """ + + def __init__( + self, + store: ControlPlaneStore, + audit_publisher: AuditPublisher | None = None, + *, + max_queue: int = DEFAULT_MAX_QUEUE, + max_batch: int = DEFAULT_MAX_BATCH, + ) -> None: + self._store = store + self._audit_publisher = audit_publisher + self._max_batch = max_batch + self._queue: queue.Queue = queue.Queue(maxsize=max_queue) + self._thread: threading.Thread | None = None + self._start_lock = threading.Lock() + self._closed = False + + # -- request path ----------------------------------------------------- + + def submit(self, row: UsageRow) -> bool: + """Enqueue one row. Never blocks, never raises. + + Returns ``False`` when the row was dropped (queue full, or the writer + is closed); the caller has already served its request either way. + """ + if self._closed: + return False + self._ensure_thread() + try: + self._queue.put_nowait(row) + except queue.Full: + USAGE_ROWS_DROPPED.inc() + logger.warning("api_usage_queue_full", endpoint=row.endpoint, tenant=row.tenant) + return False + return True + + # -- lifecycle -------------------------------------------------------- + + def _ensure_thread(self) -> None: + if self._thread is not None: + return + with self._start_lock: + if self._thread is not None: + return + thread = threading.Thread(target=self._run, name="agentflow-usage-writer", daemon=True) + thread.start() + self._thread = thread + + def flush(self, timeout: float = 5.0) -> bool: + """Block until every queued row has been written (or dropped). + + Returns ``False`` on timeout. Callers that must read their own writes + (``GET /v1/admin/usage``, tests) go through here. + """ + if self._thread is None: + return True + done = threading.Event() + try: + self._queue.put_nowait(done) + except queue.Full: + return False + return done.wait(timeout) + + def close(self, timeout: float = 5.0) -> None: + """Flush, stop the thread, and refuse further rows.""" + if self._closed: + return + self._closed = True + thread = self._thread + if thread is None: + return + try: + self._queue.put_nowait(_SHUTDOWN) + except queue.Full: # pragma: no cover - a full queue still drains + pass + thread.join(timeout) + self._thread = None + + # -- writer thread ---------------------------------------------------- + + def _run(self) -> None: + while True: + first = self._queue.get() + if first is _SHUTDOWN: + return + batch, waiters, stop = self._drain(first) + if batch: + self._write(batch) + for waiter in waiters: + waiter.set() + if stop: + return + + def _drain(self, first: object) -> tuple[list[UsageRow], list[threading.Event], bool]: + """Take ``first`` plus whatever else is already queued, up to one batch. + + Flush markers and the shutdown sentinel ride the same queue, so a + marker is only signalled after every row queued ahead of it is written. + """ + batch: list[UsageRow] = [] + waiters: list[threading.Event] = [] + stop = False + item = first + while True: + if item is _SHUTDOWN: + stop = True + elif isinstance(item, threading.Event): + waiters.append(item) + else: + batch.append(item) # type: ignore[arg-type] + if stop or len(batch) >= self._max_batch: + break + try: + item = self._queue.get_nowait() + except queue.Empty: + break + return batch, waiters, stop + + def _write(self, batch: list[UsageRow]) -> None: + try: + self._store.record_api_usage_batch(batch) + except Exception: + # The store exhausted its retries. Count the rows we dropped and + # skip their audit publish, exactly as the synchronous path did: + # a publish must never claim a row that was never inserted. + USAGE_RECORD_FAILURES.inc(len(batch)) + logger.warning("api_usage_batch_dropped", rows=len(batch), exc_info=True) + return + if self._audit_publisher is None: + return + for row in batch: + try: + self._audit_publisher.publish( + { + "event_type": "api_usage", + "tenant": row.tenant, + "key_name": row.key_name, + "endpoint": row.endpoint, + "key_id": row.key_id, + "key_slot": row.key_slot, + } + ) + except Exception: + logger.warning( + "audit_publish_failed", + tenant=row.tenant, + endpoint=row.endpoint, + key_id=row.key_id, + exc_info=True, + ) diff --git a/src/serving/api/main.py b/src/serving/api/main.py index f147393..50305b5 100644 --- a/src/serving/api/main.py +++ b/src/serving/api/main.py @@ -311,6 +311,9 @@ async def dispatch_new_events_with_cache_invalidation() -> None: if getattr(app.state, "node_emitter", None) is not None: await app.state.node_emitter.stop() await app.state.query_cache.close() + # Drain the queued api_usage rows before the process goes away; they are + # written off the request path and would otherwise die with the queue. + app.state.auth_manager.close_usage_writer() app.state.db_pool.close() logger.info("api_shutting_down") diff --git a/src/serving/api/metrics.py b/src/serving/api/metrics.py index 33d56cd..ff8b61e 100644 --- a/src/serving/api/metrics.py +++ b/src/serving/api/metrics.py @@ -33,3 +33,11 @@ "agentflow_usage_record_failures_total", "Authenticated requests served without their api_usage row being written.", ) + +# Backpressure, not failure: the writer's queue was full, so the row was shed +# rather than made to stall the request it was counting. Sustained non-zero +# means the writer cannot keep up with the request rate. +USAGE_ROWS_DROPPED = Counter( + "agentflow_usage_rows_dropped_total", + "api_usage rows dropped because the off-path writer queue was full.", +) diff --git a/src/serving/api/routers/admin.py b/src/serving/api/routers/admin.py index d83830e..4b58563 100644 --- a/src/serving/api/routers/admin.py +++ b/src/serving/api/routers/admin.py @@ -1,4 +1,5 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status +from starlette.concurrency import run_in_threadpool from src.serving.api.analytics import ( get_anomalies, @@ -34,7 +35,8 @@ async def create_api_key(payload: KeyCreateRequest, request: Request) -> dict[st @router.get("/keys", response_model=None) async def list_api_keys(request: Request) -> dict[str, object]: manager = get_auth_manager(request) - return {"keys": manager.list_keys_with_usage()} + # Blocking: flushes the usage writer, then reads DuckDB. Off the loop. + return {"keys": await run_in_threadpool(manager.list_keys_with_usage)} @router.post("/keys/{key_id}/rotate", response_model=None) @@ -88,7 +90,8 @@ async def revoke_api_key(api_key: str, request: Request) -> Response: @router.get("/usage", response_model=None) async def get_usage(request: Request) -> dict[str, object]: manager = get_auth_manager(request) - return {"usage": manager.usage_by_tenant()} + # Blocking: flushes the usage writer, then reads DuckDB. Off the loop. + return {"usage": await run_in_threadpool(manager.usage_by_tenant)} @router.get("/analytics/usage", response_model=None) diff --git a/src/serving/control_plane/embedded.py b/src/serving/control_plane/embedded.py index e327dd7..d378c5b 100644 --- a/src/serving/control_plane/embedded.py +++ b/src/serving/control_plane/embedded.py @@ -29,7 +29,14 @@ from src.db_concurrency import catalog_ddl_lock from src.serving.duckdb_connection import connect_duckdb -from .store import AUTO_RESOLVE_NOTE, ControlPlaneStore, OutboxEntry, TriageState, WebhookQueueRow +from .store import ( + AUTO_RESOLVE_NOTE, + ControlPlaneStore, + OutboxEntry, + TriageState, + UsageRow, + WebhookQueueRow, +) logger = structlog.get_logger() @@ -1483,6 +1490,51 @@ def record_api_usage( finally: conn.close() + def record_api_usage_batch(self, rows: Sequence[UsageRow]) -> None: + """One ``executemany`` inside one transaction, so a batch of N rows + costs one commit rather than N. + + DuckDB serializes writers, so the per-row form put a commit — an fsync + — on the critical path of every authenticated request, capping the API + at ``1 / commit_latency`` requests per second. Batching lifts the + accounting ceiling to ``len(rows) / commit_latency``, which is what + lets the writer keep up with the request rate off the request path + (docs/perf/usage-write-bifurcation-2026-07-09.md). + """ + if not rows: + return + params = [[r.tenant, r.key_name, r.endpoint, r.key_id, r.key_slot] for r in rows] + for attempt in range(10): + try: + conn = self._usage_cursor() + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + continue + + try: + conn.execute("BEGIN TRANSACTION") + try: + conn.executemany( + """ + INSERT INTO api_usage (tenant, key_name, endpoint, key_id, key_slot) + VALUES (?, ?, ?, ?, ?) + """, + params, + ) + except duckdb.Error: + conn.execute("ROLLBACK") + raise + conn.execute("COMMIT") + return + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + finally: + conn.close() + def get_usage_by_tenant(self) -> list[dict]: conn = self._usage_cursor() try: diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py index 556639b..a7d5c3c 100644 --- a/src/serving/control_plane/store.py +++ b/src/serving/control_plane/store.py @@ -121,6 +121,17 @@ class WebhookQueueRow: body: str | None +@dataclass(frozen=True) +class UsageRow: + """One ``api_usage`` row owed for a served authenticated request.""" + + tenant: str + key_name: str + endpoint: str + key_id: str | None + key_slot: str + + @dataclass(frozen=True) class OutboxEntry: """One pending replay-outbox row (a Kafka message owed to a topic).""" @@ -530,10 +541,31 @@ def record_api_usage( request. Raises on exhausted retries — a caller (``record_usage``) depends on the exception to skip its post-insert audit publish. - The exception stops there: ``AuthMiddleware`` catches it, increments - ``agentflow_usage_record_failures_total`` and serves the request - anyway. Accounting is a side-channel and must not fail the request it - was counting.""" + The exception stops there: the usage writer catches it, increments + ``agentflow_usage_record_failures_total`` and drops the row. + Accounting is a side-channel and must not fail the request it was + counting.""" + + def record_api_usage_batch(self, rows: Sequence[UsageRow]) -> None: + """Append many ``api_usage`` rows as **one** unit of work. + + A backend that pays a per-commit cost (an fsync, a round trip) should + override this so a batch costs one, not ``len(rows)``. The embedded + DuckDB store does; without it the accounting throughput ceiling is + ``1 / commit_latency`` rows per second, which on a slow disk is below + the request rate the API can otherwise serve (2026-07-09). + + Same failure contract as ``record_api_usage``: raise, and the caller + drops the batch and counts it. + """ + for row in rows: + self.record_api_usage( + tenant=row.tenant, + key_name=row.key_name, + endpoint=row.endpoint, + key_id=row.key_id, + key_slot=row.key_slot, + ) @abstractmethod def get_usage_by_tenant(self) -> list[dict]: diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py index 2b0b51a..a9cca97 100644 --- a/tests/unit/test_auth.py +++ b/tests/unit/test_auth.py @@ -185,17 +185,22 @@ def test_rate_limit_is_isolated_per_key(api_keys_path: Path, db_path: Path): def test_authenticated_requests_log_usage(api_keys_path: Path, db_path: Path): - client = TestClient(_build_app(api_keys_path, db_path)) + app = _build_app(api_keys_path, db_path) + client = TestClient(app) response = client.get("/v1/metrics/revenue", headers={"X-API-Key": "tenant-ops-key"}) assert response.status_code == 200 + # The row is written off the request path, so a reader that must see it + # waits for the writer — exactly as `GET /v1/admin/usage` does. + assert app.state.auth_manager.flush_usage(timeout=5.0) rows = ( duckdb.connect(str(db_path)) .execute("SELECT tenant, key_name, endpoint FROM api_usage") .fetchall() ) assert rows == [("acme", "Ops Agent", "/v1/metrics/revenue")] + app.state.auth_manager.close_usage_writer() def test_admin_endpoints_require_admin_key(api_keys_path: Path, db_path: Path): diff --git a/tests/unit/test_auth_usage_write_failure.py b/tests/unit/test_auth_usage_write_failure.py deleted file mode 100644 index 4a7bcfd..0000000 --- a/tests/unit/test_auth_usage_write_failure.py +++ /dev/null @@ -1,96 +0,0 @@ -"""A failed usage write must not fail the request it was counting. - -``ControlPlaneStore.record_api_usage`` raises on exhausted retries so that -``record_usage`` skips its post-insert audit publish. That exception used to -propagate out of ``AuthMiddleware`` through the ASGI stack, so a request that -had already authenticated and would have served a 200 came back as a 500 — -observed on all six endpoints of the 2026-07-09 Load Test when concurrent -DuckDB opens collided on the usage database. - -The middleware now counts the dropped row and serves the request. -""" - -from __future__ import annotations - -from pathlib import Path - -import duckdb -import pytest -from fastapi import FastAPI -from fastapi.testclient import TestClient - -from src.serving.api.auth import AuthManager, build_auth_middleware -from src.serving.api.metrics import USAGE_RECORD_FAILURES - -API_KEY = "tenant-order-key" - - -def _build_app(tmp_path: Path) -> FastAPI: - api_keys_path = tmp_path / "config" / "api_keys.yaml" - api_keys_path.parent.mkdir(parents=True, exist_ok=True) - api_keys_path.write_text( - f""" -keys: - - key: "{API_KEY}" - name: "Order Agent" - tenant: "acme" - rate_limit_rpm: 100 - created_at: "2026-04-10" -""".strip() - + "\n", - encoding="utf-8", - ) - - app = FastAPI() - app.state.auth_manager = AuthManager( - api_keys_path=api_keys_path, - db_path=tmp_path / "usage.duckdb", - admin_key="admin-secret", - ) - app.state.auth_manager.load() - app.state.auth_manager.ensure_usage_table() - app.middleware("http")(build_auth_middleware()) - - @app.get("/v1/metrics/revenue") - async def revenue(): - return {"metric_name": "revenue", "value": 100.0} - - return app - - -def _failures_count() -> float: - return USAGE_RECORD_FAILURES._value.get() - - -def test_request_succeeds_when_the_usage_write_raises( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - app = _build_app(tmp_path) - - def exploding_record_usage(*_args, **_kwargs): - raise duckdb.BinderException( - 'Unique file handle conflict: Cannot attach "agentflow-api-usage"' - ) - - monkeypatch.setattr( - app.state.auth_manager, "record_usage", exploding_record_usage, raising=True - ) - - before = _failures_count() - with TestClient(app) as client: - response = client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}) - - assert response.status_code == 200 - assert response.json() == {"metric_name": "revenue", "value": 100.0} - assert _failures_count() == before + 1 - - -def test_healthy_request_does_not_touch_the_failure_counter(tmp_path: Path) -> None: - app = _build_app(tmp_path) - - before = _failures_count() - with TestClient(app) as client: - response = client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}) - - assert response.status_code == 200 - assert _failures_count() == before diff --git a/tests/unit/test_usage_write_off_request_path.py b/tests/unit/test_usage_write_off_request_path.py new file mode 100644 index 0000000..f1ffe3e --- /dev/null +++ b/tests/unit/test_usage_write_off_request_path.py @@ -0,0 +1,292 @@ +"""Usage accounting must not pace the request it is counting. + +The write used to happen inline (later: on a worker thread) before the response +was produced. The embedded store serializes writers and commits per row, so a +slow disk turned every authenticated request into a queue on one fsync. The CI +``Load Test`` collapsed onto a saturated branch pinned at ``rps = 1/commit``, +which three red runs hit within 1.7% of each other while green runs spread +across 37-46 rps. Full measurement: +``docs/perf/usage-write-bifurcation-2026-07-09.md``. + +These tests pin the *guarantee* — the response does not wait on the write, and +a write that fails or is shed still serves the request — not the mechanism. + +They subsume ``test_auth_usage_write_failure.py``, which guarded the same +side-channel promise from the other direction: a `BinderException` out of the +usage write used to escape `AuthMiddleware` and turn a served request into a +500 (all six endpoints, Load Test of 2026-07-09). That test drove the failure +through `AuthManager.record_usage`, which the request path no longer calls, so +it pinned a call that no longer exists. The promise it guarded is +`test_request_succeeds_when_the_usage_write_raises` below. +""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path + +import duckdb +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from src.serving.api.auth import AuthManager, build_auth_middleware +from src.serving.api.metrics import USAGE_RECORD_FAILURES, USAGE_ROWS_DROPPED + +API_KEY = "tenant-order-key" +SLOW_WRITE_SECONDS = 0.4 + + +def _build_app(tmp_path: Path) -> FastAPI: + api_keys_path = tmp_path / "config" / "api_keys.yaml" + api_keys_path.parent.mkdir(parents=True, exist_ok=True) + api_keys_path.write_text( + f""" +keys: + - key: "{API_KEY}" + name: "Order Agent" + tenant: "acme" + rate_limit_rpm: 1000 + created_at: "2026-04-10" +""".strip() + + "\n", + encoding="utf-8", + ) + + app = FastAPI() + manager = AuthManager( + api_keys_path=api_keys_path, + db_path=tmp_path / "usage.duckdb", + admin_key="admin-secret", + ) + manager.load() + manager.ensure_usage_table() + app.state.auth_manager = manager + app.middleware("http")(build_auth_middleware()) + + @app.get("/v1/metrics/revenue") + async def revenue(): + return {"metric_name": "revenue", "value": 100.0} + + return app + + +def _failures() -> float: + return USAGE_RECORD_FAILURES._value.get() + + +def _dropped() -> float: + return USAGE_ROWS_DROPPED._value.get() + + +def test_a_slow_usage_write_does_not_delay_the_response(tmp_path: Path) -> None: + """The defect, stated as a test: with the write inline this request took + at least SLOW_WRITE_SECONDS; off the path it returns immediately.""" + app = _build_app(tmp_path) + manager = app.state.auth_manager + entered = threading.Event() + + def slow_batch(rows): + entered.set() + time.sleep(SLOW_WRITE_SECONDS) + + manager.store.record_api_usage_batch = slow_batch # type: ignore[method-assign] + + with TestClient(app) as client: + started = time.perf_counter() + response = client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}) + elapsed = time.perf_counter() - started + + assert response.status_code == 200 + assert entered.wait(2.0), "the writer thread never picked the row up" + assert elapsed < SLOW_WRITE_SECONDS / 2, ( + f"the response waited {elapsed:.3f}s on a {SLOW_WRITE_SECONDS}s usage write" + ) + manager.close_usage_writer() + + +def test_request_succeeds_when_the_usage_write_raises(tmp_path: Path) -> None: + """A store that exhausts its retries loses the row and counts it. The + request it was counting is served regardless.""" + app = _build_app(tmp_path) + manager = app.state.auth_manager + + def exploding_batch(rows): + raise duckdb.BinderException( + 'Unique file handle conflict: Cannot attach "agentflow-api-usage"' + ) + + manager.store.record_api_usage_batch = exploding_batch # type: ignore[method-assign] + + before = _failures() + with TestClient(app) as client: + response = client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}) + assert response.status_code == 200 + assert response.json() == {"metric_name": "revenue", "value": 100.0} + assert manager.flush_usage(timeout=5.0) + + assert _failures() == before + 1 + manager.close_usage_writer() + + +def test_healthy_request_does_not_touch_the_failure_counter(tmp_path: Path) -> None: + app = _build_app(tmp_path) + manager = app.state.auth_manager + + before = _failures() + with TestClient(app) as client: + response = client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}) + assert response.status_code == 200 + assert manager.flush_usage(timeout=5.0) + + assert _failures() == before + manager.close_usage_writer() + + +def test_a_failed_batch_publishes_no_audit_event(tmp_path: Path) -> None: + """A publish must never claim a row that was never inserted — the + invariant the old synchronous `record_usage` got from the raise.""" + app = _build_app(tmp_path) + manager = app.state.auth_manager + published: list[dict] = [] + + class Recorder: + def publish(self, payload: dict) -> None: + published.append(payload) + + manager._usage_writer._audit_publisher = Recorder() + + def exploding_batch(rows): + raise duckdb.IOException("disk gone") + + manager.store.record_api_usage_batch = exploding_batch # type: ignore[method-assign] + + with TestClient(app) as client: + assert client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}).status_code == 200 + assert manager.flush_usage(timeout=5.0) + + assert published == [] + manager.close_usage_writer() + + +def test_rows_are_durable_after_flush(tmp_path: Path) -> None: + """Read-your-writes: the admin usage read flushes the queue first.""" + app = _build_app(tmp_path) + manager = app.state.auth_manager + + with TestClient(app) as client: + for _ in range(5): + assert ( + client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}).status_code == 200 + ) + usage = manager.usage_by_tenant() + + assert usage == [{"tenant": "acme", "requests_last_24h": 5}] + manager.close_usage_writer() + + +def test_every_api_usage_reader_drains_the_writer_first(tmp_path: Path) -> None: + """Each of the three readers of `api_usage` must flush, or it counts a + stale total. `rotation_status` was missed on the first pass and CI's + integration suite caught it as `assert 1 == 2`.""" + app = _build_app(tmp_path) + manager = app.state.auth_manager + + with TestClient(app) as client: + for _ in range(3): + client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}) + + # No explicit flush anywhere below: each reader owns that. Called + # through the reader seam, never through `manager.store` directly — + # a store call would be served by whatever the previous reader + # happened to flush. + assert manager._key_rotator._usage_by_key()[("acme", "Order Agent")] == 3 + assert manager._key_rotator.old_key_usage_by_key_id() == {} + assert manager.usage_by_tenant() == [{"tenant": "acme", "requests_last_24h": 3}] + + manager.close_usage_writer() + + +def test_a_full_queue_sheds_the_row_and_serves_the_request(tmp_path: Path) -> None: + """Backpressure is bounded and never blocking: the counter moves, the + client does not wait.""" + app = _build_app(tmp_path) + manager = app.state.auth_manager + release = threading.Event() + + def blocking_batch(rows): + release.wait(5.0) + + manager.store.record_api_usage_batch = blocking_batch # type: ignore[method-assign] + # One slot: the first row occupies the writer, the second fills the queue, + # the third has nowhere to go. + manager._usage_writer._queue.maxsize = 1 + + before = _dropped() + with TestClient(app) as client: + for _ in range(4): + assert ( + client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}).status_code == 200 + ) + + assert _dropped() > before + release.set() + manager.close_usage_writer() + + +def test_batching_coalesces_rows_into_one_commit(tmp_path: Path) -> None: + """The writer must batch, or it just moves the 1/commit ceiling into a + queue that silently overflows.""" + app = _build_app(tmp_path) + manager = app.state.auth_manager + batches: list[int] = [] + gate = threading.Event() + + real = manager.store.record_api_usage_batch + + def counting_batch(rows): + gate.wait(2.0) + batches.append(len(rows)) + return real(rows) + + manager.store.record_api_usage_batch = counting_batch # type: ignore[method-assign] + + rows_submitted = 24 + with TestClient(app) as client: + for _ in range(rows_submitted): + client.get("/v1/metrics/revenue", headers={"X-API-Key": API_KEY}) + # Everything queued behind the gate; releasing it lets one drain pick + # up the whole backlog in a single batch. + gate.set() + assert manager.flush_usage(timeout=5.0) + + assert sum(batches) == rows_submitted + assert max(batches) > 1, f"no coalescing happened: {batches}" + assert len(batches) < rows_submitted, f"one commit per row: {batches}" + manager.close_usage_writer() + + +@pytest.mark.parametrize("rows", [0, 1, 7]) +def test_batch_write_is_atomic_and_complete(tmp_path: Path, rows: int) -> None: + """The DuckDB override wraps the batch in one transaction; every row lands.""" + from src.serving.control_plane.embedded import EmbeddedControlPlaneStore + from src.serving.control_plane.store import UsageRow + + usage_db = tmp_path / f"usage-{rows}.duckdb" + store = EmbeddedControlPlaneStore(usage_db_path_provider=lambda: usage_db) + store.ensure_usage_schema() + store.record_api_usage_batch( + [ + UsageRow( + tenant="acme", + key_name="Order Agent", + endpoint="/v1/metrics/revenue", + key_id=f"k{i}", + key_slot="current", + ) + for i in range(rows) + ] + ) + usage = store.get_usage_by_tenant() + assert usage == ([] if rows == 0 else [{"tenant": "acme", "requests_last_24h": rows}])