From 4261139dfae6f7318ab5c1c8b5629afbefd6dabf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 21:19:32 +0200 Subject: [PATCH] fix(backup): run the restore drill as a locked long pass The drill now reads every account's newest off-host object but still ran on the default fifteen-minute expiry, without a lock and with two retries, so a large instance could see it killed, run beside itself, and page every failed account three times. It now carries a four-hour expiry, stops between accounts on its budget and reports the objects it did not reach, holds a pass lock at the binding, and is not retried. --- src/lib/jobs/__tests__/restore-drill.test.ts | 32 +++++++++++++ src/lib/jobs/job-outcome.ts | 1 + src/lib/jobs/queue-runtime.ts | 10 +++- src/lib/jobs/reminder/register-maintenance.ts | 13 +++-- src/lib/jobs/restore-drill.ts | 47 +++++++++++++++++-- 5 files changed, 94 insertions(+), 9 deletions(-) diff --git a/src/lib/jobs/__tests__/restore-drill.test.ts b/src/lib/jobs/__tests__/restore-drill.test.ts index d223c5a45..81eb035ef 100644 --- a/src/lib/jobs/__tests__/restore-drill.test.ts +++ b/src/lib/jobs/__tests__/restore-drill.test.ts @@ -7,7 +7,9 @@ import { runRestoreDrill, RESTORE_DRILL_CRON, RESTORE_DRILL_QUEUE, + RESTORE_DRILL_SEND_OPTIONS, } from "../restore-drill"; +import { QUEUE_RUNTIME } from "../queue-runtime"; vi.mock("../report-worker-error", () => ({ reportWorkerError: vi.fn().mockResolvedValue(undefined), @@ -86,6 +88,36 @@ describe("runRestoreDrill", () => { expect(RESTORE_DRILL_CRON).toBe("11 4 1 * *"); }); + it("runs as a locked long pass that pages each account once per run", () => { + // Every account's object is read, so the pg-boss default of fifteen + // minutes is no ceiling for it; no retry, so a failed account is paged + // once per run instead of three times. + expect(RESTORE_DRILL_SEND_OPTIONS.expireInSeconds).toBeGreaterThan(15 * 60); + expect(RESTORE_DRILL_SEND_OPTIONS.retryLimit).toBe(0); + expect(QUEUE_RUNTIME[RESTORE_DRILL_QUEUE]).toMatchObject({ + runtime: "long", + exclusive: "lockedPass", + }); + }); + + it("stops between accounts on its budget and counts the ones it did not reach", async () => { + const s3 = makeS3Mock({ + "2026-06-01/user-aaa.json.enc": backupObject({ userId: "user-aaa" }), + "2026-06-01/user-mmm.json.enc": backupObject({ userId: "user-mmm" }), + "2026-06-01/user-zzz.json.enc": backupObject({ userId: "user-zzz" }), + }); + let asked = 0; + const report = await runRestoreDrill( + s3, + new Date("2026-06-02T04:11:00Z"), + () => ++asked > 1, + ); + expect(report.accounts.map((a) => a.objectKey)).toEqual([ + "2026-06-01/user-aaa.json.enc", + ]); + expect(report.unchecked).toBe(2); + }); + it("fetches, decrypts, and parses the newest backup object", async () => { const s3 = makeS3Mock({ "2026-05-30/user-old.json.enc": backupObject(), diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 2ef5e5cc0..33e994a32 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -200,6 +200,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "restore_drill_medications", "restore_drill_mood_entries", "restore_drill_plaintext_bytes", + "restore_drill_unchecked", "restore_drill_stale", "restore_fail_cause", "restore_failed", diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 18177c274..3dd254b8f 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -36,6 +36,7 @@ import { DOCUMENT_AI_RUN_EXPIRE_SECONDS } from "@/lib/jobs/document-ai-run"; import { DATA_BACKUP_SEND_OPTIONS } from "@/lib/jobs/data-backup-policy"; import { OFFHOST_BACKUP_EXPIRE_SECONDS } from "@/lib/jobs/offhost-backup"; import { BACKUP_RESTORE_EXPIRE_SECONDS } from "@/lib/jobs/backup-restore"; +import { RESTORE_DRILL_EXPIRE_SECONDS } from "@/lib/jobs/restore-drill"; import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "@/lib/jobs/apple-health-import-worker"; import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-intake-import"; @@ -267,7 +268,14 @@ export const QUEUE_RUNTIME: Readonly> = { exclusive: "lockedPass", why: "Uploads every account's record off-host.", }, - "data-restore-drill": short("Fetches and stream-parses one off-host object."), + "data-restore-drill": { + runtime: "long", + expireInSeconds: RESTORE_DRILL_EXPIRE_SECONDS, + expiryVia: ["RESTORE_DRILL_SEND_OPTIONS"], + stop: { file: "lib/jobs/restore-drill.ts", fn: "handleRestoreDrill" }, + exclusive: "lockedPass", + why: "Fetches and stream-parses every account's newest off-host object.", + }, "offhost-backup-purge": short("At most 200 delete requests per run."), "backup-restore": { runtime: "long", diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 6fd906114..d774495d4 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -83,6 +83,7 @@ import { handleRestoreDrill, RESTORE_DRILL_CRON, RESTORE_DRILL_QUEUE, + RESTORE_DRILL_SEND_OPTIONS, } from "@/lib/jobs/restore-drill"; import { BACKUP_RESTORE_QUEUE, @@ -551,8 +552,10 @@ const allQueues = [ // a bulk DELETE or UPDATE that timed out will time out again on an immediate // retry, so pg-boss's default two would run the same doomed statement three // times a night. The queues without it are the ones whose failure is -// transient — the TLS probe, the off-host upload, the restore drill, the -// per-user environment fetch — and those still want the default retries. +// transient — the TLS probe, the off-host upload, the per-user environment +// fetch — and those still want the default retries. The restore drill sends +// `retryLimit: 0` in its own options for the other reason: it pages each +// failed account itself, and a retry would page the same accounts again. const schedules: ScheduleEntry[] = [ // v1.25 (W-ENV) — daily 02:10 Europe/Berlin discovery tick (empty payload) // that fans out one per-user environment fetch per opted-in account. @@ -571,7 +574,7 @@ const schedules: ScheduleEntry[] = [ cronIsTheRetry, ], [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON, OFFHOST_BACKUP_SEND_OPTIONS], - [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON], + [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON, RESTORE_DRILL_SEND_OPTIONS], // Nightly backstop for the purge the deletion kicks straight away: a // request the bucket refused, or one written while the queue was down. [OFFHOST_PURGE_QUEUE, OFFHOST_PURGE_CRON, cronIsTheRetry], @@ -832,11 +835,13 @@ export async function registerMaintenanceQueues( { localConcurrency: 1, includeMetadata: true }, lockedPass(OFFHOST_BACKUP_QUEUE, () => WHOLE_PASS, handleOffhostBackup), ); + // Locked: a drill that outlives a restart or a manual run never reads the + // bucket beside another one. await createAndWork( boss, RESTORE_DRILL_QUEUE, { localConcurrency: 1 }, - handleRestoreDrill, + lockedPass(RESTORE_DRILL_QUEUE, () => WHOLE_PASS, handleRestoreDrill), ); await createAndWork( boss, diff --git a/src/lib/jobs/restore-drill.ts b/src/lib/jobs/restore-drill.ts index caec85d00..1a6854bc1 100644 --- a/src/lib/jobs/restore-drill.ts +++ b/src/lib/jobs/restore-drill.ts @@ -21,6 +21,13 @@ * Schedule: 04:11 on the 1st of each month (`11 4 1 * *`), after the * 02:30 nightly upload and the 03:xx cleanup window, on a minute slot * no other cron uses. + * + * The drill reads every account's object, so it runs as a long pass: an + * explicit expiry (`RESTORE_DRILL_EXPIRE_SECONDS`), a stop between accounts + * on the job's budget, a lock so two drills never overlap (`lockedPass` at + * the binding), and no retry. A retry would re-read the same objects and + * page the same failed accounts a second and third time; each account is + * reported once per run, and the next drill is the retry. */ import type { Job } from "pg-boss"; import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan"; @@ -38,12 +45,28 @@ import { type S3Like, } from "@/lib/jobs/offhost-backup"; import { reportWorkerError } from "@/lib/jobs/report-worker-error"; +import { jobBudget } from "@/lib/jobs/job-budget"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; export const RESTORE_DRILL_QUEUE = "data-restore-drill"; export const RESTORE_DRILL_CRON = "11 4 1 * *"; +/** + * Four hours, the off-host upload's own window: the drill downloads and + * stream-parses every object that upload wrote in one night. + */ +export const RESTORE_DRILL_EXPIRE_SECONDS = 4 * 60 * 60; + +/** + * What the monthly schedule sends with: the expiry above and no retry + * (`retryLimit: 0`, the maintenance schedules' `cronIsTheRetry`). + */ +export const RESTORE_DRILL_SEND_OPTIONS = { + expireInSeconds: RESTORE_DRILL_EXPIRE_SECONDS, + retryLimit: 0, +} as const; + /** * A drill run that finds the newest backup older than this is reported * as a failure even when decrypt + parse succeed: the nightly uploader @@ -80,6 +103,8 @@ export interface RestoreDrillReport { accounts: RestoreDrillAccount[]; /** The accounts whose object did not pass. */ failed: RestoreDrillAccount[]; + /** Objects of the newest date the run stopped before, on its budget. */ + unchecked: number; } function countArray(value: unknown): number { @@ -178,6 +203,7 @@ async function drillObject( export async function runRestoreDrill( s3Override?: S3Like, now: Date = new Date(), + shouldStop: () => boolean = () => false, ): Promise { const cfg = loadOffhostConfig(); if (!cfg) { @@ -208,6 +234,8 @@ export async function runRestoreDrill( const accounts: RestoreDrillAccount[] = []; for (const objectKey of newest) { + // Between accounts, never inside one: the job's budget is spent. + if (shouldStop()) break; const account: RestoreDrillAccount = { objectKey, ok: true, @@ -239,6 +267,7 @@ export async function runRestoreDrill( stale: ageDays > MAX_BACKUP_AGE_DAYS, accounts, failed: accounts.filter((a) => !a.ok), + unchecked: newest.length - accounts.length, }; } @@ -252,10 +281,10 @@ export async function runRestoreDrill( export async function handleRestoreDrill( jobs: Job[], ): Promise { - void jobs; + const shouldStop = jobBudget(jobs); return withBackgroundEvent("job.restore_drill", async (evt) => { try { - const report = await runRestoreDrill(); + const report = await runRestoreDrill(undefined, new Date(), shouldStop); const total = (pick: (a: RestoreDrillAccount) => number) => report.accounts.reduce((sum, a) => sum + pick(a), 0); const meta = { @@ -270,6 +299,7 @@ export async function handleRestoreDrill( restore_drill_intake_events: total((a) => a.recordCounts.intakeEvents), restore_drill_mood_entries: total((a) => a.recordCounts.moodEntries), restore_drill_stale: report.stale, + restore_drill_unchecked: report.unchecked, }; for (const [key, value] of Object.entries(meta)) evt.addMeta(key, value); evt.addMeta( @@ -301,9 +331,18 @@ export async function handleRestoreDrill( { dateKey: report.dateKey, ageDays: report.ageDays }, ); } + if (report.unchecked > 0) { + await reportWorkerError( + RESTORE_DRILL_QUEUE, + new Error( + `Restore drill stopped on its time budget with ${report.unchecked} of ${report.accounts.length + report.unchecked} backup objects unchecked.`, + ), + { dateKey: report.dateKey, unchecked: report.unchecked }, + ); + } if (report.failed.length > 0) { - // Already paged above, per account. A retry reads the same objects - // and cannot change the verdict before the next upload. + // Already paged above, per account. The queue does not retry: a + // retry reads the same objects and would page the same accounts again. return jobFailed( `restore drill: ${report.failed.length} of ${report.accounts.length} accounts failed`, new Error(report.failed.map((a) => a.objectKey).join(", ")),