feat(observability): log heap samples and a one-time pressure report

runtime.heap.sampled records heap, old space, RSS, external memory and
the work in flight every five minutes, every minute at or above 70 % of
the heap limit. At 85 % runtime.heap.pressure lists the oldest requests
and jobs in flight and the routes most of them share, once, re-armed
only after the heap falls below 70 % and thirty minutes pass. A
heap-limit crash leaves no other record of what the process was doing.

HEALTHLOG_HEAP_SNAPSHOT=1 adds one heap snapshot at that moment into
HEALTHLOG_DIAG_DIR (0700 directory, 0600 file), off by default.
docs/ops/heap-diagnostics.md warns that a snapshot holds decrypted
health data and keys and says how to delete it.

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