From 7cea33218f5b665b32e617d5380eb956552a9d4d Mon Sep 17 00:00:00 2001 From: Karan Date: Wed, 4 Nov 2026 12:33:39 -0500 Subject: [PATCH] Collector: a client that goes away no longer cancels ingest (a batch could reach R2 and never be counted) ingest() registered its counting with waitUntil only after awaiting the R2 write. A client that reset its connection during that write (about 150 ms; a browser being shut down) cancelled the handler, so the batch stayed in R2 without reaching the dashboard's counters, the export index or forwarding. A reset before the write lost the batch altogether. waitUntil now holds ingest from its first line; the response is unchanged. Under local workerd with the real entry point, 10 batches per case, the client resetting 100 ms in, R2's timing modelled: write landed, ack pending: before 10 stored, 0 counted; after 10 stored, 10 counted write not started yet: before 0 stored, 0 counted; after 10 stored, 10 counted No change when the client waits or closes normally; half a body still stores nothing; a repeated batch still counts once. Production itself is not observed (logging is off, the error was swallowed): one app showed 13 of 198 page loads in R2 and not in D1, all solo visits by an automated browser. Co-Authored-By: Claude Opus 5.5 --- telemetry-worker/README.md | 3 ++- telemetry-worker/src/collector.ts | 10 +++++++-- telemetry-worker/test/index.test.ts | 34 +++++++++++++++++++++++++++++ 3 files changed, 44 insertions(+), 3 deletions(-) diff --git a/telemetry-worker/README.md b/telemetry-worker/README.md index f2dd35a8..c613d147 100644 --- a/telemetry-worker/README.md +++ b/telemetry-worker/README.md @@ -169,7 +169,8 @@ carries no sampling settings (rule 8), so the worker cannot reweight them; the p Each batch is aggregated in one `db.batch()` (multi-row upserts `count = count + excluded.count`, at most 300 distinct event counters and 300 exact ones per batch) after one `SELECT` of its page load's state (both again, for the exact part only, when another batch of the same page load wrote first). Aggregation runs in `ctx.waitUntil` after the -202 and its failures are swallowed, so D1 can never break ingest. +202 and its failures are swallowed, so D1 can never break ingest. `ctx.waitUntil` also holds ingest itself from its first +line, so a client that goes away mid-request (a browser being shut down) cancels neither the R2 write nor the counting. ## Privacy invariants diff --git a/telemetry-worker/src/collector.ts b/telemetry-worker/src/collector.ts index 525d179a..90c7061b 100644 --- a/telemetry-worker/src/collector.ts +++ b/telemetry-worker/src/collector.ts @@ -193,7 +193,7 @@ export interface HandleOptions { uuid?: () => string; /** Outgoing requests to Google and GitHub during sign-in (the global fetch in production; tests mock it). */ fetch?: (input: string, init?: RequestInit) => Promise; - /** ctx.waitUntil in production (aggregation runs after the response); tests await inline. */ + /** ctx.waitUntil in production (it holds ingest if the client goes away; aggregation runs after the response); tests await inline. */ waitUntil?: (p: Promise) => void; /** Requests on genclass.dev that this worker does not serve go on to the site (fetch in production). */ passthrough?: (request: Request) => Promise; @@ -206,7 +206,13 @@ export async function handle(request: Request, env: Env, opts: HandleOptions = { if (path === "/v1/health" && (request.method === "GET" || request.method === "HEAD")) { return reply(200, { ok: true, schema: SCHEMA }); } - if (path === "/v1/events" || path === "/api/v1/events") return ingest(request, env, opts); + if (path === "/v1/events" || path === "/api/v1/events") { + // A client that goes away (a browser being shut down) cancels the handler unless waitUntil holds it: without this a + // batch could reach R2 and never be counted, or be lost before R2. + const res = ingest(request, env, opts); + opts.waitUntil?.(res.catch(() => undefined)); + return res; + } const billing = await routeBilling(request, url, path, env, opts); if (billing) return billing; const account = await routeAuth(request, url, path, env, opts); diff --git a/telemetry-worker/test/index.test.ts b/telemetry-worker/test/index.test.ts index efe00fe1..69d773f1 100644 --- a/telemetry-worker/test/index.test.ts +++ b/telemetry-worker/test/index.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it } from "vitest"; import { gunzipSync } from "node:zlib"; import { handle, MAX_BODY_BYTES, MAX_EVENTS, validate } from "../src/collector.js"; +import { resetTokenCache } from "../src/dashboard.js"; +import { d1 } from "./d1.js"; function bucket() { const puts: { key: string; body: Uint8Array }[] = []; @@ -124,4 +126,36 @@ describe("collector", () => { expect(res.status).toBe(204); expect(res.headers.get("access-control-allow-origin")).toBe("*"); }); + + it("a client that goes away cannot cancel ingest: waitUntil holds it from the start, and only those promises finish the batch", async () => { + resetTokenCache(); + const token = "gc_aaaaaaaaaaaaaaaaaaaaaa"; + const db = d1(); + db.db.prepare("INSERT INTO projects (token, secret_hash, name, created) VALUES (?, 'x', 'x', '2026-10-08')").run(token); + let ack = () => {}; + const acked = new Promise((res) => (ack = res)); + const puts: string[] = []; + const env = { DB: db.api, BUCKET: { async put(key: string) { puts.push(key); await acked; } } }; + const waits: Promise[] = []; + // the reply is never read, as when the client is gone + void handle(post(JSON.stringify(batch({ token }))), env, { ...opts, waitUntil: (p) => waits.push(p) }); + // held before anything is awaited, so no moment of ingest is left to the client's connection + expect(waits).toHaveLength(1); + await new Promise((res) => setTimeout(res, 5)); + expect(puts).toHaveLength(1); + expect(db.rows("SELECT sid_hash FROM sessions")).toHaveLength(0); + ack(); + for (let i = 0; i < waits.length; i++) await waits[i]; + expect(db.rows("SELECT sid_hash FROM sessions")).toHaveLength(1); + expect(db.rows("SELECT count FROM daily WHERE metric = 'sessions'")).toEqual([{ count: 1 }]); + }); + + it("a request that fails part way rejects as before, and what waitUntil holds never rejects", async () => { + const waits: Promise[] = []; + const broken = new ReadableStream({ pull(c) { c.error(new Error("client gone")); } }); + const req = new Request("https://c.example/v1/events", { method: "POST", body: broken, duplex: "half", headers: { "content-type": "text/plain" } } as RequestInit); + await expect(handle(req, bucket(), { ...opts, waitUntil: (p) => waits.push(p) })).rejects.toThrow("client gone"); + expect(waits).toHaveLength(1); + await expect(waits[0]).resolves.toBeUndefined(); + }); });