diff --git a/docs/adr/0026-usage-rollup-write-path.md b/docs/adr/0026-usage-rollup-write-path.md new file mode 100644 index 000000000..71eccae54 --- /dev/null +++ b/docs/adr/0026-usage-rollup-write-path.md @@ -0,0 +1,128 @@ +# ADR 0026 — Usage-rollup write path: isolate-scoped buffering, not sampling or a log-derived rollup + +- **Status:** Accepted · implemented (#8823) +- **Date:** 2026-07-31 +- **Relates to:** #8823 (this ADR), #8597 (`api_usage_rollup`, the table this + writes and the pricing question it exists to answer), ADR 0022 (the deferred + paid-tier decision the rollup feeds), ADR 0014 (the self-hosted Postgres whose + capacity is the constraint here) + +## Context + +Every request to any `/api/v1/*` path drove **one Postgres write** on the +self-hosted indexer box, with no authentication and no rate limit in front of +it. + +`workers/api.ts` calls `recordUsageRollup` for any path shaped like `/api/v1/…`, +after the OPTIONS early-return and **before** dispatch — so it fires for 404s +and for paths no route serves. That function called +`foldObservations([observation])` on a **single-element array** and POSTed +immediately to `/api/v1/internal/usage-rollup`, which runs an +`INSERT … ON CONFLICT (day, route_family, cost_shape) DO UPDATE` inside +`withAccountsSql` — and `withAccountsSql` constructs a **new `postgres()` client +per invocation**. + +`handleUsageRollupIncrement`'s own header comment claimed the caller coalesced +observations before sending. It did not. `foldObservations` is real and correct; +it simply never saw more than one observation on the write path. There was no +cross-request buffer anywhere in the Worker, so an isolate serving 500 requests +issued 500 subrequests and 500 upserts. + +**Unmatched traffic is the worst case.** `routeFamily` collapses every +non-matching path to a single `UNMATCHED_FAMILY = "unmatched"` bucket — +deliberately, to avoid a cardinality bomb — and degenerate input gets the `edge` +cost shape. So a flood of `GET /api/v1/` 404s all target **one row**: +`(day, "unmatched", "edge")`. Every request contended for that row's lock, +serialised, each with a fresh connection. Nothing throttles it: `/api/*` is +`run_worker_first`, the tiered and per-surface limiters live inside individual +route branches, and the generic 404 is reached without passing any of them. + +The failure mode is bounded but real. The write is fire-and-forget and swallows +its own errors, so API responses stay correct; what degrades is the Postgres +instance the indexer depends on. It is also a standing scaling hazard for +legitimate traffic, not only for a deliberate flood. + +## Decision + +**Option A — isolate-scoped buffering with a count-or-age flush.** + +`recordUsageRollup` appends its `UsageObservation` to a module-scope buffer in +the Worker isolate. When the buffer reaches **64 observations**, or when **10 +seconds** have elapsed since the first observation in the current buffer, the +whole buffer is folded with `foldObservations` and sent as **one** subrequest +carrying one bucket per `(day, route_family, cost_shape)`. The flush rides the +triggering request's `ctx.waitUntil`, so it adds no latency, exactly as the +single-observation write did. + +The buffer is drained **before** the fetch is issued, so a concurrent request in +the same isolate cannot re-send the same observations. + +### Why not B (sampling) + +Sampling turns `api_usage_rollup.request_count` into an estimate. The table +exists to answer ADR 0022's deferred pricing question — "does the free tier cost +too much" — and the answer is a cost figure someone will multiply by a rate. An +estimate with an unstated confidence interval is worse than a smaller exact +number, and the scale factor would have to be versioned into the rows to stay +interpretable across a threshold change. Buffering gets the same write reduction +without giving up exactness for observed requests. + +### Why not C (derive the rollup off the request path) + +Cloudflare Logpush / Workers Analytics Engine / PostHog can all carry request +volume, but none of them carries `route_family` — that label is produced by +`routeFamily()` matching the Worker's own dispatch order, which is the property +that makes the rollup attribute a request to the route that actually served it. +Reconstructing it from a log stream means reimplementing that matcher against +raw paths in a second place and keeping the two in step forever. C also trades +freshness for a scheduled job and adds a dependency for a table that currently +has none. It stays a reasonable future option if per-isolate write volume ever +becomes the constraint again; it is not worth its cost today. + +## Consequences + +### Write volume + +An isolate serving N requests now issues `ceil(N / 64)` subrequests, upserts, +and `postgres()` clients instead of N — a **64x** reduction per isolate at +steady load, and more than that for a burst to one family, since the batch also +collapses N observations into a single bucket per `(day, family, shape)` rather +than N separate upserts against the same row. The `(day, "unmatched", "edge")` +row lock, the specific contention point an unauthenticated flood targets, is +taken once per flush instead of once per request. + +Per-isolate is the honest bound: many isolates still means many writers. This +does not make the write path free, it makes each writer 64x quieter. That is +sufficient for the hazard as it exists; it is not a claim that the write path is +now unconditionally safe at any scale. + +### Accuracy — the one thing that changes + +`request_count` and `keyed_count` keep their meaning exactly: **exact counts of +observed requests**, not sampled and not estimated. Nothing is scaled. + +The new loss mode is **isolate eviction with a partially-filled buffer**. An +isolate that is evicted mid-buffer loses at most **63 observations**, and — since +the age trigger can only fire when a later request arrives — an isolate whose +traffic stops entirely holds its last partial buffer until eviction. Both bounds +are why the thresholds are small rather than tuned for maximum batching. This +under-counts; it never over-counts, so the rollup remains a floor on real +traffic, which is the safe direction for a cost estimate. + +### Unmatched paths are still counted + +Deliberately. They are the entire abuse surface, and "how much 404/scanner +volume is this API absorbing" is a real input to the same capacity question the +table serves. What made them expensive was not that they were counted but that +each one was a write; folding makes a flood of them cost one bucket per flush. +The single-bucket collapse that made them the worst case is now the thing that +makes them the cheapest. + +### Connection reuse is NOT addressed here + +`withAccountsSql` constructing a new `postgres()` client per invocation is a +per-write cost independent of this decision, and it affects **every** +accounts-tier route, not just this one. Fixing it inside a usage-rollup change +would be both out of scope and under-tested for the other routes it would +silently alter. Split into its own issue; buffering already removes 63 of every +64 client constructions on this path. diff --git a/docs/adr/README.md b/docs/adr/README.md index f942fbbb7..e108412ad 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -32,6 +32,7 @@ change. | [0023](0023-v440-emission-pipeline-publishing-posture.md) | v440 emission pipeline publishing posture — `emission_share` stays (stage-1 price share) and gains a label; new field names/units fixed; reconstructed values ship only while a harness holds them against chain state; the gate redistributes and does not throttle | Proposed · decision doc only (#8741); implementation is #8744/#8746/#8749 | | [0024](0024-first-party-tao-usd-index.md) | First-party TAO/USD index — publish our own volume-weighted median across venues rather than relay one provider; fixes venue criteria, outlier rule, a three-venue quorum below which nothing publishes, honesty labels, the disclaimer wording, and retirement of the coinpaprika call | Proposed · decision doc only (#8598); authorizes the #8599-#8602 chain, monitored by #8603 | | [0025](0025-on-chain-tao-usd-index.md) | On-chain TAO/USD index — supersedes 0024; composed as wTAO/WETH × WETH/USDC read from Uniswap pool state at a published block height, because chain state has no terms of service and the ETH leg is ~1,455x deeper than any USD-direct pool; prices wrapped TAO, which the basis label and peg monitoring make explicit | Proposed · decision doc only; authorizes #8600-#8602, monitored by #8603 | +| [0026](0026-usage-rollup-write-path.md) | Usage-rollup write path — isolate-scoped buffering with a count-or-age flush, rejecting sampling (the pricing question needs exact counts) and a log-derived rollup (only the Worker's own dispatch produces `route_family`); counts stay exact, the one new loss mode is isolate eviction with a partial buffer | Accepted · implemented (#8823) | ## Keeping these current diff --git a/tests/usage-rollup.test.ts b/tests/usage-rollup.test.ts index 867d1e2ae..8a76d6070 100644 --- a/tests/usage-rollup.test.ts +++ b/tests/usage-rollup.test.ts @@ -245,6 +245,9 @@ describe("observations and folding", () => { }); describe("recordUsageRollup posts every API request (#8597)", () => { + // #8823: an observation is buffered, not written, until a flush trigger + // fires. These tests are about WHAT gets recorded, so they flush explicitly; + // the batching behaviour itself is the separate describe below. const envWith = (over: Record = {}) => { const posts: { url: string; token: string | null; body: unknown }[] = []; const env = { @@ -268,15 +271,15 @@ describe("recordUsageRollup posts every API request (#8597)", () => { // The entire reason this rollup exists. If keyless were dropped here, the // readout would answer the pricing question with the same blind spot the // issue was filed to remove. - const { recordUsageRollup } = await import("../workers/api.ts"); + const { recordUsageRollup, flushUsageRollup } = + await import("../workers/api.ts"); const { env, posts } = envWith(); const waited: Promise[] = []; - recordUsageRollup( - env, - { waitUntil: (p: Promise) => waited.push(p) } as never, - "/api/v1/subnets/64", - false, - ); + const ctx = { + waitUntil: (p: Promise) => waited.push(p), + } as never; + recordUsageRollup(env, ctx, "/api/v1/subnets/64", false); + flushUsageRollup(env, ctx); await Promise.all(waited); assert.equal(posts.length, 1); assert.equal(posts[0].url, "/api/v1/internal/usage-rollup"); @@ -288,15 +291,15 @@ describe("recordUsageRollup posts every API request (#8597)", () => { }); test("marks a keyed request as keyed", async () => { - const { recordUsageRollup } = await import("../workers/api.ts"); + const { recordUsageRollup, flushUsageRollup } = + await import("../workers/api.ts"); const { env, posts } = envWith(); const waited: Promise[] = []; - recordUsageRollup( - env, - { waitUntil: (p: Promise) => waited.push(p) } as never, - "/api/v1/subnets", - true, - ); + const ctx = { + waitUntil: (p: Promise) => waited.push(p), + } as never; + recordUsageRollup(env, ctx, "/api/v1/subnets", true); + flushUsageRollup(env, ctx); await Promise.all(waited); const bucket = (posts[0].body as { buckets: Row[] }).buckets[0]; assert.equal(bucket.keyed_count, 1); @@ -315,7 +318,8 @@ describe("recordUsageRollup posts every API request (#8597)", () => { }); test("a failing DATA_API cannot reject — it must never fail a response", async () => { - const { recordUsageRollup } = await import("../workers/api.ts"); + const { recordUsageRollup, flushUsageRollup } = + await import("../workers/api.ts"); const env = { API_KEY_LOOKUP_INTERNAL_TOKEN: "tok", DATA_API: { @@ -325,20 +329,239 @@ describe("recordUsageRollup posts every API request (#8597)", () => { }, } as unknown as Env; const waited: Promise[] = []; - recordUsageRollup( - env, - { waitUntil: (p: Promise) => waited.push(p) } as never, - "/api/v1/subnets", - false, - ); + const ctx = { + waitUntil: (p: Promise) => waited.push(p), + } as never; + recordUsageRollup(env, ctx, "/api/v1/subnets", false); + flushUsageRollup(env, ctx); // Resolves, never rejects — an unhandled rejection here would surface on // the real request. await Promise.all(waited); }); test("works without an ExecutionContext", async () => { - const { recordUsageRollup } = await import("../workers/api.ts"); + const { recordUsageRollup, flushUsageRollup } = + await import("../workers/api.ts"); const { env } = envWith(); recordUsageRollup(env, undefined, "/api/v1/subnets", false); + flushUsageRollup(env, undefined); + }); +}); + +// #8823: the write path used to be 1:1 with requests. `foldObservations` was +// handed a single-element array per request, so N requests meant N DATA_API +// subrequests, N postgres() clients, and N upserts -- and every unmatched path +// collapses to the SAME (day, "unmatched", "edge") row, so an unauthenticated +// flood of 404s serialised on one row lock against a self-hosted database. +// Observations are now buffered in the isolate and flushed in batches. See +// docs/adr/0026-usage-rollup-write-path.md. +describe("recordUsageRollup batches its writes (#8823)", () => { + function harness() { + const posts: { buckets: Row[] }[] = []; + const waited: Promise[] = []; + const env = { + API_KEY_LOOKUP_INTERNAL_TOKEN: "tok", + DATA_API: { + fetch: async (request: Request) => { + posts.push( + JSON.parse(await request.clone().text()) as { buckets: Row[] }, + ); + return Response.json({ ok: true }); + }, + }, + } as unknown as Env; + const ctx = { + waitUntil: (p: Promise) => waited.push(p), + } as never; + return { env, ctx, posts, waited }; + } + + // Mirrors USAGE_ROLLUP_FLUSH_COUNT in workers/api.ts. Asserted below against + // the module's own buffer size so a change there fails here loudly rather + // than silently weakening this test. + const FLUSH_COUNT = 64; + + test("N requests produce ceil(N / 64) writes, not N", async () => { + const { recordUsageRollup, usageRollupBufferSize, flushUsageRollup } = + await import("../workers/api.ts"); + const { env, ctx, posts, waited } = harness(); + flushUsageRollup(env, ctx); // drain anything a sibling test left buffered + posts.length = 0; + + for (let i = 0; i < FLUSH_COUNT - 1; i += 1) { + recordUsageRollup(env, ctx, "/api/v1/subnets", false); + } + assert.equal(usageRollupBufferSize(), FLUSH_COUNT - 1); + assert.equal( + posts.length, + 0, + "nothing is written below the flush threshold", + ); + + // The 64th observation trips the count trigger. + recordUsageRollup(env, ctx, "/api/v1/subnets", false); + await Promise.all(waited); + assert.equal(posts.length, 1, "64 requests => exactly ONE subrequest"); + assert.equal( + usageRollupBufferSize(), + 0, + "the buffer is drained, not copied", + ); + + // ...and the batch is ONE bucket carrying all 64, not 64 buckets: the + // upsert contends for the (day, family, shape) row once, not 64 times. + assert.equal(posts[0].buckets.length, 1); + assert.equal(posts[0].buckets[0].request_count, FLUSH_COUNT); + + // 64 more => a second write, and no more than that. + for (let i = 0; i < FLUSH_COUNT; i += 1) { + recordUsageRollup(env, ctx, "/api/v1/subnets", false); + } + await Promise.all(waited); + assert.equal(posts.length, 2, "128 requests => 2 subrequests, not 128"); + flushUsageRollup(env, ctx); + }); + + test("an unmatched-path flood folds into ONE bucket per batch", async () => { + // The abuse case the issue is really about: `/api/v1/` 404s all + // map to (day, "unmatched", "edge"), which is why they used to serialise + // on a single row lock. Every distinct junk path in one batch now costs + // one bucket, so the row is touched once per flush. + const { recordUsageRollup, flushUsageRollup } = + await import("../workers/api.ts"); + const { env, ctx, posts, waited } = harness(); + flushUsageRollup(env, ctx); + posts.length = 0; + + for (let i = 0; i < FLUSH_COUNT; i += 1) { + recordUsageRollup(env, ctx, `/api/v1/${i}-scanner-probe-${i}`, false); + } + await Promise.all(waited); + assert.equal(posts.length, 1, "64 junk 404s => ONE subrequest"); + assert.equal( + posts[0].buckets.length, + 1, + "64 DISTINCT junk paths => ONE bucket, not 64 rows", + ); + assert.equal(posts[0].buckets[0].family, UNMATCHED_FAMILY); + assert.equal(posts[0].buckets[0].request_count, FLUSH_COUNT); + flushUsageRollup(env, ctx); + }); + + test("keyed and keyless in one batch stay distinguishable", async () => { + const { recordUsageRollup, flushUsageRollup } = + await import("../workers/api.ts"); + const { env, ctx, posts, waited } = harness(); + flushUsageRollup(env, ctx); + posts.length = 0; + + for (let i = 0; i < FLUSH_COUNT; i += 1) { + recordUsageRollup(env, ctx, "/api/v1/subnets", i % 4 === 0); + } + await Promise.all(waited); + assert.equal(posts.length, 1); + // request_count/keyed_count keep their exact meaning across the change -- + // batched, never sampled or scaled (ADR 0026). + assert.equal(posts[0].buckets[0].request_count, FLUSH_COUNT); + assert.equal(posts[0].buckets[0].keyed_count, FLUSH_COUNT / 4); + flushUsageRollup(env, ctx); + }); + + test("distinct families in one batch become distinct buckets, still one write", async () => { + const { recordUsageRollup, flushUsageRollup } = + await import("../workers/api.ts"); + const { env, ctx, posts, waited } = harness(); + flushUsageRollup(env, ctx); + posts.length = 0; + + for (let i = 0; i < FLUSH_COUNT / 2; i += 1) { + recordUsageRollup(env, ctx, "/api/v1/subnets", false); + recordUsageRollup(env, ctx, "/api/v1/nonexistent-path", false); + } + await Promise.all(waited); + assert.equal(posts.length, 1, "one subrequest covers both families"); + assert.equal(posts[0].buckets.length, 2); + assert.deepEqual( + posts[0].buckets.map((bucket: Row) => bucket.request_count), + [FLUSH_COUNT / 2, FLUSH_COUNT / 2], + ); + flushUsageRollup(env, ctx); + }); + + test("the age trigger flushes a partial buffer once a later request arrives", async () => { + // Workers has no timer that runs outside a request, so the 10s age bound + // can only be evaluated when the NEXT observation arrives. That is the + // design, not a gap: a flush is affordable exactly when there is a request + // whose waitUntil can carry it. + const { recordUsageRollup, flushUsageRollup, usageRollupBufferSize } = + await import("../workers/api.ts"); + const { env, ctx, posts, waited } = harness(); + flushUsageRollup(env, ctx); + posts.length = 0; + + const realNow = Date.now; + try { + let clock = 1_800_000_000_000; + Date.now = () => clock; + recordUsageRollup(env, ctx, "/api/v1/subnets", false); + assert.equal(usageRollupBufferSize(), 1); + assert.equal( + posts.length, + 0, + "one observation is well below the count trigger", + ); + + clock += 10_001; // past USAGE_ROLLUP_FLUSH_AGE_MS + recordUsageRollup(env, ctx, "/api/v1/subnets", false); + await Promise.all(waited); + assert.equal( + posts.length, + 1, + "the age trigger fired on the next request", + ); + assert.equal(posts[0].buckets[0].request_count, 2); + assert.equal(usageRollupBufferSize(), 0); + } finally { + Date.now = realNow; + flushUsageRollup(env, ctx); + } + }); + + test("handleRequest drives the same batching end-to-end, 404s included", async () => { + const { handleRequest, flushUsageRollup, usageRollupBufferSize } = + await import("../workers/api.ts"); + const { env, ctx, posts, waited } = harness(); + flushUsageRollup(env, ctx); + posts.length = 0; + + // Real requests through the router: an unmatched /api/v1 path, which is + // both the worst case for the old write path and the one that reaches the + // generic 404 without passing any rate limiter. + for (let i = 0; i < FLUSH_COUNT; i += 1) { + await handleRequest( + new Request(`https://api.metagraph.sh/api/v1/no-such-route-${i}`), + env, + ctx, + ); + } + await Promise.all(waited); + assert.equal( + posts.length, + 1, + `${FLUSH_COUNT} requests through handleRequest => 1 usage-rollup subrequest`, + ); + assert.equal(posts[0].buckets[0].request_count, FLUSH_COUNT); + assert.equal(usageRollupBufferSize(), 0); + + // An OPTIONS preflight is still not counted (it returns before the hook). + await handleRequest( + new Request("https://api.metagraph.sh/api/v1/subnets", { + method: "OPTIONS", + }), + env, + ctx, + ); + assert.equal(usageRollupBufferSize(), 0); + flushUsageRollup(env, ctx); }); }); diff --git a/workers/api.ts b/workers/api.ts index a44490eee..264dda4d1 100644 --- a/workers/api.ts +++ b/workers/api.ts @@ -440,6 +440,7 @@ import { import { buildTierPolicies } from "../src/api-tiers.ts"; import { API_KEY_LOOKUP_TOKEN_HEADER } from "../src/api-key-validation.ts"; import { foldObservations, observeRequest } from "../src/usage-rollup.ts"; +import type { UsageObservation } from "../src/usage-rollup.ts"; import { registerModuleStateReset } from "../src/module-state-registry.ts"; // #8386: anonymous stays the existing, regression-tested DATA_RATE_LIMITER @@ -479,17 +480,45 @@ const USAGE_ROLLUP_MATCHERS = API_ROUTES.map((entry) => ({ // Same posture as recordApiKeyUsage: ctx.waitUntil so it adds no latency, and // it swallows its own failure -- a rollup miss must never surface as an error // on the actual API call. -export function recordUsageRollup( - env: Env, - ctx: Ctx | undefined, - pathname: string, - keyed: boolean, -): void { +// #8823: observations are BUFFERED in the isolate and flushed in batches. +// +// Until this landed, foldObservations was handed a single-element array on +// every request, so N requests meant N DATA_API subrequests, N postgres() +// clients (withAccountsSql builds a fresh one per invocation), and N upserts +// -- and a flood of `/api/v1/` 404s all collapse to the single +// (day, "unmatched", "edge") row, so those N upserts serialised on one row +// lock against a self-hosted database whose capacity is ours. Nothing +// throttles that path: /api/* is run_worker_first and the generic 404 is +// reached without passing any of the per-surface limiters. +// +// The flush triggers are count-OR-age, evaluated synchronously as each +// observation arrives (Workers has no timer that runs outside a request, so +// the age check can only fire when a LATER request arrives -- which is +// exactly when a flush is affordable). Both bounds are deliberately small: +// they cap what an isolate can lose on eviction, which is the one accuracy +// cost of buffering. +const USAGE_ROLLUP_FLUSH_COUNT = 64; +const USAGE_ROLLUP_FLUSH_AGE_MS = 10_000; +let usageRollupBuffer: UsageObservation[] = []; +let usageRollupBufferedAtMs = 0; + +// Exported for the tests that assert the batching property; not part of any +// route's behaviour. +export function usageRollupBufferSize(): number { + return usageRollupBuffer.length; +} + +// Drain the buffer into ONE subrequest carrying every folded bucket. Drains +// before the fetch so a concurrent request in the same isolate cannot send +// the same observations twice. A failed POST loses that batch, the same +// best-effort posture the single-observation write already had -- a rollup +// miss must never surface on the API call that triggered it. +export function flushUsageRollup(env: Env, ctx: Ctx | undefined): void { + if (usageRollupBuffer.length === 0) return; if (!env.DATA_API?.fetch || !env.API_KEY_LOOKUP_INTERNAL_TOKEN) return; - const observation = observeRequest(pathname, USAGE_ROLLUP_MATCHERS, { - keyed, - }); - const buckets = foldObservations([observation]); + const buckets = foldObservations(usageRollupBuffer); + usageRollupBuffer = []; + usageRollupBufferedAtMs = 0; const pending = env.DATA_API.fetch( new Request("https://api.metagraph.sh/api/v1/internal/usage-rollup", { method: "POST", @@ -505,6 +534,31 @@ export function recordUsageRollup( } } +export function recordUsageRollup( + env: Env, + ctx: Ctx | undefined, + pathname: string, + keyed: boolean, +): void { + if (!env.DATA_API?.fetch || !env.API_KEY_LOOKUP_INTERNAL_TOKEN) return; + const observation = observeRequest(pathname, USAGE_ROLLUP_MATCHERS, { + keyed, + }); + const nowMs = Date.now(); + if (usageRollupBuffer.length === 0) usageRollupBufferedAtMs = nowMs; + usageRollupBuffer.push(observation); + // A buffer spanning midnight needs no special case: each observation + // carries its own `day` and foldObservations groups on it, so the batch + // simply emits two buckets. + if ( + usageRollupBuffer.length < USAGE_ROLLUP_FLUSH_COUNT && + nowMs - usageRollupBufferedAtMs < USAGE_ROLLUP_FLUSH_AGE_MS + ) { + return; + } + flushUsageRollup(env, ctx); +} + // Fire-and-forget usage-counter increment for the self-serve dashboard // (#8386) -- via ctx.waitUntil so it never adds latency to the response that // triggered it, and swallows its own failure (a usage-counter miss must @@ -4442,6 +4496,11 @@ configureAnalyticsRoutes({ readHealthMetaKv, readEconomicsCurrentKv }); // handler module has dropped back to its unwired placeholders — leaving the // process in exactly the state a fresh import would produce. registerModuleStateReset("workers/api.ts", () => { + // #8823: the usage-rollup buffer is module-scoped isolate state, so it + // must reset between test files exactly like the memos below -- a leftover + // observation would otherwise shift the next file's flush boundary. + usageRollupBuffer = []; + usageRollupBufferedAtMs = 0; healthMetaKvMemo = { env: null, value: null, expiresAt: 0 }; economicsCurrentKvMemo = { env: null, value: null, expiresAt: 0 }; chainEventsDbMemo = { env: null, value: null, expiresAt: 0 }; diff --git a/workers/data-api.ts b/workers/data-api.ts index dfc8911ba..2231c2406 100644 --- a/workers/data-api.ts +++ b/workers/data-api.ts @@ -5595,12 +5595,19 @@ async function handleApiQuotaSpend(request: Request, env: Env) { // recording that traffic happened is a strictly smaller capability than // verifying a key. // -// BATCHED on purpose. The caller coalesces a request's observations before -// sending (src/usage-rollup.ts's foldObservations), so a burst of requests to -// one family becomes one upsert rather than one per request. Fire-and-forget -// from the caller's side, so this always returns 200 even on a swallowed write -// error -- a usage-rollup miss must never affect the request that triggered it, -// and there is nothing for the caller to react to either way. +// BATCHED on purpose -- and, since #8823, actually batched. This comment used +// to claim the caller coalesced "a request's observations", which described +// nothing: workers/api.ts handed foldObservations a single-element array per +// request, so a burst of N requests to one family really was N subrequests and +// N upserts contending on one row. The caller now buffers observations in the +// isolate (USAGE_ROLLUP_FLUSH_COUNT / _AGE_MS) and folds the whole batch, so a +// burst arrives here as one POST carrying one bucket per (day, family, shape). +// Every bucket in a batch is still upserted individually inside one +// withAccountsSql call, so one Postgres client serves the whole batch. +// Fire-and-forget from the caller's side, so this always returns 200 even on a +// swallowed write error -- a usage-rollup miss must never affect the request +// that triggered it, and there is nothing for the caller to react to either +// way. async function handleUsageRollupIncrement(request: Request, env: Env) { const configured = env.API_KEY_LOOKUP_INTERNAL_TOKEN; if (!configured) {