From 27e83f1a83916499142de9a076ef03c57f73ef2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 22 Oct 2026 06:41:17 +0200 Subject: [PATCH] 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 | 53 +++++ src/lib/jobs/heavy-lane.ts | 24 ++- .../__tests__/container-memory.test.ts | 116 ++++++++++ src/lib/observability/container-memory.ts | 198 ++++++++++++++++++ src/lib/observability/heap-monitor.ts | 24 ++- 5 files changed, 413 insertions(+), 2 deletions(-) create mode 100644 src/lib/observability/__tests__/container-memory.test.ts create mode 100644 src/lib/observability/container-memory.ts diff --git a/src/lib/jobs/__tests__/heavy-lane.test.ts b/src/lib/jobs/__tests__/heavy-lane.test.ts index 6b1bed63b..8ebea3ea1 100644 --- a/src/lib/jobs/__tests__/heavy-lane.test.ts +++ b/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((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 = []; + }); +}); diff --git a/src/lib/jobs/heavy-lane.ts b/src/lib/jobs/heavy-lane.ts index 1437e99b8..4d3bd2c7d 100644 --- a/src/lib/jobs/heavy-lane.ts +++ b/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(); */ 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>( 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; diff --git a/src/lib/observability/__tests__/container-memory.test.ts b/src/lib/observability/__tests__/container-memory.test.ts new file mode 100644 index 000000000..c21fbde8c --- /dev/null +++ b/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) { + 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) { + let at = 0; + let i = 0; + const lines: Array<{ task: string; meta: Record }> = []; + 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); + }); +}); diff --git a/src/lib/observability/container-memory.ts b/src/lib/observability/container-memory.ts new file mode 100644 index 000000000..9a26fd658 --- /dev/null +++ b/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, + ) => void; + /** Extra figures for the line (the heap's). */ + figures: () => Record; +} + +/** 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 { + if (!reading) return {}; + return { + container_working_set_mb: megabytes(reading.workingSet), + container_limit_mb: megabytes(reading.limit), + }; +} diff --git a/src/lib/observability/heap-monitor.ts b/src/lib/observability/heap-monitor.ts index 5e049763d..9d82b0180 100644 --- a/src/lib/observability/heap-monitor.ts +++ b/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 = {}) { 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 = {}) { }); const timer = setInterval(() => { + try { + guard.check(); + } catch { + // A diagnostic must never take the process down. + } try { monitor.check(); } catch {