diff --git a/src/instrumentation.ts b/src/instrumentation.ts index ea0e46748..42e215678 100644 --- a/src/instrumentation.ts +++ b/src/instrumentation.ts @@ -3,6 +3,18 @@ import type { Instrumentation } from "next"; export async function register() { // Only start the worker on the Node.js server runtime (not Edge, not build) if (process.env.NEXT_RUNTIME === "nodejs") { + // The event-loop monitor comes first, so the loading of the database + // client and of the worker graph is inside its view, and every stall it + // reports names the boot step that was running (`boot-steps.ts`). + const [{ startEventLoopLagMonitor }, { beginBootStep }] = await Promise.all( + [ + import("@/lib/observability/event-loop-lag"), + import("@/lib/boot/boot-steps"), + ], + ); + startEventLoopLagMonitor(); + beginBootStep("readiness_summary"); + // First-boot readiness summary: a human-readable block on the core secrets // (with the exact fix on a bad one) before any fail-closed loader throws, // plus a one-glance status of the optional channels. Runs for every process @@ -18,6 +30,7 @@ export async function register() { // Does the configured encryption key open this database? A process whose // key does not refuses every API request with one clear 503 and starts // neither the producer nor the worker (a wrong key must never write). + beginBootStep("key_check"); try { const [{ runBootKeyCheck }, { prisma }] = await Promise.all([ import("@/lib/boot/key-check"), @@ -29,6 +42,7 @@ export async function register() { } // Every connection is pinned to UTC; say clearly if one is not. + beginBootStep("time_zone_check"); try { const [{ runBootTimeZoneCheck }, { prisma }] = await Promise.all([ import("@/lib/boot/time-zone-check"), @@ -41,6 +55,7 @@ export async function register() { // Every pulse reader follows the operator's pulse day-mean method; mirror // it in this process and keep the mirror current. + beginBootStep("pulse_method_sync"); try { const { startPulseDayMeanMethodSync } = await import("@/lib/measurements/pulse-day-mean-method"); @@ -49,15 +64,9 @@ export async function register() { // Without the mirror the process keeps the default method. } - // Every Node process watches its own event loop — web-only - // containers included; a stall report from the process serving - // requests is the whole point. - const { startEventLoopLagMonitor } = - await import("@/lib/observability/event-loop-lag"); - startEventLoopLagMonitor(); - - // The same for its heap: a trend line, and one report naming the work in + // Every Node process watches its heap as it watches its event loop: a trend line, and one report naming the work in // flight before the heap runs out (a heap-limit crash logs nothing else). + beginBootStep("monitors"); try { const [{ startHttpInFlightTracking }, { startHeapMonitor }] = await Promise.all([ @@ -116,6 +125,7 @@ export async function register() { // process owns no schedules, supervision, or consumers. void (async () => { try { + beginBootStep("producer_start"); const { startGlobalBossProducer } = await import("@/lib/jobs/boss-instance"); const boss = await startGlobalBossProducer(process.env.DATABASE_URL); @@ -125,6 +135,8 @@ export async function register() { registerShutdownHandlers(boss); } ready(boss !== null); + const { reportBootSteps } = await import("@/lib/boot/boot-steps"); + await reportBootSteps(); } catch (err) { ready(false); // Queue-backed garnishes degrade independently; never make the web @@ -147,6 +159,7 @@ export async function register() { startupEvt.setBackground({ task_name: "startup" }); try { + beginBootStep("worker_import"); const { startReminderWorker } = await import("@/lib/jobs/reminder-worker"); await startReminderWorker(); diff --git a/src/lib/boot/__tests__/boot-steps.test.ts b/src/lib/boot/__tests__/boot-steps.test.ts new file mode 100644 index 000000000..1c0fdf8f3 --- /dev/null +++ b/src/lib/boot/__tests__/boot-steps.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest"; + +import { + beginBootStep, + bootSteps, + currentBootStep, + endBootSteps, +} from "../boot-steps"; + +describe("boot steps", () => { + it("times each step until the next begins, and names the running one", () => { + beginBootStep("a", 100); + expect(currentBootStep()).toBe("a"); + beginBootStep("b", 130); + expect(currentBootStep()).toBe("b"); + const steps = endBootSteps(200); + expect(currentBootStep()).toBeNull(); + expect(steps.slice(-2)).toEqual([ + { name: "a", startedAtMs: 100, durationMs: 30 }, + { name: "b", startedAtMs: 130, durationMs: 70 }, + ]); + expect(bootSteps().slice(-2)).toEqual(steps.slice(-2)); + }); +}); diff --git a/src/lib/boot/boot-steps.ts b/src/lib/boot/boot-steps.ts new file mode 100644 index 000000000..387dda2cb --- /dev/null +++ b/src/lib/boot/boot-steps.ts @@ -0,0 +1,104 @@ +/** + * Named boot steps with their timings, and the step running right now. + * + * A stall at boot used to be one number: "the loop was blocked 718 ms". The + * event-loop monitor now names the step that was running when the late tick + * fired (`currentBootStep`), and the boot writes one `boot.steps` line with + * how long each step took, so a slow boot says where its time went. + * + * Kept on `globalThis`: the steps are marked from the instrumentation bundle + * and from the worker modules it loads, which are separate module copies in + * a production build. + */ +import { withBackgroundEvent } from "@/lib/logging/background"; +import { annotate } from "@/lib/logging/context"; + +export interface BootStep { + name: string; + /** Milliseconds since the process started (`performance.timeOrigin`). */ + startedAtMs: number; + durationMs: number | null; +} + +const KEY = Symbol.for("healthlog.boot.steps"); + +interface State { + steps: BootStep[]; + current: BootStep | null; +} + +function state(): State { + const slot = globalThis as Record; + let current = slot[KEY] as State | undefined; + if (!current) { + current = { steps: [], current: null }; + slot[KEY] = current; + } + return current; +} + +/** + * Begin `name`, ending the step before it. Steps run one after another; a + * step that starts work in the background ends when the next one begins. + */ +export function beginBootStep( + name: string, + now: number = performance.now(), +): void { + const s = state(); + if (s.current && s.current.durationMs === null) { + s.current.durationMs = Math.round(now - s.current.startedAtMs); + } + const step: BootStep = { + name, + startedAtMs: Math.round(now), + durationMs: null, + }; + s.steps.push(step); + s.current = step; +} + +/** End the running step without starting another. */ +export function endBootSteps(now: number = performance.now()): BootStep[] { + const s = state(); + if (s.current && s.current.durationMs === null) { + s.current.durationMs = Math.round(now - s.current.startedAtMs); + } + s.current = null; + return [...s.steps]; +} + +/** The step running now, or null once the boot is through. */ +export function currentBootStep(): string | null { + return state().current?.name ?? null; +} + +/** Every step so far, for the boot line. */ +export function bootSteps(): BootStep[] { + return [...state().steps]; +} + +/** + * End the boot and write one `boot.steps` line with every step and its + * duration, so a slow start says where its time went. Never throws. + */ +export async function reportBootSteps( + meta: Record = {}, +): Promise { + const steps = endBootSteps(); + try { + await withBackgroundEvent("boot.steps", async () => { + annotate({ + meta: { + ...meta, + boot_steps: steps + .map((s) => `${s.name}=${s.durationMs ?? 0}`) + .join(","), + boot_total_ms: Math.round(performance.now()), + }, + }); + }); + } catch { + // A diagnostic must never break boot. + } +} diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 67e4413b5..e486a92db 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -17,27 +17,35 @@ import { markWorkerStarted, recordError } from "@/lib/jobs/worker-status"; import { createWorkerBoss, setGlobalBoss } from "@/lib/jobs/boss-instance"; import { reportJobsCutOffAtBoot } from "@/lib/jobs/job-observer"; +import { beginBootStep, reportBootSteps } from "@/lib/boot/boot-steps"; import { assertSubsystemEnabled } from "@/lib/process-type"; import { DATABASE_URL, workerLog } from "./reminder/shared"; -import { - registerShutdownHandlers, - reconcileImportJobsAtBoot, - probeIntegrationStatusAtBoot, -} from "./reminder/worker-lifecycle"; -import { - registerIntegrationSyncQueues, - enqueueIntegrationSyncBootDiscovery, -} from "./reminder/register-integration-sync"; -import { registerStatusQueues } from "./reminder/register-status"; -import { - registerRollupQueues, - enqueueRollupBootDiscovery, -} from "./reminder/register-rollup"; -import { registerReminderQueues } from "./reminder/register-reminders"; -import { - registerMaintenanceQueues, - enqueueMaintenanceBootDiscovery, -} from "./reminder/register-maintenance"; +import type { PgBoss } from "pg-boss"; + +/** + * Give the event loop a turn. Each registrar's module graph is evaluated in + * one synchronous stretch when it is first imported; loading the five one + * after another, with a turn in between, keeps the longest stall to the + * largest graph instead of the sum of all five, so requests and the health + * probe are answered in between. + */ +const yieldToLoop = () => new Promise((resolve) => setImmediate(resolve)); + +/** + * Import one registrar module as its own boot step. The registrars and the + * handler graphs behind them are loaded here, after `boss.start()`, rather + * than when this module is imported. + */ +async function loadRegistrar( + name: string, + load: () => Promise, +): Promise { + beginBootStep(`load_${name}`); + await yieldToLoop(); + const mod = await load(); + await yieldToLoop(); + return mod; +} export async function startReminderWorker() { // v1.4: refuse to boot if the operator marked this container as @@ -58,6 +66,7 @@ export async function startReminderWorker() { // Read before any `boss.work` binding below can take a job: whatever is // still active now was started by a process that is gone. + beginBootStep("boss_start"); const bootedAt = new Date(); await boss.start(); await reportJobsCutOffAtBoot(bootedAt); @@ -65,20 +74,62 @@ export async function startReminderWorker() { markWorkerStarted(); // Graceful shutdown: drain in-flight jobs on SIGTERM/SIGINT. - registerShutdownHandlers(boss); + const lifecycle = await loadRegistrar( + "worker_lifecycle", + () => import("./reminder/worker-lifecycle"), + ); + lifecycle.registerShutdownHandlers(boss); // Register every domain's queues (createQueue + schedule + boss.work). Each // registrar returns the queue names it provisioned; the aggregate is the // boot-level mirror of the per-registrar `allQueues` arrays the dead-queue // guards pin. Order preserves the monolith's createQueue / schedule / work // sequence (integration → status → rollup → reminders → maintenance). - const registeredQueues = [ - ...(await registerIntegrationSyncQueues(boss)), - ...(await registerStatusQueues(boss)), - ...(await registerRollupQueues(boss)), - ...(await registerReminderQueues(boss)), - ...(await registerMaintenanceQueues(boss)), - ]; + const integration = await loadRegistrar( + "integration_sync", + () => import("./reminder/register-integration-sync"), + ); + const registeredQueues = await registerInStep( + "integration_sync", + boss, + integration.registerIntegrationSyncQueues, + ); + const status = await loadRegistrar( + "status", + () => import("./reminder/register-status"), + ); + registeredQueues.push( + ...(await registerInStep("status", boss, status.registerStatusQueues)), + ); + const rollup = await loadRegistrar( + "rollup", + () => import("./reminder/register-rollup"), + ); + registeredQueues.push( + ...(await registerInStep("rollup", boss, rollup.registerRollupQueues)), + ); + const reminders = await loadRegistrar( + "reminders", + () => import("./reminder/register-reminders"), + ); + registeredQueues.push( + ...(await registerInStep( + "reminders", + boss, + reminders.registerReminderQueues, + )), + ); + const maintenance = await loadRegistrar( + "maintenance", + () => import("./reminder/register-maintenance"), + ); + registeredQueues.push( + ...(await registerInStep( + "maintenance", + boss, + maintenance.registerMaintenanceQueues, + )), + ); workerLog( "info", `[reminder-worker] registered ${registeredQueues.length} queues across 5 domain registrars`, @@ -86,22 +137,35 @@ export async function startReminderWorker() { // v1.4.34 — reconcile any `ImportJob` rows that were mid-parse when the // worker last shut down, flipping orphaned rows to `failed`. - await reconcileImportJobsAtBoot(); + beginBootStep("boot_reconcile"); + await lifecycle.reconcileImportJobsAtBoot(); // v1.4.48 — boot probe for legacy `integration_statuses` rows that still // carry `consecutive_failures_by_kind = NULL`. - await probeIntegrationStatusAtBoot(); + await lifecycle.probeIntegrationStatusAtBoot(); // Fire-and-forget boot discovery for every self-converging backfill / // consolidation pass. Each is idempotent across reboots and never fails // worker boot on a miss (errors come back through the helper's result). - await enqueueRollupBootDiscovery(); - await enqueueIntegrationSyncBootDiscovery(); - await enqueueMaintenanceBootDiscovery(); + beginBootStep("boot_discovery"); + await rollup.enqueueRollupBootDiscovery(); + await integration.enqueueIntegrationSyncBootDiscovery(); + await maintenance.enqueueMaintenanceBootDiscovery(); + await reportBootSteps({ boot_queue_count: registeredQueues.length }); return boss; } +/** Run one registrar as its own boot step. */ +async function registerInStep( + name: string, + boss: PgBoss, + register: (boss: PgBoss) => Promise, +): Promise { + beginBootStep(`register_${name}`); + return [...(await register(boss))]; +} + // Run standalone if ( process.argv[1]?.endsWith("reminder-worker.ts") || diff --git a/src/lib/observability/event-loop-lag.ts b/src/lib/observability/event-loop-lag.ts index d5106564b..62bcdba32 100644 --- a/src/lib/observability/event-loop-lag.ts +++ b/src/lib/observability/event-loop-lag.ts @@ -1,6 +1,10 @@ +import { currentBootStep } from "@/lib/boot/boot-steps"; +import { startupPhase } from "@/lib/boot/startup-state"; import { WideEventBuilder } from "@/lib/logging/event-builder"; import { emitIfSampled } from "@/lib/logging/transports"; +import { inFlightRegistry } from "./in-flight"; + /** * Event-loop lag monitor — the instance-level "was the PROCESS stalled" * signal that per-request durations cannot carry. @@ -24,6 +28,13 @@ import { emitIfSampled } from "@/lib/logging/transports"; * same timers phase, and the stall sample can vanish into the window * that was just reset (reproduced under vitest). * + * Attribution: a stall line says what the process was doing. It carries the + * process uptime, the background role's startup phase, the boot step that + * was running (`boot-steps.ts`), and the oldest requests and jobs in flight + * with the routes most of them share, the same report the heap pressure line + * writes. The monitor is the first thing `register()` starts, so the loading + * of the database client and of the worker graph is inside its view. + * * The snapshot is published on a `globalThis` slot rather than imported * by the event builder: the builder is bundled into the Edge-runtime * proxy, and reading a global keeps the dependency one-directional @@ -56,19 +67,52 @@ type GlobalSlots = { [STARTED]?: boolean; }; +/** What a stall line says. */ +export interface EventLoopStall { + lag_ms: number; + tick_interval_ms: number; + uptime_s: number; + startup_phase: string | null; + boot_step: string | null; + in_flight_oldest: unknown[]; + in_flight_by_route: unknown[]; +} + export interface EventLoopLagMonitorOptions { tickIntervalMs?: number; stallThresholdMs?: number; /** Test seam — replaces the wide-event emit. */ - emitStall?: (stall: { lag_ms: number; tick_interval_ms: number }) => void; + emitStall?: (stall: EventLoopStall) => void; } -function defaultEmitStall(stall: { - lag_ms: number; - tick_interval_ms: number; -}): void { +/** The attribution a stall carries; never throws. */ +export function stallAttribution( + now: number = Date.now(), +): Omit { + let inFlight: { oldest: unknown[]; by_route: unknown[] } = { + oldest: [], + by_route: [], + }; + try { + inFlight = inFlightRegistry().report(now, 5); + } catch { + // The report is a garnish. + } + return { + uptime_s: Math.round(process.uptime() * 10) / 10, + startup_phase: startupPhase(now), + boot_step: currentBootStep(), + in_flight_oldest: inFlight.oldest, + in_flight_by_route: inFlight.by_route, + }; +} + +function defaultEmitStall(stall: EventLoopStall): void { const evt = new WideEventBuilder("background"); - evt.setBackground({ task_name: "event_loop_stall", result: stall }); + evt.setBackground({ + task_name: "event_loop_stall", + result: stall as unknown as Record, + }); evt.elevateLevel("warn"); evt.finish(); emitIfSampled(evt.toJSON()); @@ -110,6 +154,7 @@ export function startEventLoopLagMonitor( emitStall({ lag_ms: Math.round(lagMs), tick_interval_ms: tickIntervalMs, + ...stallAttribution(), }); }