fix(jobs): keep long passes inside their expiry and alone

pg-boss gives a job fifteen minutes unless the send says otherwise, then fails it and retries it while the handler keeps running. Several passes can legitimately need longer and had no protection at all: the seven nightly status passes and the pre-generation (a provider call per account), the period-narrative warm, the record-detection cron (every account's history, every thirty minutes), the key rotation (every encrypted column), the content-index backfill (up to 200 transcriptions) and the on-host backup. Each of them now carries an explicit expiry on every send and schedule, stops between units once three quarters of it is spent (jobBudget), and runs under a session advisory lock keyed on queue and identity (lockedPass), so the retry pg-boss starts after the expiry does nothing beside a run that is still going. The lock is the one the full-history imports already held, moved into job-lock.ts. Where a pass stops early it says so (stopped_early) and the next run continues: the status passes and the narrative warm leave the rest to the next night, the pre-generation and the narratives walk the stalest accounts first, the on-host backup now backs up the oldest copies first, the key rotation reports a partial run in its audit row and skips rotated rows when run again, and the dose-history import fails the delivery so its retry resumes from the committed chunk. The off-host backup, which already stopped on its budget, gets a four-hour expiry and the lock.
2026-11-08 12:32:43 +00:00 · 2026-10-01 09:23:45 +02:00
1 parent 0f31fd199b
commit ac3fdfccc3
@@ -33,6 +33,7 @@ import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import {
  ENCRYPTION_KEY_ROTATE_QUEUE,
  ENCRYPTION_KEY_ROTATE_SINGLETON,
+  ENCRYPTION_KEY_ROTATE_EXPIRE_SECONDS,
 } from "@/lib/jobs/encryption-key-rotate";
 import { getActiveKeyId } from "@/lib/crypto";
 import { NextRequest } from "next/server";
@@ -57,6 +58,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
      singletonKey: ENCRYPTION_KEY_ROTATE_SINGLETON,
      retryLimit: 2,
      retryDelay: 60,
+      expireInSeconds: ENCRYPTION_KEY_ROTATE_EXPIRE_SECONDS,
    },
  );

@@ -390,19 +390,29 @@ export interface RotationResult {
  dropped: number;
 }

-/** Re-encrypt one column's stale rows to the active key. */
+/** Thrown out of a column walk when the caller's time budget is spent. */
+class RotationStopped extends Error {}
+
+/**
+ * Re-encrypt one column's stale rows to the active key. `shouldStop` is the
+ * caller's time budget, checked between rows; a stopped column reports
+ * `stopped` and the rows it did not reach stay on their old key for the next
+ * run, which skips every row already rotated.
+ */
 export async function rotateColumn(
  client: CorpusClient,
  col: EncryptedColumn,
-): Promise<RotationResult> {
+  shouldStop: () => boolean = () => false,
+): Promise<RotationResult & { stopped: boolean }> {
  const delegate = getDelegate(client, col.model);
-  const result: RotationResult = {
+  const result: RotationResult & { stopped: boolean } = {
    model: col.model,
    field: col.field,
    scanned: 0,
    rotated: 0,
    errors: 0,
    dropped: 0,
+    stopped: false,
  };

  /**
@@ -432,6 +442,10 @@ export async function rotateColumn(
    // (`convertSingleValueBackup`). A row converted, replaced or deleted since
    // it was listed is gone, not an error.
    for await (const id of singleValueIds(delegate)) {
+      if (shouldStop()) {
+        result.stopped = true;
+        break;
+      }
      result.scanned += 1;
      try {
        const outcome = await convertSingleValueBackup(
@@ -452,31 +466,37 @@ export async function rotateColumn(
  // Idempotent — rows already on the active key are skipped, so an
  // interrupted run resumes cleanly on the next invocation. `scanned` counts
  // the rows that hold ciphertext, whatever the column.
-  await walkColumn(delegate, col, async ({ id, value, codec }) => {
-    result.scanned += 1;
-    if (walkedKeyId(value, codec, col.kind) === getActiveKeyId()) return;
-    try {
-      // The plaintext is never inspected, only re-sealed, so every envelope
-      // a stored value can legitimately carry survives rotation untouched.
-      const next =
-        codec !== null
-          ? reencryptBlob(value as Uint8Array, codec, col.aad)
-          : fromCiphertext(
-              encrypt(decrypt(toCiphertext(value, col.kind)!)),
-              col.kind,
-            );
-      await delegate.update({
-        where: { [pkField(col)]: id },
-        data: { [col.field]: next },
-      });
-      result.rotated += 1;
-    } catch (err) {
-      // Deleted between the read and the write (a backup replacing its
-      // pieces, an account going): nothing is left to rotate.
-      if (isRowGone(err)) return;
-      await onUnreadable(id);
-    }
-  });
+  try {
+    await walkColumn(delegate, col, async ({ id, value, codec }) => {
+      if (shouldStop()) throw new RotationStopped();
+      result.scanned += 1;
+      if (walkedKeyId(value, codec, col.kind) === getActiveKeyId()) return;
+      try {
+        // The plaintext is never inspected, only re-sealed, so every envelope
+        // a stored value can legitimately carry survives rotation untouched.
+        const next =
+          codec !== null
+            ? reencryptBlob(value as Uint8Array, codec, col.aad)
+            : fromCiphertext(
+                encrypt(decrypt(toCiphertext(value, col.kind)!)),
+                col.kind,
+              );
+        await delegate.update({
+          where: { [pkField(col)]: id },
+          data: { [col.field]: next },
+        });
+        result.rotated += 1;
+      } catch (err) {
+        // Deleted between the read and the write (a backup replacing its
+        // pieces, an account going): nothing is left to rotate.
+        if (isRowGone(err)) return;
+        await onUnreadable(id);
+      }
+    });
+  } catch (err) {
+    if (!(err instanceof RotationStopped)) throw err;
+    result.stopped = true;
+  }
  return result;
 }

@@ -488,16 +508,31 @@ export interface CorpusRotation {
  totalErrors: number;
  /** Unreadable rows deleted from `disposable` columns. */
  totalDropped: number;
+  /**
+   * The time budget ran out before every column was walked. Running again
+   * finishes the job: rows already on the active key are skipped.
+   */
+  stoppedEarly: boolean;
 }

-/** Re-encrypt the whole corpus to the active key. Idempotent + active-key-only. */
+/**
+ * Re-encrypt the whole corpus to the active key. Idempotent + active-key-only.
+ * `shouldStop` is the caller's time budget (`jobBudget` in the worker).
+ */
 export async function rotateCorpus(
  client: CorpusClient,
+  shouldStop: () => boolean = () => false,
 ): Promise<CorpusRotation> {
  const activeKeyId = getActiveKeyId();
  const results: RotationResult[] = [];
+  let stoppedEarly = false;
  for (const col of ENCRYPTED_COLUMNS) {
-    results.push(await rotateColumn(client, col));
+    const { stopped, ...result } = await rotateColumn(client, col, shouldStop);
+    results.push(result);
+    if (stopped) {
+      stoppedEarly = true;
+      break;
+    }
  }
  let totalScanned = 0;
  let totalRotated = 0;
@@ -516,5 +551,6 @@ export async function rotateCorpus(
    totalRotated,
    totalErrors,
    totalDropped,
+    stoppedEarly,
  };
 }
@@ -207,6 +207,31 @@ describe("findPregenerateCandidates", () => {
  });
 });

+describe("runInsightPregenerate — the job's time budget", () => {
+  it("stops between candidates once the budget is spent and says so", async () => {
+    capabilities = {
+      briefing: unavailable("user_disabled"),
+      statusText: unavailable("user_disabled"),
+    };
+    const { prisma } = makePrisma([
+      { id: "u1", locale: "de" },
+      { id: "u2", locale: "de" },
+      { id: "u3", locale: "de" },
+    ]);
+    let checks = 0;
+    const result = await runInsightPregenerate(prisma as never, {
+      generate: vi.fn(),
+      statusGenerators: [],
+      warmGenericMetrics: vi.fn().mockResolvedValue(0),
+      // Spent after the first candidate.
+      shouldStop: () => checks++ >= 1,
+    });
+
+    expect(result).toMatchObject({ total: 3, skipped: 1, stoppedEarly: true });
+    expect(aiCapabilityForJob).toHaveBeenCalledTimes(2);
+  });
+});
+
 describe("runInsightPregenerate — per-user capabilities", () => {
  it("skips a user with neither capability before any budget write or generation", async () => {
    capabilities = {
@@ -1337,7 +1362,7 @@ describe("queue registration", () => {

  it("schedules the cron in the schedules table (with retry policy)", () => {
    expect(workerSrc).toMatch(
-      /\[\s*INSIGHT_PREGENERATE_QUEUE\s*,\s*INSIGHT_PREGENERATE_CRON\s*,\s*insightRetryOptions\s*\]/,
+      /\[\s*INSIGHT_PREGENERATE_QUEUE\s*,\s*INSIGHT_PREGENERATE_CRON\s*,\s*\{\s*\.\.\.insightRetryOptions\s*,\s*expireInSeconds:\s*INSIGHT_PREGENERATE_EXPIRE_SECONDS\s*,?\s*\}\s*,?\s*\]/,
    );
  });

@@ -0,0 +1,74 @@
+/**
+ * `lockedPass`: each job runs under the lock `job:<queue>:<identity>`, and a
+ * delivery that finds the lock held does no work and succeeds.
+ *
+ * The lock itself (a session advisory lock on a connection of its own) is
+ * exercised against Postgres in
+ * `tests/integration/integration-backfill-lane-expiry.test.ts`; this pins the
+ * wrapper around it.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { Job } from "pg-boss";
+
+const held = vi.hoisted(() => new Set<string>());
+const keys = vi.hoisted(() => [] as string[]);
+
+vi.mock("@/lib/jobs/job-lock", () => ({
+  withJobLock: async <T>(key: string, run: () => Promise<T>) => {
+    keys.push(key);
+    if (held.has(key)) return { ran: false };
+    held.add(key);
+    try {
+      return { ran: true, result: await run() };
+    } finally {
+      held.delete(key);
+    }
+  },
+}));
+
+import { jobDone, jobFailed } from "@/lib/jobs/job-outcome";
+import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+
+type Payload = { userId?: string };
+const job = (data: Payload) => ({ id: "j", data }) as Job<Payload>;
+
+beforeEach(() => {
+  held.clear();
+  keys.length = 0;
+});
+
+describe("lockedPass", () => {
+  it("runs each job under its queue and identity", async () => {
+    const handler = vi.fn(async () => jobDone({ generated: 1 }));
+    const bound = lockedPass(
+      "q",
+      (j: Job<Payload>) =>
+        j.data.userId ? `user:${j.data.userId}` : WHOLE_PASS,
+      handler,
+    );
+    const outcome = await bound([job({ userId: "a" })]);
+    expect(outcome).toEqual(jobDone({ generated: 1 }));
+    await bound([job({})]);
+    expect(keys).toEqual(["job:q:user:a", "job:q:pass"]);
+  });
+
+  it("does nothing beside a run that still holds the lock", async () => {
+    held.add("job:q:pass");
+    const handler = vi.fn(async () => jobDone());
+    const outcome = await lockedPass("q", () => WHOLE_PASS, handler)([job({})]);
+    expect(handler).not.toHaveBeenCalled();
+    expect(outcome).toEqual(jobDone({ already_running: 1 }));
+  });
+
+  it("passes a failed outcome through", async () => {
+    const failed = jobFailed("stopped early", undefined, {
+      stopped_early: true,
+    });
+    const outcome = await lockedPass(
+      "q",
+      () => WHOLE_PASS,
+      async () => failed,
+    )([job({})]);
+    expect(outcome).toBe(failed);
+  });
+});
@@ -250,7 +250,7 @@ describe("queue registration", () => {

  it("schedules the cron in the schedules table (with retry policy)", () => {
    expect(workerSrc).toMatch(
-      /\[\s*PERIOD_NARRATIVE_QUEUE\s*,\s*PERIOD_NARRATIVE_CRON\s*,\s*insightRetryOptions\s*\]/,
+      /\[\s*PERIOD_NARRATIVE_QUEUE\s*,\s*PERIOD_NARRATIVE_CRON\s*,\s*\{\s*\.\.\.insightRetryOptions\s*,\s*expireInSeconds:\s*PERIOD_NARRATIVE_EXPIRE_SECONDS\s*,?\s*\}\s*,?\s*\]/,
    );
  });

@@ -190,9 +190,9 @@ describe("mood-status queue registration (dead-queue guard)", () => {
    expect(match![1]).toMatch(/\bMOOD_STATUS_QUEUE\b/);
  });

-  it("schedules the 02:30 cron with the insight retry policy", () => {
+  it("schedules the 02:30 cron with the nightly pass options (retries and expiry)", () => {
    expect(workerSrc).toMatch(
-      /\[\s*MOOD_STATUS_QUEUE\s*,\s*MOOD_STATUS_CRON\s*,\s*insightRetryOptions\s*\]/,
+      /\[\s*MOOD_STATUS_QUEUE\s*,\s*MOOD_STATUS_CRON\s*,\s*nightlyInsightPassOptions\s*\]/,
    );
    expect(workerSrc).toMatch(/MOOD_STATUS_CRON\s*=\s*"30 2 \* \* \*"/);
  });
@@ -811,7 +811,12 @@ export async function enqueueBackupRestoreSweepAtBoot(
    await boss.send(
      BACKUP_RESTORE_QUEUE,
      { sweep: true } satisfies BackupRestorePayload,
-      { startAfter: Math.ceil(BACKUP_RESTORE_STALE_AFTER_MS / 1000) + 15 },
+      {
+        startAfter: Math.ceil(BACKUP_RESTORE_STALE_AFTER_MS / 1000) + 15,
+        // The sweep is short, but it rides a queue whose every send carries
+        // the restore's expiry, so no job on it is cut at the default.
+        expireInSeconds: BACKUP_RESTORE_EXPIRE_SECONDS,
+      },
    );
  } catch {
    // The admission check fails an abandoned job the next time the account
@@ -64,6 +64,14 @@ export const CONTENT_INDEX_BACKFILL_CONCURRENCY = 1;
 /** Discovery page size (id-only; blobs are loaded one at a time). */
 const PAGE_SIZE = 50;

+/**
+ * The job's expiry: three hours. One job transcribes up to 200 documents with
+ * a provider call each, which outlasted pg-boss's fifteen-minute default and
+ * was retried beside itself. The run stops at three quarters of this
+ * (`jobBudget`) and holds a lock per account (`lockedPass`).
+ */
+export const CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS = 3 * 60 * 60;
+
 /** Max documents one job indexes before yielding (bounds spend + runtime). */
 const MAX_DOCS_PER_RUN = 200;

@@ -97,6 +105,11 @@ export interface ContentIndexBackfillSummary {
  failed: number;
  reason:
    "ok" | "no-provider" | "no-consent" | "budget-reached" | "unavailable";
+  /**
+   * The job's time budget ran out before the walk did. The documents not
+   * reached stay un-indexed for the next run.
+   */
+  stoppedEarly?: boolean;
 }

 /**
@@ -158,6 +171,11 @@ async function retokeniseStaleRows(
 */
 export async function runContentIndexBackfillForUser(
  userId: string,
+  /**
+   * The job's time budget (`jobBudget`), checked before each document: up to
+   * 200 transcriptions, one provider call each, can outlast the job.
+   */
+  shouldStop: () => boolean = () => false,
 ): Promise<ContentIndexBackfillSummary> {
  // Stale-tokeniser rows first: local work that owes nothing to the
  // provider, consent, or budget gates below, so a user without a usable
@@ -222,6 +240,7 @@ export async function runContentIndexBackfillForUser(
  let failed = stale.failed;
  let reason: ContentIndexBackfillSummary["reason"] = "ok";
  let cursor: string | null = null;
+  let stoppedEarly = false;

  outer: for (;;) {
    if (indexed >= MAX_DOCS_PER_RUN) break;
@@ -244,6 +263,10 @@ export async function runContentIndexBackfillForUser(

    for (const { id } of rows) {
      if (indexed >= MAX_DOCS_PER_RUN) break outer;
+      if (shouldStop()) {
+        stoppedEarly = true;
+        break outer;
+      }

      const dateKey = buildDateKey();
      const reservation = await reserveBudget(
@@ -325,9 +348,23 @@ export async function runContentIndexBackfillForUser(

  annotate({
    action: { name: "documents.contentIndex.backfill" },
-    meta: { indexed, retokenised: stale.retokenised, skipped, failed, reason },
+    meta: {
+      indexed,
+      retokenised: stale.retokenised,
+      skipped,
+      failed,
+      reason,
+      stopped_early: stoppedEarly,
+    },
  });
-  return { indexed, retokenised: stale.retokenised, skipped, failed, reason };
+  return {
+    indexed,
+    retokenised: stale.retokenised,
+    skipped,
+    failed,
+    reason,
+    ...(stoppedEarly ? { stoppedEarly } : {}),
+  };
 }

 /**
@@ -344,6 +381,7 @@ export async function enqueueContentIndexBackfill(
    enqueuedAt: new Date().toISOString(),
  };
  const jobId = await boss.send(CONTENT_INDEX_BACKFILL_QUEUE, payload, {
+    expireInSeconds: CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS,
    retryLimit: 3,
    retryDelay: 60,
    retryBackoff: true,
@@ -25,6 +25,7 @@
 import { type Job } from "pg-boss";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { auditLog } from "@/lib/auth/audit";
 import {
  rotateCorpus,
@@ -37,27 +38,40 @@ export const ENCRYPTION_KEY_ROTATE_CONCURRENCY = 1;
 /** Fixed key so duplicate admin triggers coalesce into one queued run. */
 export const ENCRYPTION_KEY_ROTATE_SINGLETON = "encryption-key-rotate";

+/**
+ * The job's expiry: twelve hours. The pass re-encrypts every encrypted column
+ * on the instance and outlasted pg-boss's fifteen-minute default, which
+ * retried it beside itself. The pass stops at three quarters of this
+ * (`jobBudget`) and holds a lock (`lockedPass`); the admin route's send
+ * carries it.
+ */
+export const ENCRYPTION_KEY_ROTATE_EXPIRE_SECONDS = 12 * 60 * 60;
+
 export interface EncryptionKeyRotatePayload {
  /** The admin who triggered the run (for the audit trail). */
  requestedByUserId?: string;
  enqueuedAt?: string;
 }

-export async function runEncryptionKeyRotation(): Promise<{
+export async function runEncryptionKeyRotation(
+  shouldStop: () => boolean = () => false,
+): Promise<{
  activeKeyId: string;
  totalScanned: number;
  totalRotated: number;
  totalErrors: number;
  totalDropped: number;
+  stoppedEarly: boolean;
 }> {
  const prisma = getWorkerPrisma();
-  const out = await rotateCorpus(prisma as unknown as CorpusClient);
+  const out = await rotateCorpus(prisma as unknown as CorpusClient, shouldStop);
  return {
    activeKeyId: out.activeKeyId,
    totalScanned: out.totalScanned,
    totalRotated: out.totalRotated,
    totalErrors: out.totalErrors,
    totalDropped: out.totalDropped,
+    stoppedEarly: out.stoppedEarly,
  };
 }

@@ -67,7 +81,10 @@ export async function handleEncryptionKeyRotate(
  return withBackgroundEvent("job.encryption_key_rotate", async (evt) => {
    const requestedBy = jobs[0]?.data?.requestedByUserId ?? null;
    try {
-      const result = await runEncryptionKeyRotation();
+      // A whole-corpus pass: it stops between rows once the job's budget is
+      // spent, and says so in the audit row, so the operator runs it again
+      // to finish (rows already rotated are skipped).
+      const result = await runEncryptionKeyRotation(jobBudget(jobs));
      evt.addMeta("rotate_active_key_id", result.activeKeyId);
      evt.addMeta("rotate_scanned", result.totalScanned);
      evt.addMeta("rotate_rotated", result.totalRotated);
@@ -19,6 +19,15 @@ import { annotate } from "@/lib/logging/context";

 export const INSIGHT_PREGENERATE_QUEUE = "insight-pregenerate";

+/**
+ * The job's expiry: three hours. The scheduled pass walks up to 200 accounts
+ * with a comprehensive generation each, which can outlast pg-boss's fifteen
+ * minute default; past it pg-boss retried the job beside the pass still
+ * running. The pass now stops itself at three quarters of this (`jobBudget`)
+ * and holds a lock (`lockedPass`), and every send and the schedule carry it.
+ */
+export const INSIGHT_PREGENERATE_EXPIRE_SECONDS = 3 * 60 * 60;
+
 export interface InsightPregeneratePayload {
  /** Set on the nightly scheduled tick (informational; the handler keys
   * its batch behaviour off the ABSENCE of `userId`/`force`). */
@@ -73,6 +82,7 @@ export async function enqueueForceWarm(payload: {
      } satisfies InsightPregeneratePayload,
      {
        singletonKey: `force:${payload.userId}`,
+        expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
        // De-dupe within a 6-minute window. v1.16.8 — widened from 120 s:
        // the client's revalidation poll runs up to ~250 s after a stale
        // page-open, so a 2-minute singleton let the tail of one poll
@@ -148,6 +158,7 @@ export async function enqueuePregenerateFailureRetry(payload: {
        // warm can never swallow the scheduled retry (and vice versa);
        // one retry per user per window.
        singletonKey: `retry:${payload.userId}`,
+        expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
        singletonSeconds: 3600,
        // A worker crash mid-warm re-runs the job; a warm that completes
        // with a failed comprehensive does NOT throw, so this cannot loop.
@@ -277,6 +277,8 @@ export interface PregenerateRunResult {
  skipped: number;
  failed: number;
  budgetBlocked: number;
+  /** The job's time budget ran out before every candidate was reached. */
+  stoppedEarly: boolean;
  /**
   * Count of per-metric assessment caches written warm across the whole
   * run. The comprehensive generator evicts the seven `*-status`
@@ -575,6 +577,12 @@ export async function runInsightPregenerate(
      userId: string;
      locale: SupportedLocale;
    }) => Promise<void>;
+    /**
+     * The job's time budget (`jobBudget`). Checked between candidates: a pass
+     * of 200 provider calls can outlast its job, and the candidates it does
+     * not reach are the stalest again tomorrow, so they lead the next pass.
+     */
+    shouldStop?: () => boolean;
  } = {},
 ): Promise<PregenerateRunResult> {
  const now = options.now ?? new Date();
@@ -606,6 +614,7 @@ export async function runInsightPregenerate(
    skipped: 0,
    failed: 0,
    budgetBlocked: 0,
+    stoppedEarly: false,
    assessmentsWarmed: 0,
    metricAssessmentsWarmed: 0,
  };
@@ -619,6 +628,10 @@ export async function runInsightPregenerate(
  result.total = candidates.length;

  for (const candidate of candidates) {
+    if (options.shouldStop?.()) {
+      result.stoppedEarly = true;
+      break;
+    }
    // v1.18.11 P3 — one feature-cache scope per candidate, so the
    // comprehensive briefing's bounded feature read is computed once and any
    // sibling consumer in this candidate's warm reuses it. Scoped per user so
@@ -819,6 +832,7 @@ export async function runInsightPregenerate(
      skipped: result.skipped,
      failed: result.failed,
      budget_blocked: result.budgetBlocked,
+      stopped_early: result.stoppedEarly,
      assessments_warmed: result.assessmentsWarmed,
      metric_assessments_warmed: result.metricAssessmentsWarmed,
    },
@@ -11,8 +11,9 @@
 * started beside it in the same worker, so the lane's "one at a time" did not
 * hold for the one case it exists for, an import too long for its expiry.
 *
- * So each run holds a session-level advisory lock keyed on the job's identity
- * (kind, provider where there is one, account) for as long as it runs. A
+ * So each run holds a session-level advisory lock (`withJobLock`) keyed on the
+ * job's identity (kind, provider where there is one, account) for as long as
+ * it runs. A
 * delivery that finds the lock taken does no work: the run that holds it is
 * still importing, and it either stamps its completion marker or leaves the
 * account for the next boot's discovery. The lock lives on a connection of its
@@ -24,47 +25,26 @@
 * One extra connection per running import; the lane admits one import per
 * process at a time, so that is one connection.
 */
-import { Client } from "pg";
+import { withJobLock, type GuardedRun } from "./job-lock";

 /** The lock key for one admission job's identity. */
 function integrationBackfillLockKey(identity: string): string {
  return `integration-backfill:${identity}`;
 }

-/** What a guarded run came to. */
-type GuardedRun<T> = { ran: true; result: T } | { ran: false };
-
 /**
 * Run `run` while holding the import lock for `identity`. Resolves
 * `{ ran: false }` at once, without calling `run`, when another run holds it.
 * Rejects exactly as `run` does otherwise.
 */
-export async function withIntegrationBackfillLock<T>(
+export function withIntegrationBackfillLock<T>(
  identity: string,
  run: () => Promise<T>,
  connectionString: string | undefined = process.env.DATABASE_URL,
 ): Promise<GuardedRun<T>> {
-  const client = new Client({ connectionString, keepAlive: true });
-  // An idle client whose server goes away emits `error`; unhandled, that
-  // takes the whole worker down. The run itself notices the database is
-  // gone through its own queries.
-  client.on("error", () => {});
-  await client.connect();
-  try {
-    const key = integrationBackfillLockKey(identity);
-    const { rows } = await client.query<{ held: boolean }>(
-      "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held",
-      [key],
-    );
-    if (!rows[0]?.held) return { ran: false };
-    try {
-      return { ran: true, result: await run() };
-    } finally {
-      await client
-        .query("SELECT pg_advisory_unlock(hashtextextended($1, 0))", [key])
-        .catch(() => {});
-    }
-  } finally {
-    await client.end().catch(() => {});
-  }
+  return withJobLock(
+    integrationBackfillLockKey(identity),
+    run,
+    connectionString,
+  );
 }
@@ -0,0 +1,58 @@
+/**
+ * A session-level advisory lock held for the length of one run, across
+ * processes.
+ *
+ * pg-boss holds a job's "one at a time" only while the job is inside its
+ * expiry. Once the expiry passes it fails the job, fires its abort signal and
+ * retries it, and it does not stop the handler: measured against pg-boss 12.34
+ * (`tests/integration/integration-backfill-lane-expiry.test.ts`), the retry of
+ * the same job started beside the run that was still going. No queue policy
+ * helps, because the policies constrain job rows and the expired row is no
+ * longer active. Only something the running handler holds can, so each long
+ * run holds an advisory lock keyed on its identity, and a delivery that finds
+ * the lock taken does no work.
+ *
+ * The lock lives on a connection of its own, outside the Prisma pool, because
+ * a session lock has to stay on one connection for the whole run and a pooled
+ * one is handed back between statements. Ending the connection releases the
+ * lock, and so does the process dying, so a crash never leaves a key locked.
+ * One extra connection per running locked job.
+ */
+import { Client } from "pg";
+
+/** What a guarded run came to. */
+export type GuardedRun<T> = { ran: true; result: T } | { ran: false };
+
+/**
+ * Run `run` while holding the advisory lock for `key`. Resolves
+ * `{ ran: false }` at once, without calling `run`, when another run holds it.
+ * Rejects exactly as `run` does otherwise.
+ */
+export async function withJobLock<T>(
+  key: string,
+  run: () => Promise<T>,
+  connectionString: string | undefined = process.env.DATABASE_URL,
+): Promise<GuardedRun<T>> {
+  const client = new Client({ connectionString, keepAlive: true });
+  // An idle client whose server goes away emits `error`; unhandled, that
+  // takes the whole worker down. The run itself notices the database is
+  // gone through its own queries.
+  client.on("error", () => {});
+  await client.connect();
+  try {
+    const { rows } = await client.query<{ held: boolean }>(
+      "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held",
+      [key],
+    );
+    if (!rows[0]?.held) return { ran: false };
+    try {
+      return { ran: true, result: await run() };
+    } finally {
+      await client
+        .query("SELECT pg_advisory_unlock(hashtextextended($1, 0))", [key])
+        .catch(() => {});
+    }
+  } finally {
+    await client.end().catch(() => {});
+  }
+}
@@ -0,0 +1,58 @@
+/**
+ * The binding for a queue whose job can outlast pg-boss's default expiry.
+ *
+ * Three things together make a long job safe, and `queue-runtime.ts` names
+ * every queue that needs them:
+ *
+ *   1. An explicit `expireInSeconds` on every send and schedule, so pg-boss
+ *      does not declare the job dead at fifteen minutes while it is working.
+ *   2. A time budget (`jobBudget` / `jobDeadline` in `job-budget.ts`) the
+ *      handler checks between units of work, so the pass stops on its own
+ *      before that expiry and the next run picks up what is left.
+ *   3. A lock the running handler holds (`withJobLock`), so the retry pg-boss
+ *      starts once the expiry has passed does nothing beside a run that is
+ *      still going. Queue policies cannot do this: they constrain job rows,
+ *      and an expired job's row is no longer active.
+ *
+ * `lockedPass` is the third. It wraps a handler so each job runs under the
+ * lock `<queue>:<identity>`, where the identity is what two runs must not
+ * share: the account for a per-user job, a constant for a pass over everyone.
+ * A delivery that finds the lock held reports `already_running` and succeeds,
+ * because the run that holds it is doing the work.
+ */
+import type { Job } from "pg-boss";
+
+import { withJobLock } from "@/lib/jobs/job-lock";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { annotate } from "@/lib/logging/context";
+
+/** The identity of a pass that walks every account: one run at a time. */
+export const WHOLE_PASS = "pass";
+
+export function lockedPass<J extends Job<unknown>>(
+  queue: string,
+  identityOf: (job: J) => string,
+  handler: (jobs: J[]) => Promise<JobOutcome>,
+): (jobs: J[]) => Promise<JobOutcome> {
+  return async (jobs) => {
+    let outcome: JobOutcome = jobDone({ jobs: 0 });
+    for (const job of jobs) {
+      const run = await withJobLock(`job:${queue}:${identityOf(job)}`, () =>
+        handler([job]),
+      );
+      if (!run.ran) {
+        annotate({
+          action: { name: "job.long_pass.already_running" },
+          meta: { queue },
+        });
+        outcome = jobDone({ already_running: 1 });
+        continue;
+      }
+      // The first failed job fails the batch; pg-boss hands one job per
+      // batch here, so this is that job's own outcome.
+      if (!run.result.ok) return run.result;
+      outcome = run.result;
+    }
+    return outcome;
+  };
+}
@@ -4,7 +4,8 @@ import { Prisma, type PrismaClient } from "@/generated/prisma/client";
 import { invalidateUserMedications } from "@/lib/cache/invalidate";
 import { auditLog } from "@/lib/auth/audit";
 import { prisma, toJson } from "@/lib/db";
-import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { consumeImportedIntakesBatch } from "@/lib/medications/inventory/consumption";
 import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
 import {
@@ -825,13 +826,25 @@ async function processNextChunk(
  );
 }

-export async function processMedicationIntakeImportJob(
+/**
+ * Work the import chunk by chunk until it is finished. `shouldStop` is the
+ * job's time budget: between chunks, once it is spent, the run returns
+ * `"stopped"` and leaves the job row as it is. Every chunk commits its own
+ * progress under the row lock, so the retry that follows picks up at the next
+ * chunk, and a second delivery running beside this one waits on that lock and
+ * continues from the progress this one wrote.
+ */
+async function workMedicationIntakeImport(
  jobId: string,
-): Promise<MedicationImportResult | null> {
+  shouldStop: () => boolean,
+): Promise<MedicationImportResult | null | "stopped"> {
  const client = getWorkerPrisma();
  for (;;) {
    const outcome = await processNextChunk(client, jobId);
-    if (!outcome.terminal) continue;
+    if (!outcome.terminal) {
+      if (shouldStop()) return "stopped";
+      continue;
+    }
    if (outcome.finalized && outcome.userId && outcome.result) {
      if (outcome.result.imported > 0) {
        invalidateUserMedications(outcome.userId, { evict: true });
@@ -843,6 +856,14 @@ export async function processMedicationIntakeImportJob(
  }
 }

+/** The whole import in one go, with no time budget. */
+export async function processMedicationIntakeImportJob(
+  jobId: string,
+): Promise<MedicationImportResult | null> {
+  const result = await workMedicationIntakeImport(jobId, () => false);
+  return result === "stopped" ? null : result;
+}
+
 async function recordWorkerFailure(
  job: Job<MedicationIntakeImportQueuePayload>,
  error: unknown,
@@ -879,7 +900,19 @@ export async function handleMedicationIntakeImport(
  job: Job<MedicationIntakeImportQueuePayload>,
 ): Promise<JobOutcome> {
  try {
-    const result = await processMedicationIntakeImportJob(job.data.jobId);
+    const result = await workMedicationIntakeImport(
+      job.data.jobId,
+      jobBudget([job]),
+    );
+    // Out of time before the last chunk: fail the delivery so pg-boss
+    // retries it, and the retry resumes from the committed progress.
+    if (result === "stopped") {
+      return jobFailed(
+        "stopped before the last chunk; the retry resumes",
+        undefined,
+        { stopped_early: true },
+      );
+    }
    // A null result is a delivery that found the row already terminal or gone.
    // There was nothing left to import, which is an absence of work rather
    // than a failure to do it.
@@ -42,6 +42,15 @@ const ALGORITHM = "aes-256-gcm";
 const IV_LENGTH = 12;
 const TAG_LENGTH = 16;

+/**
+ * The off-host pass's expiry: four hours. It uploads every account and
+ * already stopped between accounts at three quarters of the default fifteen
+ * minutes, which on an instance of any size meant a retry per night; one
+ * large account past the default was retried beside itself. Four hours with
+ * the same budget and a lock (`lockedPass`) leaves room for the night's pass.
+ */
+export const OFFHOST_BACKUP_EXPIRE_SECONDS = 4 * 60 * 60;
+
 export interface OffhostBackupConfig {
  endpoint: string;
  bucket: string;
@@ -23,6 +23,15 @@ import type { Locale } from "@/lib/i18n/config";

 export const PERIOD_NARRATIVE_QUEUE = "period-narrative-warm";

+/**
+ * The job's expiry: three hours. On a boundary night the scheduled pass
+ * writes up to two narratives for each of 200 accounts, which can outlast
+ * pg-boss's fifteen-minute default. The pass stops itself at three quarters
+ * of this (`jobBudget`) and holds a lock (`lockedPass`); every send and the
+ * schedule carry it.
+ */
+export const PERIOD_NARRATIVE_EXPIRE_SECONDS = 3 * 60 * 60;
+
 /**
 * Nightly at 05:05 Europe/Berlin — inside the existing maintenance window,
 * after the comprehensive insight pre-generation (04:30) and the computed
@@ -79,6 +88,7 @@ export async function enqueueNarrativeWarm(payload: {
      {
        singletonKey: `warm:${payload.userId}:${payload.period}:${payload.locale}`,
        singletonSeconds: 120,
+        expireInSeconds: PERIOD_NARRATIVE_EXPIRE_SECONDS,
        // Transient provider / pool failures retry with backoff instead
        // of leaving the narrative cold until the next boundary night.
        retryLimit: 3,
@@ -63,6 +63,8 @@ export interface NarrativeWarmRunResult {
  insufficient: number;
  failed: number;
  budgetBlocked: number;
+  /** The job's time budget ran out before every candidate was reached. */
+  stoppedEarly: boolean;
 }

 /**
@@ -152,6 +154,12 @@ export async function runPeriodNarrativeWarm(
    cap?: number;
    /** Injected for the test — defaults to the real generator. */
    generate?: typeof generatePeriodNarrative;
+    /**
+     * The job's time budget (`jobBudget`), checked between candidates. The
+     * candidates are ordered least recently narrated first, so the ones a
+     * stopped pass did not reach lead the next one.
+     */
+    shouldStop?: () => boolean;
  } = {},
 ): Promise<NarrativeWarmRunResult> {
  const now = options.now ?? new Date();
@@ -167,6 +175,7 @@ export async function runPeriodNarrativeWarm(
    insufficient: 0,
    failed: 0,
    budgetBlocked: 0,
+    stoppedEarly: false,
  };

  const flags = await getAssistantFlags();
@@ -180,6 +189,10 @@ export async function runPeriodNarrativeWarm(
  result.total = candidates.length;

  for (const candidate of candidates) {
+    if (options.shouldStop?.()) {
+      result.stoppedEarly = true;
+      break;
+    }
    // The nightly pass exists for the model-written narrative. Unavailable,
    // it skips the user before the context build and the budget write; the
    // deterministic narrative is written on the next read instead (the
@@ -233,6 +246,7 @@ export async function runPeriodNarrativeWarm(
      insufficient: result.insufficient,
      failed: result.failed,
      budget_blocked: result.budgetBlocked,
+      stopped_early: result.stoppedEarly,
    },
  });

@@ -15,6 +15,16 @@ import { getGlobalBoss } from "@/lib/jobs/boss-instance";

 export const PR_DETECTION_QUEUE = "pr-detection";

+/**
+ * The job's expiry: twenty-five minutes. The fallback cron walks every
+ * account's whole history every thirty minutes, which on a large instance
+ * outlasts pg-boss's fifteen-minute default; past it pg-boss retried the pass
+ * beside itself. The pass stops at three quarters of this (`jobBudget`) and
+ * holds a lock (`lockedPass`), so it ends before the next tick and a retry or
+ * a late tick never runs beside it. The ingest sends carry the same expiry.
+ */
+export const PR_DETECTION_EXPIRE_SECONDS = 25 * 60;
+
 /** Concurrency budget for the worker process — five jobs in flight
 *  is enough to drain a multi-user backfill spike without crowding
 *  the reminder check or the daily insights workload that runs on the
@@ -62,6 +72,7 @@ export async function enqueuePrDetection(
    triggeredAt: new Date().toISOString(),
  };
  await boss.send(PR_DETECTION_QUEUE, payload, {
+    expireInSeconds: PR_DETECTION_EXPIRE_SECONDS,
    // Stagger retries gently — the detector is purely read-heavy on
    // the user's own data, but a stampede of failed jobs against a
    // briefly-flaky DB would compound the problem.
@@ -70,6 +70,7 @@ function buildPrismaMock(
 ) {
  return {
    user: { findMany: vi.fn().mockResolvedValue(users) },
+    dataBackup: { findMany: vi.fn().mockResolvedValue([]) },
  };
 }

@@ -236,3 +237,44 @@ describe("handleDataBackup outcome", () => {
    expect(outcome).toMatchObject({ ok: true, did: { backed: 0, total: 0 } });
  });
 });
+
+describe("handleDataBackup on a large instance", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mocks.store.mockResolvedValue(undefined);
+  });
+
+  it("backs up the accounts with the oldest copies first and stops on the budget", async () => {
+    const prisma = buildPrismaMock([
+      { id: "fresh", username: "fresh" },
+      { id: "never", username: "never" },
+      { id: "stale", username: "stale" },
+    ]);
+    prisma.dataBackup.findMany.mockResolvedValue([
+      { userId: "fresh", createdAt: new Date("2026-09-30T00:00:00Z") },
+      { userId: "stale", createdAt: new Date("2026-09-01T00:00:00Z") },
+    ]);
+    mocks.getWorkerPrisma.mockReturnValue(prisma);
+    const controller = new AbortController();
+    // Two accounts in, pg-boss gives up on the job.
+    mocks.store.mockImplementation(async () => {
+      if (mocks.store.mock.calls.length === 2) controller.abort();
+    });
+
+    const outcome = await handleDataBackup([
+      {
+        expireInSeconds: 7200,
+        signal: controller.signal,
+      } as unknown as Parameters<typeof handleDataBackup>[0][number],
+    ]);
+
+    const order = mocks.store.mock.calls.map(
+      ([input]) => (input as { userId: string }).userId,
+    );
+    expect(order).toEqual(["never", "stale"]);
+    expect(outcome).toMatchObject({
+      ok: true,
+      did: { backed: 2, stopped_early: true },
+    });
+  });
+});
@@ -121,16 +121,39 @@ export async function handleOffhostBackup(
  });
 }

+/**
+ * The weekly on-host copy of every account.
+ *
+ * The pass serialises each account's whole record, so on a large instance it
+ * can outlast its job. It walks the accounts whose weekly copy is oldest
+ * first and stops between accounts once the job's budget is spent
+ * (`jobBudget`), reporting `stopped_early`. The accounts it did not reach have
+ * the oldest copies, so they lead the next pass (the weekly cron or the
+ * admin's "run now"), and every pass makes progress where it matters most.
+ * The binding holds a lock (`lockedPass`), so a retry never runs beside it.
+ */
 export async function handleDataBackup(
  jobs: Job<DataBackupPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
+  const shouldStop = jobBudget(jobs);
  return withBackgroundEvent("job.data_backup", async (evt) => {
    const prisma = getWorkerPrisma();
    try {
-      const users = await prisma.user.findMany({
-        select: { id: true, username: true },
-      });
+      const [accounts, copies] = await Promise.all([
+        prisma.user.findMany({ select: { id: true, username: true } }),
+        prisma.dataBackup.findMany({
+          where: { type: "WEEKLY_AUTO" },
+          select: { userId: true, createdAt: true },
+        }),
+      ]);
+      const copiedAt = new Map(
+        copies.map((copy) => [copy.userId, copy.createdAt.getTime()]),
+      );
+      // No copy yet sorts first, then the oldest copy.
+      const users = [...accounts].sort(
+        (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1),
+      );
+      let stoppedEarly = false;

      let backed = 0;
      let usersFailed = 0;
@@ -142,6 +165,10 @@ export async function handleDataBackup(
      // six weeks in.
      let lastError: unknown;
      for (const user of users) {
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
        reportJobProgress({
          backup_user: user.id,
          backup_users_done: backed + usersFailed + skippedRestoring,
@@ -207,6 +234,7 @@ export async function handleDataBackup(
        users_failed: usersFailed,
        records_oversized: oversized,
        users_skipped_restoring: skippedRestoring,
+        stopped_early: stoppedEarly,
      };

      // A pass that wrote nothing for anybody protected nobody, and saying
@@ -21,6 +21,7 @@ import {
  runInsightStatusGenerate,
 } from "@/lib/jobs/insight-status-generate";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status";
 import { generateWeightStatusForUser } from "@/lib/insights/weight-status";
@@ -77,6 +78,12 @@ export interface MedicationComplianceStatusPayload {
 * The pass is what succeeds or fails. One user whose generation threw is that
 * user's cold card, reported as the `failed` count; failing the job would
 * re-run the whole cohort over it.
+ *
+ * The pass walks every candidate with a provider call each, so on a large
+ * instance it can outlast its job. `shouldStop` is the job's time budget: the
+ * pass stops between users once it is spent and reports `stopped_early`. The
+ * users it did not reach keep yesterday's note until the next night or their
+ * next visit; nothing is retried beside a pass that is still running.
 */
 export async function runStatusCronGenerate(
  taskName: string,
@@ -84,6 +91,7 @@ export async function runStatusCronGenerate(
    userId: string,
    options: { locale: string | null; force: boolean },
  ) => Promise<unknown>,
+  shouldStop: () => boolean = () => false,
 ): Promise<JobOutcome> {
  return withBackgroundEvent(taskName, async (evt) => {
    const prisma = getWorkerPrisma();
@@ -96,8 +104,13 @@ export async function runStatusCronGenerate(

      let generated = 0;
      let failed = 0;
+      let stoppedEarly = false;

      for (const user of users) {
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
        // The capability before the generator builds a snapshot. The batch
        // entry below checks for itself.
        const capability = await aiCapabilityForJob(user.id, "statusText");
@@ -124,9 +137,14 @@ export async function runStatusCronGenerate(

      evt.setBackground({
        task_name: taskName,
-        result: { generated, failed, total: users.length },
+        result: { generated, failed, total: users.length, stoppedEarly },
+      });
+      return jobDone({
+        total: users.length,
+        generated,
+        failed,
+        stopped_early: stoppedEarly,
      });
-      return jobDone({ total: users.length, generated, failed });
    } catch (err) {
      evt.setError(err);
      recordError();
@@ -153,7 +171,10 @@ export async function runStatusCronGenerate(
 * inside the batch itself). Net effect: the nightly ladder pays one call per
 * user instead of seven, with no queue/registry churn.
 */
-async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {
+async function runStatusBatchCron(
+  taskName: string,
+  shouldStop: () => boolean,
+): Promise<JobOutcome> {
  return withBackgroundEvent(taskName, async (evt) => {
    const prisma = getWorkerPrisma();
    try {
@@ -166,7 +187,14 @@ async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {
      let generated = 0;
      let served = 0;
      let failed = 0;
+      let stoppedEarly = false;
      for (const user of users) {
+        // Same budget as the per-metric passes: stop between users, and
+        // leave the rest to the later crons and the next night.
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
        try {
          const result = await generateStatusBatchForUser(user.id, {
            // The stored locale, then the operator default — the prompt names
@@ -186,16 +214,34 @@ async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {

      annotate({
        action: { name: "insights.status.batch.cron" },
-        meta: { generated, served, failed, total: users.length },
+        meta: {
+          generated,
+          served,
+          failed,
+          total: users.length,
+          stopped_early: stoppedEarly,
+        },
      });
      evt.setBackground({
        task_name: taskName,
-        result: { generated, served, failed, total: users.length },
+        result: {
+          generated,
+          served,
+          failed,
+          total: users.length,
+          stoppedEarly,
+        },
      });
      // A user whose batch call threw already fell back to the single-card
      // path inside the batch, and the six later per-metric crons cover what
      // is still cold. The batch pass ran, so it reports the count.
-      return jobDone({ total: users.length, generated, served, failed });
+      return jobDone({
+        total: users.length,
+        generated,
+        served,
+        failed,
+        stopped_early: stoppedEarly,
+      });
    } catch (err) {
      evt.setError(err);
      recordError();
@@ -208,61 +254,66 @@ async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {
 export function handleGeneralStatusGenerate(
  jobs: Job<GeneralStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
-  return runStatusBatchCron("job.insights.batch");
+  return runStatusBatchCron("job.insights.batch", jobBudget(jobs));
 }

 export function handleBloodPressureStatusGenerate(
  jobs: Job<BloodPressureStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
  return runStatusCronGenerate(
    "job.insights.blood_pressure",
    generateBloodPressureStatusForUser,
+    jobBudget(jobs),
  );
 }

 export function handleWeightStatusGenerate(
  jobs: Job<WeightStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
  return runStatusCronGenerate(
    "job.insights.weight",
    generateWeightStatusForUser,
+    jobBudget(jobs),
  );
 }

 export function handlePulseStatusGenerate(
  jobs: Job<PulseStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
  return runStatusCronGenerate(
    "job.insights.pulse",
    generatePulseStatusForUser,
+    jobBudget(jobs),
  );
 }

 export function handleBmiStatusGenerate(
  jobs: Job<BmiStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
-  return runStatusCronGenerate("job.insights.bmi", generateBmiStatusForUser);
+  return runStatusCronGenerate(
+    "job.insights.bmi",
+    generateBmiStatusForUser,
+    jobBudget(jobs),
+  );
 }

 export function handleMoodStatusGenerate(
  jobs: Job<MoodStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
-  return runStatusCronGenerate("job.insights.mood", generateMoodStatusForUser);
+  return runStatusCronGenerate(
+    "job.insights.mood",
+    generateMoodStatusForUser,
+    jobBudget(jobs),
+  );
 }

 export function handleMedicationComplianceStatusGenerate(
  jobs: Job<MedicationComplianceStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
  return runStatusCronGenerate(
    "job.insights.medication_compliance",
    generateMedicationComplianceStatusForUser,
+    jobBudget(jobs),
  );
 }

@@ -310,7 +361,9 @@ export async function handleInsightPregenerateJob(
      return jobDone({ forced: forced.length, scheduled: 0 });
    }
    try {
-      const summary = await runInsightPregenerate(getWorkerPrisma());
+      const summary = await runInsightPregenerate(getWorkerPrisma(), {
+        shouldStop: jobBudget(jobs),
+      });
      evt.setBackground({
        task_name: "job.insight_pregenerate",
        result: { ...summary },
@@ -325,6 +378,7 @@ export async function handleInsightPregenerateJob(
        skipped: summary.skipped,
        failed: summary.failed,
        budget_blocked: summary.budgetBlocked,
+        stopped_early: summary.stoppedEarly,
        assessments_warmed: summary.assessmentsWarmed,
        metric_assessments_warmed: summary.metricAssessmentsWarmed,
      });
@@ -4,6 +4,7 @@
 * Extracted from reminder-worker.ts, which owns the queue names, cron
 * schedules, and boss.work registrations.
 */
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { type Job } from "pg-boss";
 import { runGeoBackfill } from "@/lib/jobs/geo-backfill";
 import { fetchGeoLite2Databases } from "@/lib/geo/geolite2-fetch";
@@ -223,6 +224,11 @@ export async function handlePrDetection(
  let insertedAcrossJobs = 0;
  let tiesAcrossJobs = 0;
  let usersFailed = 0;
+  let stoppedEarly = false;
+  // The cron pass walks every account's all-time history and ticks every
+  // thirty minutes. It stops between accounts once the job's budget is spent;
+  // the ingest jobs and the next tick carry the rest.
+  const shouldStop = jobBudget(jobs);
  for (const job of jobs) {
    await withBackgroundEvent("job.pr_detection", async (evt) => {
      const p = getWorkerPrisma();
@@ -241,6 +247,10 @@ export async function handlePrDetection(
      let insertedTotal = 0;
      let tiesTotal = 0;
      for (const userId of userIds) {
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
        try {
          const result = await detectPersonalRecordsForUser(userId, {
            silent,
@@ -274,5 +284,6 @@ export async function handlePrDetection(
    pr_detection_inserted: insertedAcrossJobs,
    pr_detection_ties: tiesAcrossJobs,
    pr_detection_users_failed: usersFailed,
+    stopped_early: stoppedEarly,
  });
 }
@@ -31,8 +31,12 @@ import {
  PR_DETECTION_QUEUE,
  PR_DETECTION_CONCURRENCY,
  PR_DETECTION_FALLBACK_CRON,
+  PR_DETECTION_EXPIRE_SECONDS,
  type PrDetectionPayload,
 } from "@/lib/jobs/pr-detection";
+import { jobBudget } from "@/lib/jobs/job-budget";
+import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+import { OFFHOST_BACKUP_EXPIRE_SECONDS } from "@/lib/jobs/offhost-backup";
 import {
  MEDICATION_INVENTORY_EXPIRE_QUEUE,
  MEDICATION_INVENTORY_EXPIRE_CRON,
@@ -531,7 +535,11 @@ const schedules: ScheduleEntry[] = [
    STEP_UP_ELEVATION_CLEANUP_CRON,
    cronIsTheRetry,
  ],
-  [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON],
+  [
+    OFFHOST_BACKUP_QUEUE,
+    OFFHOST_BACKUP_CRON,
+    { expireInSeconds: OFFHOST_BACKUP_EXPIRE_SECONDS },
+  ],
  [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON],
  // Nightly backstop for the purge the deletion kicks straight away: a
  // request the bucket refused, or one written while the queue was down.
@@ -557,7 +565,11 @@ const schedules: ScheduleEntry[] = [
  // cron payload deliberately omits a `userId` so the handler iterates
  // every user; per-user push-suppression cannot apply on the cron
  // path (the silent flag is set by the ingest hooks).
-  [PR_DETECTION_QUEUE, PR_DETECTION_FALLBACK_CRON],
+  [
+    PR_DETECTION_QUEUE,
+    PR_DETECTION_FALLBACK_CRON,
+    { expireInSeconds: PR_DETECTION_EXPIRE_SECONDS },
+  ],
  // v1.4.25 W19b — daily expire-stale pass for the per-pen inventory
  // entities. Flips IN_USE rows whose 30-day clock has blown to
  // EXPIRED at 03:30 Europe/Berlin (in the existing 02:xx–03:xx
@@ -747,7 +759,7 @@ export async function registerMaintenanceQueues(
    boss,
    DATA_BACKUP_QUEUE,
    { localConcurrency: 1 },
-    handleDataBackup,
+    lockedPass(DATA_BACKUP_QUEUE, () => WHOLE_PASS, handleDataBackup),
  );
  await createAndWork<RateLimitCleanupPayload>(
    boss,
@@ -778,7 +790,7 @@ export async function registerMaintenanceQueues(
    boss,
    OFFHOST_BACKUP_QUEUE,
    { localConcurrency: 1, includeMetadata: true },
-    handleOffhostBackup,
+    lockedPass(OFFHOST_BACKUP_QUEUE, () => WHOLE_PASS, handleOffhostBackup),
  );
  await createAndWork(
    boss,
@@ -935,7 +947,16 @@ export async function registerMaintenanceQueues(
    boss,
    PR_DETECTION_QUEUE,
    { localConcurrency: PR_DETECTION_CONCURRENCY },
-    handlePrDetection,
+    // An ingest job is one account; the cron pass is every account, one run
+    // at a time.
+    lockedPass(
+      PR_DETECTION_QUEUE,
+      (job) => {
+        const userId = (job.data as PrDetectionPayload | undefined)?.userId;
+        return userId ? `user:${userId}` : WHOLE_PASS;
+      },
+      handlePrDetection,
+    ),
  );
  await createAndWork<MedicationInventoryExpirePayload>(
    boss,
@@ -1159,7 +1180,11 @@ export async function registerMaintenanceQueues(
    boss,
    ENCRYPTION_KEY_ROTATE_QUEUE,
    { localConcurrency: ENCRYPTION_KEY_ROTATE_CONCURRENCY },
-    handleEncryptionKeyRotate,
+    lockedPass(
+      ENCRYPTION_KEY_ROTATE_QUEUE,
+      () => WHOLE_PASS,
+      handleEncryptionKeyRotate,
+    ),
  );

  // v1.25 — medication-note encryption backfill worker. The boot enqueue
@@ -1318,56 +1343,65 @@ export async function registerMaintenanceQueues(
    boss,
    CONTENT_INDEX_BACKFILL_QUEUE,
    { localConcurrency: CONTENT_INDEX_BACKFILL_CONCURRENCY },
-    async (jobs) => {
-      let users = 0;
-      let skipped = 0;
-      let documentsIndexed = 0;
-      let documentsRetokenised = 0;
-      let documentsSkipped = 0;
-      let documentsFailed = 0;
-      for (const job of jobs) {
-        const { userId } = job.data;
-        if (!userId) {
-          skipped++;
-          continue;
+    lockedPass(
+      CONTENT_INDEX_BACKFILL_QUEUE,
+      (job) => `user:${job.data.userId ?? ""}`,
+      async (jobs) => {
+        let users = 0;
+        let stoppedEarly = false;
+        const shouldStop = jobBudget(jobs);
+        let skipped = 0;
+        let documentsIndexed = 0;
+        let documentsRetokenised = 0;
+        let documentsSkipped = 0;
+        let documentsFailed = 0;
+        for (const job of jobs) {
+          const { userId } = job.data;
+          if (!userId) {
+            skipped++;
+            continue;
+          }
+          try {
+            const {
+              indexed,
+              retokenised,
+              skipped: docsSkipped,
+              failed: docsFailed,
+              reason,
+              stoppedEarly: stopped,
+            } = await runContentIndexBackfillForUser(userId, shouldStop);
+            users++;
+            stoppedEarly ||= stopped === true;
+            documentsIndexed += indexed;
+            documentsRetokenised += retokenised;
+            documentsSkipped += docsSkipped;
+            documentsFailed += docsFailed;
+            workerLog(
+              "info",
+              `[document-content-index-backfill] user=${userId} indexed=${indexed} retokenised=${retokenised} skipped=${docsSkipped} failed=${docsFailed} reason=${reason}`,
+            );
+          } catch (err) {
+            recordError();
+            workerLog(
+              "error",
+              `[document-content-index-backfill] user=${userId} failed`,
+              err,
+            );
+            throw err;
+          }
        }
-        try {
-          const {
-            indexed,
-            retokenised,
-            skipped: docsSkipped,
-            failed: docsFailed,
-            reason,
-          } = await runContentIndexBackfillForUser(userId);
-          users++;
-          documentsIndexed += indexed;
-          documentsRetokenised += retokenised;
-          documentsSkipped += docsSkipped;
-          documentsFailed += docsFailed;
-          workerLog(
-            "info",
-            `[document-content-index-backfill] user=${userId} indexed=${indexed} retokenised=${retokenised} skipped=${docsSkipped} failed=${docsFailed} reason=${reason}`,
-          );
-        } catch (err) {
-          recordError();
-          workerLog(
-            "error",
-            `[document-content-index-backfill] user=${userId} failed`,
-            err,
-          );
-          throw err;
-        }
-      }
-      return jobDone({
-        jobs: jobs.length,
-        users,
-        skipped,
-        documents_indexed: documentsIndexed,
-        documents_retokenised: documentsRetokenised,
-        documents_skipped: documentsSkipped,
-        documents_failed: documentsFailed,
-      });
-    },
+        return jobDone({
+          jobs: jobs.length,
+          users,
+          skipped,
+          documents_indexed: documentsIndexed,
+          documents_retokenised: documentsRetokenised,
+          documents_skipped: documentsSkipped,
+          documents_failed: documentsFailed,
+          stopped_early: stoppedEarly,
+        });
+      },
+    ),
  );

  // Document AI — automatic per-document content indexing. Enqueued on upload
@@ -102,11 +102,16 @@ import {
 } from "@/lib/jobs/insight-status-generate";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
+import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+import { INSIGHT_PREGENERATE_EXPIRE_SECONDS } from "@/lib/jobs/insight-pregenerate-shared";
+import { PERIOD_NARRATIVE_EXPIRE_SECONDS } from "@/lib/jobs/period-narrative-shared";
 import { getWorkerPrisma, workerLog } from "./shared";
 import {
  createAndSchedule,
  createAndWork,
  insightRetryOptions,
+  nightlyInsightPassOptions,
  type QueuePolicyTable,
  type ScheduleEntry,
 } from "./registrar-shared";
@@ -232,25 +237,32 @@ const allQueues = [
 ];

 const schedules: ScheduleEntry[] = [
-  [GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON, insightRetryOptions],
+  [GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON, nightlyInsightPassOptions],
  [
    BLOOD_PRESSURE_STATUS_QUEUE,
    BLOOD_PRESSURE_STATUS_CRON,
-    insightRetryOptions,
+    nightlyInsightPassOptions,
  ],
-  [WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON, insightRetryOptions],
-  [PULSE_STATUS_QUEUE, PULSE_STATUS_CRON, insightRetryOptions],
-  [BMI_STATUS_QUEUE, BMI_STATUS_CRON, insightRetryOptions],
+  [WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON, nightlyInsightPassOptions],
+  [PULSE_STATUS_QUEUE, PULSE_STATUS_CRON, nightlyInsightPassOptions],
+  [BMI_STATUS_QUEUE, BMI_STATUS_CRON, nightlyInsightPassOptions],
  // v1.15.20 — mood status nightly, continuing the 02:xx ladder.
-  [MOOD_STATUS_QUEUE, MOOD_STATUS_CRON, insightRetryOptions],
+  [MOOD_STATUS_QUEUE, MOOD_STATUS_CRON, nightlyInsightPassOptions],
  [
    MEDICATION_COMPLIANCE_STATUS_QUEUE,
    MEDICATION_COMPLIANCE_STATUS_CRON,
-    insightRetryOptions,
+    nightlyInsightPassOptions,
  ],
  // v1.7.0 — nightly 04:30 Europe/Berlin comprehensive-insight
  // pre-generation. Budget-gated per user inside the handler.
-  [INSIGHT_PREGENERATE_QUEUE, INSIGHT_PREGENERATE_CRON, insightRetryOptions],
+  [
+    INSIGHT_PREGENERATE_QUEUE,
+    INSIGHT_PREGENERATE_CRON,
+    {
+      ...insightRetryOptions,
+      expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
+    },
+  ],
  // v1.10.0 — computed scores (WX-C). Nightly 04:45 Europe/Berlin
  // Recovery-score compute + store, after the rollup-feeding consolidation
  // + drain so the signals it reads are already folded.
@@ -266,7 +278,14 @@ const schedules: ScheduleEntry[] = [
  // v1.11.0 — nightly 05:05 Europe/Berlin period-narrative warm. The
  // handler only fans out on a week (Mon) / month (1st) boundary; every
  // other night is a cheap no-op. Budget-gated per user inside the runner.
-  [PERIOD_NARRATIVE_QUEUE, PERIOD_NARRATIVE_CRON, insightRetryOptions],
+  [
+    PERIOD_NARRATIVE_QUEUE,
+    PERIOD_NARRATIVE_CRON,
+    {
+      ...insightRetryOptions,
+      expireInSeconds: PERIOD_NARRATIVE_EXPIRE_SECONDS,
+    },
+  ],
  // v1.15.20 — daily 05:15 Europe/Berlin proactive Coach nudge, after
  // the 04:45–04:55 score crons so the recovery-score trigger reads
  // settled rows. Deterministic triggers only — no AI call on this path.
@@ -369,43 +388,59 @@ export async function registerStatusQueues(
    boss,
    GENERAL_STATUS_QUEUE,
    { localConcurrency: 1 },
-    handleGeneralStatusGenerate,
+    lockedPass(
+      GENERAL_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleGeneralStatusGenerate,
+    ),
  );
  await createAndWork<BloodPressureStatusPayload>(
    boss,
    BLOOD_PRESSURE_STATUS_QUEUE,
    { localConcurrency: 1 },
-    handleBloodPressureStatusGenerate,
+    lockedPass(
+      BLOOD_PRESSURE_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleBloodPressureStatusGenerate,
+    ),
  );
  await createAndWork<WeightStatusPayload>(
    boss,
    WEIGHT_STATUS_QUEUE,
    { localConcurrency: 1 },
-    handleWeightStatusGenerate,
+    lockedPass(
+      WEIGHT_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleWeightStatusGenerate,
+    ),
  );
  await createAndWork<PulseStatusPayload>(
    boss,
    PULSE_STATUS_QUEUE,
    { localConcurrency: 1 },
-    handlePulseStatusGenerate,
+    lockedPass(PULSE_STATUS_QUEUE, () => WHOLE_PASS, handlePulseStatusGenerate),
  );
  await createAndWork<BmiStatusPayload>(
    boss,
    BMI_STATUS_QUEUE,
    { localConcurrency: 1 },
-    handleBmiStatusGenerate,
+    lockedPass(BMI_STATUS_QUEUE, () => WHOLE_PASS, handleBmiStatusGenerate),
  );
  await createAndWork<MoodStatusPayload>(
    boss,
    MOOD_STATUS_QUEUE,
    { localConcurrency: 1 },
-    handleMoodStatusGenerate,
+    lockedPass(MOOD_STATUS_QUEUE, () => WHOLE_PASS, handleMoodStatusGenerate),
  );
  await createAndWork<MedicationComplianceStatusPayload>(
    boss,
    MEDICATION_COMPLIANCE_STATUS_QUEUE,
    { localConcurrency: 1 },
-    handleMedicationComplianceStatusGenerate,
+    lockedPass(
+      MEDICATION_COMPLIANCE_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleMedicationComplianceStatusGenerate,
+    ),
  );
  // v1.7.0 — nightly comprehensive-insight pre-generation.
  //
@@ -422,7 +457,16 @@ export async function registerStatusQueues(
    boss,
    INSIGHT_PREGENERATE_QUEUE,
    { localConcurrency: 2 },
-    handleInsightPregenerateJob,
+    // The cohort walk is one pass at a time; a forced warm is one per
+    // account, so it never waits behind the walk.
+    lockedPass(
+      INSIGHT_PREGENERATE_QUEUE,
+      (job) =>
+        job.data?.force && job.data.userId
+          ? `user:${job.data.userId}`
+          : WHOLE_PASS,
+      handleInsightPregenerateJob,
+    ),
  );
  // S4 — event-driven morning digest refresh. One forced comprehensive
  // regeneration per user per local morning, enqueued by the sleep-arrival
@@ -749,44 +793,53 @@ export async function registerStatusQueues(
    boss,
    PERIOD_NARRATIVE_QUEUE,
    { localConcurrency: 1 },
-    async (jobs): Promise<JobOutcome> => {
-      // Two modes on one queue, so the outcome counts them separately: a
-      // non-boundary night is a scheduled pass that generated nothing, which
-      // is a success with zero counts, not an idle queue.
-      let singleUser = 0;
-      let scheduled = 0;
-      let generated = 0;
-      let failed = 0;
-      for (const job of jobs) {
-        try {
-          if (job.data?.userId) {
-            await warmOneNarrative(job.data);
-            singleUser++;
-          } else {
-            const summary = await runPeriodNarrativeWarm(getWorkerPrisma());
-            scheduled++;
-            generated += summary.generated;
-            failed += summary.failed;
-            workerLog(
-              "info",
-              `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
-            );
+    lockedPass(
+      PERIOD_NARRATIVE_QUEUE,
+      (job) => (job.data?.userId ? `user:${job.data.userId}` : WHOLE_PASS),
+      async (jobs): Promise<JobOutcome> => {
+        // Two modes on one queue, so the outcome counts them separately: a
+        // non-boundary night is a scheduled pass that generated nothing, which
+        // is a success with zero counts, not an idle queue.
+        let singleUser = 0;
+        let scheduled = 0;
+        let generated = 0;
+        let failed = 0;
+        let stoppedEarly = false;
+        for (const job of jobs) {
+          try {
+            if (job.data?.userId) {
+              await warmOneNarrative(job.data);
+              singleUser++;
+            } else {
+              const summary = await runPeriodNarrativeWarm(getWorkerPrisma(), {
+                shouldStop: jobBudget(jobs),
+              });
+              scheduled++;
+              generated += summary.generated;
+              failed += summary.failed;
+              stoppedEarly ||= summary.stoppedEarly;
+              workerLog(
+                "info",
+                `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
+              );
+            }
+          } catch (err) {
+            recordError();
+            await reportWorkerError(PERIOD_NARRATIVE_QUEUE, err, {
+              mode: job.data?.userId ? "single-user" : "scheduled",
+            });
+            throw err;
          }
-        } catch (err) {
-          recordError();
-          await reportWorkerError(PERIOD_NARRATIVE_QUEUE, err, {
-            mode: job.data?.userId ? "single-user" : "scheduled",
-          });
-          throw err;
        }
-      }
-      return jobDone({
-        single_user: singleUser,
-        scheduled,
-        generated,
-        failed,
-      });
-    },
+        return jobDone({
+          single_user: singleUser,
+          scheduled,
+          generated,
+          failed,
+          stopped_early: stoppedEarly,
+        });
+      },
+    ),
  );
  // v1.11.1 — combined Coach memory refresh: rolling conversation summary +
  // durable fact extraction for one long conversation. localConcurrency 1 so a
@@ -168,6 +168,22 @@ export const insightRetryOptions = {
  retryBackoff: true,
 } as const;

+/**
+ * Send options for a nightly AI pass over every account (the status crons).
+ *
+ * The retries of `insightRetryOptions`, plus a two-hour expiry. Each pass
+ * makes a provider call per account, and on a large instance that outlasted
+ * pg-boss's fifteen-minute default: the job was declared dead and retried
+ * beside the pass still running. The pass now stops itself at three quarters
+ * of this (`jobBudget`) and holds a lock (`lockedPass`). The options ride on
+ * the schedule because the cron is the only sender.
+ */
+export const NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS = 2 * 60 * 60;
+export const nightlyInsightPassOptions = {
+  ...insightRetryOptions,
+  expireInSeconds: NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS,
+} as const;
+
 /**
 * Retry policy for a cron-driven pass whose failure mode is deterministic.
 *