mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 13:01:00 +00:00
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.
This commit is contained in:
1 parent
4be7109a2b
commit
4261139dfa
5 files changed
+94
-9
No files matched your search
@@ -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(),
|
||||
|
||||
@@ -200,6 +200,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
|
||||
"restore_drill_medications",
|
||||
"restore_drill_mood_entries",
|
||||
"restore_drill_plaintext_bytes",
|
||||
"restore_drill_unchecked",
|
||||
"restore_drill_stale",
|
||||
"restore_fail_cause",
|
||||
"restore_failed",
|
||||
|
||||
@@ -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<Record<string, QueueRuntime>> = {
|
||||
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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<RestoreDrillReport> {
|
||||
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<object>[],
|
||||
): Promise<JobOutcome> {
|
||||
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(", ")),
|
||||
|
||||
Reference in new issue
Block a user