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

@@ -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<string, string> = {};
+    const notLong = (table: Readonly<Record<string, QueueRuntime>>) =>
+      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

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<Record<string, QueueRuntime>> = {
 
   // ── 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.",

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

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

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(<granularity>, 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,