diff --git a/src/__tests__/job-queue-runtime-guard.test.ts b/src/__tests__/job-queue-runtime-guard.test.ts index 22a8b991a..71dd3a2fe 100644 --- a/src/__tests__/job-queue-runtime-guard.test.ts +++ b/src/__tests__/job-queue-runtime-guard.test.ts @@ -417,16 +417,6 @@ describe("a cron-fed queue polls slowly, a sent one does not", () => { }); }); -/** 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)); @@ -437,7 +427,10 @@ describe("heavy jobs share the lane", () => { ); }); - it("holds a lockedPass outside the lane, so an expiry retry does not queue twice", () => { + it("holds a lockedPass outside the lane, so an expiry retry finds the run going", () => { + // Outside, a retry pg-boss starts while the expired run still holds the + // lane finds the lock and settles; inside, the lane would defer it and + // run the job a second time once the first is done. const inverted = laned .filter((b) => /\blockedPass\s*\(/.test(b.text)) .filter((b) => b.text.indexOf("lockedPass") > b.text.indexOf("heavyLane")) @@ -445,17 +438,45 @@ describe("heavy jobs share the lane", () => { 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([]); + it("defers only on queues where a job can go back to created", () => { + // A deferred job returns from active to created. Under `short` (one + // queued job per key) and `stately` (one per state) a twin already + // queued would refuse it, and the job would fail instead of waiting. + const policies = policiesOf(source); + expect(policies.size).toBeGreaterThanOrEqual(40); + const refused = Object.keys(HEAVY_QUEUES) + .filter((queue) => /^(?:short|stately)$/.test(policies.get(queue) ?? "")) + .sort(); + expect(refused).toEqual([]); + // The planted case: the same check over a table that declares one. + const planted = new Map(source); + const file = REGISTRARS[2]; + planted.set( + file, + `const DATA_BACKUP_QUEUE = "data-backup";\nconst queuePolicies: QueuePolicyTable = {\n [DATA_BACKUP_QUEUE]: { policy: "short", reason: "x" },\n};\n`, + ); + expect(policiesOf(planted).get("data-backup")).toBe("short"); }); }); + +/** Every `[QUEUE]: { policy: "…" }` in a registrar's `queuePolicies`. */ +function policiesOf(texts: ReadonlyMap): Map { + const definitions = definitionsOf(texts); + const out = new Map(); + for (const file of REGISTRARS) { + const text = code(texts.get(file) ?? ""); + const m = /const\s+queuePolicies\s*:\s*QueuePolicyTable\s*=\s*\{/.exec( + text, + ); + if (!m) continue; + const open = m.index + m[0].length - 1; + const body = text.slice(open, closeOf(text, open)); + for (const entry of body.matchAll( + /\[\s*([A-Z][A-Z0-9_]*)\s*\]\s*:\s*\{\s*policy\s*:\s*"([a-z]+)"/g, + )) { + const queue = resolve(definitions, file, entry[1]); + if (queue) out.set(queue, entry[2]); + } + } + return out; +} diff --git a/src/lib/jobs/__tests__/heavy-lane.test.ts b/src/lib/jobs/__tests__/heavy-lane.test.ts index 9f9e7a2d7..0b97297ed 100644 --- a/src/lib/jobs/__tests__/heavy-lane.test.ts +++ b/src/lib/jobs/__tests__/heavy-lane.test.ts @@ -1,19 +1,21 @@ /** - * 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. + * The heavy lane runs one heavy job at a time per process. A job that finds + * the lane taken goes back to its queue at once instead of waiting in the + * handler, where its expiry would run down while it stood in line. */ import { describe, expect, it, vi } from "vitest"; import type { Job } from "pg-boss"; vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); +vi.mock("@/lib/db", () => ({ prisma: {} })); 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; +function job(id = "j", data: { force?: boolean } = {}): TestJob { + return { id, data } as unknown as TestJob; } /** A handler that runs until `finish` is called, recording the overlap. */ @@ -29,87 +31,127 @@ function gate(log: string[], name: string) { return { handler, finish }; } +/** A stand-in for the way back to the queue, recording what went back. */ +function deferrals() { + const back: Array<{ queue: string; ids: string[] }> = []; + const defer = vi.fn(async (queue: string, jobs: readonly TestJob[]) => { + back.push({ queue, ids: jobs.map((j) => j.id) }); + return jobs.length; + }); + return { back, defer }; +} + const tick = () => new Promise((r) => setTimeout(r, 0)); describe("heavyLane", () => { - it("runs two heavy jobs of different queues one after the other", async () => { + it("sends a heavy job back to its queue while another holds the lane", 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()]); + const a = gate(log, "import"); + const b = gate(log, "backup"); + const { back, defer } = deferrals(); + const running = heavyLane( + "apple-health-import-v2", + a.handler, + undefined, + defer, + )([job("upload")]); await tick(); - expect(log).toEqual(["start:backup"]); + const outcome = await heavyLane( + "data-backup", + b.handler, + undefined, + defer, + )([job("weekly")]); + // It returned at once, without running and without waiting. + expect(outcome).toMatchObject({ ok: true, did: { lane_deferred: 1 } }); + expect(b.handler).not.toHaveBeenCalled(); + expect(back).toEqual([{ queue: "data-backup", ids: ["weekly"] }]); a.finish(); - await first; + await running; + // Once the lane is free the redelivered job runs. + const again = heavyLane( + "data-backup", + b.handler, + undefined, + defer, + )([job("weekly")]); await tick(); - expect(log).toEqual(["start:backup", "end:backup", "start:import"]); b.finish(); - await second; + await again; + expect(log).toEqual([ + "start:import", + "end:import", + "start:backup", + "end:backup", + ]); + expect(defer).toHaveBeenCalledTimes(1); }); - it("lets a light job of a heavy queue pass the line", async () => { + it("lets a light job of a heavy queue pass the lane", async () => { const log: string[] = []; const a = gate(log, "walk"); const b = gate(log, "warm"); + const { defer } = deferrals(); const notForced = (j: TestJob) => !j.data.force; const walk = heavyLane( "insight-pregenerate", a.handler, notForced, + defer, )([job()]); const warm = heavyLane( "insight-pregenerate", b.handler, notForced, - )([job({ force: true })]); + defer, + )([job("w", { 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; + expect(defer).not.toHaveBeenCalled(); }); it("frees the lane when a handler throws", async () => { - const failing = heavyLane("rollup-full-backfill", async () => { - throw new Error("boom"); - }); + const { defer } = deferrals(); + const failing = heavyLane( + "rollup-full-backfill", + async () => { + throw new Error("boom"); + }, + undefined, + defer, + ); await expect(failing([job()])).rejects.toThrow("boom"); const next = vi.fn(async () => jobDone({ jobs: 1 })); await expect( - heavyLane("data-backup", next)([job()]), + heavyLane("data-backup", next, undefined, defer)([job()]), ).resolves.toMatchObject({ ok: true }); + expect(next).toHaveBeenCalledTimes(1); + expect(defer).not.toHaveBeenCalled(); + }); + + it("fails the delivery when the way back to the queue fails", async () => { + const log: string[] = []; + const holder = gate(log, "holder"); + const { defer } = deferrals(); + const held = heavyLane( + "data-backup-offhost", + holder.handler, + undefined, + defer, + )([job()]); + await tick(); + const broken = vi.fn(async () => { + throw new Error("db down"); + }); + // pg-boss then retries it, rather than completing a job that never ran. + await expect( + heavyLane("data-backup", vi.fn(), undefined, broken)([job()]), + ).rejects.toThrow("db down"); + holder.finish(); + await held; }); it("refuses a queue that is not declared heavy", () => { diff --git a/src/lib/jobs/heavy-lane.ts b/src/lib/jobs/heavy-lane.ts index 032df827c..0f924880d 100644 --- a/src/lib/jobs/heavy-lane.ts +++ b/src/lib/jobs/heavy-lane.ts @@ -7,47 +7,44 @@ * 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. + * heavy job runs only while no other heavy job does. * * 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. + * A job that arrives while the lane is taken does not wait in the handler. It + * goes back to the queue (`deferLaneJobs`): its row returns to `created` with + * `start_after` a little ahead, the claim cleared so the next fetch does not + * count an attempt, and pg-boss's completion of the handler then finds no + * active row and leaves it alone. Waiting in the handler instead ran the + * job's expiry down while it stood in line, because pg-boss counts the expiry + * from the fetch and ends the handler at it: a six-hour import let a + * two-hour backup expire through all its retries, and an upload that waited + * longer than its own expiry was failed, stopped counting as in use, and had + * its staged file swept from under it. A deferred job keeps its id, its + * retries and its staged upload, and its expiry starts when it runs. */ import type { Job } from "pg-boss"; +import { prisma } from "@/lib/db"; 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": { @@ -68,78 +65,74 @@ export const HEAVY_QUEUES: Readonly> = { "period-narrative-warm": { why: "The boundary-night walk: a narrative per account and period.", }, + "integration-backfill-admission": { + why: "A provider's full history for one account, page by page; the Google Health import is the largest heap in the tree.", + }, }; -type Waiter = () => void; +/** How far ahead a deferred job is put back in the queue. */ +export const LANE_DEFER_SECONDS = 60; -const NEVER = new AbortController().signal; +/** + * Put `jobs` of `queue` back in the queue to run `seconds` from now. The same + * reset pg-boss's own `restore()` makes, with `start_after` moved in the same + * statement: a restore followed by a separate update leaves a moment in which + * another fetch could claim the row, and the completion of this handler would + * then settle that newer claim, since a restored job's next claim keeps its + * attempt number. Answers how many rows went back. + */ +export async function deferLaneJobs( + queue: string, + jobs: readonly Job[], + seconds: number = LANE_DEFER_SECONDS, +): Promise { + const ids = jobs.map((job) => job.id); + return prisma.$executeRaw` + UPDATE pgboss.job + SET state = 'created', + started_on = NULL, + heartbeat_on = NULL, + upsert_by_key = NULL, + start_after = now() + make_interval(secs => ${seconds}::int) + WHERE name = ${queue} + AND id = ANY(${ids}::uuid[]) + AND state = 'active' + `; +} -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. + * once. `defer` is the way back to the queue; tests replace it. */ export function heavyLane>( queue: string, handler: (jobs: J[]) => Promise, inLane: (job: J) => boolean = () => true, + defer: (queue: string, jobs: readonly J[]) => Promise = deferLaneJobs, ): (jobs: J[]) => Promise { - const heavy = HEAVY_QUEUES[queue]; - if (!heavy) throw new Error(`heavyLane: ${queue} is not in HEAVY_QUEUES`); + if (!HEAVY_QUEUES[queue]) { + 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 }); + if (busy) { + const deferred = await defer(queue, jobs); + annotate({ + action: { name: "job.heavy_lane.deferred" }, + meta: { queue, jobs: jobs.length, deferred }, + }); + return jobDone({ lane_deferred: deferred }); + } + busy = true; + annotate({ action: { name: "job.heavy_lane.entered" }, meta: { queue } }); try { return await handler(jobs); } finally { - leave(); + busy = false; } }; } - -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 04f85eb8e..672cc8ba3 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -143,8 +143,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", + // Heavy jobs that found the lane taken and went back to their queue. + "lane_deferred", "legacy_rows_soft_deleted", "linked", "malformed", diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index 7fd137473..afb14ee91 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -70,6 +70,7 @@ import { type IntegrationBackfillAdmissionPayload, } from "@/lib/jobs/integration-backfill-admission"; import { drainIntegrationBackfillAdmission } from "@/lib/jobs/integration-backfill-drain"; +import { heavyLane } from "@/lib/jobs/heavy-lane"; import { INTEGRATION_WATCHDOG_DAILY_CRON, INTEGRATION_WATCHDOG_DAILY_QUEUE, @@ -491,7 +492,9 @@ export async function registerIntegrationSyncQueues( await createAndSchedule(boss, allQueues, schedules, queuePolicies); // pg-boss scopes groupConcurrency to one queue name, so every provider // reaches this shared durable queue before doing full-history work. The - // common group is enforced in PostgreSQL across worker processes/nodes. + // common group is enforced in PostgreSQL across worker processes/nodes; + // the heavy lane keeps an import off the heap while a backup, a restore + // or an upload is on it. await createAndWork( boss, @@ -500,7 +503,10 @@ export async function registerIntegrationSyncQueues( localConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY, groupConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY, }, - drainIntegrationBackfillAdmission, + heavyLane( + INTEGRATION_BACKFILL_ADMISSION_QUEUE, + drainIntegrationBackfillAdmission, + ), ); await createAndWork( diff --git a/tests/integration/heavy-lane-defer.test.ts b/tests/integration/heavy-lane-defer.test.ts new file mode 100644 index 000000000..bbd0f0485 --- /dev/null +++ b/tests/integration/heavy-lane-defer.test.ts @@ -0,0 +1,147 @@ +/** + * The heavy lane against a real pg-boss on a real Postgres: a heavy job that + * finds the lane taken goes back to its queue, keeps its id and its attempt, + * is not settled by pg-boss's completion of the handler that sent it back, + * and does not spend its expiry while it waits. + * + * The case it replaces: a job that waited in the handler ran its expiry down + * while it stood in line, because pg-boss counts the expiry from the fetch. + * Here the holder runs for six seconds against a two-second expiry on the + * waiting job; the waiting job still runs once, after the holder, without a + * single failed attempt. + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { PgBoss, type Job } from "pg-boss"; + +import { prisma } from "@/lib/db"; +import { deferLaneJobs, heavyLane } from "@/lib/jobs/heavy-lane"; +import { jobDone } from "@/lib/jobs/job-outcome"; + +let boss: PgBoss; + +const HOLDER_QUEUE = "data-backup-offhost"; +const WAITER_QUEUE = "data-backup"; +const WORK = { localConcurrency: 1, pollingIntervalSeconds: 0.5 } as const; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +beforeAll(async () => { + boss = new PgBoss({ connectionString: process.env.DATABASE_URL }); + await boss.start(); + await boss.createQueue(HOLDER_QUEUE); + await boss.createQueue(WAITER_QUEUE); +}, 120_000); + +afterAll(async () => { + await boss?.stop({ graceful: false }); +}); + +async function row(id: string) { + const rows = await prisma.$queryRaw< + Array<{ + state: string; + retry_count: number; + started_on: Date | null; + future: boolean; + }> + >` + SELECT state::text AS state, retry_count, started_on, + start_after > now() AS future + FROM pgboss.job WHERE id = ${id}::uuid + `; + return rows[0]; +} + +describe("deferLaneJobs", () => { + it("returns an active job to created, later, at the same attempt", async () => { + const id = await boss.send(WAITER_QUEUE, { probe: 1 }); + const [fetched] = await boss.fetch(WAITER_QUEUE); + expect(fetched?.id).toBe(id); + expect((await row(id!))?.state).toBe("active"); + + expect(await deferLaneJobs(WAITER_QUEUE, [fetched!], 30)).toBe(1); + expect(await row(id!)).toMatchObject({ + state: "created", + retry_count: 0, + started_on: null, + future: true, + }); + // Not claimable before its time. + expect(await boss.fetch(WAITER_QUEUE)).toEqual([]); + // pg-boss's completion of the handler finds no active row to settle. + await boss.complete(WAITER_QUEUE, id!); + expect((await row(id!))?.state).toBe("created"); + await boss.deleteJob(WAITER_QUEUE, id!); + }); + + it("leaves a job alone that is no longer active", async () => { + const id = await boss.send(WAITER_QUEUE, { probe: 2 }); + const job = { id, name: WAITER_QUEUE, data: {} } as unknown as Job; + expect(await deferLaneJobs(WAITER_QUEUE, [job], 30)).toBe(0); + expect((await row(id!))?.future).toBe(false); + await boss.deleteJob(WAITER_QUEUE, id!); + }); +}); + +describe("a heavy job behind a long one", () => { + it("waits in the queue, not in the handler, and runs once after it", async () => { + const log: string[] = []; + const deliveries: string[] = []; + const shortDefer = (queue: string, jobs: readonly Job[]) => + deferLaneJobs(queue, jobs, 1); + + await boss.work( + HOLDER_QUEUE, + WORK, + heavyLane( + HOLDER_QUEUE, + async () => { + log.push("start:holder"); + await sleep(6_000); + log.push("end:holder"); + return jobDone({ jobs: 1 }); + }, + undefined, + shortDefer, + ), + ); + await boss.work( + WAITER_QUEUE, + WORK, + heavyLane( + WAITER_QUEUE, + async () => { + log.push("run:waiter"); + return jobDone({ jobs: 1 }); + }, + undefined, + async (queue, jobs) => { + deliveries.push(jobs[0]!.id); + return shortDefer(queue, jobs); + }, + ), + ); + + await boss.send(HOLDER_QUEUE, {}, { expireInSeconds: 60 }); + await sleep(1_000); + const waiter = await boss.send( + WAITER_QUEUE, + {}, + { expireInSeconds: 2, retryLimit: 0 }, + ); + + await sleep(9_000); + await boss.offWork(WAITER_QUEUE); + await boss.offWork(HOLDER_QUEUE); + + // Three times its expiry in line, and not a single failed attempt. + expect(await row(waiter!)).toMatchObject({ + state: "completed", + retry_count: 0, + }); + // Sent back several times while the holder ran, then run once after it. + expect(deliveries.length).toBeGreaterThanOrEqual(2); + expect(new Set(deliveries)).toEqual(new Set([waiter])); + expect(log).toEqual(["start:holder", "end:holder", "run:waiter"]); + }, 60_000); +});