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

@@ -138,6 +138,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "notifications_dispatched",
   "notifications_failed",
   "notified",
+  "offhost_backup_already_uploaded",
   "offhost_backup_configured",
   "offhost_backup_failed",
   "offhost_backup_oversized",

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

src/lib/jobs/reminder/__tests__/offhost-backup-handler.test.ts

@@ -53,6 +53,8 @@ function report(over: Partial<Record<string, unknown>> = {}) {
     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);
+  });
 });

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<OffhostBackupPayload>[],
+  jobs: JobWithMetadata<OffhostBackupPayload>[],
 ): Promise<JobOutcome> {
-  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.

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<OffhostBackupPayload>(
     boss,
     OFFHOST_BACKUP_QUEUE,
-    { localConcurrency: 1 },
+    { localConcurrency: 1, includeMetadata: true },
     handleOffhostBackup,
   );
   await createAndWork(

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