Merge branch 'fix/v1436-heap' into release/v1.43.6

src/instrumentation.ts

@@ -56,6 +56,20 @@ export async function register() {
       await import("@/lib/observability/event-loop-lag");
     startEventLoopLagMonitor();
 
+    // The same for its heap: a trend line, and one report naming the work in
+    // flight before the heap runs out (a heap-limit crash logs nothing else).
+    try {
+      const [{ startHttpInFlightTracking }, { startHeapMonitor }] =
+        await Promise.all([
+          import("@/lib/observability/in-flight"),
+          import("@/lib/observability/heap-monitor"),
+        ]);
+      startHttpInFlightTracking();
+      startHeapMonitor();
+    } catch {
+      // A diagnostic must never break boot.
+    }
+
     const { shouldRunWorker } = await import("@/lib/process-type");
     if (!shouldRunWorker()) {
       // Split deployment: web requests still produce queue jobs, but this

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__/heap-monitor.test.ts

@@ -0,0 +1,232 @@
+import {
+  mkdtempSync,
+  readdirSync,
+  readFileSync,
+  rmSync,
+  statSync,
+} from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { describe, expect, it } from "vitest";
+
+import { walkSourceFiles } from "@/__tests__/helpers/source-files";
+
+import {
+  HEAP_CHECK_INTERVAL_MS,
+  HEAP_PRESSURE_COOLDOWN_MS,
+  HEAP_PRESSURE_LIST_MAX,
+  HEAP_SAMPLE_INTERVAL_ELEVATED_MS,
+  HEAP_SAMPLE_INTERVAL_MS,
+  HEAP_TASK_NAMES,
+  HeapMonitor,
+  type HeapFigures,
+  type HeapMonitorDeps,
+  writeDiagnosticHeapSnapshot,
+} from "../heap-monitor";
+import { InFlightRegistry } from "../in-flight";
+
+const LIMIT = 1000;
+
+const FIGURE_KEYS: Array<keyof HeapFigures> = [
+  "heap_used_mb",
+  "heap_total_mb",
+  "heap_limit_mb",
+  "heap_used_ratio",
+  "old_space_mb",
+  "rss_mb",
+  "external_mb",
+  "array_buffers_mb",
+  "active_requests",
+  "active_jobs",
+  "in_flight_dropped",
+  "uptime_s",
+];
+
+function harness(overrides: Partial<HeapMonitorDeps> = {}) {
+  const state = { now: 0, used: 100 };
+  const emitted: Array<{
+    task: string;
+    level: string;
+    meta: Record<string, unknown>;
+  }> = [];
+  let snapshots = 0;
+  const registry = new InFlightRegistry();
+  const figures = (): HeapFigures =>
+    Object.fromEntries(
+      FIGURE_KEYS.map((k) => [k, 1]),
+    ) as unknown as HeapFigures;
+  const monitor = new HeapMonitor({
+    now: () => state.now,
+    readHeap: () => ({ used: state.used, limit: LIMIT }),
+    readFigures: figures,
+    registry,
+    emit: (task, level, meta) => emitted.push({ task, level, meta }),
+    snapshotEnabled: false,
+    writeSnapshot: () => {
+      snapshots++;
+      return { snapshot: "written", snapshot_path: "/diag/x.heapsnapshot" };
+    },
+    ...overrides,
+  });
+  /** Advance in five-second checks for `ms`. */
+  const run = (ms: number) => {
+    for (let t = 0; t < ms; t += HEAP_CHECK_INTERVAL_MS) {
+      monitor.check();
+      state.now += HEAP_CHECK_INTERVAL_MS;
+    }
+  };
+  const named = (task: string) => emitted.filter((e) => e.task === task);
+  return {
+    state,
+    emitted,
+    named,
+    run,
+    registry,
+    snapshots: () => snapshots,
+  };
+}
+
+describe("heap monitor sampling", () => {
+  it("samples every five minutes under 70 % and every minute above", () => {
+    const h = harness();
+    h.run(30 * 60_000);
+    expect(h.named(HEAP_TASK_NAMES.sampled)).toHaveLength(
+      (30 * 60_000) / HEAP_SAMPLE_INTERVAL_MS,
+    );
+
+    const elevated = harness();
+    elevated.state.used = 750;
+    elevated.run(10 * 60_000);
+    expect(elevated.named(HEAP_TASK_NAMES.sampled)).toHaveLength(
+      (10 * 60_000) / HEAP_SAMPLE_INTERVAL_ELEVATED_MS,
+    );
+  });
+
+  it("carries a fixed meta shape at info", () => {
+    const h = harness();
+    h.run(HEAP_CHECK_INTERVAL_MS);
+    const [line] = h.named(HEAP_TASK_NAMES.sampled);
+    expect(line.level).toBe("info");
+    expect(Object.keys(line.meta).sort()).toEqual([...FIGURE_KEYS].sort());
+  });
+});
+
+describe("heap pressure report", () => {
+  it("fires once at 85 % with the work in flight, at warn", () => {
+    const h = harness();
+    for (let i = 0; i < 80; i++) {
+      h.registry.begin({
+        kind: "http",
+        method: "POST",
+        target: "/",
+        startedAt: i,
+      });
+    }
+    h.state.used = 900;
+    h.run(20 * 60_000);
+    const pressure = h.named(HEAP_TASK_NAMES.pressure);
+    expect(pressure).toHaveLength(1);
+    expect(pressure[0].level).toBe("warn");
+    expect(pressure[0].meta.snapshot).toBe("disabled");
+    expect(pressure[0].meta.in_flight_oldest).toHaveLength(
+      HEAP_PRESSURE_LIST_MAX,
+    );
+    expect(pressure[0].meta.in_flight_by_route).toEqual([
+      { kind: "http", route: "/", count: 80 },
+    ]);
+    for (const key of FIGURE_KEYS) expect(pressure[0].meta).toHaveProperty(key);
+  });
+
+  it("stays quiet while the heap hovers, and re-arms only after it fell back and the cooldown passed", () => {
+    const h = harness();
+    h.state.used = 900;
+    h.run(60_000);
+    // Falls back, but the cooldown has not passed.
+    h.state.used = 500;
+    h.run(60_000);
+    h.state.used = 900;
+    h.run(60_000);
+    expect(h.named(HEAP_TASK_NAMES.pressure)).toHaveLength(1);
+
+    // Stays above 70 % past the cooldown: still armed off.
+    h.state.used = 800;
+    h.run(HEAP_PRESSURE_COOLDOWN_MS);
+    h.state.used = 900;
+    h.run(60_000);
+    expect(h.named(HEAP_TASK_NAMES.pressure)).toHaveLength(1);
+
+    // Falls under 70 % after the cooldown: armed again.
+    h.state.used = 500;
+    h.run(HEAP_CHECK_INTERVAL_MS);
+    h.state.used = 900;
+    h.run(HEAP_CHECK_INTERVAL_MS);
+    expect(h.named(HEAP_TASK_NAMES.pressure)).toHaveLength(2);
+  });
+
+  it("writes no snapshot unless opted in, and at most one when it is", () => {
+    const off = harness();
+    off.state.used = 900;
+    off.run(60_000);
+    expect(off.snapshots()).toBe(0);
+    expect(off.named(HEAP_TASK_NAMES.snapshot)).toHaveLength(0);
+
+    const on = harness({ snapshotEnabled: true });
+    for (let round = 0; round < 3; round++) {
+      on.state.used = 900;
+      on.run(60_000);
+      on.state.used = 100;
+      on.run(HEAP_PRESSURE_COOLDOWN_MS + HEAP_CHECK_INTERVAL_MS);
+    }
+    expect(on.named(HEAP_TASK_NAMES.pressure)).toHaveLength(3);
+    expect(on.snapshots()).toBe(1);
+    const [line] = on.named(HEAP_TASK_NAMES.snapshot);
+    expect(line.meta.snapshot_path).toBe("/diag/x.heapsnapshot");
+    // The report is written before the snapshot is taken.
+    const order = on.emitted
+      .filter((e) => e.task !== HEAP_TASK_NAMES.sampled)
+      .map((e) => e.task);
+    expect(order.slice(0, 2)).toEqual([
+      HEAP_TASK_NAMES.pressure,
+      HEAP_TASK_NAMES.snapshot,
+    ]);
+  });
+});
+
+describe("runtime.heap task names", () => {
+  it("every runtime.heap.* literal in src is a registered task name", () => {
+    const registered = new Set<string>(Object.values(HEAP_TASK_NAMES));
+    const found = new Set<string>();
+    const src = join(process.cwd(), "src");
+    for (const file of walkSourceFiles(src, { floor: 1000 })) {
+      if (file.includes("__tests__") || file.startsWith("generated/")) continue;
+      const text = readFileSync(join(src, file), "utf8");
+      for (const match of text.matchAll(/["'`](runtime\.heap\.[a-z_.]+)["'`]/g))
+        found.add(match[1]);
+    }
+    // A sweep that matches nothing agrees with any register.
+    expect(found.size).toBeGreaterThan(0);
+    for (const name of found) expect(registered).toContain(name);
+    for (const name of registered) expect(found).toContain(name);
+  });
+});
+
+describe("default snapshot writer", () => {
+  it("writes into a 0700 directory, the file 0600", () => {
+    const dir = mkdtempSync(join(tmpdir(), "hl-heap-test-"));
+    const previous = process.env.HEALTHLOG_DIAG_DIR;
+    process.env.HEALTHLOG_DIAG_DIR = dir;
+    try {
+      const outcome = writeDiagnosticHeapSnapshot();
+      expect(outcome.snapshot).toBe("written");
+      if (outcome.snapshot !== "written") return;
+      const sub = readdirSync(dir)[0];
+      expect(statSync(join(dir, sub)).mode & 0o777).toBe(0o700);
+      expect(statSync(outcome.snapshot_path).mode & 0o777).toBe(0o600);
+    } finally {
+      if (previous === undefined) delete process.env.HEALTHLOG_DIAG_DIR;
+      else process.env.HEALTHLOG_DIAG_DIR = previous;
+      rmSync(dir, { recursive: true, force: true });
+    }
+  }, 90_000);
+});

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/heap-monitor.ts

@@ -0,0 +1,272 @@
+import { chmodSync, mkdirSync } from "node:fs";
+import { join } from "node:path";
+import {
+  getHeapSpaceStatistics,
+  getHeapStatistics,
+  writeHeapSnapshot,
+} from "node:v8";
+
+import { envFlag, envOr } from "@/lib/env";
+import { WideEventBuilder } from "@/lib/logging/event-builder";
+import { emitIfSampled } from "@/lib/logging/transports";
+
+import { inFlightRegistry, type InFlightRegistry } from "./in-flight";
+
+/**
+ * Heap monitor: the line a process writes before it runs out of memory.
+ *
+ * A Node process that hits its heap limit dies with `JavaScript heap out of
+ * memory` and nothing else: no JavaScript stack, no wide event, no record of
+ * the requests it was serving. The only memory figures in the log came from
+ * `job.progress`, written by jobs that run longer than two minutes. That made
+ * a crash impossible to attribute after the fact.
+ *
+ * Three task names, all background wide events:
+ *
+ *   - `runtime.heap.sampled`, every five minutes, and every minute while the
+ *     heap stands at or above 70 % of its limit. About 290 lines a day on a
+ *     quiet process, a trend line rather than a flood.
+ *   - `runtime.heap.pressure`, at `warn`, the first time the heap reaches 85 %
+ *     of its limit: the heap figures plus the oldest requests and jobs in
+ *     flight and the routes most of them share (see `in-flight.ts`). It
+ *     re-arms only after the heap has fallen back under 70 % AND thirty minutes
+ *     have passed, so a heap hovering at the line writes one report, not one
+ *     every check.
+ *   - `runtime.heap.snapshot`, at `warn`, only with the opt-in below: where
+ *     the snapshot was written, or that writing it failed.
+ *
+ * The heap is read every five seconds, which costs one
+ * `v8.getHeapStatistics()` call (a few microseconds); the fuller read for an
+ * emitted line runs at most once a minute. A pressure climb therefore cannot
+ * hide between two minute ticks.
+ *
+ * Opt-in heap snapshot: with `HEALTHLOG_HEAP_SNAPSHOT=1`, the first pressure
+ * report is followed by one `v8.writeHeapSnapshot()` into
+ * `HEALTHLOG_DIAG_DIR` (default `/tmp`), inside a directory only the process
+ * user can open, the file chmod 0600. One per process, ever. A snapshot is a
+ * copy of the whole heap, decrypted health records and keys included: it is
+ * never uploaded, never logged, and only its path is reported. Writing it
+ * blocks the process for seconds and needs memory of its own, so it is a tool
+ * for chasing one crash, not a setting to leave on. See
+ * `docs/ops/heap-diagnostics.md`.
+ */
+
+export const HEAP_TASK_NAMES = {
+  sampled: "runtime.heap.sampled",
+  pressure: "runtime.heap.pressure",
+  snapshot: "runtime.heap.snapshot",
+} as const;
+
+export const HEAP_CHECK_INTERVAL_MS = 5_000;
+export const HEAP_SAMPLE_INTERVAL_MS = 5 * 60_000;
+export const HEAP_SAMPLE_INTERVAL_ELEVATED_MS = 60_000;
+export const HEAP_ELEVATED_RATIO = 0.7;
+export const HEAP_PRESSURE_RATIO = 0.85;
+export const HEAP_PRESSURE_COOLDOWN_MS = 30 * 60_000;
+export const HEAP_PRESSURE_LIST_MAX = 50;
+
+export interface HeapFigures {
+  heap_used_mb: number;
+  heap_total_mb: number;
+  heap_limit_mb: number;
+  heap_used_ratio: number;
+  old_space_mb: number;
+  rss_mb: number;
+  external_mb: number;
+  array_buffers_mb: number;
+  active_requests: number;
+  active_jobs: number;
+  in_flight_dropped: number;
+  uptime_s: number;
+}
+
+export type SnapshotOutcome =
+  | { snapshot: "written"; snapshot_path: string }
+  | { snapshot: "failed"; snapshot_error: string };
+
+export interface HeapMonitorDeps {
+  now: () => number;
+  /** Used heap and limit, in bytes: the five-second check. */
+  readHeap: () => { used: number; limit: number };
+  /** The full figures for an emitted line. */
+  readFigures: () => HeapFigures;
+  registry: InFlightRegistry;
+  emit: (
+    taskName: string,
+    level: "info" | "warn",
+    meta: Record<string, unknown>,
+  ) => void;
+  snapshotEnabled: boolean;
+  writeSnapshot: () => SnapshotOutcome;
+}
+
+function megabytes(bytes: number): number {
+  return Math.round(bytes / 1024 / 1024);
+}
+
+function defaultReadHeap(): { used: number; limit: number } {
+  const stats = getHeapStatistics();
+  return { used: stats.used_heap_size, limit: stats.heap_size_limit };
+}
+
+function defaultReadFigures(registry: InFlightRegistry): HeapFigures {
+  const memory = process.memoryUsage();
+  const limit = getHeapStatistics().heap_size_limit;
+  const oldSpace = getHeapSpaceStatistics().find(
+    (space) => space.space_name === "old_space",
+  );
+  const counts = registry.counts();
+  return {
+    heap_used_mb: megabytes(memory.heapUsed),
+    heap_total_mb: megabytes(memory.heapTotal),
+    heap_limit_mb: megabytes(limit),
+    heap_used_ratio:
+      limit > 0 ? Math.round((memory.heapUsed / limit) * 1000) / 1000 : 0,
+    old_space_mb: megabytes(oldSpace?.space_used_size ?? 0),
+    rss_mb: megabytes(memory.rss),
+    external_mb: megabytes(memory.external),
+    array_buffers_mb: megabytes(memory.arrayBuffers),
+    active_requests: counts.http,
+    active_jobs: counts.job,
+    in_flight_dropped: registry.dropped,
+    uptime_s: Math.round(process.uptime()),
+  };
+}
+
+function defaultEmit(
+  taskName: string,
+  level: "info" | "warn",
+  meta: Record<string, unknown>,
+): void {
+  const event = new WideEventBuilder("background");
+  event.setBackground({ task_name: taskName });
+  for (const [key, value] of Object.entries(meta)) event.addMeta(key, value);
+  if (level === "warn") event.elevateLevel("warn");
+  event.finish();
+  emitIfSampled(event.toJSON());
+}
+
+/** The opt-in heap snapshot; exported for its file-mode test. */
+export function writeDiagnosticHeapSnapshot(): SnapshotOutcome {
+  try {
+    const dir = join(
+      envOr("HEALTHLOG_DIAG_DIR", "/tmp"),
+      `healthlog-heap-${process.pid}`,
+    );
+    mkdirSync(dir, { recursive: true, mode: 0o700 });
+    chmodSync(dir, 0o700);
+    const stamp = new Date().toISOString().replace(/[:.]/g, "-");
+    const path = writeHeapSnapshot(join(dir, `heap-${stamp}.heapsnapshot`));
+    chmodSync(path, 0o600);
+    return { snapshot: "written", snapshot_path: path };
+  } catch (err) {
+    return {
+      snapshot: "failed",
+      snapshot_error:
+        (err as NodeJS.ErrnoException | null)?.code ??
+        (err instanceof Error ? err.name : "Unknown"),
+    };
+  }
+}
+
+/**
+ * The monitor's decisions, one `check()` per five-second tick. Separate from
+ * the timer so the tests drive it with a fake clock and fake readings.
+ */
+export class HeapMonitor {
+  private lastSampleAt: number | null = null;
+  private pressureArmed = true;
+  private lastPressureAt: number | null = null;
+  private snapshotTaken = false;
+
+  constructor(private readonly deps: HeapMonitorDeps) {}
+
+  check(): void {
+    const { now, readHeap } = this.deps;
+    const at = now();
+    const { used, limit } = readHeap();
+    const ratio = limit > 0 ? used / limit : 0;
+
+    if (
+      !this.pressureArmed &&
+      ratio < HEAP_ELEVATED_RATIO &&
+      this.lastPressureAt !== null &&
+      at - this.lastPressureAt >= HEAP_PRESSURE_COOLDOWN_MS
+    ) {
+      this.pressureArmed = true;
+    }
+
+    if (this.pressureArmed && ratio >= HEAP_PRESSURE_RATIO) {
+      this.pressureArmed = false;
+      this.lastPressureAt = at;
+      this.lastSampleAt = at;
+      this.reportPressure(at);
+      return;
+    }
+
+    const interval =
+      ratio >= HEAP_ELEVATED_RATIO
+        ? HEAP_SAMPLE_INTERVAL_ELEVATED_MS
+        : HEAP_SAMPLE_INTERVAL_MS;
+    if (this.lastSampleAt === null || at - this.lastSampleAt >= interval) {
+      this.lastSampleAt = at;
+      this.deps.emit(HEAP_TASK_NAMES.sampled, "info", {
+        ...this.deps.readFigures(),
+      });
+    }
+  }
+
+  private reportPressure(at: number): void {
+    const { registry, emit, readFigures } = this.deps;
+    const inFlight = registry.report(at, HEAP_PRESSURE_LIST_MAX);
+    const willSnapshot = this.deps.snapshotEnabled && !this.snapshotTaken;
+    // Written before the snapshot: if writing the snapshot is what tips the
+    // process over, the report is already on stdout.
+    emit(HEAP_TASK_NAMES.pressure, "warn", {
+      ...readFigures(),
+      pressure_ratio: HEAP_PRESSURE_RATIO,
+      in_flight_oldest: inFlight.oldest,
+      in_flight_by_route: inFlight.by_route,
+      snapshot: willSnapshot ? "pending" : "disabled",
+    });
+    if (!willSnapshot) return;
+    this.snapshotTaken = true;
+    emit(HEAP_TASK_NAMES.snapshot, "warn", {
+      ...readFigures(),
+      ...this.deps.writeSnapshot(),
+    });
+  }
+}
+
+const STARTED = Symbol.for("healthlog.heapMonitor.started");
+
+/**
+ * Start the monitor. Idempotent per process, like the event-loop monitor
+ * beside it; the timer is unref'd and never holds the process open.
+ */
+export function startHeapMonitor(overrides: Partial<HeapMonitorDeps> = {}) {
+  const slots = globalThis as unknown as { [STARTED]?: boolean };
+  if (slots[STARTED]) return;
+  slots[STARTED] = true;
+
+  const registry = overrides.registry ?? inFlightRegistry();
+  const monitor = new HeapMonitor({
+    now: Date.now,
+    readHeap: defaultReadHeap,
+    readFigures: () => defaultReadFigures(registry),
+    registry,
+    emit: defaultEmit,
+    snapshotEnabled: envFlag("HEALTHLOG_HEAP_SNAPSHOT"),
+    writeSnapshot: writeDiagnosticHeapSnapshot,
+    ...overrides,
+  });
+
+  const timer = setInterval(() => {
+    try {
+      monitor.check();
+    } catch {
+      // A diagnostic must never take the process down.
+    }
+  }, HEAP_CHECK_INTERVAL_MS);
+  timer.unref();
+}

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.
+    }
+  });
+}