From c6bd27f06feeeb955e76f7b887eeffb050157da2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 25 Oct 2026 00:35:36 +0200 Subject: [PATCH] fix(jobs): load every job handler once at worker boot and report failures Since handlers load on their queue's first job, a module the image cannot load no longer fails the worker's boot. The worker comes up healthy, the queue burns its retries when it first has work, and a one-shot job (a restore, an import tail) ends failed with nobody told at deploy time. After binding every queue the worker now runs `preloadHandlers` in the background, without awaiting it, so readiness is unchanged. It loads each handler once through the same memoised loader its queue is bound to, so a module that loads stays loaded for the first job. Each failure is reported on the `job.run.failed` line at error, with the queue's name and `reason: "handler_load_failed"`, plus the worker error report (stderr and GlitchTip). A summary line on stderr names every queue that failed. The sweep never throws, and a failing report does not stop it. docs/ops/deploy.md says what to look for after a deploy. --- docs/ops/deploy.md | 15 ++ .../__tests__/preload-handlers.test.ts | 168 ++++++++++++++++++ src/lib/jobs/registry/index.ts | 14 +- src/lib/jobs/registry/register.ts | 96 ++++++++-- src/lib/jobs/reminder-worker.ts | 13 +- src/lib/jobs/run-job.ts | 14 ++ 6 files changed, 305 insertions(+), 15 deletions(-) create mode 100644 src/lib/jobs/registry/__tests__/preload-handlers.test.ts diff --git a/docs/ops/deploy.md b/docs/ops/deploy.md index 046be2905..ff425a19a 100644 --- a/docs/ops/deploy.md +++ b/docs/ops/deploy.md @@ -237,6 +237,21 @@ Pass `--only=prod` / `--only=demo` to check a single target, or `--attempts=N --interval=ms` to widen the retry window while a slow pull settles. +The worker loads each job's handler module on that job's first run, so a +module the image cannot load does not stop the worker from starting. Once +it has bound its queues, the worker loads every handler a single time in +the background and reports each one that fails, normally within a minute +of start. Each failure is a `job.run.failed` event at `error` with +`reason: "handler_load_failed"` and the queue's name, and stderr also gets +`[pg-boss] [registry] preloaded N/M job handlers; failed to load: `. +If either shows up after a deploy, the image is broken for those queues, +and their jobs will fail when they run. Roll back (see Rollback above) +instead of waiting for the retries to run out: + +```sh +docker compose logs --since 5m app 2>&1 | grep -E 'handler_load_failed|failed to load:' +``` + ## Walk the synthetic journey The version check proves the instance serves the right build. It does not diff --git a/src/lib/jobs/registry/__tests__/preload-handlers.test.ts b/src/lib/jobs/registry/__tests__/preload-handlers.test.ts new file mode 100644 index 000000000..6f73883e9 --- /dev/null +++ b/src/lib/jobs/registry/__tests__/preload-handlers.test.ts @@ -0,0 +1,168 @@ +/** + * The worker's boot preload of every job handler module. + * + * Handlers load on their queue's first job, so a module that cannot load + * stopped failing the boot and started failing jobs instead: the queue burned + * its retries, and a one-shot job (a restore, an import tail) ended failed + * with nobody told at deploy time. The worker now loads every handler once in + * the background after binding and reports each failure by queue name. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job, PgBoss } from "pg-boss"; + +const bound = vi.hoisted( + () => new Map Promise>(), +); + +vi.mock("@/lib/jobs/reminder/registrar-shared", () => ({ + createAndSchedule: vi.fn(async () => undefined), + retireQueues: vi.fn(async () => undefined), + createAndWork: vi.fn( + async ( + _boss: unknown, + queue: string, + _opts: unknown, + handler: (jobs: unknown[]) => Promise, + ) => { + bound.set(queue, handler); + }, + ), +})); + +vi.mock("@/lib/jobs/reminder/shared", () => ({ workerLog: vi.fn() })); + +vi.mock("@/lib/logging/signal", () => ({ emitSignal: vi.fn() })); +vi.mock("@/lib/jobs/report-worker-error", () => ({ + reportWorkerError: vi.fn(async () => undefined), +})); +vi.mock("@/lib/jobs/job-failures", () => ({ + readConsecutiveRunFailures: vi.fn(async () => 0), +})); + +import { workerLog } from "@/lib/jobs/reminder/shared"; +import { emitSignal } from "@/lib/logging/signal"; +import { reportWorkerError } from "@/lib/jobs/report-worker-error"; +import { reportHandlerLoadFailure } from "@/lib/jobs/run-job"; + +import type { JobDefinition, JobGroup } from "../define-job"; +import { + preloadHandlers, + registerJobGroup, + type BoundLoader, +} from "../register"; + +const boss = {} as PgBoss; + +function definition( + queue: string, + load: () => Promise, +): JobDefinition { + return { queue, work: { batchSize: 1 }, load } as unknown as JobDefinition; +} + +beforeEach(() => { + bound.clear(); + vi.clearAllMocks(); +}); + +describe("preloading the job handlers at worker boot", () => { + it("reports a module that fails to load, by queue, and loads the rest", async () => { + const ok = vi.fn(async () => async () => ({ ok: true as const })); + const broken = vi.fn(async () => { + throw new Error("Cannot find module './gone'"); + }); + const group = { + name: "maintenance", + jobs: [ + definition("first-ok", ok), + definition("broken-module", broken), + definition("second-ok", ok), + ], + } as unknown as JobGroup; + + const loaders: BoundLoader[] = []; + await registerJobGroup(boss, group, (b) => loaders.push(b)); + expect(loaders.map((b) => b.queue)).toEqual([ + "first-ok", + "broken-module", + "second-ok", + ]); + + const report = vi.fn(async () => undefined); + const failed = await preloadHandlers(loaders, report); + + expect(failed).toEqual(["broken-module"]); + expect(report).toHaveBeenCalledTimes(1); + expect(report).toHaveBeenCalledWith( + "broken-module", + expect.objectContaining({ message: "Cannot find module './gone'" }), + ); + expect(ok).toHaveBeenCalledTimes(2); + expect(workerLog).toHaveBeenCalledWith( + "error", + expect.stringContaining("failed to load: broken-module"), + ); + }); + + it("preloads through the loader the queue is bound to, so the first job does not import again", async () => { + const handler = vi.fn(async () => ({ ok: true as const })); + const load = vi.fn(async () => handler); + const group = { + name: "maintenance", + jobs: [definition("warm", load)], + } as unknown as JobGroup; + + const loaders: BoundLoader[] = []; + await registerJobGroup(boss, group, (b) => loaders.push(b)); + await preloadHandlers(loaders, vi.fn()); + await bound.get("warm")!([{ id: "j1", data: {} } as Job]); + + expect(load).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledTimes(1); + }); + + it("does not stop the sweep when a report itself fails", async () => { + const loaders: BoundLoader[] = [ + { queue: "a", load: () => Promise.reject(new Error("a")) }, + { queue: "b", load: () => Promise.reject(new Error("b")) }, + ]; + const report = vi.fn(async () => { + throw new Error("sink down"); + }); + await expect(preloadHandlers(loaders, report)).resolves.toEqual(["a", "b"]); + expect(report).toHaveBeenCalledTimes(2); + }); + + it("reports a load failure on the job.run.failed line, at error, naming the queue", async () => { + const error = new Error("SyntaxError in handler"); + await reportHandlerLoadFailure("broken-module", error); + + expect(emitSignal).toHaveBeenCalledWith({ + action: "job.run.failed", + level: "error", + meta: { + queue: "broken-module", + reason: "handler_load_failed", + consecutiveFailures: 0, + }, + error, + }); + expect(reportWorkerError).toHaveBeenCalledWith("broken-module", error, { + phase: "handler_preload", + }); + }); + + it("is started by the worker after binding, without holding up the boot", () => { + const source = readFileSync( + join(process.cwd(), "src/lib/jobs/reminder-worker.ts"), + "utf8", + ); + const register = source.indexOf("registry.registerAllJobGroups("); + const preload = source.indexOf("void registry.preloadHandlers(bound)"); + expect(register).toBeGreaterThan(-1); + expect(preload).toBeGreaterThan(register); + }); +}); diff --git a/src/lib/jobs/registry/index.ts b/src/lib/jobs/registry/index.ts index a6d02a764..6ab42fe0c 100644 --- a/src/lib/jobs/registry/index.ts +++ b/src/lib/jobs/registry/index.ts @@ -10,7 +10,13 @@ import type { JobDefinition, JobGroup } from "./define-job"; import { INTEGRATION_SYNC_GROUP } from "./integration-sync"; import { MAINTENANCE_GROUP } from "./maintenance"; import { REMINDERS_GROUP } from "./reminders"; -import { registerJobGroup, runBootDiscovery } from "./register"; +import { + registerJobGroup, + runBootDiscovery, + type BoundLoader, +} from "./register"; + +export { preloadHandlers, type BoundLoader } from "./register"; import { ROLLUP_GROUP } from "./rollup"; import { STATUS_GROUP } from "./status"; @@ -29,16 +35,18 @@ export function allJobDefinitions(): readonly JobDefinition[] { /** * Register every group in order. `onGroup` runs before each one, so the boot - * can time the groups as steps. Returns every queue name created. + * can time the groups as steps; `onBound` receives every queue's handler + * loader as it is bound. Returns every queue name created. */ export async function registerAllJobGroups( boss: PgBoss, onGroup: (group: JobGroup) => void = () => {}, + onBound: (bound: BoundLoader) => void = () => {}, ): Promise { const queues: string[] = []; for (const group of JOB_GROUPS) { onGroup(group); - queues.push(...(await registerJobGroup(boss, group))); + queues.push(...(await registerJobGroup(boss, group, onBound))); } return queues; } diff --git a/src/lib/jobs/registry/register.ts b/src/lib/jobs/registry/register.ts index 3a2a36bef..c11052fdb 100644 --- a/src/lib/jobs/registry/register.ts +++ b/src/lib/jobs/registry/register.ts @@ -13,6 +13,14 @@ * queue's protections do not depend on whether its module has loaded yet. A * module that fails to load fails the job that asked for it, with the queue's * name on the error, and the next job tries the import again. + * + * Loading lazily moved a broken module from a failed boot to a failed job: the + * worker comes up healthy, and the queue burns its retries the first time it + * has work, which for a one-shot job (a restore, an import tail) means the + * job ends failed with nobody told at deploy time. `preloadHandlers` closes + * that gap. The worker runs it once after binding, in the background, through + * the same memoised loader the queue uses, and reports every module that does + * not load before any job needs it. */ import type { Job, PgBoss } from "pg-boss"; @@ -26,6 +34,7 @@ import { type ScheduleEntry, } from "@/lib/jobs/reminder/registrar-shared"; import { workerLog } from "@/lib/jobs/reminder/shared"; +import { reportHandlerLoadFailure } from "@/lib/jobs/run-job"; import type { BatchHandler, @@ -35,6 +44,34 @@ import type { JobGroup, } from "./define-job"; +/** A queue's handler loader, memoised: one import, retried after a failure. */ +export type HandlerLoader = () => Promise>>; + +/** One bound queue and the loader its handler goes through. */ +export interface BoundLoader { + readonly queue: string; + readonly load: HandlerLoader; +} + +/** + * The loader a definition's binding shares: the first call imports, every + * later call answers from the same promise, and a failed import is forgotten + * so the next call tries again. + */ +export function handlerLoader( + definition: JobDefinition, + ctx: JobContext, +): HandlerLoader { + let loading: Promise>> | null = null; + return () => { + loading ??= definition.load(ctx).catch((err: unknown) => { + loading = null; + throw err; + }); + return loading; + }; +} + /** * The handler a definition binds: its loaded module behind the lock and the * lane it declares. @@ -42,15 +79,8 @@ import type { export function bindableHandler( definition: JobDefinition, ctx: JobContext, + load: HandlerLoader = handlerLoader(definition, ctx), ): BatchHandler> { - let loading: Promise>> | null = null; - const load = (): Promise>> => { - loading ??= definition.load(ctx).catch((err: unknown) => { - loading = null; - throw err; - }); - return loading; - }; let handler: BatchHandler> = async (jobs) => (await load())(jobs); if (definition.lane) { @@ -88,11 +118,13 @@ export function policiesOf(jobs: readonly JobDefinition[]): QueuePolicyTable { /** * Register one group: create, schedule, retire, bind. Returns the queue names - * it created. + * it created. `onBound` is handed each queue's loader as it is bound, so the + * boot can preload through the very loader the queue will use. */ export async function registerJobGroup( boss: PgBoss, group: JobGroup, + onBound: (bound: BoundLoader) => void = () => {}, ): Promise { const queues = group.jobs.map((job) => job.queue); await createAndSchedule( @@ -104,7 +136,14 @@ export async function registerJobGroup( if (group.retired) await retireQueues(boss, group.retired); const ctx: JobContext = { boss }; for (const job of group.jobs) { - await createAndWork(boss, job.queue, job.work, bindableHandler(job, ctx)); + const load = handlerLoader(job, ctx); + await createAndWork( + boss, + job.queue, + job.work, + bindableHandler(job, ctx, load), + ); + onBound({ queue: job.queue, load }); } if (group.afterBind) await group.afterBind(ctx); return queues; @@ -144,3 +183,40 @@ export async function runBootDiscovery( } } } + +/** + * Load every bound handler once, in declaration order, and report each one + * that fails: `job.run.failed` at error with the queue's name and the reason + * `handler_load_failed`, the line the repeated-failure alert already keys on, + * plus the worker error report (stderr and GlitchTip). Never throws, and the + * caller does not wait for it, so a broken module costs the boot nothing but + * the report. A module that loads stays loaded, and the queue's first job + * finds it in the loader's memo. Returns the queues that failed. + */ +export async function preloadHandlers( + bound: readonly BoundLoader[], + report: ( + queue: string, + error: unknown, + ) => Promise = reportHandlerLoadFailure, +): Promise { + const failed: string[] = []; + for (const { queue, load } of bound) { + try { + await load(); + } catch (error) { + failed.push(queue); + try { + await report(queue, error); + } catch { + // The report is the last line of defence; it must not stop the sweep. + } + } + } + workerLog( + failed.length === 0 ? "info" : "error", + `[registry] preloaded ${bound.length - failed.length}/${bound.length} job handlers` + + (failed.length === 0 ? "" : `; failed to load: ${failed.join(", ")}`), + ); + return failed; +} diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index c882ca002..715ce80b2 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -17,6 +17,7 @@ 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 type { BoundLoader } from "./registry/index"; /** * Give the event loop a turn around a module load, so requests and the @@ -76,8 +77,11 @@ export async function startReminderWorker() { "registry", () => import("./registry/index"), ); - const registeredQueues = await registry.registerAllJobGroups(boss, (group) => - beginBootStep(`register_${group.name}`), + const bound: BoundLoader[] = []; + const registeredQueues = await registry.registerAllJobGroups( + boss, + (group) => beginBootStep(`register_${group.name}`), + (loader) => bound.push(loader), ); workerLog( "info", @@ -100,6 +104,11 @@ export async function startReminderWorker() { await registry.runAllBootDiscovery(); await reportBootSteps({ boot_queue_count: registeredQueues.length }); + + // Every handler module is loaded once now, in the background, so one that + // cannot load is reported at deploy time rather than on its queue's first + // job. Not awaited: readiness does not wait on it, and it never throws. + void registry.preloadHandlers(bound); return boss; } diff --git a/src/lib/jobs/run-job.ts b/src/lib/jobs/run-job.ts index f4df20089..6a7b74716 100644 --- a/src/lib/jobs/run-job.ts +++ b/src/lib/jobs/run-job.ts @@ -90,6 +90,20 @@ async function signalRunFailed( }); } +/** + * A queue's handler module that failed to load before any job asked for it + * (the worker's boot preload, `preloadHandlers`). Reported the way a failed + * run is, so the operator hears about it at deploy time on the line the + * repeated-failure alert already reads, instead of on the queue's first job. + */ +export async function reportHandlerLoadFailure( + queue: string, + error: unknown, +): Promise { + await reportWorkerError(queue, error, { phase: "handler_preload" }); + await signalRunFailed(queue, "handler_load_failed", error); +} + /** * Wrap a handler for `boss.work`. Resolves with the bounded, redacted * persistence representation on `ok: true`; reports and rethrows on