diff --git a/src/lib/jobs/__tests__/drain-continuation.test.ts b/src/lib/jobs/__tests__/drain-continuation.test.ts new file mode 100644 index 000000000..c73e7ddeb --- /dev/null +++ b/src/lib/jobs/__tests__/drain-continuation.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from "vitest"; + +import { + DRAIN_MAX_CONTINUATIONS, + nextDrainContinuation, +} from "@/lib/jobs/drain-continuation"; + +// A nightly drain tick that ran out of budget hands the rest to a follow-up +// job, so a first-time backlog drains the same night (issue #1031). The chain +// must not loop on a pass that cannot move, and must end. + +describe("nextDrainContinuation", () => { + it("hands on a tick that stopped early after folding something", () => { + expect( + nextDrainContinuation({ + stoppedEarly: true, + daysFolded: 12, + continuation: 0, + }), + ).toBe(1); + expect( + nextDrainContinuation({ + stoppedEarly: true, + daysFolded: 1, + continuation: 5, + }), + ).toBe(6); + }); + + it("does not hand on a tick that finished", () => { + expect( + nextDrainContinuation({ + stoppedEarly: false, + daysFolded: 400, + continuation: 0, + }), + ).toBeNull(); + }); + + it("does not hand on a tick that stopped without progress", () => { + expect( + nextDrainContinuation({ + stoppedEarly: true, + daysFolded: 0, + continuation: 0, + }), + ).toBeNull(); + }); + + it("ends the chain at the cap", () => { + expect( + nextDrainContinuation({ + stoppedEarly: true, + daysFolded: 10, + continuation: DRAIN_MAX_CONTINUATIONS - 1, + }), + ).toBe(DRAIN_MAX_CONTINUATIONS); + expect( + nextDrainContinuation({ + stoppedEarly: true, + daysFolded: 10, + continuation: DRAIN_MAX_CONTINUATIONS, + }), + ).toBeNull(); + }); +}); diff --git a/src/lib/jobs/__tests__/drain-cumulative-queue.test.ts b/src/lib/jobs/__tests__/drain-cumulative-queue.test.ts index b904a30dd..4421c5ebb 100644 --- a/src/lib/jobs/__tests__/drain-cumulative-queue.test.ts +++ b/src/lib/jobs/__tests__/drain-cumulative-queue.test.ts @@ -52,7 +52,7 @@ describe("reminder-worker — drainPerSampleCumulative nightly schedule", () => it("registers a createAndWork handler against the drain queue", () => { expect(source).toMatch( - /createAndWork[\s\S]{0,200}DRAIN_CUMULATIVE_QUEUE[\s\S]{0,900}drainPerSampleCumulative/, + /createAndWork[\s\S]{0,200}DRAIN_CUMULATIVE_QUEUE[\s\S]{0,1100}drainPerSampleCumulative/, ); }); diff --git a/src/lib/jobs/__tests__/job-budget.test.ts b/src/lib/jobs/__tests__/job-budget.test.ts new file mode 100644 index 000000000..27eec6bf6 --- /dev/null +++ b/src/lib/jobs/__tests__/job-budget.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; + +import { JOB_BUDGET_SHARE, jobBudget } from "@/lib/jobs/job-budget"; + +// pg-boss declares a job dead at its expiry but does not stop the handler; +// the long passes stop themselves on this budget instead (issue #1031). + +function job(expireInSeconds: number, signal = new AbortController().signal) { + return { expireInSeconds, signal }; +} + +describe("jobBudget", () => { + it("stops at the budget share of the job's expiry, not at the expiry", () => { + let now = 1_000_000; + const shouldStop = jobBudget([job(900)], () => now); + + now += 900 * 1000 * JOB_BUDGET_SHARE - 1; + expect(shouldStop()).toBe(false); + now += 1; + expect(shouldStop()).toBe(true); + expect(JOB_BUDGET_SHARE).toBeLessThan(1); + }); + + it("stops as soon as pg-boss aborts the job, whatever the clock says", () => { + const controller = new AbortController(); + const shouldStop = jobBudget([job(900, controller.signal)], () => 0); + expect(shouldStop()).toBe(false); + controller.abort(); + expect(shouldStop()).toBe(true); + }); + + it("takes the tightest expiry of a batch", () => { + let now = 0; + const shouldStop = jobBudget([job(7200), job(60)], () => now); + now = 60 * 1000 * JOB_BUDGET_SHARE; + expect(shouldStop()).toBe(true); + }); + + it("never stops on the clock when there is no job to take an expiry from", () => { + let now = 0; + const shouldStop = jobBudget([], () => now); + now = Number.MAX_SAFE_INTEGER; + expect(shouldStop()).toBe(false); + }); +}); diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index bd86101b7..30eb96b14 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/src/lib/jobs/dense-intraday-retention.ts @@ -89,10 +89,13 @@ export interface DenseIntradayRetentionPayload { */ export async function runDenseIntradayRetentionForUser( userId: string, + shouldStop?: () => boolean, ): Promise<{ daysConsolidated: number; perSampleRowsSoftDeleted: number; derivedRestingRowsUpserted: number; + /** The job's budget ran out first; the next run continues from here. */ + stoppedEarly: boolean; }> { // Kill-switch: when an operator opts out, no-op so any already-queued // backlog completes cleanly (drains the queue) instead of running the fold. @@ -101,10 +104,12 @@ export async function runDenseIntradayRetentionForUser( daysConsolidated: 0, perSampleRowsSoftDeleted: 0, derivedRestingRowsUpserted: 0, + stoppedEarly: false, }; } const summary = await runDenseIntradayRetention(prisma, { userId, + shouldStop, log: () => { // Silent inside the queue handler — the worker logs the totals. }, @@ -121,6 +126,7 @@ export async function runDenseIntradayRetentionForUser( // days for proxy users, preserving the resting signal post-fold. derived_resting_rows_upserted: summary.totals.derivedRestingRowsUpserted, + stopped_early: summary.stoppedEarly, }, }, }); @@ -128,6 +134,7 @@ export async function runDenseIntradayRetentionForUser( daysConsolidated: summary.totals.daysConsolidated, perSampleRowsSoftDeleted: summary.totals.perSampleRowsSoftDeleted, derivedRestingRowsUpserted: summary.totals.derivedRestingRowsUpserted, + stoppedEarly: summary.stoppedEarly, }; } diff --git a/src/lib/jobs/drain-continuation.ts b/src/lib/jobs/drain-continuation.ts new file mode 100644 index 000000000..35947dfcc --- /dev/null +++ b/src/lib/jobs/drain-continuation.ts @@ -0,0 +1,33 @@ +/** + * When the nightly drain tick hands its remaining work to a follow-up job. + * + * The tick's three passes (cumulative drain, daily-mean consolidation, dense + * intra-day retention) stop at the job's time budget (`jobBudget`) instead of + * running into pg-boss's expiry. On a normal night that never happens. After + * a multi-year import it does, and without a follow-up the backlog would drain + * one budget per night. So a tick that stopped early sends itself again a + * minute later, and the chain is bounded twice: a follow-up is sent only when + * the run folded at least one day, so a pass that makes no progress cannot + * loop, and never more than `DRAIN_MAX_CONTINUATIONS` times in a row. + */ + +/** Follow-ups one nightly tick may chain: about four and a half hours of budget. */ +export const DRAIN_MAX_CONTINUATIONS = 24; + +/** Delay before a follow-up tick starts, in seconds. */ +export const DRAIN_CONTINUATION_DELAY_SECONDS = 60; + +/** + * The follow-up's continuation number, or `null` when no follow-up is due. + * `continuation` is the current job's own number (0 for the cron's job). + */ +export function nextDrainContinuation(input: { + stoppedEarly: boolean; + daysFolded: number; + continuation: number; +}): number | null { + if (!input.stoppedEarly) return null; + if (input.daysFolded <= 0) return null; + if (input.continuation >= DRAIN_MAX_CONTINUATIONS) return null; + return input.continuation + 1; +} diff --git a/src/lib/jobs/job-budget.ts b/src/lib/jobs/job-budget.ts new file mode 100644 index 000000000..8607004b1 --- /dev/null +++ b/src/lib/jobs/job-budget.ts @@ -0,0 +1,48 @@ +/** + * A time budget for a long pass, derived from the pg-boss job it runs under. + * + * pg-boss gives every job an expiry (`expireInSeconds`, 900 s unless the send + * says otherwise). When it passes, pg-boss marks the job failed with + * `job timed out`, retries it, and fires the job's abort signal. It does NOT + * stop the handler: a handler that never looks at the signal keeps writing + * after its job has been declared dead, and the retry then runs beside it over + * the same rows. On a large account that is how one pass that could not + * finish turned into two or three copies of itself contending for the same + * days. + * + * The passes that walk a whole history therefore stop on their own, between + * units of work, once a share of the expiry has gone by, and report that they + * stopped early. Their units commit one at a time and a finished unit drops + * out of the next scan, so the following run starts where this one stopped. + * The share leaves room for the unit in flight and for the handler's own + * bookkeeping before pg-boss's deadline. + */ +import type { Job } from "pg-boss"; + +/** Share of the job's expiry a pass may spend before it stops itself. */ +export const JOB_BUDGET_SHARE = 0.75; + +/** What the budget reads from a pg-boss job. */ +type BudgetedJob = Pick, "expireInSeconds" | "signal">; + +/** + * Build the `shouldStop` predicate for a batch of jobs handled together. + * + * True once the budget has elapsed (the tightest job's expiry decides) or once + * pg-boss has aborted any of the jobs. `now` is injectable for tests. + */ +export function jobBudget( + jobs: readonly BudgetedJob[], + now: () => number = Date.now, +): () => boolean { + const expirySeconds = Math.min( + ...jobs.map((job) => + Number.isFinite(job.expireInSeconds) && job.expireInSeconds > 0 + ? job.expireInSeconds + : Number.POSITIVE_INFINITY, + ), + ); + const deadline = now() + expirySeconds * 1000 * JOB_BUDGET_SHARE; + return () => + now() >= deadline || jobs.some((job) => job.signal?.aborted === true); +} diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 7d372364a..f57210e09 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -55,6 +55,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "connections_deleted", "considered", "context_surfaced", + "continuation", + "continued", "created", "daily_rows_retired", "days_consolidated", @@ -200,6 +202,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "skipped_outside_window", "slots_collapsed", "states_deleted", + "stopped_early", "stored", "subscription_repair_failed", "summaries_enqueued", diff --git a/src/lib/jobs/mean-consolidation.ts b/src/lib/jobs/mean-consolidation.ts index a001dc2b9..bbe5ab219 100644 --- a/src/lib/jobs/mean-consolidation.ts +++ b/src/lib/jobs/mean-consolidation.ts @@ -44,9 +44,15 @@ export interface MeanConsolidationPayload { */ export async function runMeanConsolidationForUser( userId: string, -): Promise<{ daysConsolidated: number; perSampleRowsSoftDeleted: number }> { + shouldStop?: () => boolean, +): Promise<{ + daysConsolidated: number; + perSampleRowsSoftDeleted: number; + stoppedEarly: boolean; +}> { const summary = await consolidateDailyMean(prisma, { userId, + shouldStop, cutoffHours: MEAN_CONSOLIDATION_CUTOFF_HOURS, log: () => { // Silent inside the queue handler — the worker logs the totals. @@ -59,12 +65,14 @@ export async function runMeanConsolidationForUser( days: summary.totals.daysConsolidated, per_sample_rows_soft_deleted: summary.totals.perSampleRowsSoftDeleted, daily_rows_upserted: summary.totals.dailyRowsUpserted, + stopped_early: summary.stoppedEarly, }, }, }); return { daysConsolidated: summary.totals.daysConsolidated, perSampleRowsSoftDeleted: summary.totals.perSampleRowsSoftDeleted, + stoppedEarly: summary.stoppedEarly, }; } diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index eca1bf4ff..3ddd5ea92 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -110,6 +110,11 @@ import { type ScheduleEntry, } from "./registrar-shared"; import { jobDone, jobFailed } from "@/lib/jobs/job-outcome"; +import { jobBudget } from "@/lib/jobs/job-budget"; +import { + DRAIN_CONTINUATION_DELAY_SECONDS, + nextDrainContinuation, +} from "@/lib/jobs/drain-continuation"; // v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative rows. // Collapses each user × cumulative-type × calendar-day bucket into one // `stats:…` row so the list view stops painting hundreds of step chunks per @@ -123,7 +128,12 @@ const DRAIN_CUMULATIVE_QUEUE = "drain-per-sample-cumulative"; const DRAIN_CUMULATIVE_CRON = "45 3 * * *"; interface DrainCumulativePayload { - triggeredAt: string; + triggeredAt?: string; + /** + * How many times this tick has already handed itself on. Absent on the + * cron's own job. See `nextDrainContinuation`. + */ + continuation?: number; } const allQueues = [ @@ -437,12 +447,15 @@ export async function registerRollupQueues( MEAN_CONSOLIDATION_QUEUE, { localConcurrency: MEAN_CONSOLIDATION_CONCURRENCY }, async (jobs) => { + const shouldStop = jobBudget(jobs); let daysTotal = 0; let perSampleRowsTotal = 0; + let stoppedEarly = false; for (const job of jobs) { const { userId } = job.data; - const { daysConsolidated, perSampleRowsSoftDeleted } = - await runMeanConsolidationForUser(userId); + const summary = await runMeanConsolidationForUser(userId, shouldStop); + const { daysConsolidated, perSampleRowsSoftDeleted } = summary; + stoppedEarly ||= summary.stoppedEarly; daysTotal += daysConsolidated; perSampleRowsTotal += perSampleRowsSoftDeleted; workerLog( @@ -450,10 +463,14 @@ export async function registerRollupQueues( `[mean-consolidation] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted}`, ); } + // Out of budget is a completed run, not a failed one: the folded days + // are committed, and the nightly tick carries on from the first day + // this run did not reach. return jobDone({ users: jobs.length, days_consolidated: daysTotal, per_sample_rows_soft_deleted: perSampleRowsTotal, + stopped_early: stoppedEarly, }); }, ); @@ -470,17 +487,24 @@ export async function registerRollupQueues( DENSE_INTRADAY_RETENTION_QUEUE, { localConcurrency: DENSE_INTRADAY_RETENTION_CONCURRENCY }, async (jobs) => { + const shouldStop = jobBudget(jobs); let daysTotal = 0; let perSampleRowsTotal = 0; let derivedRestingRowsTotal = 0; + let stoppedEarly = false; for (const job of jobs) { const { userId } = job.data; try { + const summary = await runDenseIntradayRetentionForUser( + userId, + shouldStop, + ); const { daysConsolidated, perSampleRowsSoftDeleted, derivedRestingRowsUpserted, - } = await runDenseIntradayRetentionForUser(userId); + } = summary; + stoppedEarly ||= summary.stoppedEarly; daysTotal += daysConsolidated; perSampleRowsTotal += perSampleRowsSoftDeleted; derivedRestingRowsTotal += derivedRestingRowsUpserted; @@ -503,6 +527,10 @@ export async function registerRollupQueues( days_consolidated: daysTotal, per_sample_rows_soft_deleted: perSampleRowsTotal, derived_resting_rows_upserted: derivedRestingRowsTotal, + // Out of budget completes the job: the folded days are committed and + // the nightly tick, which hands itself on while work remains, + // continues from the first day this run did not reach. + stopped_early: stoppedEarly, }); }, ); @@ -679,6 +707,9 @@ export async function registerRollupQueues( DRAIN_CUMULATIVE_QUEUE, { localConcurrency: 1 }, async (jobs) => { + // One budget shared by all three passes, run in turn. + const shouldStop = jobBudget(jobs); + let stoppedEarly = false; // The three drains below are independent passes sharing one nightly // tick, so one failing pass must not stop the other two from running. // The first failure is held back and reported once the tick is over — @@ -699,8 +730,10 @@ export async function registerRollupQueues( const summary = await drainPerSampleCumulative(getWorkerPrisma(), { dryRun: false, cutoffHours: DRAIN_CUMULATIVE_CUTOFF_HOURS, + shouldStop, log: (line) => workerLog("info", line), }); + stoppedEarly ||= summary.stoppedEarly; bucketsCollapsed += summary.totals.bucketsCollapsed; perSampleRowsDeleted += summary.totals.perSampleRowsDeleted; workerLog( @@ -729,8 +762,10 @@ export async function registerRollupQueues( const meanSummary = await consolidateDailyMean(getWorkerPrisma(), { dryRun: false, cutoffHours: MEAN_CONSOLIDATION_CUTOFF_HOURS, + shouldStop, log: (line) => workerLog("info", line), }); + stoppedEarly ||= meanSummary.stoppedEarly; meanDaysConsolidated += meanSummary.totals.daysConsolidated; meanRowsSoftDeleted += meanSummary.totals.perSampleRowsSoftDeleted; workerLog( @@ -769,9 +804,11 @@ export async function registerRollupQueues( getWorkerPrisma(), { dryRun: false, + shouldStop, log: (line) => workerLog("info", line), }, ); + stoppedEarly ||= denseSummary.stoppedEarly; denseDaysConsolidated += denseSummary.totals.daysConsolidated; denseRowsSoftDeleted += denseSummary.totals.perSampleRowsSoftDeleted; @@ -790,7 +827,41 @@ export async function registerRollupQueues( } } + // Hand the remaining work to a follow-up job when this one ran out of + // budget (bounded; see `nextDrainContinuation`). + const continuation = Math.max( + 0, + ...jobs.map((job) => job.data.continuation ?? 0), + ); + const next = nextDrainContinuation({ + stoppedEarly, + daysFolded: + bucketsCollapsed + meanDaysConsolidated + denseDaysConsolidated, + continuation, + }); + let continued = false; + if (next !== null) { + try { + const followUp: DrainCumulativePayload = { + triggeredAt: new Date().toISOString(), + continuation: next, + }; + continued = + (await boss.send(DRAIN_CUMULATIVE_QUEUE, followUp, { + ...cronIsTheRetry, + startAfter: DRAIN_CONTINUATION_DELAY_SECONDS, + })) !== null; + } catch (err) { + // The next nightly tick picks the work up regardless; a lost + // follow-up only delays it. + workerLog("error", "[drain-cumulative] follow-up send failed", err); + } + } + const did = { + stopped_early: stoppedEarly, + continuation, + continued, passes_failed: passesFailed, buckets_collapsed: bucketsCollapsed, per_sample_rows_deleted: perSampleRowsDeleted,