From 5a6c78efe5e72b3601f5efa4888ad3fd9d909f3c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 11:06:17 +0200 Subject: [PATCH] fix(jobs): give the full rollup fold an expiry, a budget and a lock The fold was declared short on the strength of its 60-second statements, but four aggregates and four persist transactions of up to 120 s come to about thirteen minutes at worst under the defaults, and nothing bounds them once an operator lifts the statement timeout. Past pg-boss's fifteen-minute default the retry folded the account a second time. The send now carries an hour's expiry, the fold stops between granularities on the job's budget, and the binding holds a per-account lockedPass outside the lane. The runtime guard now requires every heavy queue to be declared long. --- src/__tests__/job-queue-runtime-guard.test.ts | 22 +++++++++ src/lib/jobs/queue-runtime.ts | 12 +++-- src/lib/jobs/reminder/register-rollup.ts | 48 ++++++++++++------- .../__tests__/measurement-rollups.test.ts | 13 +++++ src/lib/rollups/measurement-rollups.ts | 27 ++++++++++- 5 files changed, 102 insertions(+), 20 deletions(-) diff --git a/src/__tests__/job-queue-runtime-guard.test.ts b/src/__tests__/job-queue-runtime-guard.test.ts index 71dd3a2fe..48da4c47d 100644 --- a/src/__tests__/job-queue-runtime-guard.test.ts +++ b/src/__tests__/job-queue-runtime-guard.test.ts @@ -438,6 +438,28 @@ describe("heavy jobs share the lane", () => { expect(inverted).toEqual([]); }); + it("declares every heavy queue long, so its expiry, stop and exclusivity are proven", () => { + // A heavy job is the one most likely to run past pg-boss's default + // expiry, and the retry that follows an expiry runs it a second time. + // `long` is the declaration the checks above hold to an explicit expiry, + // a stop on the budget and a lock (or a named marker). A heavy queue + // that is genuinely bounded far below the default may be named here + // with its reason instead. + const HEAVY_BUT_SHORT: Record = {}; + const notLong = (table: Readonly>) => + Object.keys(HEAVY_QUEUES) + .filter((q) => table[q]?.runtime !== "long" && !(q in HEAVY_BUT_SHORT)) + .sort(); + expect(notLong(QUEUE_RUNTIME)).toEqual([]); + // The planted case: the fold as it was declared before, `short`. + expect( + notLong({ + ...QUEUE_RUNTIME, + "rollup-full-backfill": { runtime: "short", why: "four statements" }, + }), + ).toEqual(["rollup-full-backfill"]); + }); + 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 diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index e4922f110..a5546c376 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -42,6 +42,7 @@ import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-int import { HEALTH_CONNECT_IMPORT_SEND_OPTIONS } from "@/lib/jobs/health-connect-import-worker"; import { MEASUREMENT_MAINTENANCE_SEND_OPTIONS } from "@/lib/jobs/measurement-maintenance"; import { PULSE_DAY_MEAN_SEND_OPTIONS } from "@/lib/jobs/pulse-day-mean"; +import { ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS } from "@/lib/rollups/measurement-rollups"; /** pg-boss's default job expiry, in seconds. */ export const DEFAULT_JOB_EXPIRE_SECONDS = 15 * 60; @@ -190,9 +191,14 @@ export const QUEUE_RUNTIME: Readonly> = { // ── Rollups ──────────────────────────────────────────────────────── "rollup-recompute": short("One bucket."), - "rollup-full-backfill": short( - "One account's rollup tiers: four aggregate statements, each held to the 60-second statement timeout.", - ), + "rollup-full-backfill": { + runtime: "long", + expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, + expiryVia: ["ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS"], + stop: "binding", + exclusive: "lockedPass", + why: "One account's five-year fold: four aggregates under the statement timeout and four persist transactions of up to 120 s each, about thirteen minutes at worst under the defaults and unbounded when the operator lifts the statement timeout.", + }, "mood-rollup-recompute": short("One bucket."), "mood-rollup-full-backfill": short( "One account's mood entries, a few a day.", diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index e779f0f48..27534267c 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -109,6 +109,7 @@ import { BOOT_BACKFILL_STAGGER_SECONDS, } from "./shared"; import { heavyLane } from "@/lib/jobs/heavy-lane"; +import { lockedPass } from "@/lib/jobs/long-pass"; import { createAndSchedule, createAndWork, @@ -332,22 +333,37 @@ export async function registerRollupQueues( boss, ROLLUP_FULL_BACKFILL_QUEUE, { localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY }, - heavyLane(ROLLUP_FULL_BACKFILL_QUEUE, async (jobs) => { - let rowsUpsertedTotal = 0; - for (const job of jobs) { - const { userId } = job.data; - const { rowsUpserted, durationMs } = await recomputeUserRollups(userId); - rowsUpsertedTotal += rowsUpserted; - workerLog( - "info", - `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, - ); - } - return jobDone({ - users: jobs.length, - rows_upserted: rowsUpsertedTotal, - }); - }), + // The lock sits outside the lane: a retry pg-boss starts after the expiry + // while the first fold still runs finds the account's lock and settles, + // instead of waiting in the lane and folding the account again. + lockedPass( + ROLLUP_FULL_BACKFILL_QUEUE, + (job) => job.data.userId, + heavyLane(ROLLUP_FULL_BACKFILL_QUEUE, async (jobs) => { + const shouldStop = jobBudget(jobs); + let rowsUpsertedTotal = 0; + for (const job of jobs) { + const { userId } = job.data; + const { rowsUpserted, durationMs, stopped } = + await recomputeUserRollups(userId, { shouldStop }); + if (stopped) { + // Each granularity committed on its own; the retry folds again. + throw new Error( + `rollup-full-backfill for user ${userId} stopped on its job budget; the retry folds the rest`, + ); + } + rowsUpsertedTotal += rowsUpserted; + workerLog( + "info", + `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, + ); + } + return jobDone({ + users: jobs.length, + rows_upserted: rowsUpsertedTotal, + }); + }), + ), ); // v1.5.6 — legacy step consolidation worker. The boot enqueue helper diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index 850e55fa3..3426575fa 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -496,6 +496,19 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { * instant, a different oldest bucket was corrupted on each run and nothing * repaired it. The window must be snapped out to whole buckets first. */ + it("stops between granularities once the job's budget is spent", async () => { + queryRawUnsafe.mockResolvedValue([]); + let asked = 0; + const result = await recomputeUserRollups("user-1", { + // The budget runs out after the first granularity was folded. + shouldStop: () => ++asked > 1, + }); + expect(result.stopped).toBe(true); + // The first granularity's aggregate, then nothing more. + expect(queryRawUnsafe).toHaveBeenCalledTimes(1); + queryRawUnsafe.mockReset(); + }); + describe("bucket alignment of the recompute window", () => { it("snaps a mid-day `from` down to the start of its DAY bucket", async () => { queryRawUnsafe.mockResolvedValueOnce(syntheticRows(1)); diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 9dd95cc2c..cf33a2c6d 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -126,6 +126,17 @@ export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill"; /** Boot-fold worker concurrency. Serial so we don't crowd the pool. */ export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1; +/** + * The full fold's expiry: an hour. One fold is four granularities, each an + * aggregate held to the connection's statement timeout (60 s by default) and + * a persist transaction of up to 120 s, so the worst case under the defaults + * is close to thirteen minutes, too near pg-boss's fifteen-minute default to + * rely on, and an operator who raises `DATABASE_STATEMENT_TIMEOUT_MS` (or + * sets it to 0) lifts that bound altogether. Past the expiry pg-boss retries + * the job, and the retry folded the same account a second time. + */ +export const ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS = 60 * 60; + /** Payload the boot-time backfill helper enqueues. */ export interface RollupFullBackfillPayload { userId: string; @@ -203,8 +214,14 @@ export async function recomputeUserRollups( * statement group, so it ends with it. */ statementTimeoutMs?: number; + /** + * Asked before each granularity: a fold that runs under a job's budget + * (`jobBudget`) stops between granularities, each of which commits on + * its own, and reports `stopped`. + */ + shouldStop?: () => boolean; } = {}, -): Promise<{ rowsUpserted: number; durationMs: number }> { +): Promise<{ rowsUpserted: number; durationMs: number; stopped?: true }> { const startedAt = Date.now(); const granularities = opts.granularities ?? ALL_GRANULARITIES; const fiveYearsAgo = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); @@ -214,6 +231,13 @@ export async function recomputeUserRollups( let rowsUpserted = 0; for (const granularity of granularities) { + if (opts.shouldStop?.()) { + return { + rowsUpserted, + durationMs: Date.now() - startedAt, + stopped: true, + }; + } // Snap the window out to whole buckets before aggregating. The // aggregate filters rows by `measured_at` but GROUPS by // `date_trunc(, measured_at)`, and `persistRollupRows` @@ -1219,6 +1243,7 @@ export async function enqueueBootTimeRollupBackfill( enqueuedAt: new Date().toISOString(), }; const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, { + expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, retryLimit: 3, retryDelay: 60, retryBackoff: true,