fix(diagnostics): end a connection report body that takes over 10 s to arrive

changelog.d/connection-diagnostics-read-deadline.md

@@ -0,0 +1,3 @@
+### Fixed
+- A connection diagnostics report whose body takes more than 10 seconds to arrive is cut off, so a
+  device sending one slowly can no longer hold every report slot and get other reports refused.

server/client-connection-log.test.ts

@@ -76,6 +76,27 @@ describe("operator-local client diagnostic endpoint", () => {
     expect((await log.handle(request({ ...body(), target: "x" }), "after")).status).toBe(400);
   });
 
+  it("ends a body that keeps trickling in once the deadline for the whole body passes", async () => {
+    let now = 1000;
+    const log = new ClientConnectionLog(() => {}, () => now);
+    const space = new TextEncoder().encode(" ");
+    // each chunk arrives 4 s after the last: no gap is long, but the body never ends on its own
+    const trickle = new ReadableStream<Uint8Array>({ pull(controller) { now += 4_000; controller.enqueue(space); } });
+    const response = await log.handle(new Request("http://fixture/api/connection-diagnostics", { method: "POST", headers: { "content-type": "application/json" }, body: trickle }), "slow");
+    expect(response.status).toBe(408);
+    expect(now).toBeLessThan(1000 + 30_000);
+  });
+
+  it("gives every read slot back when the bodies holding them stop arriving", async () => {
+    const log = new ClientConnectionLog(() => {}, () => 1000, 1);
+    const stalled = Array.from({ length: 16 }, (_, i) => log.handle(new Request("http://fixture/api/connection-diagnostics", {
+      method: "POST", headers: { "content-type": "application/json" },
+      body: new ReadableStream<Uint8Array>({ start(controller) { controller.enqueue(new TextEncoder().encode("{")); } }),
+    }), `stalled-${i}`));
+    expect((await Promise.all(stalled)).map((response) => response.status)).toEqual(Array(16).fill(408));
+    expect((await log.handle(request(body()), "after")).status).toBe(204);
+  });
+
   it("returns method/JSON refusals and never fails a client because the log writer threw", async () => {
     const log = new ClientConnectionLog(() => { throw new Error("write failed"); });
     expect((await log.handle(new Request("http://fixture/api/connection-diagnostics"), "one")).status).toBe(405);

server/client-connection-log.ts

@@ -5,6 +5,8 @@ const KINDS: readonly ClientConnectionEventKind[] = ["connect", "open", "connect
 const MAX_BYTES = 8192;
 // Report bodies read at once, server-wide: 16 x 8 KiB at most, whatever peer keys callers rotate.
 const MAX_READING = 16;
+// A body still arriving after this long gives up its read slot, however steadily it trickles in.
+const READ_DEADLINE_MS = 10_000;
 const object = (value: unknown): value is Record<string, unknown> => value !== null && typeof value === "object" && !Array.isArray(value);
 const integer = (value: unknown, max = Number.MAX_SAFE_INTEGER): value is number => typeof value === "number" && Number.isSafeInteger(value) && value >= 0 && value <= max;
 
@@ -43,7 +45,7 @@ export class ClientConnectionLog {
   private peers = new Map<string, { at: number; count: number }>();
   private global = { at: 0, count: 0 };
   private reading = 0;
-  constructor(private write: (line: string) => void = (line) => console.warn(line), private now: () => number = Date.now) {}
+  constructor(private write: (line: string) => void = (line) => console.warn(line), private now: () => number = Date.now, private readDeadlineMs = READ_DEADLINE_MS) {}
 
   async handle(request: Request, peer: string): Promise<Response> {
     if (request.method !== "POST") return jsonResponse({ error: { code: "method_not_allowed", message: "use POST" } }, 405, { allow: "POST" });
@@ -72,16 +74,25 @@ export class ClientConnectionLog {
     if (!request.body) return badRequest("invalid_trace", "Connection diagnostics are required");
     const reader = request.body.getReader();
     const chunks: Uint8Array[] = []; let size = 0;
+    // One deadline for the whole body, not per chunk: the timer covers a body that stops arriving,
+    // the clock check after each chunk a body that keeps arriving slowly.
+    const deadline = this.now() + this.readDeadlineMs;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const expired = new Promise<"expired">((resolve) => { timer = setTimeout(() => resolve("expired"), this.readDeadlineMs); });
+    const timedOut = () => { void reader.cancel().catch(() => {}); return jsonResponse({ error: { code: "trace_timeout", message: "Connection diagnostics took too long to arrive" } }, 408); };
     try {
       for (;;) {
-        const { value, done } = await reader.read();
+        const next = await Promise.race([reader.read(), expired]);
+        if (next === "expired") return timedOut();
+        const { value, done } = next;
         if (done) break;
         size += value.byteLength;
         if (size > MAX_BYTES) { void reader.cancel().catch(() => {}); return tooLarge(); }
         chunks.push(value);
+        if (this.now() >= deadline) return timedOut();
       }
     } catch { return badRequest("invalid_trace", "Connection diagnostics could not be read"); }
-    finally { reader.releaseLock(); }
+    finally { clearTimeout(timer); reader.releaseLock(); }
     const bytes = new Uint8Array(size); let offset = 0;
     for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.byteLength; }
     let trace: ClientConnectionTrace | null;

shared/protocol.ts

@@ -124,8 +124,9 @@ export type { Machine, MachineEvent, MachineRoster, PaneTarget, SetupJob, SetupR
  *         Update POSTs require X-Herdr-Update: 1, same-origin browser requests,
  *         and the usual token gate. Managed starts only; status is polled during restart.
  *  POST   /api/connection-diagnostics    -> 204; ClientConnectionTrace, metadata only, same-origin
- *         and authenticated (watch devices included). 400 invalid_trace, 413 trace_too_large,
- *         429 trace_rate_limited. At most 16 events and 8192 bytes; never input or credentials.
+ *         and authenticated (watch devices included). 400 invalid_trace, 408 trace_timeout (the
+ *         body took over 10 s to arrive), 413 trace_too_large, 429 trace_rate_limited. At most 16
+ *         events and 8192 bytes; never input or credentials.
  *  GET    /api/telemetry                 -> TelemetryStatus (shared/telemetry.ts), no-store
  *  POST   /api/telemetry                 -> TelemetryStatus; TelemetryChange (shared/telemetry.ts):
  *         { enabled?, notice_seen?: true, notice_version?, settings? }. A notice acknowledgement

src/lib/api-signin.test.ts

@@ -0,0 +1,58 @@
+import { afterAll, afterEach, describe, expect, it, spyOn } from "bun:test";
+import { ApiError, authenticate, pairDevice } from "./api.ts";
+import { ConnectionTrace } from "./connectionTrace.ts";
+
+let status = 204;
+const calls: { url: string; init?: RequestInit }[] = [];
+const fetchSpy = spyOn(globalThis, "fetch").mockImplementation(Object.assign(async (url: URL | RequestInfo, init?: RequestInit) => {
+  calls.push({ url: String(url), init });
+  return status === 204 ? new Response(null, { status }) : Response.json({ error: { code: "invalid_token", message: "fictional refusal" } }, { status });
+}, { preconnect: globalThis.fetch.preconnect }));
+const opened: ConnectionTrace[] = [];
+afterAll(() => fetchSpy.mockRestore());
+afterEach(() => { status = 204; calls.length = 0; fetchSpy.mockClear(); for (const trace of opened.splice(0)) trace.close(); });
+
+const context = () => ({ connection: 1, socket_state: 1, visibility: "visible" as const, online: true });
+// A macrotask turn: every promise continuation of the report has run, whatever its depth.
+const settle = () => new Promise<void>((resolve) => setTimeout(resolve, 0));
+async function signedOutTrace() {
+  let now = 1000; let reports = 0;
+  const trace = new ConnectionTrace("ws://fixture/ws", context, async () => { reports++; return 401; }, () => now);
+  opened.push(trace);
+  trace.record("close", {}, true); await settle();
+  now += 60_000; trace.record("close", {}, true);
+  expect(reports).toBe(1);
+  return { reported: () => reports, incident: () => { now += 60_000; trace.record("close", {}, true); } };
+}
+
+describe("sign-in calls", () => {
+  const signIns = [
+    { name: "authenticate", url: "/api/auth", call: () => authenticate("fictional-token"), body: { token: "fictional-token" } },
+    { name: "pairDevice", url: "/api/devices/pair", call: () => pairDevice("fictional-code", "fictional-label"), body: { code: "fictional-code", label: "fictional-label" } },
+  ];
+
+  for (const { name, url, call, body } of signIns) {
+    it(`${name} posts its JSON body and lets reporting paused by a 401 resume`, async () => {
+      const trace = await signedOutTrace();
+      await call();
+      expect(calls).toHaveLength(1);
+      expect(calls[0]!.url).toBe(url);
+      expect(calls[0]!.init?.method).toBe("POST");
+      expect(new Headers(calls[0]!.init?.headers).get("content-type")).toBe("application/json");
+      expect(new Headers(calls[0]!.init?.headers).get("x-herdr-machine")).toBe("1");
+      expect(JSON.parse(String(calls[0]!.init?.body))).toEqual(body);
+      trace.incident();
+      expect(trace.reported()).toBe(2);
+    });
+
+    it(`${name} throws the server's refusal and leaves reporting paused`, async () => {
+      const trace = await signedOutTrace();
+      status = 401;
+      const error = await call().then(() => null, (reason: unknown) => reason);
+      expect(error).toBeInstanceOf(ApiError);
+      expect(error).toMatchObject({ status: 401, code: "invalid_token" });
+      trace.incident();
+      expect(trace.reported()).toBe(1);
+    });
+  }
+});

src/lib/connectionTrace.test.ts

@@ -1,14 +1,18 @@
-import { describe, expect, it, spyOn } from "bun:test";
+import { afterEach, describe, expect, it, spyOn } from "bun:test";
 import type { ClientConnectionTrace } from "../../shared/protocol.ts";
 import { ConnectionTrace, resumeConnectionTraces } from "./connectionTrace.ts";
 
 const context = () => ({ connection: 1, socket_state: 1, visibility: "visible" as const, online: true });
 // A macrotask turn: every promise continuation of the report has run, whatever its depth.
 const settle = () => new Promise<void>((resolve) => setTimeout(resolve, 0));
+// Every trace a test opens is closed after it, so resumeConnectionTraces() in a later test reaches only its own.
+const opened: ConnectionTrace[] = [];
+const open = (...args: ConstructorParameters<typeof ConnectionTrace>) => { const trace = new ConnectionTrace(...args); opened.push(trace); return trace; };
+afterEach(() => { for (const trace of opened.splice(0)) trace.close(); });
 function fixture(status = 204) {
   let now = 1000;
   const sent: ClientConnectionTrace[] = [];
-  const trace = new ConnectionTrace("wss://private-host/ws?machine_id=private-id&token=private-credential", context, async (batch) => { sent.push(batch); return status; }, () => now);
+  const trace = open("wss://private-host/ws?machine_id=private-id&token=private-credential", context, async (batch) => { sent.push(batch); return status; }, () => now);
   return { trace, sent, advance: (ms: number) => { now += ms; } };
 }
 
@@ -64,7 +68,7 @@ describe("client connection breadcrumbs", () => {
   it("retains an incident recorded while an earlier report is in flight", async () => {
     let finish!: (status: number) => void; let now = 0;
     const sent: ClientConnectionTrace[] = [];
-    const trace = new ConnectionTrace("ws://fixture/ws", context, (batch) => { sent.push(batch); return new Promise((resolve) => { finish = resolve; }); }, () => now);
+    const trace = open("ws://fixture/ws", context, (batch) => { sent.push(batch); return new Promise((resolve) => { finish = resolve; }); }, () => now);
     trace.record("probe_retry", {}, true);
     trace.record("probe_drop", {}, true);
     finish(204); await settle(); now += 10_000;
@@ -76,7 +80,7 @@ describe("client connection breadcrumbs", () => {
 
   it("defers a failed HTTP report until a later connection event, with no new timer", async () => {
     let now = 0, calls = 0;
-    const trace = new ConnectionTrace("ws://fixture/ws", context, async () => { calls++; throw new Error("network unavailable"); }, () => now);
+    const trace = open("ws://fixture/ws", context, async () => { calls++; throw new Error("network unavailable"); }, () => now);
     trace.record("pagehide", {}, true); await settle();
     trace.record("pageshow"); expect(calls).toBe(1);
     now += 10_000; trace.record("open"); expect(calls).toBe(2);
@@ -125,7 +129,7 @@ describe("client connection breadcrumbs", () => {
   it("ignores a 401 that answers a report sent before signing in again", async () => {
     let finish!: (status: number) => void; let now = 0;
     const sent: ClientConnectionTrace[] = [];
-    const trace = new ConnectionTrace("ws://fixture/ws", context, (batch) => { sent.push(batch); return new Promise((resolve) => { finish = resolve; }); }, () => now);
+    const trace = open("ws://fixture/ws", context, (batch) => { sent.push(batch); return new Promise((resolve) => { finish = resolve; }); }, () => now);
     trace.record("close", {}, true);
     resumeConnectionTraces();
     finish(401); await settle();
@@ -136,7 +140,7 @@ describe("client connection breadcrumbs", () => {
   });
 
   it("does not throw into the socket when a reporter fails synchronously", () => {
-    const trace = new ConnectionTrace("ws://fixture/ws", context, () => { throw new Error("reporter failed"); });
+    const trace = open("ws://fixture/ws", context, () => { throw new Error("reporter failed"); });
     expect(() => trace.record("close", {}, true)).not.toThrow();
   });
 
@@ -148,7 +152,7 @@ describe("client connection breadcrumbs", () => {
       const random = spyOn(crypto, "getRandomValues").mockImplementation(() => { throw new Error("crypto unavailable"); });
       try {
         let calls = 0;
-        const trace = new ConnectionTrace("ws://fixture/ws", context, async () => { calls++; return 204; });
+        const trace = open("ws://fixture/ws", context, async () => { calls++; return 204; });
         expect(() => trace.record("close", {}, true)).not.toThrow();
         expect(calls).toBe(0);
       } finally { random.mockRestore(); }
@@ -162,7 +166,7 @@ describe("client connection breadcrumbs", () => {
     globals.window = { fetch: async (url: string, init: RequestInit) => { requests.push({ url, init }); return { status: 204 }; } };
     globals.document = { createElement() {} };
     try {
-      const trace = new ConnectionTrace("wss://private-host/ws?token=private-credential", context);
+      const trace = open("wss://private-host/ws?token=private-credential", context);
       trace.record("pagehide", {}, true); await settle();
       expect(requests).toHaveLength(1);
       expect(requests[0]).toMatchObject({ url: "/api/connection-diagnostics", init: { method: "POST", credentials: "same-origin", mode: "same-origin", redirect: "error", keepalive: true } });