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); + }); });