fix(jobs): stop the long drains on a budget instead of at the expiry

pg-boss marks a job "job timed out" at its expiry and fires its abort
signal, but does not stop the handler. A drain that could not finish kept
writing after its job was declared dead, and the retry then ran beside it
over the same days.

The nightly drain tick and the per-user dense-retention and daily-mean
jobs now stop between days once 75 % of the job's expiry has passed, or
as soon as pg-boss aborts the job, and complete with `stopped_early`
instead of failing. What they folded is committed. A nightly tick that
stopped early after folding something sends itself again a minute later,
up to 24 times, so a first-time backlog after a large import drains the
same night; a run that folded nothing does not hand on, so a pass that
cannot move does not loop.

The expiries themselves stay at their defaults: with the passes bounded
per day and budgeted, a longer limit would only delay the signal.

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();
+  });
+});

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/,
     );
   });
 

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

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,
   };
 }
 

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;
+}

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<Job<unknown>, "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);
+}

src/lib/jobs/job-outcome.ts

@@ -55,6 +55,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = 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<string> = new Set([
   "skipped_outside_window",
   "slots_collapsed",
   "states_deleted",
+  "stopped_early",
   "stored",
   "subscription_repair_failed",
   "summaries_enqueued",

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,
   };
 }
 

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,