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

@@ -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: <queues>`.
+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

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<string, (jobs: unknown[]) => Promise<unknown>>(),
+);
+
+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<unknown>,
+    ) => {
+      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<unknown>,
+): 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<unknown>]);
+
+    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);
+  });
+});

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<string[]> {
   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;
 }

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<BatchHandler<Job<unknown>>>;
+
+/** 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<BatchHandler<Job<unknown>>> | 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<Job<unknown>> {
-  let loading: Promise<BatchHandler<Job<unknown>>> | null = null;
-  const load = (): Promise<BatchHandler<Job<unknown>>> => {
-    loading ??= definition.load(ctx).catch((err: unknown) => {
-      loading = null;
-      throw err;
-    });
-    return loading;
-  };
   let handler: BatchHandler<Job<unknown>> = 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<readonly string[]> {
   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<void> = reportHandlerLoadFailure,
+): Promise<string[]> {
+  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;
+}

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;
 }
 

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<void> {
+  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