Collector: a client that goes away no longer cancels ingest

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
 

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<Response>;
-  /** 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<unknown>) => void;
   /** Requests on genclass.dev that this worker does not serve go on to the site (fetch in production). */
   passthrough?: (request: Request) => Promise<Response>;
@@ -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);

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<void>((res) => (ack = res));
+    const puts: string[] = [];
+    const env = { DB: db.api, BUCKET: { async put(key: string) { puts.push(key); await acked; } } };
+    const waits: Promise<unknown>[] = [];
+    // 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<unknown>[] = [];
+    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();
+  });
 });