fix(jobs): give the hourly rebuild the same time budget

The one-shot hourly rebuild of already-folded heart-rate, HRV and SpO2
days walks one account's whole folded history in one job with the
default fifteen-minute expiry, about fifty statements per day. Like the
drains it now stops between days once 75 % of the expiry has passed or
pg-boss aborts the job, and fails so pg-boss retries it (the boot send
allows three retries); a rebuilt day leaves the candidate set, so the
retry starts at the first day not yet rebuilt.

src/lib/jobs/dense-intraday-hourly-rebuild.ts

@@ -65,11 +65,14 @@ export interface DenseIntradayHourlyRebuildPayload {
  */
 export async function runDenseIntradayHourlyRebuildForUser(
   userId: string,
+  shouldStop?: () => boolean,
 ): Promise<{
   daysRebuilt: number;
   hourlyRowsUpserted: number;
   dailyRowsRetired: number;
   daysSkippedNoTombstones: number;
+  /** The job's budget ran out first; see the queue handler. */
+  stoppedEarly: boolean;
 }> {
   // Shares the retention drain's kill-switch: when an operator disables the
   // dense tier, no-op so any already-queued backlog drains cleanly.
@@ -79,10 +82,12 @@ export async function runDenseIntradayHourlyRebuildForUser(
       hourlyRowsUpserted: 0,
       dailyRowsRetired: 0,
       daysSkippedNoTombstones: 0,
+      stoppedEarly: false,
     };
   }
   const summary = await runDenseIntradayHourlyRebuild(prisma, {
     userId,
+    shouldStop,
     log: () => {
       // Silent inside the queue handler — the worker logs the totals.
     },
@@ -104,6 +109,7 @@ export async function runDenseIntradayHourlyRebuildForUser(
     hourlyRowsUpserted: summary.totals.hourlyRowsUpserted,
     dailyRowsRetired: summary.totals.dailyRowsRetired,
     daysSkippedNoTombstones: summary.totals.daysSkippedNoTombstones,
+    stoppedEarly: summary.stoppedEarly,
   };
 }
 

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

@@ -547,6 +547,8 @@ export async function registerRollupQueues(
     DENSE_INTRADAY_HOURLY_REBUILD_QUEUE,
     { localConcurrency: DENSE_INTRADAY_HOURLY_REBUILD_CONCURRENCY },
     async (jobs) => {
+      const shouldStop = jobBudget(jobs);
+      let stoppedEarly = false;
       let daysRebuiltTotal = 0;
       let hourlyRowsTotal = 0;
       let dailyRowsRetiredTotal = 0;
@@ -554,12 +556,17 @@ export async function registerRollupQueues(
       for (const job of jobs) {
         const { userId } = job.data;
         try {
+          const summary = await runDenseIntradayHourlyRebuildForUser(
+            userId,
+            shouldStop,
+          );
           const {
             daysRebuilt,
             hourlyRowsUpserted,
             dailyRowsRetired,
             daysSkippedNoTombstones,
-          } = await runDenseIntradayHourlyRebuildForUser(userId);
+          } = summary;
+          stoppedEarly ||= summary.stoppedEarly;
           daysRebuiltTotal += daysRebuilt;
           hourlyRowsTotal += hourlyRowsUpserted;
           dailyRowsRetiredTotal += dailyRowsRetired;
@@ -578,13 +585,25 @@ export async function registerRollupQueues(
           throw err;
         }
       }
-      return jobDone({
+      const did = {
         users: jobs.length,
+        stopped_early: stoppedEarly,
         days_rebuilt: daysRebuiltTotal,
         hourly_rows_upserted: hourlyRowsTotal,
         daily_rows_retired: dailyRowsRetiredTotal,
         days_skipped_no_tombstones: daysSkippedTotal,
-      });
+      };
+      // Out of budget: fail so pg-boss retries (the boot send allows three
+      // retries), and the retry starts at the first day not yet rebuilt. The
+      // queue is `exclusive`, so the job cannot hand itself on while active.
+      if (stoppedEarly) {
+        return jobFailed(
+          "stopped before every day was rebuilt; the retry resumes",
+          undefined,
+          did,
+        );
+      }
+      return jobDone(did);
     },
   );
 

src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts

@@ -463,4 +463,44 @@ describe("runDenseIntradayHourlyRebuild — rebuild flow", () => {
     expect(summary.totals.daysFailed).toBe(1);
     expect(summary.totals.daysRebuilt).toBe(1);
   });
+
+  it("stops before the next day once its budget is spent (#1031)", async () => {
+    const { mock, $transaction } = buildPrismaMock({
+      dailyByType: {
+        HEART_RATE_VARIABILITY: [
+          { id: "daily-1", externalId: HRV_DAILY_ID },
+          {
+            id: "daily-2",
+            externalId:
+              "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-02",
+          },
+        ],
+      },
+      tombstonesByType: {
+        HEART_RATE_VARIABILITY: [
+          tombstone(
+            "a",
+            40,
+            "2026-05-01T08:00:00.000Z",
+            "HEART_RATE_VARIABILITY",
+          ),
+          tombstone(
+            "b",
+            44,
+            "2026-05-02T08:00:00.000Z",
+            "HEART_RATE_VARIABILITY",
+          ),
+        ],
+      },
+    });
+    let asked = 0;
+    const summary = await runDenseIntradayHourlyRebuild(mock, {
+      log: () => {},
+      shouldStop: () => asked++ >= 1,
+    });
+
+    expect(summary.stoppedEarly).toBe(true);
+    expect(summary.totals.daysRebuilt).toBe(1);
+    expect($transaction).toHaveBeenCalledTimes(1);
+  });
 });

src/lib/measurements/dense-intraday-hourly-rebuild.ts

@@ -111,6 +111,8 @@ export interface DenseIntradayHourlyRebuildSummary {
     /** Days whose rebuild threw and was stepped over (retried next run). */
     daysFailed: number;
   };
+  /** `shouldStop` ended the pass before every candidate day was reached. */
+  stoppedEarly: boolean;
 }
 
 export interface DenseIntradayHourlyRebuildOptions {
@@ -126,6 +128,12 @@ export interface DenseIntradayHourlyRebuildOptions {
    * pass `0` to lift the bound.
    */
   retentionDays?: number;
+  /**
+   * Asked before each day. Returning `true` ends the pass cleanly; the
+   * summary reports `stoppedEarly`. A rebuilt day leaves the candidate set,
+   * so the next run starts at the first day this one did not reach.
+   */
+  shouldStop?: () => boolean;
 }
 
 /**
@@ -147,6 +155,7 @@ export async function runDenseIntradayHourlyRebuild(
 
   const summary: DenseIntradayHourlyRebuildSummary = {
     dryRun,
+    stoppedEarly: false,
     totals: {
       usersScanned: 0,
       daysRebuilt: 0,
@@ -160,7 +169,7 @@ export async function runDenseIntradayHourlyRebuild(
   const users = await loadConsolidationUsers(prismaClient, options.userId);
   summary.totals.usersScanned = users.length;
 
-  for (const user of users) {
+  walk: for (const user of users) {
     const tz = resolveUserTimezone(user.timezone);
 
     for (const type of DENSE_INTRADAY_RETENTION_TYPES) {
@@ -195,6 +204,10 @@ export async function runDenseIntradayHourlyRebuild(
       if (candidates.length === 0) continue;
 
       for (const candidate of candidates) {
+        if (options.shouldStop?.()) {
+          summary.stoppedEarly = true;
+          break walk;
+        }
         // Per-day failure boundary: one poisoned day is stepped over so the
         // walk keeps rebuilding every other user / type / day; the failed
         // day keeps its live daily row, so the discovery re-finds it.