diff --git a/src/lib/jobs/__tests__/backup-pass-attempts.test.ts b/src/lib/jobs/__tests__/backup-pass-attempts.test.ts new file mode 100644 index 000000000..722fec5ff --- /dev/null +++ b/src/lib/jobs/__tests__/backup-pass-attempts.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, it } from "vitest"; + +import { + heldBackAccounts, + orderInterruptedLast, +} from "../backup-pass-attempts"; + +type Row = { userId: string; startedAt: Date; finishedAt: Date | null }; + +/** + * A weekly-style pass over `ids`: oldest copy first, held-back accounts last, + * at most `budget` accounts per run, `kills` the accounts whose record kills + * the process. Returns the accounts copied in each run. + */ +function simulate(opts: { + ids: string[]; + budget: number; + kills: Set; + runs: number; + interruptedFirst?: string; +}): string[][] { + let clock = 0; + const attempts = new Map(); + const copiedAt = new Map(); + if (opts.interruptedFirst) { + attempts.set(opts.interruptedFirst, { + userId: opts.interruptedFirst, + startedAt: new Date(++clock), + finishedAt: null, + }); + } + const copied: string[][] = []; + for (let run = 0; run < opts.runs; run += 1) { + const order = orderInterruptedLast( + [...opts.ids] + .map((id) => ({ id })) + .sort( + (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1), + ), + heldBackAccounts([...attempts.values()]), + ); + const thisRun: string[] = []; + for (const { id } of order.slice(0, opts.budget)) { + const row = attempts.get(id) ?? { + userId: id, + startedAt: new Date(0), + finishedAt: null, + }; + row.startedAt = new Date(++clock); + attempts.set(id, row); + if (opts.kills.has(id)) break; // the process dies; the run ends here + row.finishedAt = new Date(++clock); + copiedAt.set(id, clock); + thisRun.push(id); + } + copied.push(thisRun); + } + return copied; +} + +describe("heldBackAccounts", () => { + const at = (ms: number) => new Date(ms); + + it("holds back an interrupted attempt that is the newest of the pass", () => { + const held = heldBackAccounts([ + { userId: "x", startedAt: at(30), finishedAt: at(10) }, + { userId: "a", startedAt: at(20), finishedAt: at(21) }, + ]); + expect([...held.keys()]).toEqual(["x"]); + }); + + it("releases it once a later run has started on another account", () => { + const held = heldBackAccounts([ + { userId: "x", startedAt: at(30), finishedAt: null }, + { userId: "a", startedAt: at(40), finishedAt: at(41) }, + ]); + expect(held.size).toBe(0); + }); + + it("ignores finished attempts", () => { + const held = heldBackAccounts([ + { userId: "a", startedAt: at(40), finishedAt: at(41) }, + ]); + expect(held.size).toBe(0); + }); +}); + +describe("the order across runs", () => { + const ids = ["x", "a", "b", "c", "d"]; + + it("keeps a record that kills the process from costing the others their copies", () => { + // `x` has the oldest copy and kills every run that reaches it. + const runs = simulate({ ids, budget: 99, kills: new Set(["x"]), runs: 6 }); + // The first run meets it unwarned; from then on it goes last every run, + // because each crash makes its attempt the newest again. + expect(runs[0]).toEqual([]); + for (const run of runs.slice(1)) { + expect([...run].sort()).toEqual(["a", "b", "c", "d"]); + } + }); + + it("still copies an interrupted account when every run stops on its budget", () => { + // `x` was interrupted once (a restart, not its record). Every run stops + // after three accounts, so kept last for good it would never be reached. + const runs = simulate({ + ids, + budget: 3, + kills: new Set(), + runs: 4, + interruptedFirst: "x", + }); + const firstCopy = runs.findIndex((run) => run.includes("x")); + expect(firstCopy).toBeGreaterThanOrEqual(0); + expect(firstCopy).toBeLessThanOrEqual(1); + // And every account gets a copy within the four runs. + expect(new Set(runs.flat())).toEqual(new Set(ids)); + }); +}); diff --git a/src/lib/jobs/backup-pass-attempts.ts b/src/lib/jobs/backup-pass-attempts.ts index 05c0d919b..389252334 100644 --- a/src/lib/jobs/backup-pass-attempts.ts +++ b/src/lib/jobs/backup-pass-attempts.ts @@ -16,6 +16,16 @@ * takes that account last, after everybody else has their copy, and the admin * backups page names it. Every write and read here is best effort: the marker * is bookkeeping about a backup, and must never be the reason one fails. + * + * Last only once. A pass that regularly stops on its time budget never reaches + * the end of its list, so an account kept last for as long as its attempt + * stays open would never be attempted again, and would never get a copy. An + * interrupted account is therefore held back only until a later run has + * started on another account (`accountsToTakeLast`): from then on it takes its + * ordinary place again. If it kills the process again, its new attempt is the + * newest one and it goes last once more. So a crash loop costs the other + * accounts at most every second run, and the account itself is attempted at + * least every second run. */ import type { PrismaClient } from "@/generated/prisma/client"; @@ -83,7 +93,51 @@ export async function readInterruptedBackupAttempts( } /** - * `accounts` in their order, with every interrupted one moved behind the + * The interrupted accounts a pass should take last: those whose interrupted + * attempt is still the newest attempt of the pass, i.e. no later run has + * started on another account since. Rows are one per account. Pure, so the + * rule can be tested apart from the database. + */ +export function heldBackAccounts( + rows: ReadonlyArray<{ + userId: string; + startedAt: Date; + finishedAt: Date | null; + }>, +): Map { + const out = new Map(); + for (const row of rows) { + const open = row.finishedAt === null || row.finishedAt < row.startedAt; + if (!open) continue; + const passedOver = rows.some( + (other) => other.userId !== row.userId && other.startedAt > row.startedAt, + ); + if (!passedOver) out.set(row.userId, row.startedAt); + } + return out; +} + +/** + * What a pass orders by: the interrupted accounts it takes last + * (`heldBackAccounts`). Empty when the table cannot be read. + */ +export async function readAccountsToTakeLast( + prisma: Db, + pass: BackupPass, +): Promise> { + try { + const rows = await prisma.backupPassAttempt.findMany({ + where: { pass }, + select: { userId: true, startedAt: true, finishedAt: true }, + }); + return heldBackAccounts(rows); + } catch { + return new Map(); + } +} + +/** + * `accounts` in their order, with every held-back one moved behind the * rest; among those, the one interrupted longest ago first. */ export function orderInterruptedLast( diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index ad9cf04dd..05a9f23b5 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -39,7 +39,7 @@ import { markBackupAttemptFinished, markBackupAttemptStarted, orderInterruptedLast, - readInterruptedBackupAttempts, + readAccountsToTakeLast, } from "@/lib/jobs/backup-pass-attempts"; import { BACKUP_HEARTBEAT_SECONDS } from "@/lib/jobs/data-backup-policy"; import { annotate, getEvent } from "@/lib/logging/context"; @@ -930,14 +930,15 @@ export async function runOffhostBackup( // By id, with an account whose last walk never came back behind the rest: // a record that killed the process once would otherwise lead every retry - // and keep every account after it out of the bucket - // (`backup-pass-attempts.ts`). + // and keep every account after it out of the bucket. Behind them for one + // run only, so a pass that stops on its budget still reaches it the run + // after (`backup-pass-attempts.ts`). const users = orderInterruptedLast( await prisma.user.findMany({ select: { id: true }, orderBy: { id: "asc" }, }), - await readInterruptedBackupAttempts(prisma, OFFHOST_BACKUP_QUEUE), + await readAccountsToTakeLast(prisma, OFFHOST_BACKUP_QUEUE), ); const doneThisRun = new Set(); if (runStartedAt) { diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index d4bb4b1a4..62e1adeda 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -27,7 +27,7 @@ import { markBackupAttemptFinished, markBackupAttemptStarted, orderInterruptedLast, - readInterruptedBackupAttempts, + readAccountsToTakeLast, } from "@/lib/jobs/backup-pass-attempts"; import { DATA_BACKUP_QUEUE } from "@/lib/jobs/data-backup-policy"; @@ -154,7 +154,7 @@ export async function handleDataBackup( where: { type: "WEEKLY_AUTO" }, select: { userId: true, createdAt: true }, }), - readInterruptedBackupAttempts(prisma, DATA_BACKUP_QUEUE), + readAccountsToTakeLast(prisma, DATA_BACKUP_QUEUE), ]); const copiedAt = new Map( copies.map((copy) => [copy.userId, copy.createdAt.getTime()]), @@ -163,7 +163,9 @@ export async function handleDataBackup( // attempt never came back goes behind all of them: it is the one with // the oldest copy, so it would lead every retry, and a record that // killed the process once would keep every account after it from a - // copy (`backup-pass-attempts.ts`). + // copy. Only until a later run has started on another account, so a + // pass that stops on its budget still reaches it the run after + // (`backup-pass-attempts.ts`). const users = orderInterruptedLast( [...accounts].sort( (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1),