feat(observability): keep a bounded register of requests and jobs in flight

Every HTTP request the server accepts is registered through Node's
http.server.request.start diagnostics channel and removed when its
response closes, finished or aborted, so pages, RSC payloads and server
actions are covered alongside API routes. Jobs register around their
handler in observeJob. The register holds strings only, cuts the query
string on entry, caps at 2000 entries and counts what it refuses; route
labels are scrubbed and collapsed only when a report is written.

src/lib/jobs/job-observer.ts

@@ -30,6 +30,7 @@ import { AsyncLocalStorage } from "node:async_hooks";
 import { WideEventBuilder } from "@/lib/logging/event-builder";
 import { emitIfSampled } from "@/lib/logging/transports";
 import { readActiveJobsStartedBefore } from "@/lib/jobs/job-failures";
+import { trackInFlight } from "@/lib/observability/in-flight";
 
 /** How often a running job writes `job.progress`. */
 export const JOB_PROGRESS_INTERVAL_MS = 2 * 60 * 1000;
@@ -160,7 +161,9 @@ export async function observeJob<R>(
   };
 
   try {
-    const result = await progressStore.run(cell, run);
+    const result = await trackInFlight({ kind: "job", target: queue }, () =>
+      progressStore.run(cell, run),
+    );
     const did =
       result !== null && typeof result === "object" && "did" in result
         ? (result as { did?: unknown }).did

src/lib/observability/__tests__/in-flight.test.ts

@@ -0,0 +1,217 @@
+import { EventEmitter } from "node:events";
+import { createServer, request as httpRequest } from "node:http";
+import type { AddressInfo } from "node:net";
+
+import { describe, expect, it } from "vitest";
+
+import {
+  InFlightRegistry,
+  routeLabel,
+  startHttpInFlightTracking,
+  inFlightRegistry,
+  trackHttpRequest,
+  trackInFlight,
+} from "../in-flight";
+
+/**
+ * The register is only worth anything if it is accurate at the moment the
+ * heap runs out: an entry that outlives its work is a leak that grows with
+ * every request, and a register that grows without bound is itself the
+ * memory problem it is meant to name.
+ */
+
+function fakeExchange(url: string, headers: Record<string, string> = {}) {
+  const response = new EventEmitter();
+  return {
+    message: {
+      request: { url, method: "POST", headers },
+      response,
+    } as unknown as Parameters<typeof trackHttpRequest>[0],
+    close: () => response.emit("close"),
+  };
+}
+
+describe("InFlightRegistry", () => {
+  it("removes the entry when the work resolves", async () => {
+    const registry = new InFlightRegistry();
+    await trackInFlight(
+      { kind: "job", target: "q" },
+      async () => {
+        expect(registry.size).toBe(1);
+      },
+      registry,
+    );
+    expect(registry.size).toBe(0);
+  });
+
+  it("removes the entry when the work throws", async () => {
+    const registry = new InFlightRegistry();
+    await expect(
+      trackInFlight(
+        { kind: "job", target: "q" },
+        async () => {
+          throw new Error("boom");
+        },
+        registry,
+      ),
+    ).rejects.toThrow("boom");
+    expect(registry.size).toBe(0);
+  });
+
+  it("refuses entries past its bound and counts them", () => {
+    const registry = new InFlightRegistry(3);
+    const ids = [1, 2, 3, 4, 5].map(() =>
+      registry.begin({ kind: "http", target: "/", startedAt: 0 }),
+    );
+    expect(registry.size).toBe(3);
+    expect(registry.dropped).toBe(2);
+    expect(ids.slice(3)).toEqual([0, 0]);
+    // A refused id ends as a no-op and never removes someone else's entry.
+    registry.end(0);
+    expect(registry.size).toBe(3);
+    registry.end(ids[0]);
+    expect(registry.size).toBe(2);
+  });
+
+  it("reports the oldest entries and the routes most of them share", () => {
+    const registry = new InFlightRegistry();
+    registry.begin({ kind: "job", target: "backup", startedAt: 1_000 });
+    for (let i = 0; i < 3; i++) {
+      registry.begin({
+        kind: "http",
+        method: "POST",
+        target: "/login",
+        serverAction: true,
+        startedAt: 50_000 + i,
+      });
+    }
+    const report = registry.report(61_000, 2);
+    expect(report.oldest).toEqual([
+      { kind: "job", route: "backup", age_s: 60 },
+      {
+        kind: "http",
+        method: "POST",
+        route: "/login",
+        server_action: true,
+        age_s: 11,
+      },
+    ]);
+    expect(report.by_route[0]).toEqual({
+      kind: "http",
+      route: "/login",
+      count: 3,
+    });
+  });
+});
+
+describe("HTTP request tracking", () => {
+  it("keeps the request until its response closes, without the query", () => {
+    const registry = new InFlightRegistry();
+    const exchange = fakeExchange("/api/x?token=abc", { "next-action": "1" });
+    trackHttpRequest(exchange.message, registry);
+    expect(registry.size).toBe(1);
+    const [entry] = registry.report(Date.now(), 1).oldest;
+    expect(entry.route).toBe("/api/x");
+    expect(entry.server_action).toBe(true);
+    expect(JSON.stringify(registry.report(Date.now(), 5))).not.toContain("abc");
+    exchange.close();
+    expect(registry.size).toBe(0);
+  });
+
+  it("removes an aborted request: close fires once, a second close is harmless", () => {
+    const registry = new InFlightRegistry();
+    const first = fakeExchange("/a");
+    const second = fakeExchange("/b");
+    trackHttpRequest(first.message, registry);
+    trackHttpRequest(second.message, registry);
+    first.close();
+    first.close();
+    expect(registry.size).toBe(1);
+    second.close();
+    expect(registry.size).toBe(0);
+  });
+
+  it("sees a real request through Node's server channel, and lets it go", async () => {
+    startHttpInFlightTracking();
+    const registry = inFlightRegistry();
+    const before = registry.size;
+    let during = -1;
+    const server = createServer((_req, res) => {
+      during = registry.size;
+      res.end("ok");
+    });
+    await new Promise<void>((resolve) => server.listen(0, resolve));
+    const { port } = server.address() as AddressInfo;
+    await new Promise<void>((resolve, reject) => {
+      httpRequest({ port, path: "/probe?x=1" }, (res) => {
+        res.resume();
+        res.on("end", resolve);
+      })
+        .on("error", reject)
+        .end();
+    });
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+    expect(during).toBe(before + 1);
+    expect(registry.size).toBe(before);
+  });
+});
+
+describe("HTTP request tracking, client gone", () => {
+  it("removes a request whose client hung up before any answer", async () => {
+    startHttpInFlightTracking();
+    const registry = inFlightRegistry();
+    const before = registry.size;
+    let serverSawClose: () => void = () => {};
+    const closed = new Promise<void>((resolve) => (serverSawClose = resolve));
+    let during = -1;
+    const server = createServer((_req, res) => {
+      during = registry.size;
+      // Never answers; the client aborts.
+      res.on("close", () => setImmediate(serverSawClose));
+    });
+    await new Promise<void>((resolve) => server.listen(0, resolve));
+    const { port } = server.address() as AddressInfo;
+    const req = httpRequest({ port, path: "/hang" });
+    req.on("error", () => {});
+    req.end();
+    while (during === -1) await new Promise((r) => setTimeout(r, 5));
+    req.destroy();
+    await closed;
+    server.closeAllConnections();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+    expect(during).toBe(before + 1);
+    expect(registry.size).toBe(before);
+  });
+});
+
+describe("routeLabel", () => {
+  it("scrubs secret segments and collapses ids and payloads", () => {
+    expect(
+      routeLabel({ kind: "http", target: "/api/withings/webhook/s3cret" }),
+    ).toBe("/api/withings/webhook/[REDACTED]");
+    expect(
+      routeLabel({
+        kind: "http",
+        target: "/api/measurements/clx9k2abcdefghij12345",
+      }),
+    ).toBe("/api/measurements/:id");
+    expect(routeLabel({ kind: "http", target: "/api/items/42" })).toBe(
+      "/api/items/:id",
+    );
+    expect(
+      routeLabel({ kind: "http", target: "/%27%20OR%201=1--/login" }),
+    ).toBe("/:x/login");
+  });
+
+  it("is bounded in segments and length", () => {
+    const label = routeLabel({
+      kind: "http",
+      target: "/" + Array.from({ length: 30 }, () => "abc").join("/"),
+    });
+    expect(label.split("/").length).toBeLessThanOrEqual(10);
+    expect(label.endsWith("/…")).toBe(true);
+    expect(
+      routeLabel({ kind: "http", target: "/" + "a/".repeat(500) }).length,
+    ).toBeLessThanOrEqual(120);
+  });
+});

src/lib/observability/in-flight.ts

@@ -0,0 +1,233 @@
+import { subscribe } from "node:diagnostics_channel";
+import type { IncomingMessage, ServerResponse } from "node:http";
+
+import { redactSecrets } from "@/lib/logging/redact";
+
+/**
+ * What the process is doing right now: every HTTP request it has accepted and
+ * not yet answered, and every queue job a handler is running.
+ *
+ * It exists for one moment, the heap-pressure report in `heap-monitor.ts`. A
+ * process that runs out of heap dies without a JavaScript stack and without a
+ * word about the work it was carrying, so the list has to be in memory and
+ * ready before the heap is gone, not reconstructed afterwards.
+ *
+ * ## Why the HTTP side hooks the server and not `apiHandler` or the proxy
+ *
+ * `apiHandler` sees API routes only. Pages, RSC payloads and server actions,
+ * which is what an unauthenticated scanner mostly hits, never pass through it.
+ * The proxy sees those, but only on the way in: it hands the request on and
+ * never learns when the render finished, so it cannot keep a live set. Node's
+ * own `http.server.request.start` diagnostics channel fires for every request
+ * the server accepts, in this process, before any Next.js code runs, and the
+ * response's `close` event fires exactly once whether it finished or the
+ * client went away. One map insert and one delete per request, nothing on the
+ * request path itself.
+ *
+ * ## Bounds and privacy
+ *
+ * At most `IN_FLIGHT_MAX` entries; a request arriving past it is counted in
+ * `dropped` and not kept. An entry holds strings and a number only, never the
+ * request object, so a response whose `close` never fired cannot pin a body.
+ * The query string is cut on entry and never stored. The path is kept as
+ * received and turned into a label only when a report is written: secret path
+ * segments scrubbed by `redactSecrets`, id-shaped and odd segments collapsed.
+ */
+
+export const IN_FLIGHT_MAX = 2000;
+
+export type InFlightKind = "http" | "job";
+
+export interface InFlightEntry {
+  kind: InFlightKind;
+  /** HTTP method; absent for jobs. */
+  method?: string;
+  /** Path without query for HTTP, queue name for jobs. */
+  target: string;
+  /** Set when the request carried a `Next-Action` header. */
+  serverAction?: boolean;
+  startedAt: number;
+}
+
+export interface InFlightReportEntry {
+  kind: InFlightKind;
+  method?: string;
+  route: string;
+  server_action?: true;
+  age_s: number;
+}
+
+export class InFlightRegistry {
+  private readonly entries = new Map<number, InFlightEntry>();
+  private nextId = 1;
+  private droppedCount = 0;
+
+  constructor(private readonly max: number = IN_FLIGHT_MAX) {}
+
+  /** Record a unit of work. Returns its id, or 0 when the register is full. */
+  begin(entry: InFlightEntry): number {
+    if (this.entries.size >= this.max) {
+      this.droppedCount++;
+      return 0;
+    }
+    const id = this.nextId++;
+    this.entries.set(id, entry);
+    return id;
+  }
+
+  /** Forget a unit of work. Idempotent; 0 is a no-op. */
+  end(id: number): void {
+    if (id !== 0) this.entries.delete(id);
+  }
+
+  get size(): number {
+    return this.entries.size;
+  }
+
+  /** Requests and jobs refused because the register was full. */
+  get dropped(): number {
+    return this.droppedCount;
+  }
+
+  counts(): Record<InFlightKind, number> {
+    const out: Record<InFlightKind, number> = { http: 0, job: 0 };
+    for (const entry of this.entries.values()) out[entry.kind]++;
+    return out;
+  }
+
+  /**
+   * The report a pressure event carries: the `limit` oldest entries, and the
+   * `limit` most frequent labels with how many entries share each. A flood of
+   * young requests to one route shows in the second list even when the first
+   * is all slow jobs.
+   */
+  report(
+    now: number,
+    limit: number,
+  ): {
+    oldest: InFlightReportEntry[];
+    by_route: Array<{ kind: InFlightKind; route: string; count: number }>;
+  } {
+    const all = [...this.entries.values()];
+    all.sort((a, b) => a.startedAt - b.startedAt);
+    const oldest = all.slice(0, limit).map((entry) => ({
+      kind: entry.kind,
+      ...(entry.method ? { method: entry.method } : {}),
+      route: routeLabel(entry),
+      ...(entry.serverAction ? { server_action: true as const } : {}),
+      age_s: Math.max(0, Math.round((now - entry.startedAt) / 1000)),
+    }));
+
+    const tally = new Map<
+      string,
+      { kind: InFlightKind; route: string; count: number }
+    >();
+    for (const entry of all) {
+      const route = routeLabel(entry);
+      const key = `${entry.kind} ${entry.method ?? ""} ${route}`;
+      const cell = tally.get(key);
+      if (cell) cell.count++;
+      else tally.set(key, { kind: entry.kind, route, count: 1 });
+    }
+    const by_route = [...tally.values()]
+      .sort((a, b) => b.count - a.count)
+      .slice(0, limit);
+
+    return { oldest, by_route };
+  }
+}
+
+const ROUTE_LABEL_MAX = 120;
+const SEGMENTS_MAX = 8;
+const PLAIN_SEGMENT = /^[A-Za-z0-9._~-]{1,40}$/;
+const ID_SEGMENT = /^(?=.*\d)[A-Za-z0-9_-]{16,}$|^\d+$/;
+
+/**
+ * A request path as a report label: secret segments scrubbed, id-shaped
+ * segments (`:id`) and anything that is not a plain path word (`:x`, which
+ * catches the encoded payloads a scanner sends) collapsed, at most eight
+ * segments and 120 characters. A job's label is its queue name as is.
+ */
+export function routeLabel(entry: Pick<InFlightEntry, "kind" | "target">) {
+  if (entry.kind === "job") return entry.target.slice(0, ROUTE_LABEL_MAX);
+  const segments = redactSecrets(entry.target).split("/");
+  const label = segments
+    .slice(0, SEGMENTS_MAX + 1)
+    .map((segment) => {
+      if (segment === "" || segment === "[REDACTED]") return segment;
+      if (ID_SEGMENT.test(segment)) return ":id";
+      return PLAIN_SEGMENT.test(segment) ? segment : ":x";
+    })
+    .join("/");
+  const suffix = segments.length > SEGMENTS_MAX + 1 ? "/…" : "";
+  return (label + suffix).slice(0, ROUTE_LABEL_MAX);
+}
+
+const REGISTRY = Symbol.for("healthlog.inFlight.registry");
+const HTTP_HOOKED = Symbol.for("healthlog.inFlight.httpHooked");
+
+type Slots = { [REGISTRY]?: InFlightRegistry; [HTTP_HOOKED]?: boolean };
+
+/**
+ * The process-wide register. Kept on `globalThis` because Next.js bundles
+ * instrumentation, route handlers and the job runner into separate module
+ * graphs, and each would otherwise get a register of its own.
+ */
+export function inFlightRegistry(): InFlightRegistry {
+  const slots = globalThis as unknown as Slots;
+  return (slots[REGISTRY] ??= new InFlightRegistry());
+}
+
+/** Run `fn` as one registered unit of work; the entry goes on any exit. */
+export async function trackInFlight<R>(
+  entry: Omit<InFlightEntry, "startedAt">,
+  fn: () => Promise<R>,
+  registry: InFlightRegistry = inFlightRegistry(),
+): Promise<R> {
+  const id = registry.begin({ ...entry, startedAt: Date.now() });
+  try {
+    return await fn();
+  } finally {
+    registry.end(id);
+  }
+}
+
+interface RequestStartMessage {
+  request: IncomingMessage;
+  response: ServerResponse;
+}
+
+/**
+ * Register one accepted request until its response closes. Exported for the
+ * tests; the channel subscriber below is the only production caller.
+ */
+export function trackHttpRequest(
+  message: RequestStartMessage,
+  registry: InFlightRegistry = inFlightRegistry(),
+): void {
+  const { request, response } = message;
+  const url = request.url ?? "";
+  const queryAt = url.indexOf("?");
+  const id = registry.begin({
+    kind: "http",
+    method: request.method ?? "GET",
+    target: queryAt === -1 ? url : url.slice(0, queryAt),
+    serverAction: request.headers["next-action"] !== undefined,
+    startedAt: Date.now(),
+  });
+  if (id !== 0) response.once("close", () => registry.end(id));
+}
+
+/** Hook every HTTP request this process serves. Idempotent per process. */
+export function startHttpInFlightTracking(): void {
+  const slots = globalThis as unknown as Slots;
+  if (slots[HTTP_HOOKED]) return;
+  slots[HTTP_HOOKED] = true;
+  subscribe("http.server.request.start", (message) => {
+    try {
+      trackHttpRequest(message as RequestStartMessage);
+    } catch {
+      // A diagnostic must never break a request.
+    }
+  });
+}