From 737795d7fba84d84732613562622986953b255d8 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Thu, 9 Jul 2026 07:40:32 +0300 Subject: [PATCH 1/3] docs: state the real Spaces quota, not the one we guessed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The docs claimed the free tier keeps three cpu-basic Spaces awake. It does not: four Docker Spaces run concurrently under this account today, and HF refuses `POST .../restart` on a paused one with an explicit cpu-basic quota error. A second account is not an escape hatch either — creating another free Docker Space returns 402 (only static Spaces are free). So `ekb` and the standalone demo stay paused until a running Space is paused to make room. Two of the four running Spaces belong to other projects, so that trade is not this project's to make. Nothing about the deployment is missing; the constraint is concurrent compute, and it is now stated as measured rather than assumed. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 5 +++-- docs/dv2-multi-branch/RELEASE_STATUS.md | 17 ++++++++++++----- 2 files changed, 15 insertions(+), 7 deletions(-) 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. From ab7257d672adf0332702b392f9f8bd60d1e7f1c1 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Thu, 9 Jul 2026 09:14:44 +0300 Subject: [PATCH 2/3] fix(serving): take the api_usage write off the request path The Load Test's bimodality was not the runner. Every authenticated request wrote its own api_usage row before the response was produced, and the embedded store serializes writers and commits per row -- so each request queued behind one fsync and the API was capped at `1 / commit_latency` requests per second. The load client is closed-loop (15 users, 0.1-0.5s think time), which gives it two equilibria: an unsaturated one bounded by think time (~46 rps), and a saturated one pinned at `rps = 1/s`, independent of user count. The runner's disk only decided which side of `s ~ 20ms` the commit landed on. That is why three red runs agreed to within 1.7% (29.4 / 29.1 / 28.9 rps) while nine green runs spread across 37-46, and why rps varied 1.5x while p99 varied 10x -- a slower machine cannot do that, a queue can. Reproduced on the shipped code: sleep(34ms) inside record_api_usage yields 31.4 rps / p50 160ms against CI's red branch of 29.1 / 161. Moving the write off the request path makes throughput flat in s (34ms -> 60ms: 37.9 -> 37.2 rps, versus 31.4 -> 8.3 in-path). The request now enqueues a UsageRow and returns; one background thread drains the queue. Batching is part of the fix, not a tuning knob: a per-row background writer still commits at 1/s rows per second, so the ceiling would just move from request latency into a queue that silently overflows. record_api_usage_batch puts one commit under N rows. Accounting is a side-channel. It already could not fail the request it counted; now it cannot pace it either. 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 and was already droppable on exhausted retries, and reads that must see their own writes call flush_usage. A full queue sheds rows into agentflow_usage_rows_dropped_total rather than stalling the request. test_auth_usage_write_failure.py is removed: it drove its failure through AuthManager.record_usage, which the request path no longer calls, so it guarded a call rather than a promise. Its promise -- a failed usage write must not fail the request -- is now test_request_succeeds_when_the_usage_write_raises, and the new suite additionally pins the defect itself (a slow write must not delay the response), batching, backpressure, and read-your-writes. Finding N1 needs no threshold normalisation: the gate was reporting a real defect. Full measurement in docs/perf/usage-write-bifurcation-2026-07-09.md. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../usage-write-bifurcation-2026-07-09.md | 124 ++++++++ src/serving/api/auth/manager.py | 30 ++ src/serving/api/auth/middleware.py | 28 +- src/serving/api/auth/usage_writer.py | 224 +++++++++++++++ src/serving/api/main.py | 3 + src/serving/api/metrics.py | 8 + src/serving/api/routers/admin.py | 7 +- src/serving/control_plane/embedded.py | 54 +++- src/serving/control_plane/store.py | 40 ++- tests/unit/test_auth.py | 7 +- tests/unit/test_auth_usage_write_failure.py | 96 ------- .../unit/test_usage_write_off_request_path.py | 270 ++++++++++++++++++ 12 files changed, 771 insertions(+), 120 deletions(-) create mode 100644 docs/perf/usage-write-bifurcation-2026-07-09.md create mode 100644 src/serving/api/auth/usage_writer.py delete mode 100644 tests/unit/test_auth_usage_write_failure.py create mode 100644 tests/unit/test_usage_write_off_request_path.py 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/manager.py b/src/serving/api/auth/manager.py index a142143..7b75012 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]: + self.flush_usage() 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..4198193 --- /dev/null +++ b/tests/unit/test_usage_write_off_request_path.py @@ -0,0 +1,270 @@ +"""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_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}]) From 98ec4e76600ead79534d6413039ae650784c4d62 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Thu, 9 Jul 2026 09:31:24 +0300 Subject: [PATCH 3/3] fix(serving): flush the usage writer in every api_usage reader MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three call sites read api_usage; the first pass flushed two of them. The one it missed, `KeyRotator.old_key_usage_by_key_id`, backs `GET /v1/admin/keys/ {id}/rotation-status`, and CI's integration suite caught it immediately: `assert 1 == 2` — one row had been written, the other was still queued. Move the flush to the reader seam rather than the callers, so a future reader of the table cannot forget it. `AuthManager.list_keys_with_usage` no longer flushes on its own: `KeyRotator._usage_by_key` does it on the way through. The regression test calls the readers, never `manager.store` directly — a store call is served by whatever the previous reader happened to flush, and would have passed against the bug. Verified to fail with the flushes removed. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/serving/api/auth/key_rotation.py | 4 ++++ src/serving/api/auth/manager.py | 2 +- .../unit/test_usage_write_off_request_path.py | 22 +++++++++++++++++++ 3 files changed, 27 insertions(+), 1 deletion(-) 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 7b75012..5c7c815 100644 --- a/src/serving/api/auth/manager.py +++ b/src/serving/api/auth/manager.py @@ -477,7 +477,7 @@ def close_usage_writer(self, timeout: float = 5.0) -> None: self._usage_writer.close(timeout) def list_keys_with_usage(self) -> list[dict]: - self.flush_usage() + # 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]: diff --git a/tests/unit/test_usage_write_off_request_path.py b/tests/unit/test_usage_write_off_request_path.py index 4198193..f1ffe3e 100644 --- a/tests/unit/test_usage_write_off_request_path.py +++ b/tests/unit/test_usage_write_off_request_path.py @@ -186,6 +186,28 @@ def test_rows_are_durable_after_flush(tmp_path: Path) -> None: 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."""