diff --git a/src/lib/jobs/dense-intraday-hourly-rebuild.ts b/src/lib/jobs/dense-intraday-hourly-rebuild.ts index 9f451d577..a90de1ea7 100644 --- a/src/lib/jobs/dense-intraday-hourly-rebuild.ts +++ b/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, }; } diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index 3ddd5ea92..ef38e1ddd 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/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); }, ); diff --git a/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts b/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts index c69715be9..9a149a2f4 100644 --- a/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts +++ b/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); + }); }); diff --git a/src/lib/measurements/dense-intraday-hourly-rebuild.ts b/src/lib/measurements/dense-intraday-hourly-rebuild.ts index 15e116724..a5cbb7d78 100644 --- a/src/lib/measurements/dense-intraday-hourly-rebuild.ts +++ b/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.