diff --git a/src/app/api/admin/encryption/rotate/route.ts b/src/app/api/admin/encryption/rotate/route.ts index 0beacfe7f..e3f8dc063 100644 --- a/src/app/api/admin/encryption/rotate/route.ts +++ b/src/app/api/admin/encryption/rotate/route.ts @@ -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, }, ); diff --git a/src/lib/crypto/encryption-corpus.ts b/src/lib/crypto/encryption-corpus.ts index d470a163b..c55240731 100644 --- a/src/lib/crypto/encryption-corpus.ts +++ b/src/lib/crypto/encryption-corpus.ts @@ -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 { + shouldStop: () => boolean = () => false, +): Promise { 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 { 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, }; } diff --git a/src/lib/jobs/__tests__/insight-pregenerate.test.ts b/src/lib/jobs/__tests__/insight-pregenerate.test.ts index 0d0026a0a..163d0d209 100644 --- a/src/lib/jobs/__tests__/insight-pregenerate.test.ts +++ b/src/lib/jobs/__tests__/insight-pregenerate.test.ts @@ -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*\]/, ); }); diff --git a/src/lib/jobs/__tests__/long-pass.test.ts b/src/lib/jobs/__tests__/long-pass.test.ts new file mode 100644 index 000000000..31a750fc3 --- /dev/null +++ b/src/lib/jobs/__tests__/long-pass.test.ts @@ -0,0 +1,74 @@ +/** + * `lockedPass`: each job runs under the lock `job::`, 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()); +const keys = vi.hoisted(() => [] as string[]); + +vi.mock("@/lib/jobs/job-lock", () => ({ + withJobLock: async (key: string, run: () => Promise) => { + 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; + +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) => + 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); + }); +}); diff --git a/src/lib/jobs/__tests__/period-narrative-warm.test.ts b/src/lib/jobs/__tests__/period-narrative-warm.test.ts index 58e270c6d..a41071ad5 100644 --- a/src/lib/jobs/__tests__/period-narrative-warm.test.ts +++ b/src/lib/jobs/__tests__/period-narrative-warm.test.ts @@ -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*\]/, ); }); diff --git a/src/lib/jobs/__tests__/status-cron-candidates.test.ts b/src/lib/jobs/__tests__/status-cron-candidates.test.ts index fcd97978a..075c531ab 100644 --- a/src/lib/jobs/__tests__/status-cron-candidates.test.ts +++ b/src/lib/jobs/__tests__/status-cron-candidates.test.ts @@ -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 \* \* \*"/); }); diff --git a/src/lib/jobs/backup-restore.ts b/src/lib/jobs/backup-restore.ts index 0e76db75e..e2af9e0dd 100644 --- a/src/lib/jobs/backup-restore.ts +++ b/src/lib/jobs/backup-restore.ts @@ -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 diff --git a/src/lib/jobs/document-content-index-backfill.ts b/src/lib/jobs/document-content-index-backfill.ts index 2dc1bceb9..357619313 100644 --- a/src/lib/jobs/document-content-index-backfill.ts +++ b/src/lib/jobs/document-content-index-backfill.ts @@ -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 { // 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, diff --git a/src/lib/jobs/encryption-key-rotate.ts b/src/lib/jobs/encryption-key-rotate.ts index 86eb671d9..cd9c93eb8 100644 --- a/src/lib/jobs/encryption-key-rotate.ts +++ b/src/lib/jobs/encryption-key-rotate.ts @@ -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); diff --git a/src/lib/jobs/insight-pregenerate-shared.ts b/src/lib/jobs/insight-pregenerate-shared.ts index 22aca36c8..e6353b37a 100644 --- a/src/lib/jobs/insight-pregenerate-shared.ts +++ b/src/lib/jobs/insight-pregenerate-shared.ts @@ -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. diff --git a/src/lib/jobs/insight-pregenerate.ts b/src/lib/jobs/insight-pregenerate.ts index 37cac27f8..cdf911576 100644 --- a/src/lib/jobs/insight-pregenerate.ts +++ b/src/lib/jobs/insight-pregenerate.ts @@ -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; + /** + * 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 { 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, }, diff --git a/src/lib/jobs/integration-backfill-lock.ts b/src/lib/jobs/integration-backfill-lock.ts index 1e6a90905..c42403720 100644 --- a/src/lib/jobs/integration-backfill-lock.ts +++ b/src/lib/jobs/integration-backfill-lock.ts @@ -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 = { 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( +export function withIntegrationBackfillLock( identity: string, run: () => Promise, connectionString: string | undefined = process.env.DATABASE_URL, ): Promise> { - 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, + ); } diff --git a/src/lib/jobs/job-lock.ts b/src/lib/jobs/job-lock.ts new file mode 100644 index 000000000..5db0d75ba --- /dev/null +++ b/src/lib/jobs/job-lock.ts @@ -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 = { 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( + key: string, + run: () => Promise, + connectionString: string | undefined = process.env.DATABASE_URL, +): Promise> { + 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(() => {}); + } +} diff --git a/src/lib/jobs/long-pass.ts b/src/lib/jobs/long-pass.ts new file mode 100644 index 000000000..0d4ca0da4 --- /dev/null +++ b/src/lib/jobs/long-pass.ts @@ -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 `:`, 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>( + queue: string, + identityOf: (job: J) => string, + handler: (jobs: J[]) => Promise, +): (jobs: J[]) => Promise { + 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; + }; +} diff --git a/src/lib/jobs/medication-intake-import.ts b/src/lib/jobs/medication-intake-import.ts index b7c92dfd0..a79a71ed4 100644 --- a/src/lib/jobs/medication-intake-import.ts +++ b/src/lib/jobs/medication-intake-import.ts @@ -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 { + shouldStop: () => boolean, +): Promise { 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 { + const result = await workMedicationIntakeImport(jobId, () => false); + return result === "stopped" ? null : result; +} + async function recordWorkerFailure( job: Job, error: unknown, @@ -879,7 +900,19 @@ export async function handleMedicationIntakeImport( job: Job, ): Promise { 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. diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index a2f2e34fa..e1c89d864 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -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; diff --git a/src/lib/jobs/period-narrative-shared.ts b/src/lib/jobs/period-narrative-shared.ts index 52dad1371..56742c1ea 100644 --- a/src/lib/jobs/period-narrative-shared.ts +++ b/src/lib/jobs/period-narrative-shared.ts @@ -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, diff --git a/src/lib/jobs/period-narrative-warm.ts b/src/lib/jobs/period-narrative-warm.ts index 6ade00411..15eb409c9 100644 --- a/src/lib/jobs/period-narrative-warm.ts +++ b/src/lib/jobs/period-narrative-warm.ts @@ -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 { 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, }, }); diff --git a/src/lib/jobs/pr-detection.ts b/src/lib/jobs/pr-detection.ts index 8062dc74e..94f933fd3 100644 --- a/src/lib/jobs/pr-detection.ts +++ b/src/lib/jobs/pr-detection.ts @@ -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. diff --git a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts index fda4afc67..30b92dca3 100644 --- a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts +++ b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts @@ -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[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 }, + }); + }); +}); diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 5d57c2e27..f02714de6 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -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[], ): Promise { - 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 diff --git a/src/lib/jobs/reminder/insights-handlers.ts b/src/lib/jobs/reminder/insights-handlers.ts index 4fd293222..4f529f204 100644 --- a/src/lib/jobs/reminder/insights-handlers.ts +++ b/src/lib/jobs/reminder/insights-handlers.ts @@ -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, + shouldStop: () => boolean = () => false, ): Promise { 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 { +async function runStatusBatchCron( + taskName: string, + shouldStop: () => boolean, +): Promise { return withBackgroundEvent(taskName, async (evt) => { const prisma = getWorkerPrisma(); try { @@ -166,7 +187,14 @@ async function runStatusBatchCron(taskName: string): Promise { 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 { 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 { export function handleGeneralStatusGenerate( jobs: Job[], ): Promise { - void jobs; - return runStatusBatchCron("job.insights.batch"); + return runStatusBatchCron("job.insights.batch", jobBudget(jobs)); } export function handleBloodPressureStatusGenerate( jobs: Job[], ): Promise { - void jobs; return runStatusCronGenerate( "job.insights.blood_pressure", generateBloodPressureStatusForUser, + jobBudget(jobs), ); } export function handleWeightStatusGenerate( jobs: Job[], ): Promise { - void jobs; return runStatusCronGenerate( "job.insights.weight", generateWeightStatusForUser, + jobBudget(jobs), ); } export function handlePulseStatusGenerate( jobs: Job[], ): Promise { - void jobs; return runStatusCronGenerate( "job.insights.pulse", generatePulseStatusForUser, + jobBudget(jobs), ); } export function handleBmiStatusGenerate( jobs: Job[], ): Promise { - void jobs; - return runStatusCronGenerate("job.insights.bmi", generateBmiStatusForUser); + return runStatusCronGenerate( + "job.insights.bmi", + generateBmiStatusForUser, + jobBudget(jobs), + ); } export function handleMoodStatusGenerate( jobs: Job[], ): Promise { - void jobs; - return runStatusCronGenerate("job.insights.mood", generateMoodStatusForUser); + return runStatusCronGenerate( + "job.insights.mood", + generateMoodStatusForUser, + jobBudget(jobs), + ); } export function handleMedicationComplianceStatusGenerate( jobs: Job[], ): Promise { - 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, }); diff --git a/src/lib/jobs/reminder/ops-handlers.ts b/src/lib/jobs/reminder/ops-handlers.ts index 7560e70f5..7f944c1b8 100644 --- a/src/lib/jobs/reminder/ops-handlers.ts +++ b/src/lib/jobs/reminder/ops-handlers.ts @@ -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, }); } diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 460ac70ec..e9fdc896d 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -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( 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( 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 diff --git a/src/lib/jobs/reminder/register-status.ts b/src/lib/jobs/reminder/register-status.ts index 70ece8c3d..bd8d3c09a 100644 --- a/src/lib/jobs/reminder/register-status.ts +++ b/src/lib/jobs/reminder/register-status.ts @@ -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( boss, BLOOD_PRESSURE_STATUS_QUEUE, { localConcurrency: 1 }, - handleBloodPressureStatusGenerate, + lockedPass( + BLOOD_PRESSURE_STATUS_QUEUE, + () => WHOLE_PASS, + handleBloodPressureStatusGenerate, + ), ); await createAndWork( boss, WEIGHT_STATUS_QUEUE, { localConcurrency: 1 }, - handleWeightStatusGenerate, + lockedPass( + WEIGHT_STATUS_QUEUE, + () => WHOLE_PASS, + handleWeightStatusGenerate, + ), ); await createAndWork( boss, PULSE_STATUS_QUEUE, { localConcurrency: 1 }, - handlePulseStatusGenerate, + lockedPass(PULSE_STATUS_QUEUE, () => WHOLE_PASS, handlePulseStatusGenerate), ); await createAndWork( boss, BMI_STATUS_QUEUE, { localConcurrency: 1 }, - handleBmiStatusGenerate, + lockedPass(BMI_STATUS_QUEUE, () => WHOLE_PASS, handleBmiStatusGenerate), ); await createAndWork( boss, MOOD_STATUS_QUEUE, { localConcurrency: 1 }, - handleMoodStatusGenerate, + lockedPass(MOOD_STATUS_QUEUE, () => WHOLE_PASS, handleMoodStatusGenerate), ); await createAndWork( 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 => { - // 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 => { + // 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 diff --git a/src/lib/jobs/reminder/registrar-shared.ts b/src/lib/jobs/reminder/registrar-shared.ts index c46497fb8..6841b0111 100644 --- a/src/lib/jobs/reminder/registrar-shared.ts +++ b/src/lib/jobs/reminder/registrar-shared.ts @@ -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. *