fix(jobs): run one heavy job at a time per process

Web and worker share one process and one heap limit, and the jobs that
hold a large working set (the Apple Health and Health Connect imports,
restore, the weekly and off-host backups, the restore drill, the rollup
backfill, the nightly status batch, the pre-generation walk and the
narrative walk) ran side by side whenever their schedules or an upload
met, so the peak was their sum. `heavyLane` queues them in the process,
first come first served. A waiter leaves the line when pg-boss expires it
and its retry takes the place; a one-shot whose input lives outside
pg-boss keeps its place. The runtime guard binds every heavy queue in the
lane, keeps `lockedPass` outside it, and allows the one-shot rule only
where the send options forbid a retry.

src/__tests__/job-queue-runtime-guard.test.ts

@@ -42,6 +42,7 @@ import {
   type SourceSite,
 } from "@/lib/jobs/queue-runtime";
 import { SCHEDULED_AND_SENT_QUEUES } from "@/lib/jobs/reminder/registrar-shared";
+import { HEAVY_QUEUES } from "@/lib/jobs/heavy-lane";
 
 const SRC = join(process.cwd(), "src");
 const REGISTRARS = [
@@ -415,3 +416,46 @@ describe("a cron-fed queue polls slowly, a sent one does not", () => {
     expect(dynamic.has("whoop-recovery-sync")).toBe(true);
   });
 });
+
+/** The text of the first `const NAME = …` in the tree, up to its end. */
+function constBody(texts: ReadonlyMap<string, string>, name: string): string {
+  for (const [file, raw] of texts) {
+    if (new RegExp(`\\bconst\\s+${name}\\b`).test(raw)) {
+      return bodyOf(texts, { file, fn: name }).split(/\n\n/)[0];
+    }
+  }
+  return "";
+}
+
+describe("heavy jobs share the lane", () => {
+  const all = scan(source);
+  const laned = all.bindings.filter((b) => /\bheavyLane\s*\(/.test(b.text));
+
+  it("binds every heavy queue in the lane, and nothing else", () => {
+    expect(laned.map((b) => b.queue).sort()).toEqual(
+      Object.keys(HEAVY_QUEUES).sort(),
+    );
+  });
+
+  it("holds a lockedPass outside the lane, so an expiry retry does not queue twice", () => {
+    const inverted = laned
+      .filter((b) => /\blockedPass\s*\(/.test(b.text))
+      .filter((b) => b.text.indexOf("lockedPass") > b.text.indexOf("heavyLane"))
+      .map((b) => b.queue);
+    expect(inverted).toEqual([]);
+  });
+
+  it("keeps a waiter past its expiry only where nothing retries the job", () => {
+    const problems = Object.entries(HEAVY_QUEUES)
+      .filter(([, heavy]) => heavy.oneShot)
+      .flatMap(([queue]) => {
+        const entry = QUEUE_RUNTIME[queue];
+        if (entry?.runtime !== "long") return [`${queue}: not a long queue`];
+        const noRetry = entry.expiryVia.some((v) =>
+          /\bretryLimit:\s*0\b/.test(constBody(source, v)),
+        );
+        return noRetry ? [] : [`${queue}: its send options allow a retry`];
+      });
+    expect(problems).toEqual([]);
+  });
+});

src/lib/jobs/__tests__/heavy-lane.test.ts

@@ -0,0 +1,120 @@
+/**
+ * The heavy lane runs one heavy job at a time per process, first come first
+ * served; a waiter pg-boss expired leaves the line, a one-shot keeps its place.
+ */
+import { describe, expect, it, vi } from "vitest";
+import type { Job } from "pg-boss";
+
+vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
+
+import { heavyLane } from "../heavy-lane";
+import { jobDone } from "../job-outcome";
+
+type TestJob = Job<{ force?: boolean }>;
+
+function job(data: { force?: boolean } = {}, signal?: AbortSignal): TestJob {
+  return { id: "j", data, signal } as unknown as TestJob;
+}
+
+/** A handler that runs until `finish` is called, recording the overlap. */
+function gate(log: string[], name: string) {
+  let finish!: () => void;
+  const done = new Promise<void>((r) => (finish = r));
+  const handler = vi.fn(async () => {
+    log.push(`start:${name}`);
+    await done;
+    log.push(`end:${name}`);
+    return jobDone({ jobs: 1 });
+  });
+  return { handler, finish };
+}
+
+const tick = () => new Promise((r) => setTimeout(r, 0));
+
+describe("heavyLane", () => {
+  it("runs two heavy jobs of different queues one after the other", async () => {
+    const log: string[] = [];
+    const a = gate(log, "backup");
+    const b = gate(log, "import");
+    const first = heavyLane("data-backup", a.handler)([job()]);
+    const second = heavyLane("apple-health-import-v2", b.handler)([job()]);
+    await tick();
+    expect(log).toEqual(["start:backup"]);
+    a.finish();
+    await first;
+    await tick();
+    expect(log).toEqual(["start:backup", "end:backup", "start:import"]);
+    b.finish();
+    await second;
+  });
+
+  it("lets a light job of a heavy queue pass the line", async () => {
+    const log: string[] = [];
+    const a = gate(log, "walk");
+    const b = gate(log, "warm");
+    const notForced = (j: TestJob) => !j.data.force;
+    const walk = heavyLane(
+      "insight-pregenerate",
+      a.handler,
+      notForced,
+    )([job()]);
+    const warm = heavyLane(
+      "insight-pregenerate",
+      b.handler,
+      notForced,
+    )([job({ force: true })]);
+    await tick();
+    b.finish();
+    a.finish();
+    await Promise.all([walk, warm]);
+    // The warm started while the walk held the lane.
+    expect(log.indexOf("start:warm")).toBeLessThan(log.indexOf("end:walk"));
+  });
+
+  it("drops a waiter pg-boss expired, and keeps a one-shot past its expiry", async () => {
+    const log: string[] = [];
+    const holder = gate(log, "holder");
+    const pass = gate(log, "pass");
+    const upload = gate(log, "upload");
+    const held = heavyLane("data-backup-offhost", holder.handler)([job()]);
+    const expiry = new AbortController();
+    const waitingPass = heavyLane(
+      "data-backup",
+      pass.handler,
+    )([job({}, expiry.signal)]);
+    const uploadExpiry = new AbortController();
+    const waitingUpload = heavyLane(
+      "apple-health-import-v2",
+      upload.handler,
+    )([job({}, uploadExpiry.signal)]);
+    await tick();
+    expiry.abort();
+    uploadExpiry.abort();
+    const outcome = await waitingPass;
+    expect(outcome).toMatchObject({ ok: true, did: { lane_expired: 1 } });
+    expect(pass.handler).not.toHaveBeenCalled();
+    holder.finish();
+    await held;
+    await tick();
+    expect(log).toEqual(["start:holder", "end:holder", "start:upload"]);
+    upload.finish();
+    await waitingUpload;
+  });
+
+  it("frees the lane when a handler throws", async () => {
+    const failing = heavyLane("rollup-full-backfill", async () => {
+      throw new Error("boom");
+    });
+    await expect(failing([job()])).rejects.toThrow("boom");
+    const next = vi.fn(async () => jobDone({ jobs: 1 }));
+    await expect(
+      heavyLane("data-backup", next)([job()]),
+    ).resolves.toMatchObject({ ok: true });
+  });
+
+  it("refuses a queue that is not declared heavy", () => {
+    expect(() =>
+      heavyLane("host-metric-sample", async () => jobDone()),
+    ).toThrow(/not in HEAVY_QUEUES/);
+  });
+});

src/lib/jobs/heavy-lane.ts

@@ -0,0 +1,145 @@
+/**
+ * One heavy job at a time per process.
+ *
+ * The queues in `HEAVY_QUEUES` each hold a large working set in the heap while
+ * they run: an export being parsed, a whole account being restored or
+ * serialised, a provider pass over every account. Web and worker share one
+ * process and one heap limit, and these jobs used to run side by side when
+ * their schedules or a person's upload happened to meet, so the peak was the
+ * sum of whatever overlapped rather than the largest one. Under the lane a
+ * heavy job waits for the one before it, first come first served.
+ *
+ * The lane lives in the process, not the database: the heap it protects is
+ * the process's own, and another process has a heap of its own.
+ *
+ * A job waits until its turn or until pg-boss gives up on it. At the job's
+ * expiry pg-boss aborts its signal and applies the queue's retry policy, so a
+ * waiter that sees the abort leaves the line without running and reports
+ * `lane_expired`; the retry takes its place. A queue bound with `lockedPass`
+ * holds that lock outside the lane, so the retry finds the lock taken while
+ * the original still waits instead of queueing a second time
+ * (`heavy-lane-guard.test.ts`). The exception is a one-shot: its input lives
+ * outside pg-boss (a staged upload, a restore row) and nothing retries it, so
+ * it keeps its place past the expiry and runs when its turn comes, as pg-boss
+ * lets a handler do.
+ */
+import type { Job } from "pg-boss";
+
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { annotate } from "@/lib/logging/context";
+
+export interface HeavyQueue {
+  /** What makes the job heavy. */
+  why: string;
+  /** A one-shot keeps its place past its expiry; see the module note. */
+  oneShot?: true;
+}
+
+/** The queues that take the lane. */
+export const HEAVY_QUEUES: Readonly<Record<string, HeavyQueue>> = {
+  "apple-health-import-v2": {
+    why: "Parses an export of up to several gigabytes.",
+    oneShot: true,
+  },
+  "health-connect-import": {
+    why: "Reads a Health Connect export database.",
+    oneShot: true,
+  },
+  "backup-restore": {
+    why: "Holds a batch of a whole account's readings per insert.",
+    oneShot: true,
+  },
+  "data-backup": { why: "Serialises every account's whole record." },
+  "data-backup-offhost": {
+    why: "Serialises and uploads every account's record.",
+  },
+  "data-restore-drill": {
+    why: "Stream-parses every account's newest off-host object.",
+  },
+  "rollup-full-backfill": {
+    why: "Folds five years of one account's readings into buckets.",
+  },
+  "insights-general-status": {
+    why: "Builds every status snapshot for every account, a provider call each.",
+  },
+  "insight-pregenerate": {
+    why: "The nightly walk: a comprehensive snapshot and generation per account.",
+  },
+  "period-narrative-warm": {
+    why: "The boundary-night walk: a narrative per account and period.",
+  },
+};
+
+type Waiter = () => void;
+
+const NEVER = new AbortController().signal;
+
+const waiting: Waiter[] = [];
+let busy = false;
+
+/** Resolves true once the lane is ours, false if `signal` aborts first. */
+function enter(signal: AbortSignal): Promise<boolean> {
+  if (!busy) {
+    busy = true;
+    return Promise.resolve(true);
+  }
+  return new Promise((resolve) => {
+    const onAbort = () => {
+      const at = waiting.indexOf(grant);
+      if (at >= 0) waiting.splice(at, 1);
+      resolve(false);
+    };
+    const grant: Waiter = () => {
+      signal.removeEventListener("abort", onAbort);
+      resolve(true);
+    };
+    if (signal.aborted) return resolve(false);
+    signal.addEventListener("abort", onAbort, { once: true });
+    waiting.push(grant);
+  });
+}
+
+function leave(): void {
+  const next = waiting.shift();
+  if (next) next();
+  else busy = false;
+}
+
+/**
+ * Wrap `handler` so the jobs of `queue` run in the lane. `inLane` picks the
+ * jobs that are heavy where a queue carries light ones too (a single
+ * account's warm beside the nightly walk); a batch with none of them runs at
+ * once.
+ */
+export function heavyLane<J extends Job<unknown>>(
+  queue: string,
+  handler: (jobs: J[]) => Promise<JobOutcome>,
+  inLane: (job: J) => boolean = () => true,
+): (jobs: J[]) => Promise<JobOutcome> {
+  const heavy = HEAVY_QUEUES[queue];
+  if (!heavy) throw new Error(`heavyLane: ${queue} is not in HEAVY_QUEUES`);
+  return async (jobs) => {
+    if (!jobs.some(inLane)) return handler(jobs);
+    const queuedAt = Date.now();
+    const entered = await enter(
+      heavy.oneShot ? NEVER : AbortSignal.any(signalsOf(jobs)),
+    );
+    const waitedMs = Date.now() - queuedAt;
+    annotate({
+      action: {
+        name: entered ? "job.heavy_lane.entered" : "job.heavy_lane.expired",
+      },
+      meta: { queue, waited_ms: waitedMs },
+    });
+    if (!entered) return jobDone({ lane_expired: jobs.length });
+    try {
+      return await handler(jobs);
+    } finally {
+      leave();
+    }
+  };
+}
+
+function signalsOf(jobs: readonly Job<unknown>[]): AbortSignal[] {
+  return jobs.flatMap((job) => (job.signal ? [job.signal] : []));
+}

src/lib/jobs/job-outcome.ts

@@ -140,6 +140,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "inventory_expired_count",
   "inventory_items_migrated",
   "jobs",
+  // A heavy job pg-boss expired while it waited for the heavy lane.
+  "lane_expired",
   "legacy_rows_soft_deleted",
   "linked",
   "malformed",

src/lib/jobs/reminder/register-maintenance.ts

@@ -36,6 +36,7 @@ import {
 } from "@/lib/jobs/pr-detection";
 import { jobBudget } from "@/lib/jobs/job-budget";
 import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+import { heavyLane } from "@/lib/jobs/heavy-lane";
 import {
   OFFHOST_BACKUP_QUEUE,
   OFFHOST_BACKUP_SEND_OPTIONS,
@@ -931,7 +932,11 @@ export async function registerMaintenanceQueues(
     boss,
     DATA_BACKUP_QUEUE,
     { localConcurrency: 1 },
-    lockedPass(DATA_BACKUP_QUEUE, () => WHOLE_PASS, handleDataBackup),
+    lockedPass(
+      DATA_BACKUP_QUEUE,
+      () => WHOLE_PASS,
+      heavyLane(DATA_BACKUP_QUEUE, handleDataBackup),
+    ),
   );
   await createAndWork<RateLimitCleanupPayload>(
     boss,
@@ -962,7 +967,11 @@ export async function registerMaintenanceQueues(
     boss,
     OFFHOST_BACKUP_QUEUE,
     { localConcurrency: 1, includeMetadata: true },
-    lockedPass(OFFHOST_BACKUP_QUEUE, () => WHOLE_PASS, handleOffhostBackup),
+    lockedPass(
+      OFFHOST_BACKUP_QUEUE,
+      () => WHOLE_PASS,
+      heavyLane(OFFHOST_BACKUP_QUEUE, handleOffhostBackup),
+    ),
   );
   // Locked: a drill that outlives a restart or a manual run never reads the
   // bucket beside another one.
@@ -970,7 +979,11 @@ export async function registerMaintenanceQueues(
     boss,
     RESTORE_DRILL_QUEUE,
     { localConcurrency: 1 },
-    lockedPass(RESTORE_DRILL_QUEUE, () => WHOLE_PASS, handleRestoreDrill),
+    lockedPass(
+      RESTORE_DRILL_QUEUE,
+      () => WHOLE_PASS,
+      heavyLane(RESTORE_DRILL_QUEUE, handleRestoreDrill),
+    ),
   );
   await createAndWork(
     boss,
@@ -985,7 +998,7 @@ export async function registerMaintenanceQueues(
     boss,
     BACKUP_RESTORE_QUEUE,
     { localConcurrency: 1 },
-    handleBackupRestore,
+    heavyLane(BACKUP_RESTORE_QUEUE, handleBackupRestore),
   );
   // A restore the previous process was running is picked up again once its
   // heartbeat has gone stale; see `sweepInterruptedRestores`.
@@ -1192,7 +1205,7 @@ export async function registerMaintenanceQueues(
     boss,
     APPLE_HEALTH_IMPORT_V2_QUEUE,
     { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY },
-    async (jobs) => {
+    heavyLane(APPLE_HEALTH_IMPORT_V2_QUEUE, async (jobs) => {
       // pg-boss v12 work callbacks always receive an array (batched
       // worker mode); for our concurrency-1 + batchSize-1 case we just
       // process each job sequentially.
@@ -1205,7 +1218,7 @@ export async function registerMaintenanceQueues(
         if (!outcome.ok && failure === null) failure = outcome;
       }
       return failure ?? jobDone({ jobs: jobs.length });
-    },
+    }),
   );
   // Drain pre-revision-2 backlog by moving each claimed legacy job onto the
   // isolated v2 queue. This preserves the ImportJob status id without letting
@@ -1228,7 +1241,7 @@ export async function registerMaintenanceQueues(
     boss,
     HEALTH_CONNECT_IMPORT_QUEUE,
     { localConcurrency: HEALTH_CONNECT_IMPORT_CONCURRENCY },
-    handleHealthConnectImport,
+    heavyLane(HEALTH_CONNECT_IMPORT_QUEUE, handleHealthConnectImport),
   );
   // v1.42 — compaction-tombstone backlog purge. Serial: each batch holds
   // an account against a concurrent restore.

src/lib/jobs/reminder/register-rollup.ts

@@ -108,6 +108,7 @@ import {
   workerLog,
   BOOT_BACKFILL_STAGGER_SECONDS,
 } from "./shared";
+import { heavyLane } from "@/lib/jobs/heavy-lane";
 import {
   createAndSchedule,
   createAndWork,
@@ -331,7 +332,7 @@ export async function registerRollupQueues(
     boss,
     ROLLUP_FULL_BACKFILL_QUEUE,
     { localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY },
-    async (jobs) => {
+    heavyLane(ROLLUP_FULL_BACKFILL_QUEUE, async (jobs) => {
       let rowsUpsertedTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
@@ -346,7 +347,7 @@ export async function registerRollupQueues(
         users: jobs.length,
         rows_upserted: rowsUpsertedTotal,
       });
-    },
+    }),
   );
 
   // v1.5.6 — legacy step consolidation worker. The boot enqueue helper

src/lib/jobs/reminder/register-status.ts

@@ -104,6 +104,7 @@ import { withBackgroundEvent } from "@/lib/logging/background";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { jobBudget } from "@/lib/jobs/job-budget";
 import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+import { heavyLane } from "@/lib/jobs/heavy-lane";
 import { INSIGHT_PREGENERATE_EXPIRE_SECONDS } from "@/lib/jobs/insight-pregenerate-shared";
 import { PERIOD_NARRATIVE_EXPIRE_SECONDS } from "@/lib/jobs/period-narrative-shared";
 import { getWorkerPrisma, workerLog } from "./shared";
@@ -346,7 +347,7 @@ export async function registerStatusQueues(
     lockedPass(
       GENERAL_STATUS_QUEUE,
       () => WHOLE_PASS,
-      handleGeneralStatusGenerate,
+      heavyLane(GENERAL_STATUS_QUEUE, handleGeneralStatusGenerate),
     ),
   );
   // v1.7.0 — nightly comprehensive-insight pre-generation.
@@ -372,7 +373,11 @@ export async function registerStatusQueues(
         job.data?.force && job.data.userId
           ? `user:${job.data.userId}`
           : WHOLE_PASS,
-      handleInsightPregenerateJob,
+      heavyLane(
+        INSIGHT_PREGENERATE_QUEUE,
+        handleInsightPregenerateJob,
+        (job) => !(job.data?.force && job.data.userId),
+      ),
     ),
   );
   // S4 — event-driven morning digest refresh. One forced comprehensive
@@ -703,49 +708,56 @@ export async function registerStatusQueues(
     lockedPass(
       PERIOD_NARRATIVE_QUEUE,
       (job) => (job.data?.userId ? `user:${job.data.userId}` : WHOLE_PASS),
-      async (jobs): Promise<JobOutcome> => {
-        // Two modes on one queue, so the outcome counts them separately: a
-        // non-boundary night is a scheduled pass that generated nothing, which
-        // is a success with zero counts, not an idle queue.
-        let singleUser = 0;
-        let scheduled = 0;
-        let generated = 0;
-        let failed = 0;
-        let stoppedEarly = false;
-        for (const job of jobs) {
-          try {
-            if (job.data?.userId) {
-              await warmOneNarrative(job.data);
-              singleUser++;
-            } else {
-              const summary = await runPeriodNarrativeWarm(getWorkerPrisma(), {
-                shouldStop: jobBudget(jobs),
+      heavyLane(
+        PERIOD_NARRATIVE_QUEUE,
+        async (jobs): Promise<JobOutcome> => {
+          // Two modes on one queue, so the outcome counts them separately: a
+          // non-boundary night is a scheduled pass that generated nothing, which
+          // is a success with zero counts, not an idle queue.
+          let singleUser = 0;
+          let scheduled = 0;
+          let generated = 0;
+          let failed = 0;
+          let stoppedEarly = false;
+          for (const job of jobs) {
+            try {
+              if (job.data?.userId) {
+                await warmOneNarrative(job.data);
+                singleUser++;
+              } else {
+                const summary = await runPeriodNarrativeWarm(
+                  getWorkerPrisma(),
+                  {
+                    shouldStop: jobBudget(jobs),
+                  },
+                );
+                scheduled++;
+                generated += summary.generated;
+                failed += summary.failed;
+                stoppedEarly ||= summary.stoppedEarly;
+                workerLog(
+                  "info",
+                  `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
+                );
+              }
+            } catch (err) {
+              recordError();
+              await reportWorkerError(PERIOD_NARRATIVE_QUEUE, err, {
+                mode: job.data?.userId ? "single-user" : "scheduled",
               });
-              scheduled++;
-              generated += summary.generated;
-              failed += summary.failed;
-              stoppedEarly ||= summary.stoppedEarly;
-              workerLog(
-                "info",
-                `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
-              );
+              throw err;
             }
-          } catch (err) {
-            recordError();
-            await reportWorkerError(PERIOD_NARRATIVE_QUEUE, err, {
-              mode: job.data?.userId ? "single-user" : "scheduled",
-            });
-            throw err;
           }
-        }
-        return jobDone({
-          single_user: singleUser,
-          scheduled,
-          generated,
-          failed,
-          stopped_early: stoppedEarly,
-        });
-      },
+          return jobDone({
+            single_user: singleUser,
+            scheduled,
+            generated,
+            failed,
+            stopped_early: stoppedEarly,
+          });
+        },
+        (job) => !job.data?.userId,
+      ),
     ),
   );
   // v1.11.1 — combined Coach memory refresh: rolling conversation summary +