diff --git a/src/lib/jobs/job-observer.ts b/src/lib/jobs/job-observer.ts index 18c4166d6..7b463bdec 100644 --- a/src/lib/jobs/job-observer.ts +++ b/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( }; 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 diff --git a/src/lib/observability/__tests__/in-flight.test.ts b/src/lib/observability/__tests__/in-flight.test.ts new file mode 100644 index 000000000..8b55b9d84 --- /dev/null +++ b/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 = {}) { + const response = new EventEmitter(); + return { + message: { + request: { url, method: "POST", headers }, + response, + } as unknown as Parameters[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((resolve) => server.listen(0, resolve)); + const { port } = server.address() as AddressInfo; + await new Promise((resolve, reject) => { + httpRequest({ port, path: "/probe?x=1" }, (res) => { + res.resume(); + res.on("end", resolve); + }) + .on("error", reject) + .end(); + }); + await new Promise((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((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((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((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); + }); +}); diff --git a/src/lib/observability/in-flight.ts b/src/lib/observability/in-flight.ts new file mode 100644 index 000000000..de48c9609 --- /dev/null +++ b/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(); + 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 { + const out: Record = { 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) { + 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( + entry: Omit, + fn: () => Promise, + registry: InFlightRegistry = inFlightRegistry(), +): Promise { + 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. + } + }); +}