feat(observability): guard the container's memory, not only the heap

The heap monitor's tick now also reads the cgroup working set (usage minus
reclaimable page cache) against the container limit. At 85 % it writes one
runtime.memory.pressure line with the work in flight, and the heavy lane
sends new heavy jobs back to the queue for a minute and asks a sliced
holder to give way, for at most fifteen minutes per episode. Heap lines
carry the working set and the limit. Outside a cgroup with a limit nothing
changes.

src/lib/jobs/__tests__/heavy-lane.test.ts

@@ -9,6 +9,10 @@ import type { Job } from "pg-boss";
 vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
 vi.mock("@/lib/db", () => ({ prisma: {} }));
 vi.mock("@/lib/logging/signal", () => ({ emitSignal: vi.fn() }));
+const memory = vi.hoisted(() => ({ pressure: false }));
+vi.mock("@/lib/observability/container-memory", () => ({
+  memoryPressureActive: () => memory.pressure,
+}));
 
 import { emitSignal } from "@/lib/logging/signal";
 import { heavyLane, LANE_WAIT_WARN_MS, laneWanted } from "../heavy-lane";
@@ -266,3 +270,52 @@ describe("a long wait for the lane", () => {
     await held;
   });
 });
+
+describe("the lane under container memory pressure", () => {
+  it("sends a heavy job back while the container is near its limit", async () => {
+    const log: string[] = [];
+    const a = gate(log, "import");
+    const { back, defer } = deferrals();
+    memory.pressure = true;
+    try {
+      const outcome = await heavyLane(
+        "apple-health-import-v2",
+        a.handler,
+        undefined,
+        defer,
+      )([job("upload")]);
+      expect(outcome).toMatchObject({ ok: true, did: { lane_deferred: 1 } });
+      expect(a.handler).not.toHaveBeenCalled();
+      expect(back).toEqual([
+        { queue: "apple-health-import-v2", ids: ["upload"] },
+      ]);
+    } finally {
+      memory.pressure = false;
+    }
+  });
+
+  it("asks a holder to give way at its next boundary", async () => {
+    let seen: boolean[] = [];
+    const { defer } = deferrals();
+    let release!: () => void;
+    const gateDone = new Promise<void>((r) => (release = r));
+    const running = heavyLane(
+      "apple-health-import-v2",
+      async () => {
+        seen.push(laneWanted());
+        await gateDone;
+        seen.push(laneWanted());
+        return jobDone({ jobs: 1 });
+      },
+      undefined,
+      defer,
+    )([job("upload")]);
+    await tick();
+    memory.pressure = true;
+    release();
+    await running;
+    memory.pressure = false;
+    expect(seen).toEqual([false, true]);
+    seen = [];
+  });
+});

src/lib/jobs/heavy-lane.ts

@@ -33,6 +33,13 @@
  * job that has stood in line for `LANE_WAIT_WARN_MS` is reported at `warn`
  * (`job.heavy_lane.waiting_long`), once per holder, so a lane that is held
  * for hours shows in the alerts instead of only in `info` lines.
+ *
+ * The container memory guard (`observability/container-memory.ts`) closes
+ * the lane for a while when the cgroup's working set reaches 85 % of its
+ * limit: a heavy job that arrives then goes back to the queue the same way,
+ * and a holder that runs in slices gives way at its next boundary, so the
+ * peak of an import does not land on top of a memory peak and end in a
+ * SIGKILL. The guard holds jobs back for fifteen minutes at most.
  */
 import { AsyncLocalStorage } from "node:async_hooks";
 import type { Job } from "pg-boss";
@@ -40,6 +47,7 @@ import type { Job } from "pg-boss";
 import { prisma } from "@/lib/db";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { annotate } from "@/lib/logging/context";
+import { memoryPressureActive } from "@/lib/observability/container-memory";
 import { emitSignal } from "@/lib/logging/signal";
 
 export interface HeavyQueue {
@@ -155,7 +163,8 @@ const holdScope = new AsyncLocalStorage<LaneHold>();
  */
 export function laneWanted(): boolean {
   const current = holdScope.getStore();
-  return current !== undefined && current === hold && current.wantedBy !== null;
+  if (current === undefined || current !== hold) return false;
+  return current.wantedBy !== null || memoryPressureActive();
 }
 
 function reportLongWait(
@@ -201,6 +210,19 @@ export function heavyLane<J extends Job<unknown>>(
   return async (jobs) => {
     if (!jobs.some(inLane)) return handler(jobs);
     const current = hold;
+    if (!current && memoryPressureActive()) {
+      const deferral = await defer(queue, jobs);
+      annotate({
+        action: { name: "job.heavy_lane.deferred" },
+        meta: {
+          queue,
+          holder: "memory_pressure",
+          jobs: jobs.length,
+          deferred: deferral.deferred,
+        },
+      });
+      return jobDone({ lane_deferred: deferral.deferred });
+    }
     if (current) {
       const deferral = await defer(queue, jobs);
       if (queue !== current.queue) current.wantedBy ??= queue;

src/lib/observability/__tests__/container-memory.test.ts

@@ -0,0 +1,116 @@
+import { describe, expect, it } from "vitest";
+
+import { InFlightRegistry } from "../in-flight";
+import {
+  CONTAINER_BACKOFF_MAX_MS,
+  CONTAINER_MEMORY_TASK,
+  ContainerMemoryGuard,
+  memoryPressureActive,
+  readContainerMemory,
+  type ContainerMemoryReading,
+} from "../container-memory";
+
+const MB = 1024 * 1024;
+
+function files(map: Record<string, string>) {
+  return (path: string) => {
+    if (!(path in map)) throw new Error(`ENOENT ${path}`);
+    return map[path]!;
+  };
+}
+
+describe("readContainerMemory", () => {
+  it("reads cgroup v2 and subtracts the reclaimable page cache", () => {
+    const reading = readContainerMemory(
+      files({
+        "/sys/fs/cgroup/memory.current": `${900 * MB}\n`,
+        "/sys/fs/cgroup/memory.max": `${1536 * MB}\n`,
+        "/sys/fs/cgroup/memory.stat": `anon 1\ninactive_file ${100 * MB}\nactive_file 5\n`,
+      }),
+    );
+    expect(reading).toEqual({ workingSet: 800 * MB, limit: 1536 * MB });
+  });
+
+  it("falls back to cgroup v1", () => {
+    const reading = readContainerMemory(
+      files({
+        "/sys/fs/cgroup/memory/memory.usage_in_bytes": `${500 * MB}`,
+        "/sys/fs/cgroup/memory/memory.limit_in_bytes": `${1024 * MB}`,
+        "/sys/fs/cgroup/memory/memory.stat": `total_inactive_file ${20 * MB}`,
+      }),
+    );
+    expect(reading).toEqual({ workingSet: 480 * MB, limit: 1024 * MB });
+  });
+
+  it("reads nothing without a limit or outside a cgroup", () => {
+    expect(
+      readContainerMemory(
+        files({
+          "/sys/fs/cgroup/memory.current": "123",
+          "/sys/fs/cgroup/memory.max": "max",
+        }),
+      ),
+    ).toBeNull();
+    expect(readContainerMemory(files({}))).toBeNull();
+  });
+});
+
+describe("ContainerMemoryGuard", () => {
+  function guard(readings: Array<ContainerMemoryReading | null>) {
+    let at = 0;
+    let i = 0;
+    const lines: Array<{ task: string; meta: Record<string, unknown> }> = [];
+    const g = new ContainerMemoryGuard({
+      now: () => at,
+      read: () => readings[Math.min(i++, readings.length - 1)] ?? null,
+      registry: new InFlightRegistry(),
+      emit: (task, _level, meta) => lines.push({ task, meta }),
+      figures: () => ({ heap_used_mb: 1 }),
+    });
+    return {
+      tick: (ms = 5_000) => {
+        at += ms;
+        g.check();
+      },
+      lines,
+    };
+  }
+  const reading = (ratio: number) => ({
+    workingSet: Math.round(ratio * 1000 * MB),
+    limit: 1000 * MB,
+  });
+
+  it("reports once at 85 % and holds heavy jobs back while it lasts", () => {
+    const g = guard([reading(0.5), reading(0.9), reading(0.9), reading(0.6)]);
+    g.tick();
+    expect(memoryPressureActive()).toBe(false);
+    g.tick();
+    expect(memoryPressureActive()).toBe(true);
+    expect(g.lines).toHaveLength(1);
+    expect(g.lines[0]!.task).toBe(CONTAINER_MEMORY_TASK);
+    expect(g.lines[0]!.meta).toMatchObject({
+      container_working_set_mb: 900,
+      container_limit_mb: 1000,
+      heap_used_mb: 1,
+    });
+    g.tick();
+    expect(g.lines).toHaveLength(1);
+    g.tick();
+    expect(memoryPressureActive()).toBe(false);
+  });
+
+  it("lets heavy jobs through once the pressure has lasted fifteen minutes", () => {
+    const g = guard([reading(0.9)]);
+    g.tick();
+    expect(memoryPressureActive()).toBe(true);
+    g.tick(CONTAINER_BACKOFF_MAX_MS);
+    expect(memoryPressureActive()).toBe(false);
+  });
+
+  it("never raises the flag outside a cgroup", () => {
+    const g = guard([null]);
+    g.tick();
+    expect(memoryPressureActive()).toBe(false);
+    expect(g.lines).toHaveLength(0);
+  });
+});

src/lib/observability/container-memory.ts

@@ -0,0 +1,198 @@
+import { readFileSync } from "node:fs";
+
+import { megabytes } from "@/lib/support/heap-limit";
+
+import type { InFlightRegistry } from "./in-flight";
+
+/**
+ * Container memory guard: the process's whole footprint against the cgroup
+ * limit, not just the V8 heap.
+ *
+ * The heap monitor sees the V8 heap only. A large share of this process's
+ * memory lives outside it: the GeoLite2 databases read into Buffers, the
+ * query compiler's WebAssembly memory, canvas and pdf.js for document
+ * thumbnails, zlib streams of a backup. When the cgroup runs out the kernel
+ * kills the process with SIGKILL and nothing is logged at all, however calm
+ * the heap line looked.
+ *
+ * Every five seconds (the heap monitor's tick) the guard reads the cgroup's
+ * working set: `memory.current` minus the reclaimable `inactive_file` page
+ * cache, the figure `docker stats` and the kubelet report. At 85 % of the
+ * limit it writes one `runtime.memory.pressure` line at `warn` with the work
+ * in flight, re-armed once the working set is back under 70 % and thirty
+ * minutes have passed, and raises a flag the heavy job lane reads to send
+ * new heavy jobs back to the queue for a minute instead of starting them
+ * (`memoryPressureActive`). A pressure that does not ease within fifteen
+ * minutes is taken as the container's normal size rather than a peak: the
+ * flag drops, so a container sized at its limit is never starved of its
+ * nightly jobs, and the pressure line has said so.
+ *
+ * Outside a cgroup with a limit (a developer machine, a host without one)
+ * the guard reads nothing and never raises the flag.
+ */
+
+export const CONTAINER_MEMORY_TASK = "runtime.memory.pressure";
+export const CONTAINER_PRESSURE_RATIO = 0.85;
+export const CONTAINER_REARM_RATIO = 0.7;
+export const CONTAINER_PRESSURE_COOLDOWN_MS = 30 * 60_000;
+/** How long new heavy jobs are held back before the pressure counts as normal. */
+export const CONTAINER_BACKOFF_MAX_MS = 15 * 60_000;
+
+export interface ContainerMemoryReading {
+  /** Working set: usage minus reclaimable page cache, bytes. */
+  workingSet: number;
+  /** The cgroup's limit, bytes. */
+  limit: number;
+}
+
+const UNLIMITED_BYTES = 1024 ** 4;
+
+function readNumber(raw: string): number | null {
+  const text = raw.trim();
+  if (!/^\d+$/.test(text)) return null;
+  const n = Number(text);
+  return Number.isFinite(n) ? n : null;
+}
+
+function statField(stat: string, field: string): number {
+  const line = stat.split("\n").find((l) => l.startsWith(`${field} `));
+  return line ? (readNumber(line.slice(field.length + 1)) ?? 0) : 0;
+}
+
+/**
+ * The cgroup working set and limit, cgroup v2 first, then v1. `null` when
+ * there is no limit or the files cannot be read.
+ */
+export function readContainerMemory(
+  read: (path: string) => string = (path) => readFileSync(path, "utf8"),
+): ContainerMemoryReading | null {
+  const layouts = [
+    {
+      usage: "/sys/fs/cgroup/memory.current",
+      limit: "/sys/fs/cgroup/memory.max",
+      stat: "/sys/fs/cgroup/memory.stat",
+      inactive: "inactive_file",
+    },
+    {
+      usage: "/sys/fs/cgroup/memory/memory.usage_in_bytes",
+      limit: "/sys/fs/cgroup/memory/memory.limit_in_bytes",
+      stat: "/sys/fs/cgroup/memory/memory.stat",
+      inactive: "total_inactive_file",
+    },
+  ];
+  for (const layout of layouts) {
+    let usage: number | null;
+    let limit: number | null;
+    try {
+      usage = readNumber(read(layout.usage));
+      limit = readNumber(read(layout.limit));
+    } catch {
+      continue;
+    }
+    if (usage === null || limit === null) return null;
+    if (limit <= 0 || limit >= UNLIMITED_BYTES) return null;
+    let inactive = 0;
+    try {
+      inactive = statField(read(layout.stat), layout.inactive);
+    } catch {
+      // Without the stat file the usage stands as it is.
+    }
+    return { workingSet: Math.max(0, usage - inactive), limit };
+  }
+  return null;
+}
+
+const PRESSURE_FLAG = Symbol.for("healthlog.containerMemory.pressure");
+
+type FlagSlot = { [PRESSURE_FLAG]?: boolean };
+
+/**
+ * Whether new heavy jobs should wait: the working set is at or above 85 % of
+ * the limit and has not stayed there for fifteen minutes yet. Kept on
+ * `globalThis` because the lane and the guard can sit in different server
+ * bundles of one process.
+ */
+export function memoryPressureActive(): boolean {
+  return (globalThis as FlagSlot)[PRESSURE_FLAG] === true;
+}
+
+export interface ContainerMemoryDeps {
+  now: () => number;
+  read: () => ContainerMemoryReading | null;
+  registry: InFlightRegistry;
+  emit: (
+    taskName: string,
+    level: "info" | "warn",
+    meta: Record<string, unknown>,
+  ) => void;
+  /** Extra figures for the line (the heap's). */
+  figures: () => Record<string, unknown>;
+}
+
+/** The guard's decisions, one `check()` per tick; separate for the tests. */
+export class ContainerMemoryGuard {
+  private armed = true;
+  private lastReportAt: number | null = null;
+  private pressureSince: number | null = null;
+
+  constructor(private readonly deps: ContainerMemoryDeps) {}
+
+  /** The latest reading, for the heap line. */
+  last: ContainerMemoryReading | null = null;
+
+  check(): void {
+    const { now, read } = this.deps;
+    const at = now();
+    const reading = read();
+    this.last = reading;
+    const slot = globalThis as FlagSlot;
+    if (!reading) {
+      slot[PRESSURE_FLAG] = false;
+      return;
+    }
+    const ratio = reading.workingSet / reading.limit;
+
+    if (ratio >= CONTAINER_PRESSURE_RATIO) {
+      this.pressureSince ??= at;
+      slot[PRESSURE_FLAG] = at - this.pressureSince < CONTAINER_BACKOFF_MAX_MS;
+    } else {
+      this.pressureSince = null;
+      slot[PRESSURE_FLAG] = false;
+    }
+
+    if (
+      !this.armed &&
+      ratio < CONTAINER_REARM_RATIO &&
+      this.lastReportAt !== null &&
+      at - this.lastReportAt >= CONTAINER_PRESSURE_COOLDOWN_MS
+    ) {
+      this.armed = true;
+    }
+    if (this.armed && ratio >= CONTAINER_PRESSURE_RATIO) {
+      this.armed = false;
+      this.lastReportAt = at;
+      const inFlight = this.deps.registry.report(at, 50);
+      this.deps.emit(CONTAINER_MEMORY_TASK, "warn", {
+        ...this.deps.figures(),
+        container_working_set_mb: megabytes(reading.workingSet),
+        container_limit_mb: megabytes(reading.limit),
+        container_used_ratio: Math.round(ratio * 1000) / 1000,
+        pressure_ratio: CONTAINER_PRESSURE_RATIO,
+        heavy_jobs_held_back_max_minutes: CONTAINER_BACKOFF_MAX_MS / 60_000,
+        in_flight_oldest: inFlight.oldest,
+        in_flight_by_route: inFlight.by_route,
+      });
+    }
+  }
+}
+
+/** The working set and limit as heap-line fields, or nothing outside a cgroup. */
+export function containerMemoryFields(
+  reading: ContainerMemoryReading | null,
+): Record<string, number> {
+  if (!reading) return {};
+  return {
+    container_working_set_mb: megabytes(reading.workingSet),
+    container_limit_mb: megabytes(reading.limit),
+  };
+}

src/lib/observability/heap-monitor.ts

@@ -11,6 +11,11 @@ import { WideEventBuilder } from "@/lib/logging/event-builder";
 import { emitIfSampled } from "@/lib/logging/transports";
 import { megabytes } from "@/lib/support/heap-limit";
 
+import {
+  ContainerMemoryGuard,
+  containerMemoryFields,
+  readContainerMemory,
+} from "./container-memory";
 import { inFlightRegistry, type InFlightRegistry } from "./in-flight";
 
 /**
@@ -351,10 +356,22 @@ export function startHeapMonitor(overrides: Partial<HeapMonitorDeps> = {}) {
   slots[STARTED] = true;
 
   const registry = overrides.registry ?? inFlightRegistry();
+  // The container guard rides the same tick, and its latest reading rides
+  // every heap line, so the trend shows the cgroup beside the heap.
+  const guard = new ContainerMemoryGuard({
+    now: Date.now,
+    read: () => readContainerMemory(),
+    registry,
+    emit: emitHeapLine,
+    figures: () => ({ ...defaultReadFigures(registry) }),
+  });
   const monitor = new HeapMonitor({
     now: Date.now,
     readHeap: defaultReadHeap,
-    readFigures: () => defaultReadFigures(registry),
+    readFigures: () => ({
+      ...defaultReadFigures(registry),
+      ...containerMemoryFields(guard.last),
+    }),
     registry,
     emit: emitHeapLine,
     snapshotEnabled: envFlag("HEALTHLOG_HEAP_SNAPSHOT"),
@@ -363,6 +380,11 @@ export function startHeapMonitor(overrides: Partial<HeapMonitorDeps> = {}) {
   });
 
   const timer = setInterval(() => {
+    try {
+      guard.check();
+    } catch {
+      // A diagnostic must never take the process down.
+    }
     try {
       monitor.check();
     } catch {