diff --git a/src/__tests__/job-queue-runtime-guard.test.ts b/src/__tests__/job-queue-runtime-guard.test.ts index 7f5adbf72..22a8b991a 100644 --- a/src/__tests__/job-queue-runtime-guard.test.ts +++ b/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, 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([]); + }); +}); diff --git a/src/lib/jobs/__tests__/heavy-lane.test.ts b/src/lib/jobs/__tests__/heavy-lane.test.ts new file mode 100644 index 000000000..9f9e7a2d7 --- /dev/null +++ b/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((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/); + }); +}); diff --git a/src/lib/jobs/heavy-lane.ts b/src/lib/jobs/heavy-lane.ts new file mode 100644 index 000000000..032df827c --- /dev/null +++ b/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> = { + "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 { + 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>( + queue: string, + handler: (jobs: J[]) => Promise, + inLane: (job: J) => boolean = () => true, +): (jobs: J[]) => Promise { + 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[]): AbortSignal[] { + return jobs.flatMap((job) => (job.signal ? [job.signal] : [])); +} diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index b09cb08e8..4a29444fd 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -140,6 +140,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = 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", diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 7e6dd1d1a..4f25c5679 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/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( 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. diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index 41609001e..7a40496bc 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/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 diff --git a/src/lib/jobs/reminder/register-status.ts b/src/lib/jobs/reminder/register-status.ts index 1f3d961c6..0b14a2d75 100644 --- a/src/lib/jobs/reminder/register-status.ts +++ b/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 => { - // 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 => { + // 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 +