fix(account-deletion): one failing erasure no longer holds up the others

The hourly purge rethrew any error that was not the Guardian refusal, so the
oldest due account, failing on every run, stopped every later erasure and the
sweep of expired pre-wipe copies for good.

Each account is now its own try: a failure is counted, reported through the
worker error path with its first error, and told to the admins once per
request (user.account.erasure_failed), and the run goes on. The pre-wipe
sweep and the off-host kick run in a finally. Every account the run starts is
marked in backup_pass_attempts under the purge's queue name and the mark
closes only on erasure or when the account is no longer due, so an account
that failed or was refused goes behind every other due account next time.
The run stops starting accounts once its job budget is spent, since fifty
erasures of up to two minutes each did not fit the queue's expiry.

src/lib/jobs/__tests__/account-deletion-purge.test.ts

@@ -0,0 +1,229 @@
+/**
+ * One account whose erasure throws must not hold up any other erasure, the
+ * sweep of expired pre-wipe copies, or the off-host kick; and it goes behind
+ * the other due accounts next time.
+ *
+ * Mutations that turn this red: rethrow a non-Guardian error from the loop
+ * (b and c stay, the sweep never runs); move the sweep out of the `finally`;
+ * drop the attempt marks or `orderInterruptedLast` (the failing account is
+ * first again in the second run).
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+type Attempt = {
+  userId: string;
+  pass: string;
+  startedAt: Date;
+  finishedAt: Date | null;
+};
+
+const state = vi.hoisted(() => ({
+  due: [] as Array<{ id: string; managedProfileAt: Date | null }>,
+  poison: new Set<string>(),
+  refuse: new Set<string>(),
+  attempted: [] as string[],
+  erased: [] as string[],
+  preWipeSweeps: 0,
+  attempts: [] as Attempt[],
+  workerErrors: [] as unknown[],
+  notices: [] as Array<Record<string, unknown>>,
+  audits: [] as Array<{ action: string; userId: string }>,
+  clock: 0,
+}));
+
+vi.mock("@/lib/db", () => {
+  const tick = () => new Date(1_000_000 + ++state.clock);
+  return {
+    prisma: {
+      user: {
+        findMany: vi.fn(async (args: { where?: { role?: string } }) =>
+          args.where?.role ? [{ id: "admin" }] : state.due,
+        ),
+        findUnique: vi.fn(async ({ where }: { where: { id: string } }) => ({
+          username: where.id,
+          deletionRequestedAt: new Date(0),
+        })),
+      },
+      auditLog: {
+        count: vi.fn(
+          async ({ where }: { where: { userId: string; action: string } }) =>
+            state.audits.filter(
+              (a) => a.userId === where.userId && a.action === where.action,
+            ).length,
+        ),
+      },
+      backupRestoreJob: { findMany: vi.fn(async () => []) },
+      dataBackup: {
+        deleteMany: vi.fn(async () => {
+          state.preWipeSweeps++;
+          return { count: 3 };
+        }),
+      },
+      backupPassAttempt: {
+        upsert: vi.fn(
+          async ({ create }: { create: { userId: string; pass: string } }) => {
+            const row = state.attempts.find(
+              (a) => a.userId === create.userId && a.pass === create.pass,
+            );
+            if (row) row.startedAt = tick();
+            else
+              state.attempts.push({
+                ...create,
+                startedAt: tick(),
+                finishedAt: null,
+              });
+          },
+        ),
+        updateMany: vi.fn(
+          async ({ where }: { where: { userId: string; pass: string } }) => {
+            for (const a of state.attempts) {
+              if (a.userId === where.userId && a.pass === where.pass) {
+                a.finishedAt = tick();
+              }
+            }
+          },
+        ),
+        findMany: vi.fn(async ({ where }: { where: { pass: string } }) =>
+          state.attempts.filter((a) => a.pass === where.pass),
+        ),
+      },
+      $transaction: vi.fn(),
+    },
+  };
+});
+vi.mock("@/lib/managed-profiles/lifecycle", () => {
+  class LastManagedGuardianError extends Error {}
+  return {
+    ACCOUNT_DELETION_TRANSACTION_OPTIONS: {},
+    LastManagedGuardianError,
+    withManagedProfileLock: vi.fn(),
+    deleteGuardianAccountWithLifecycle: vi.fn(
+      async (id: string, fn: (tx: unknown) => Promise<void>) => {
+        state.attempted.push(id);
+        if (state.poison.has(id)) {
+          throw new Error(
+            "Transaction already closed: timeout 120000ms exceeded",
+          );
+        }
+        if (state.refuse.has(id)) throw new LastManagedGuardianError();
+        await fn({});
+      },
+    ),
+  };
+});
+vi.mock("@/lib/account-deletion/lifecycle", () => ({
+  lockPendingAccount: vi.fn(async () => ({
+    pending: true,
+    purgeAt: new Date(0),
+  })),
+  eraseAccount: vi.fn(async (_tx: unknown, id: string) => {
+    state.erased.push(id);
+  }),
+}));
+vi.mock("@/lib/jobs/offhost-purge", () => ({ kickOffhostPurge: vi.fn() }));
+vi.mock("@/lib/auth/audit", () => ({
+  auditLog: vi.fn(async (action: string, opts: { userId: string }) => {
+    state.audits.push({ action, userId: opts.userId });
+  }),
+}));
+vi.mock("@/lib/notifications/dispatch-localised", () => ({
+  dispatchLocalisedNotification: vi.fn(async (o: Record<string, unknown>) => {
+    state.notices.push(o);
+  }),
+}));
+vi.mock("@/lib/jobs/report-worker-error", () => ({
+  reportWorkerError: vi.fn(async (_q: string, err: unknown) => {
+    state.workerErrors.push(err);
+  }),
+}));
+vi.mock("@/lib/logging/background", () => ({
+  withBackgroundEvent: (_n: string, f: (e: unknown) => unknown) =>
+    f({ addMeta() {}, addWarning() {} }),
+}));
+
+const { processDueAccountDeletions, handleAccountDeletionPurge } =
+  await import("@/lib/jobs/account-deletion-purge");
+const { kickOffhostPurge } = await import("@/lib/jobs/offhost-purge");
+
+const account = (id: string) => ({ id, managedProfileAt: null });
+
+beforeEach(() => {
+  state.due = [account("poison"), account("b"), account("c")];
+  state.poison = new Set(["poison"]);
+  state.refuse = new Set();
+  state.attempted = [];
+  state.erased = [];
+  state.preWipeSweeps = 0;
+  state.attempts = [];
+  state.workerErrors = [];
+  state.notices = [];
+  state.audits = [];
+  vi.mocked(kickOffhostPurge).mockClear();
+});
+
+describe("the account purge with one erasure that keeps failing", () => {
+  it("erases every other due account, sweeps pre-wipe copies and kicks the off-host purge", async () => {
+    const report = await processDueAccountDeletions(new Date());
+    expect(state.erased).toEqual(["b", "c"]);
+    expect(report).toMatchObject({ purged: 2, failed: 1, refused: 0 });
+    expect(state.preWipeSweeps).toBe(1);
+    expect(kickOffhostPurge).toHaveBeenCalledTimes(1);
+  });
+
+  it("takes the failing account after the others in the next run", async () => {
+    await processDueAccountDeletions(new Date());
+    state.due = [account("poison"), account("d")];
+    state.attempted = [];
+    await processDueAccountDeletions(new Date());
+    expect(state.attempted).toEqual(["d", "poison"]);
+  });
+
+  it("puts refused accounts behind the others too, longest waiting first", async () => {
+    state.poison = new Set();
+    state.refuse = new Set(["r1", "r2"]);
+    state.due = [account("r1"), account("r2")];
+    await processDueAccountDeletions(new Date());
+    state.due = [account("r1"), account("r2"), account("e")];
+    state.attempted = [];
+    await processDueAccountDeletions(new Date());
+    expect(state.attempted).toEqual(["e", "r1", "r2"]);
+  });
+
+  it("sweeps pre-wipe copies even when the due accounts cannot be read", async () => {
+    const { prisma } = await import("@/lib/db");
+    vi.mocked(prisma.user.findMany).mockRejectedValueOnce(
+      new Error("connection reset"),
+    );
+    await expect(processDueAccountDeletions(new Date())).rejects.toThrow(
+      /connection reset/,
+    );
+    expect(state.preWipeSweeps).toBe(1);
+  });
+
+  it("stops starting accounts when its budget is spent, and says how many it left", async () => {
+    let calls = 0;
+    const report = await processDueAccountDeletions(new Date(), {
+      shouldStop: () => ++calls > 1,
+    });
+    expect(state.attempted).toEqual(["poison"]);
+    expect(report.deferred).toBe(2);
+    expect(state.preWipeSweeps).toBe(1);
+  });
+
+  it("reports the failure through the worker error path and tells the admins once", async () => {
+    const outcome = await handleAccountDeletionPurge([]);
+    expect(outcome).toMatchObject({ ok: true });
+    expect(String(state.workerErrors[0])).toMatch(
+      /1 account erasure\(s\) failed.*timeout 120000ms/,
+    );
+    expect(state.notices).toEqual([
+      expect.objectContaining({
+        userId: "admin",
+        messageKey: "notifications.admin.accountErasureFailedBody",
+        params: { username: "poison" },
+      }),
+    ]);
+    await handleAccountDeletionPurge([]);
+    expect(state.notices).toHaveLength(1);
+  });
+});

src/lib/jobs/account-deletion-purge.ts

@@ -18,10 +18,28 @@
  * window would leave a managed profile nobody can reach, so that erasure is
  * refused, the account stays locked and pending, the run reports it, and
  * every admin is told once through the localized admin notification.
+ *
+ * One account never holds up the others. An erasure that throws (a record
+ * too large for the transaction's time limit, a lock a restore holds) is
+ * caught, counted, reported through the worker error path and to the admins
+ * once, and the run goes on with the next account. Every account the run
+ * starts is marked (`backup_pass_attempts`, pass `account-deletion-purge`)
+ * and the mark is closed only when the account is erased or no longer due,
+ * so an account that failed or was refused is taken after every other due
+ * account in the following runs, the longest-waiting first. The run stops
+ * starting accounts once its time budget is spent; the sweep of expired
+ * pre-wipe copies and the off-host kick run whatever happened before them.
  */
 import type { Job } from "pg-boss";
 
 import { prisma } from "@/lib/db";
+import {
+  markBackupAttemptFinished,
+  markBackupAttemptStarted,
+  orderInterruptedLast,
+  readInterruptedBackupAttempts,
+} from "@/lib/jobs/backup-pass-attempts";
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { kickOffhostPurge } from "@/lib/jobs/offhost-purge";
 import { reportWorkerError } from "@/lib/jobs/report-worker-error";
@@ -43,14 +61,26 @@ import { dispatchLocalisedNotification } from "@/lib/notifications/dispatch-loca
 import { PRE_WIPE_BACKUP_TYPE_PREFIX } from "@/lib/account-deletion/pre-wipe-type";
 import { ACCOUNT_DELETION_PURGE_QUEUE } from "./account-deletion-purge-queue";
 
-/** Accounts one run erases; the next run takes the rest. */
-const ACCOUNTS_PER_RUN = 50;
+/**
+ * Due accounts one run reads. The run erases as many as its time budget
+ * allows; the next run takes the rest.
+ */
+const ACCOUNTS_PER_RUN = 200;
+
+/** The attempt ledger's key for this job. */
+const PASS = ACCOUNT_DELETION_PURGE_QUEUE;
 
 interface AccountDeletionPurgeReport {
   due: number;
   purged: number;
   /** Erasures refused by the Guardian invariant; the accounts stay pending. */
   refused: number;
+  /** Erasures that threw; the accounts stay pending and are tried again. */
+  failed: number;
+  /** The first error of the failed erasures, for the worker error report. */
+  firstFailure: unknown;
+  /** Due accounts the run left for the next one when its budget ran out. */
+  deferred: number;
   preWipeCopiesDeleted: number;
 }
 
@@ -132,18 +162,33 @@ async function deleteExpiredPreWipeCopies(now: Date): Promise<number> {
   return count;
 }
 
-/** The audit action that marks a refused erasure as already reported. */
-const REFUSED_ERASURE_ACTION = "user.account.deletion_refused";
+/**
+ * Why an erasure did not happen, the audit action that marks it as already
+ * reported, and the notice the admins get.
+ */
+const NOT_ERASED = {
+  last_managed_guardian: {
+    action: "user.account.deletion_refused",
+    messageKey: "notifications.admin.accountErasureRefusedBody",
+  },
+  erasure_failed: {
+    action: "user.account.erasure_failed",
+    messageKey: "notifications.admin.accountErasureFailedBody",
+  },
+} as const;
 
 /**
  * Tell every admin, once, that an account whose window ended could not be
  * erased, and why. The run retries the account every hour; the audit row on
- * the account (`user.account.deletion_refused`, newer than the deletion
- * request) is what keeps the second hour from repeating the message. A
- * deletion undone and asked for again is a new request and is reported
- * afresh.
+ * the account (one action per reason, newer than the deletion request) is
+ * what keeps the second hour from repeating the message. A deletion undone
+ * and asked for again is a new request and is reported afresh.
  */
-async function notifyAdminsOfRefusedErasure(accountId: string): Promise<void> {
+async function notifyAdminsOfUnerasedAccount(
+  accountId: string,
+  reason: keyof typeof NOT_ERASED,
+): Promise<void> {
+  const { action, messageKey } = NOT_ERASED[reason];
   const account = await prisma.user.findUnique({
     where: { id: accountId },
     select: { username: true, deletionRequestedAt: true },
@@ -152,15 +197,15 @@ async function notifyAdminsOfRefusedErasure(accountId: string): Promise<void> {
   const reported = await prisma.auditLog.count({
     where: {
       userId: accountId,
-      action: REFUSED_ERASURE_ACTION,
+      action,
       createdAt: { gte: account.deletionRequestedAt },
     },
   });
   if (reported > 0) return;
-  await auditLog(REFUSED_ERASURE_ACTION, {
+  await auditLog(action, {
     userId: accountId,
     actorUserId: null,
-    details: { reason: "last_managed_guardian" },
+    details: { reason },
   });
   const admins = await prisma.user.findMany({
     where: activeAccount({ role: "ADMIN" }),
@@ -170,65 +215,118 @@ async function notifyAdminsOfRefusedErasure(accountId: string): Promise<void> {
     await dispatchLocalisedNotification({
       userId: admin.id,
       titleKey: "notifications.admin.accountErasureRefusedTitle",
-      messageKey: "notifications.admin.accountErasureRefusedBody",
+      messageKey,
       params: { username: account.username },
       metadata: { source: "account-deletion-purge", url: "/admin/users" },
     });
   }
 }
 
+/** A notice that could not be sent must not stop the run. */
+async function notifyQuietly(
+  accountId: string,
+  reason: keyof typeof NOT_ERASED,
+): Promise<void> {
+  try {
+    await notifyAdminsOfUnerasedAccount(accountId, reason);
+  } catch (err) {
+    await reportWorkerError(ACCOUNT_DELETION_PURGE_QUEUE, err);
+  }
+}
+
 export async function processDueAccountDeletions(
   now: Date = new Date(),
+  options: { shouldStop?: () => boolean } = {},
 ): Promise<AccountDeletionPurgeReport> {
-  // account-deletion: the purge's own query: exactly the accounts pending deletion.
-  const due = await prisma.user.findMany({
-    where: {
-      deletionRequestedAt: { not: null },
-      OR: [{ deletionPurgeAt: null }, { deletionPurgeAt: { lte: now } }],
-    },
-    select: { id: true, managedProfileAt: true },
-    orderBy: { deletionPurgeAt: "asc" },
-    take: ACCOUNTS_PER_RUN,
-  });
-
+  const shouldStop = options.shouldStop ?? (() => false);
   const report: AccountDeletionPurgeReport = {
-    due: due.length,
+    due: 0,
     purged: 0,
     refused: 0,
+    failed: 0,
+    firstFailure: null,
+    deferred: 0,
     preWipeCopiesDeleted: 0,
   };
-  for (const account of due) {
-    try {
-      if (
-        await purgeAccount(account.id, account.managedProfileAt !== null, now)
-      ) {
-        report.purged++;
+  try {
+    // account-deletion: the purge's own query: exactly the accounts pending deletion.
+    const due = await prisma.user.findMany({
+      where: {
+        deletionRequestedAt: { not: null },
+        OR: [{ deletionPurgeAt: null }, { deletionPurgeAt: { lte: now } }],
+      },
+      select: { id: true, managedProfileAt: true },
+      orderBy: [{ deletionPurgeAt: "asc" }, { id: "asc" }],
+      take: ACCOUNTS_PER_RUN,
+    });
+    report.due = due.length;
+    // An account whose last erasure failed, was refused or was cut off goes
+    // behind every other due account.
+    const ordered = orderInterruptedLast(
+      due,
+      await readInterruptedBackupAttempts(prisma, PASS),
+    );
+    for (const [index, account] of ordered.entries()) {
+      if (shouldStop()) {
+        report.deferred = ordered.length - index;
+        break;
+      }
+      await markBackupAttemptStarted(prisma, PASS, account.id);
+      try {
+        if (
+          await purgeAccount(account.id, account.managedProfileAt !== null, now)
+        ) {
+          report.purged++;
+        }
+        // Erased (the mark went with the account) or no longer due.
+        await markBackupAttemptFinished(prisma, PASS, account.id);
+      } catch (error) {
+        if (error instanceof LastManagedGuardianError) {
+          report.refused++;
+          await notifyQuietly(account.id, "last_managed_guardian");
+        } else {
+          report.failed++;
+          report.firstFailure ??= error;
+          await notifyQuietly(account.id, "erasure_failed");
+        }
       }
-    } catch (error) {
-      if (!(error instanceof LastManagedGuardianError)) throw error;
-      report.refused++;
-      await notifyAdminsOfRefusedErasure(account.id);
     }
+  } finally {
+    // The window of a pre-wipe copy ends whatever happened to the accounts.
+    report.preWipeCopiesDeleted = await deleteExpiredPreWipeCopies(now);
+    // The off-host copies of every erased account, now rather than at the
+    // nightly slot. The requests committed with the erasures.
+    if (report.purged > 0) await kickOffhostPurge();
   }
-  report.preWipeCopiesDeleted = await deleteExpiredPreWipeCopies(now);
-
-  // The off-host copies of every erased account, now rather than at the
-  // nightly slot. The requests committed with the erasures.
-  if (report.purged > 0) await kickOffhostPurge();
   return report;
 }
 
 export async function handleAccountDeletionPurge(
   jobs: Job<object>[],
 ): Promise<JobOutcome> {
-  void jobs;
   return withBackgroundEvent("job.account_deletion_purge", async (evt) => {
     try {
-      const report = await processDueAccountDeletions();
+      const report = await processDueAccountDeletions(new Date(), {
+        shouldStop: jobBudget(jobs),
+      });
       evt.addMeta("account_deletions_due", report.due);
       evt.addMeta("account_deletions_purged", report.purged);
       evt.addMeta("account_deletions_refused", report.refused);
+      evt.addMeta("account_deletions_failed", report.failed);
+      evt.addMeta("account_deletions_deferred", report.deferred);
       evt.addMeta("pre_wipe_copies_deleted", report.preWipeCopiesDeleted);
+      if (report.failed > 0) {
+        const cause =
+          report.firstFailure instanceof Error
+            ? report.firstFailure.message
+            : String(report.firstFailure);
+        await reportWorkerError(
+          ACCOUNT_DELETION_PURGE_QUEUE,
+          new Error(
+            `${report.failed} account erasure(s) failed and will be tried again next hour, after the other due accounts. The accounts stay locked and pending. First error: ${cause}`,
+          ),
+        );
+      }
       if (report.refused > 0) {
         await reportWorkerError(
           ACCOUNT_DELETION_PURGE_QUEUE,
@@ -241,6 +339,8 @@ export async function handleAccountDeletionPurge(
         account_deletions_due: report.due,
         account_deletions_purged: report.purged,
         account_deletions_refused: report.refused,
+        account_deletions_failed: report.failed,
+        account_deletions_deferred: report.deferred,
         pre_wipe_copies_deleted: report.preWipeCopiesDeleted,
       });
     } catch (err) {

src/lib/jobs/backup-pass-attempts.ts

@@ -31,8 +31,13 @@
 import type { PrismaClient } from "@/generated/prisma/client";
 import { logCaught } from "@/lib/logging/signal";
 
-/** The pass's queue name, which is also the marker's key. */
-export type BackupPass = "data-backup" | "data-backup-offhost";
+/**
+ * The pass's queue name, which is also the marker's key. The account-deletion
+ * purge walks the due accounts the same way and keeps its marks here too: an
+ * account whose erasure failed or was refused stays open and is taken last.
+ */
+export type BackupPass =
+  "data-backup" | "data-backup-offhost" | "account-deletion-purge";
 
 type Db = Pick<PrismaClient, "backupPassAttempt">;