fix(serving): stop a usage-accounting write from 500-ing a served request (#172)

Every authenticated request appends an api_usage row from a worker thread,
and the analytics/admin routers build a throwaway store per request. Each
of those opened its own duckdb.connect() on the usage file. The last close
destroys the DuckDB instance, so a close racing an open leaves the file
attached by two instances and DuckDB raises

    BinderException: Unique file handle conflict

That escaped AuthMiddleware and turned requests that had already succeeded
into 500s: the 2026-07-09 Load Test on main saw 19 of 1712 fail across all
six endpoints, every traceback ending in record_api_usage -> connect_duckdb.

Root cause: EmbeddedControlPlaneStore now holds one owning connection per
usage-db path for the life of the process and hands out .cursor() children,
the shape DuckDBPool already uses for the serving database. Call sites are
untouched — they still close() what they are given, and closing a cursor
leaves the connection alive. 80 concurrent usage writes: 80 connects -> 1.

Defence in depth: record_api_usage still raises on exhausted retries (its
caller depends on that to skip the audit publish), but the middleware now
catches it, counts agentflow_usage_record_failures_total, logs
api_usage_record_skipped, and serves the request. Accounting is a
side-channel and must not fail the request it was counting.

Four tests pinned the old per-call-connect mechanism rather than the
behaviour; they now drop the cached connection before injecting their
fault and assert on retry attempts instead of connect counts.

Verified: 1670 unit + 263 integration pass locally; the middleware
regression test fails on the pre-fix tree with the exact BinderException.
The race's timing reproduces only on the CI runner, so the new tests pin
the mechanism that removes it rather than the timing.

Co-authored-by: JuliaEdom <uedomskikh@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

CHANGELOG.md

@@ -4,6 +4,33 @@ All notable changes to AgentFlow are documented in this file.
 
 ## [Unreleased]
 
+### Fixed — a usage-accounting write could turn a served request into a 500
+
+- **The usage database is now opened once per process.** Every authenticated
+  request appends an `api_usage` row from a worker thread, and the
+  analytics/admin routers build a throwaway `EmbeddedControlPlaneStore` per
+  request; each of those used to call `duckdb.connect()` on the same file. The
+  last close destroys the DuckDB instance, so a close racing an open left the
+  file briefly attached by two instances and DuckDB raised
+  `BinderException: Unique file handle conflict`. `EmbeddedControlPlaneStore`
+  now keeps one owning connection per usage-db path and hands out `.cursor()`
+  children — the shape `DuckDBPool` already uses for the serving database.
+  Callers are unchanged: they still `close()` what they are given, and closing
+  a cursor leaves the connection alive. Measured on the store's own path: 80
+  concurrent usage writes went from 80 physical connects to 1.
+- **The exception no longer reaches the client.** `record_api_usage` still
+  raises on exhausted retries — `record_usage` depends on that to skip its
+  audit publish — but `AuthMiddleware` now catches it, increments the new
+  `agentflow_usage_record_failures_total` counter, logs
+  `api_usage_record_skipped`, and serves the request. Accounting is a
+  side-channel; a dropped row must not fail the request it was counting.
+- Caught by the Load Test on `main` (2026-07-09): 19 of 1712 requests returned
+  500 across all six endpoints, each with `record_api_usage → connect_duckdb`
+  in the traceback. Regression tests pin both invariants
+  (`tests/unit/test_usage_db_connection_reuse.py`,
+  `tests/unit/test_auth_usage_write_failure.py`); the race's timing reproduces
+  only on the CI runner, so the tests pin the mechanism that removes it.
+
 ### Changed — one Flink version across pip extra and container runtime (audit 07.07 F2)
 
 - **The Docker runtime moves 2.2.1 → 2.3.0**, matching the `[flink]` extra

src/serving/api/auth/middleware.py

@@ -12,7 +12,7 @@
 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
+from src.serving.api.metrics import AUTH_FAILURES, USAGE_RECORD_FAILURES
 from src.serving.api.security import redact_sensitive_headers
 
 from .manager import _CURRENT_TENANT_ID, TenantKey, get_auth_manager
@@ -110,7 +110,26 @@ async def __call__(
         # 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)
-        await run_in_threadpool(manager.record_usage, tenant_key, path)
+        #
+        # 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:
+            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 = {
             "X-RateLimit-Limit": str(tenant_key.rate_limit_rpm),

src/serving/api/metrics.py

@@ -26,3 +26,10 @@
     "HTTP requests served by the API, labelled by method, route template, and status code.",
     labelnames=("method", "route", "status"),
 )
+
+# Usage accounting is a side-channel: a dropped row must never fail the request
+# it was counting. Non-zero means per-tenant request counters under-report.
+USAGE_RECORD_FAILURES = Counter(
+    "agentflow_usage_record_failures_total",
+    "Authenticated requests served without their api_usage row being written.",
+)