From b7a26d3a4d105f5f5c1613c9b35a3e71602be565 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 25 Sep 2026 01:58:06 +0200 Subject: [PATCH] fix(backup): resume an off-host run where its last attempt stopped The off-host backup walked every account in one job with the default fifteen-minute expiry, and a retry started again from the first account. On an instance with one large record next to others, an attempt that ran out of time spent its retry uploading the accounts it had already done (#1031). The run now stops between accounts once the job's time budget is spent and fails, so pg-boss retries it. Every attempt of a job carries the same creation time; an attempt skips the accounts whose ledger shows a success since then, and keys its objects on that day, so a retry after midnight still writes under the run's date. A new job uploads every account again, and a retry whose accounts were all done passes. --- src/lib/jobs/job-outcome.ts | 1 + src/lib/jobs/offhost-backup.ts | 54 +++++++++++++++++-- .../__tests__/offhost-backup-handler.test.ts | 23 ++++++++ src/lib/jobs/reminder/backup-handlers.ts | 31 +++++++++-- src/lib/jobs/reminder/register-maintenance.ts | 3 +- .../integration/offhost-backup-ledger.test.ts | 46 ++++++++++++++++ 6 files changed, 149 insertions(+), 9 deletions(-) diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index f57210e09..2f25991f4 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -138,6 +138,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "notifications_dispatched", "notifications_failed", "notified", + "offhost_backup_already_uploaded", "offhost_backup_configured", "offhost_backup_failed", "offhost_backup_oversized", diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index 19bb54f84..1805448bf 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -593,9 +593,33 @@ interface BackupRunReport { largestObjectBytes: number; /** Accounts refused for size rather than failed for a reason. */ oversized: number; + /** + * Accounts an earlier attempt of the same run already uploaded, and this + * one therefore skipped (see `RunOffhostBackupOptions.runStartedAt`). + */ + alreadyUploaded: number; + /** `shouldStop` ended the run before every account was reached. */ + stoppedEarly: boolean; } -export type RunOffhostBackupOptions = UploadBackupOptions; +export interface RunOffhostBackupOptions extends UploadBackupOptions { + /** + * When this run first started: the pg-boss job's creation time, the same + * on every attempt of the job. An attempt skips each account whose ledger + * shows a successful upload since then, because an earlier attempt of the + * same run already put its object in the bucket, and the objects are keyed + * on this date so a retry after midnight still writes under the run's day. + * Without it a retry started again from the first account (#1031), and on + * a cohort with one large record the retry never got past the accounts the + * first attempt had already done. + */ + runStartedAt?: Date; + /** + * Asked before each account. Returning `true` ends the run cleanly between + * accounts, before the job's expiry would cut one off halfway. + */ + shouldStop?: () => boolean; +} export async function runOffhostBackup( prisma: PrismaClient, @@ -610,9 +634,23 @@ export async function runOffhostBackup( ); } const s3 = s3Override ?? (await getS3Client(cfg)); - const dateKey = now.toISOString().slice(0, 10); + const runStartedAt = options.runStartedAt; + const dateKey = (runStartedAt ?? now).toISOString().slice(0, 10); - const users = await prisma.user.findMany({ select: { id: true } }); + const users = await prisma.user.findMany({ + select: { id: true }, + orderBy: { id: "asc" }, + }); + const doneThisRun = new Set(); + if (runStartedAt) { + const done = await prisma.offhostBackupState.findMany({ + where: { lastSuccessAt: { gte: runStartedAt } }, + select: { userId: true }, + }); + for (const row of done) doneThisRun.add(row.userId); + } + let alreadyUploaded = 0; + let stoppedEarly = false; let uploaded = 0; let failed = 0; let oversized = 0; @@ -621,6 +659,14 @@ export async function runOffhostBackup( let ledgerWriteFailures = 0; const evt = getEvent(); for (const user of users) { + if (doneThisRun.has(user.id)) { + alreadyUploaded++; + continue; + } + if (options.shouldStop?.()) { + stoppedEarly = true; + break; + } let objectBytes: number | null = null; try { objectBytes = await uploadEncryptedBackup( @@ -721,6 +767,8 @@ export async function runOffhostBackup( totalUsers: users.length, largestObjectBytes, oversized, + alreadyUploaded, + stoppedEarly, }; } diff --git a/src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts b/src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts index 41bfb935b..b1d880103 100644 --- a/src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts +++ b/src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts @@ -53,6 +53,8 @@ function report(over: Partial> = {}) { totalUsers: 2, largestObjectBytes: 9_000, oversized: 0, + alreadyUploaded: 0, + stoppedEarly: false, ...over, }; } @@ -67,6 +69,8 @@ describe("handleOffhostBackup", () => { const outcome = await handleOffhostBackup([]); expect(outcome.ok).toBe(true); expect(outcome.ok && outcome.did).toEqual({ + offhost_backup_already_uploaded: 0, + stopped_early: false, offhost_backup_uploaded: 2, offhost_backup_failed: 0, offhost_backup_total_users: 2, @@ -138,4 +142,23 @@ describe("handleOffhostBackup", () => { offhost_backup_configured: false, }); }); + + it("fails an attempt that ran out of budget, so pg-boss retries and the retry resumes (#1031)", async () => { + mocks.runOffhostBackup.mockResolvedValue( + report({ uploaded: 1, totalUsers: 3, stoppedEarly: true }), + ); + const outcome = await handleOffhostBackup([]); + expect(outcome).toMatchObject({ + ok: false, + reason: expect.stringContaining("retry resumes"), + }); + }); + + it("passes a retry whose accounts an earlier attempt had all uploaded", async () => { + mocks.runOffhostBackup.mockResolvedValue( + report({ uploaded: 0, alreadyUploaded: 2, totalUsers: 2 }), + ); + const outcome = await handleOffhostBackup([]); + expect(outcome.ok).toBe(true); + }); }); diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 41a114b1d..0a3b69ac5 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -4,7 +4,8 @@ * Extracted from reminder-worker.ts, which owns the queue names, cron * schedules, and boss.work registrations. */ -import { type Job } from "pg-boss"; +import { type Job, type JobWithMetadata } from "pg-boss"; +import { jobBudget } from "@/lib/jobs/job-budget"; import { recordError } from "@/lib/jobs/worker-status"; import { BackupBlobTooLargeError } from "@/lib/export/backup-blob"; import { storeBackupBlob } from "@/lib/export/store-backup-blob"; @@ -27,13 +28,18 @@ export interface OffhostBackupPayload { } export async function handleOffhostBackup( - jobs: Job[], + jobs: JobWithMetadata[], ): Promise { - void jobs; return withBackgroundEvent("job.offhost_backup", async (evt) => { const p = getWorkerPrisma(); try { - const report = await runOffhostBackup(p); + // A retry of this job carries the same creation time, so it resumes + // after the accounts an earlier attempt uploaded, and the run stops + // between accounts before the job's expiry would cut one off. + const report = await runOffhostBackup(p, undefined, new Date(), { + runStartedAt: jobs[0]?.createdOn, + shouldStop: jobBudget(jobs), + }); evt.addMeta("offhost_backup_uploaded", report.uploaded); evt.addMeta("offhost_backup_failed", report.failed); evt.addMeta("offhost_backup_total_users", report.totalUsers); @@ -55,6 +61,8 @@ export async function handleOffhostBackup( } const did = { + offhost_backup_already_uploaded: report.alreadyUploaded, + stopped_early: report.stoppedEarly, offhost_backup_uploaded: report.uploaded, offhost_backup_failed: report.failed, offhost_backup_total_users: report.totalUsers, @@ -69,7 +77,20 @@ export async function handleOffhostBackup( // than one. Per-account failures still ride out as counts when SOME // account got a copy: that is the fan-out rule, and retrying the whole // cohort over one object would re-upload everybody's. - if (report.totalUsers > 0 && report.uploaded === 0) { + // Out of budget before every account: fail, so pg-boss retries the job, + // and the retry resumes after the accounts this attempt uploaded. + if (report.stoppedEarly) { + return jobFailed( + "stopped before every account was uploaded; the retry resumes", + undefined, + did, + ); + } + + if ( + report.totalUsers > 0 && + report.uploaded + report.alreadyUploaded === 0 + ) { // The SDK's own words, not a stack: `runJob` puts the cause message in // the reported meta, and "SignatureDoesNotMatch" is the sentence an // operator can act on. diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 4d60ab352..30507c800 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -724,10 +724,11 @@ export async function registerMaintenanceQueues( { localConcurrency: 1 }, handleStepUpElevationCleanup, ); + // With metadata: the handler resumes a retry from the job's creation time. await createAndWork( boss, OFFHOST_BACKUP_QUEUE, - { localConcurrency: 1 }, + { localConcurrency: 1, includeMetadata: true }, handleOffhostBackup, ); await createAndWork( diff --git a/tests/integration/offhost-backup-ledger.test.ts b/tests/integration/offhost-backup-ledger.test.ts index f2047fc9a..a4718865c 100644 --- a/tests/integration/offhost-backup-ledger.test.ts +++ b/tests/integration/offhost-backup-ledger.test.ts @@ -228,4 +228,50 @@ describe("off-host backup ledger (real Postgres)", () => { await prisma.user.delete({ where: { id: "account-one" } }); expect(await prisma.offhostBackupState.count()).toBe(0); }); + + it("a retry of the same run resumes after the accounts it already uploaded (#1031)", async () => { + const prisma = getPrismaClient(); + for (const id of ["acct-a", "acct-b", "acct-c"]) await seedUser(id); + const bucket = makeBucket(); + const puts: string[] = []; + const putStream = bucket.putStream; + bucket.putStream = async (key, body) => { + puts.push(key); + return putStream(key, body); + }; + // The job was created a moment ago; its retries share that instant. + const runStartedAt = new Date(Date.now() - 60_000); + const day = runStartedAt.toISOString().slice(0, 10); + + // First attempt: out of budget after one account. + let asked = 0; + const first = await runOffhostBackup(prisma, bucket, new Date(), { + runStartedAt, + shouldStop: () => asked++ >= 1, + }); + expect(first.stoppedEarly).toBe(true); + expect(first.uploaded).toBe(1); + + // The retry: same run, so the account already done is skipped. + const retry = await runOffhostBackup(prisma, bucket, new Date(), { + runStartedAt, + }); + expect(retry.stoppedEarly).toBe(false); + expect(retry.alreadyUploaded).toBe(1); + expect(retry.uploaded).toBe(2); + + // Each account uploaded exactly once, all under the run's own day. + expect(puts.sort()).toEqual([ + `${day}/user-acct-a.json.enc`, + `${day}/user-acct-b.json.enc`, + `${day}/user-acct-c.json.enc`, + ]); + + // A new run (a later job) uploads everybody again. + const next = await runOffhostBackup(prisma, bucket, new Date(), { + runStartedAt: new Date(), + }); + expect(next.alreadyUploaded).toBe(0); + expect(next.uploaded).toBe(3); + }); });