diff --git a/docs/self-hosting/scaling.md b/docs/self-hosting/scaling.md index 6d703d9ab..6ecac0731 100644 --- a/docs/self-hosting/scaling.md +++ b/docs/self-hosting/scaling.md @@ -268,8 +268,10 @@ turns the limit off). `canceling statement due to statement timeout`. - The admin status page lists `rollup-full-backfill` under failing background jobs, and every admin gets a push once a system job has - failed three times in a row, a reminder at most once a day while it keeps - failing, and a notice when it runs again. + failed three jobs in a row (retries of one job count once; a job that runs + per account counts per account), a reminder at most once a day while it + keeps failing, and a notice when the failing job runs again. A one-off job + that failed and does not run again is announced once. ### What the app does about it diff --git a/src/lib/jobs/__tests__/system-job-alert.test.ts b/src/lib/jobs/__tests__/system-job-alert.test.ts index 9719308d4..224fb0209 100644 --- a/src/lib/jobs/__tests__/system-job-alert.test.ts +++ b/src/lib/jobs/__tests__/system-job-alert.test.ts @@ -13,68 +13,140 @@ import { } from "@/lib/jobs/system-job-alert"; const at = new Date("2026-10-29T10:00:00.000Z"); +const HOUR_MS = 3_600_000; -function run(overrides: Partial = {}): TerminalRun { +/** The `ago`-th newest job (0 = newest), an hour apart. */ +function run(ago = 0, overrides: Partial = {}): TerminalRun { return { + key: "", state: "failed", - retryCount: 0, allFailed: false, - completedOn: at, + completedOn: new Date(at.getTime() - ago * HOUR_MS), error: "boom", expireSeconds: null, ...overrides, }; } +/** Jobs newest first, one hour apart, from a list of outcomes. */ +function history(key: string, outcomes: Array<"fail" | "ok">, from = 0) { + return outcomes.map((o, i) => + run(from + i, { + key, + state: o === "fail" ? "failed" : "completed", + error: o === "fail" ? `boom ${key}` : null, + }), + ); +} + describe("judgeQueueRuns", () => { it("is inactive with no run left in the ledger", () => { expect(judgeQueueRuns([]).state).toBe("inactive"); }); - it("counts every attempt of a run, so one job that used up its retries is a streak", () => { - const verdict = judgeQueueRuns([run({ retryCount: 3 })]); - expect(verdict).toMatchObject({ state: "failing", failures: 4 }); - expect(verdict.lastError).toBe("boom"); - }); - - it("calls the streak at the threshold, not before", () => { - const below = Array.from({ length: SYSTEM_JOB_FAILURE_STREAK - 1 }, () => - run(), - ); - expect(judgeQueueRuns(below).state).toBe("pending"); - expect(judgeQueueRuns([...below, run()]).state).toBe("failing"); - }); - - it("ends the streak at the newest success", () => { - const verdict = judgeQueueRuns([ - run({ error: "newest" }), - run({ state: "completed", error: null }), - run(), - run(), - run(), - ]); - expect(verdict).toMatchObject({ - state: "ok", - failures: 1, - lastError: "newest", + it("counts jobs, not attempts: one job that used up its retries is one failure", () => { + expect(judgeQueueRuns([run(0)])).toMatchObject({ + state: "pending", + failures: 0, }); }); - it("counts a completed run that says every candidate failed", () => { - const allFailed = run({ state: "completed", allFailed: true }); - expect(judgeQueueRuns([allFailed, allFailed, allFailed]).state).toBe( - "failing", + it("calls the streak at the threshold, not before", () => { + const below = Array.from( + { length: SYSTEM_JOB_FAILURE_STREAK - 1 }, + (_, i) => run(i), ); + expect(judgeQueueRuns(below).state).toBe("pending"); + const verdict = judgeQueueRuns([...below, run(below.length)]); + expect(verdict).toMatchObject({ + state: "failing", + failures: SYSTEM_JOB_FAILURE_STREAK, + lastError: "boom", + lastFailureAt: at, + }); + }); + + it("does not report a recovery while the newest job failed", () => { + expect( + judgeQueueRuns(history("", ["fail", "ok", "fail", "fail", "fail"])), + ).toMatchObject({ state: "pending", failures: 0 }); + expect( + judgeQueueRuns(history("", ["ok", "fail", "fail", "fail"])).state, + ).toBe("ok"); + }); + + it("has nothing to report for a queue that never failed", () => { + expect(judgeQueueRuns(history("", ["ok", "ok"])).state).toBe("inactive"); + }); + + it("counts a completed job that says every candidate failed", () => { + const allFailed = (ago: number) => + run(ago, { state: "completed", allFailed: true }); + expect( + judgeQueueRuns([allFailed(0), allFailed(1), allFailed(2)]).state, + ).toBe("failing"); }); it("explains a bare timeout the way the admin card does", () => { - const verdict = judgeQueueRuns([ - run({ error: "job timed out", expireSeconds: 3600, retryCount: 2 }), - ]); + const verdict = judgeQueueRuns( + [0, 1, 2].map((ago) => + run(ago, { error: "job timed out", expireSeconds: 3600 }), + ), + ); expect(verdict.lastError).toMatch( /^job timed out: did not finish within its 1 h limit/, ); }); + + describe("per account", () => { + it("is failing while one account fails every hour and the others succeed", () => { + // Interleaved: the queue's newest job is always another account's + // success, which used to read as a recovery every other hour. + const verdict = judgeQueueRuns([ + ...history("a", ["fail", "fail", "fail"], 1), + ...history("b", ["ok", "ok", "ok", "ok"], 0), + ...history("c", ["ok", "ok", "ok"], 0), + ]); + expect(verdict).toMatchObject({ + state: "failing", + failures: 3, + lastError: "boom a", + }); + }); + + it("does not build a streak from three accounts failing once each", () => { + const verdict = judgeQueueRuns([ + ...history("a", ["fail"], 0), + ...history("b", ["fail"], 1), + ...history("c", ["fail"], 2), + ]); + expect(verdict.state).toBe("pending"); + }); + + it("recovers only when the failing account itself succeeds", () => { + const others = history("b", ["ok", "ok"], 0); + expect( + judgeQueueRuns([ + ...history("a", ["fail", "fail", "fail"], 0), + ...others, + ]).state, + ).toBe("failing"); + expect( + judgeQueueRuns([ + ...history("a", ["ok", "fail", "fail", "fail"], 0), + ...others, + ]).state, + ).toBe("ok"); + }); + + it("reports the account that failed most recently", () => { + const verdict = judgeQueueRuns([ + ...history("old", ["fail", "fail", "fail", "fail"], 5), + ...history("new", ["fail", "fail", "fail"], 0), + ]); + expect(verdict).toMatchObject({ failures: 3, lastError: "boom new" }); + }); + }); }); describe("watchedQueues", () => { diff --git a/src/lib/jobs/system-job-alert.ts b/src/lib/jobs/system-job-alert.ts index d6174eec7..ff6187462 100644 --- a/src/lib/jobs/system-job-alert.ts +++ b/src/lib/jobs/system-job-alert.ts @@ -1,6 +1,6 @@ /** * Push to every admin when a background job of the instance keeps failing: - * once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` runs in a row, a + * once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` jobs in a row, a * reminder at most once a day while it keeps failing, and once when it runs * successfully again. Nothing else. * @@ -12,11 +12,28 @@ * account-facing status of its own (the `system` audience in * `subsystem-surface.ts`). * - * The streak is read from pg-boss's own ledger, newest terminal run first: a - * run that failed (or completed and said every candidate failed, - * `did.all_failed`) adds its attempts, retries included, and the first run - * that succeeded ends the streak. A queue with no terminal run left in the - * ledger is not judged at all. + * The streak is read from pg-boss's own ledger, newest terminal job first, + * and it counts jobs, not attempts: a job that used up its retries is one + * failure. A job that failed, or completed and said every candidate failed + * (`did.all_failed`), adds one; the first job that succeeded ends it. + * + * Many of these queues run one job per account (a rollup recompute, a + * backfill, a score), so the streak is kept per job key: the account the job + * carries (`data.userId`), or the whole queue for a job that carries none. + * Not the `singletonKey`: it often names a bucket, a window or a single + * workout, which makes nearly every job its own key, and a queue failing on + * every job would then never build a streak. A queue is failing while any + * key has failed `SYSTEM_JOB_FAILURE_STREAK` jobs in a row, and it has + * recovered only when no key is failing and a key that had failed ran + * successfully since; another account's success says nothing about the one + * that fails. One account failing every hour while the rest succeed is + * therefore one problem notice and one recovery, not a notice each hour. + * + * A reminder needs a failure newer than the notice before it. A queue that + * keeps failing is reminded once a day; a one-shot job (a boot backfill) that + * failed and does not run again is announced once and then left to the admin + * page, instead of a reminder every day until pg-boss deletes its row. A + * queue with nothing left in the ledger to judge is forgotten silently. * * The memory is one row per announced queue (`SystemJobAlertNotice`), claimed * with a compare-and-set before the notice is sent, exactly as the readiness @@ -41,21 +58,24 @@ import { type StoredNotice, } from "@/lib/recovery/readiness-notify"; -/** Failed runs in a row, retries included, before the admins are told. */ +/** Failed jobs in a row, of one job key, before the admins are told. */ export const SYSTEM_JOB_FAILURE_STREAK = 3; -/** Newest terminal runs read per queue. */ -const LOOKBACK_ROWS = 50; +/** Newest terminal jobs read per queue, across every key. */ +const LOOKBACK_ROWS = 5000; + +/** Newest terminal jobs judged per job key. */ +const LOOKBACK_ROWS_PER_KEY = 20; /** Where the notice takes the admin: the page with the failing-jobs card. */ const ADMIN_STATUS_PATH = "/admin/system-status"; -/** One terminal run as the ledger holds it, newest first. */ +/** One terminal job as the ledger holds it. */ export interface TerminalRun { + /** The account the job ran for (`data.userId`), or "" for the queue. */ + key: string; state: "completed" | "failed"; - /** Retries pg-boss ran before this run settled. */ - retryCount: number; - /** The run completed but reported that every candidate failed. */ + /** The job completed but reported that every candidate failed. */ allFailed: boolean; completedOn: Date; error: string | null; @@ -65,39 +85,66 @@ export interface TerminalRun { /** What the ledger says about one queue. */ export interface QueueVerdict { /** - * `failing`: the streak reached the threshold. `ok`: the newest run, or a - * run after the last failures, succeeded. `pending`: fewer failures than - * the threshold and no success to judge by. `inactive`: no run left. + * `failing`: some key reached the streak. `ok`: no key is failing, and a + * key that had failed has since run successfully. `pending`: a failure + * short of the streak, and no recovery to report. `inactive`: nothing to + * report, no job left or none that failed. */ state: "failing" | "ok" | "pending" | "inactive"; - /** Failed attempts in a row, newest first. */ + /** Failed jobs in a row of the key that failed most recently. */ failures: number; - /** The newest failure's message, redacted and truncated. */ + /** That key's newest failure, redacted and truncated. */ lastError: string; + /** When the newest failure of a failing key settled, if any key is. */ + lastFailureAt: Date | null; } -/** Pure: judge one queue's newest terminal runs. */ +function failed(run: TerminalRun): boolean { + return run.state === "failed" || run.allFailed; +} + +/** Pure: judge one queue's terminal jobs, in any order. */ export function judgeQueueRuns(runs: readonly TerminalRun[]): QueueVerdict { - if (runs.length === 0) - return { state: "inactive", failures: 0, lastError: "" }; - let failures = 0; - let lastError: string | null = null; - let succeeded = false; + const byKey = new Map(); for (const run of runs) { - if (run.state === "completed" && !run.allFailed) { - succeeded = true; - break; - } - failures += run.retryCount + 1; - lastError ??= presentError(run.error, run.expireSeconds); + const list = byKey.get(run.key) ?? []; + list.push(run); + byKey.set(run.key, list); } - const state = - failures >= SYSTEM_JOB_FAILURE_STREAK - ? "failing" - : succeeded - ? "ok" - : "pending"; - return { state, failures, lastError: lastError ?? "" }; + let failing: { streak: number; newest: TerminalRun } | null = null; + let recovered = false; + let anyFailure = false; + for (const list of byKey.values()) { + list.sort((a, b) => b.completedOn.getTime() - a.completedOn.getTime()); + let streak = 0; + while (streak < list.length && failed(list[streak])) streak += 1; + if (list.some(failed)) anyFailure = true; + if (streak === 0) { + // Newest job succeeded: a recovery, when this key had failed before. + if (list.some(failed)) recovered = true; + continue; + } + if ( + streak >= SYSTEM_JOB_FAILURE_STREAK && + (failing === null || + list[0].completedOn.getTime() > failing.newest.completedOn.getTime()) + ) { + failing = { streak, newest: list[0] }; + } + } + if (failing) { + return { + state: "failing", + failures: failing.streak, + lastError: presentError( + failing.newest.error, + failing.newest.expireSeconds, + ), + lastFailureAt: failing.newest.completedOn, + }; + } + const state = recovered ? "ok" : anyFailure ? "pending" : "inactive"; + return { state, failures: 0, lastError: "", lastFailureAt: null }; } /** The queues this alert watches: every one whose failure has no other surface. */ @@ -108,9 +155,11 @@ export function watchedQueues(): string[] { } /** - * The newest terminal runs of each queue, newest first, one statement over - * pg-boss's `(name, completed_on)` index. `null` when the queue schema is not - * there (a web-only deployment): no queue to ask is not a healthy queue. + * The newest terminal jobs of each queue, keyed by account, one statement + * over pg-boss's `(name, completed_on)` index: the newest `LOOKBACK_ROWS` of + * the queue, and of those the newest `LOOKBACK_ROWS_PER_KEY` of each key. + * `null` when the queue schema is not there (a web-only deployment): no + * queue to ask is not a healthy queue. */ export async function readTerminalRuns( queues: readonly string[], @@ -119,40 +168,48 @@ export async function readTerminalRuns( const rows = await prisma.$queryRaw< Array<{ name: string; + job_key: string; state: string; - retry_count: number; all_failed: string | null; completed_on: Date; error: string | null; expire_seconds: number | null; }> >` - SELECT q.name, j.state, j.retry_count, j.all_failed, j.completed_on, - j.error, j.expire_seconds + SELECT q.name, k.job_key, k.state, k.all_failed, k.completed_on, + k.error, k.expire_seconds FROM unnest(${queues as string[]}::text[]) AS q(name) CROSS JOIN LATERAL ( - SELECT state, retry_count, completed_on, expire_seconds, - output->'did'->>'all_failed' AS all_failed, - COALESCE( - output->>'message', - output->'value'->>'message', - output::text - ) AS error - FROM pgboss.job - WHERE name = q.name - AND completed_on IS NOT NULL - AND state IN ('completed', 'failed') - ORDER BY completed_on DESC - LIMIT ${LOOKBACK_ROWS} - ) j - ORDER BY q.name, j.completed_on DESC + SELECT j.*, + ROW_NUMBER() OVER ( + PARTITION BY j.job_key ORDER BY j.completed_on DESC + ) AS key_rank + FROM ( + SELECT state, completed_on, expire_seconds, + COALESCE(data->>'userId', '') AS job_key, + output->'did'->>'all_failed' AS all_failed, + COALESCE( + output->>'message', + output->'value'->>'message', + output::text + ) AS error + FROM pgboss.job + WHERE name = q.name + AND completed_on IS NOT NULL + AND state IN ('completed', 'failed') + ORDER BY completed_on DESC + LIMIT ${LOOKBACK_ROWS} + ) j + ) k + WHERE k.key_rank <= ${LOOKBACK_ROWS_PER_KEY} + ORDER BY q.name, k.completed_on DESC `; const byQueue = new Map(); for (const row of rows) { const runs = byQueue.get(row.name) ?? []; runs.push({ + key: row.job_key, state: row.state === "failed" ? "failed" : "completed", - retryCount: Number(row.retry_count ?? 0), allFailed: row.all_failed === "true", completedOn: row.completed_on, error: row.error, @@ -222,6 +279,16 @@ export async function notifySystemJobFailures( // With no admin yet there is nobody to tell, so nothing is claimed: the // first admin hears about a queue that is still failing. if (!decision.next || admins.length === 0) continue; + // Nothing failed since the last notice: a job that does not run again is + // not news a second time. + if ( + decision.send === "reminder" && + stored?.notifiedAt && + verdict.lastFailureAt && + verdict.lastFailureAt.getTime() <= stored.notifiedAt.getTime() + ) { + continue; + } // Claim: only the evaluation that moves the row from what it read sends. const claimed = row diff --git a/tests/integration/system-job-alert.test.ts b/tests/integration/system-job-alert.test.ts index bb267bf76..dd9518143 100644 --- a/tests/integration/system-job-alert.test.ts +++ b/tests/integration/system-job-alert.test.ts @@ -4,7 +4,10 @@ * A background job of the instance that keeps failing pushes every admin * once, a reminder at most once a day while it keeps failing, and once when it * runs successfully again. Fewer failures than the streak, and queues that - * have an account-facing surface of their own, push nothing. + * have an account-facing surface of their own, push nothing. The streak + * counts jobs, not retries, and is kept per account: one account failing + * every hour beside accounts that succeed is one problem and one recovery, + * and a job that does not run again is announced once, not every day. * * pg-boss provisions its own schema here and the jobs are failed and completed * through its own API, as in `job-failures.test.ts`, so the reader is proven @@ -54,19 +57,32 @@ const HOUR_MS = 3_600_000; let boss: PgBoss; -async function runOnce(queue: string, outcome: "fail" | "complete") { - const id = await boss.send(queue, { userId: "u" }, { retryLimit: 0 }); +async function runOnce( + queue: string, + outcome: "fail" | "complete", + userId = "u", + retryLimit = 0, +) { + const id = await boss.send( + queue, + { userId }, + { retryLimit, retryDelay: 0, retryBackoff: false }, + ); if (!id) throw new Error("queue rejected the job"); - const [job] = await boss.fetch(queue); - if (!job) throw new Error("nothing to fetch"); - if (outcome === "fail") { - await boss.fail( - queue, - job.id, - new Error("canceling statement due to statement timeout"), - ); - } else { - await boss.complete(queue, job.id, { ok: true }); + // Every attempt, retries included, until pg-boss settles the job. + for (let attempt = 0; attempt <= retryLimit; attempt++) { + const [job] = await boss.fetch(queue); + if (!job) throw new Error("nothing to fetch"); + if (outcome === "fail") { + await boss.fail( + queue, + job.id, + new Error(`canceling statement due to statement timeout (${userId})`), + ); + } else { + await boss.complete(queue, job.id, { ok: true }); + return; + } } } @@ -127,7 +143,8 @@ describe("notifySystemJobFailures", () => { expect(again.sent).toEqual([]); expect(dispatched.calls).toEqual([]); - // A day later: one reminder. + // It fails once more; a day after the notice: one reminder. + await runOnce(SYSTEM_QUEUE, "fail"); const reminder = await notifySystemJobFailures( new Date(now.getTime() + 25 * HOUR_MS), ); @@ -175,4 +192,58 @@ describe("notifySystemJobFailures", () => { expect(outcome.failing).toBe(0); expect(outcome.sent).toEqual([]); }); + + it("counts a job that used up its retries as one failure", async () => { + await makeUser({ role: "ADMIN" }); + await runOnce(SYSTEM_QUEUE, "fail", "u", 2); + const outcome = await notifySystemJobFailures(); + expect(outcome.failing).toBe(0); + expect(outcome.sent).toEqual([]); + }); + + it("tells once about one account failing every hour beside accounts that succeed", async () => { + await makeUser({ role: "ADMIN" }); + const start = Date.now(); + const sent: string[] = []; + // Ten hours: account `a` fails every hour, the others succeed after it, + // so the queue's newest job is always a success. + for (let hour = 0; hour < 10; hour++) { + await runOnce(SYSTEM_QUEUE, "fail", "a"); + await runOnce(SYSTEM_QUEUE, "complete", "b"); + await runOnce(SYSTEM_QUEUE, "complete", "c"); + const outcome = await notifySystemJobFailures( + new Date(start + hour * HOUR_MS), + ); + sent.push(...outcome.sent.map((n) => n.kind)); + } + expect(sent).toEqual(["problem"]); + expect(String(dispatched.calls[0].params?.error)).toContain("(a)"); + + // `a` succeeds: one recovery. + await runOnce(SYSTEM_QUEUE, "complete", "a"); + const recovered = await notifySystemJobFailures( + new Date(start + 10 * HOUR_MS), + ); + expect(recovered.sent).toEqual([ + { queue: SYSTEM_QUEUE, kind: "recovered" }, + ]); + }); + + it("does not remind about a job that has not run again since the notice", async () => { + await makeUser({ role: "ADMIN" }); + for (let i = 0; i < SYSTEM_JOB_FAILURE_STREAK; i++) { + await runOnce(SYSTEM_QUEUE, "fail"); + } + const now = new Date(Date.now() + 1_000); + expect((await notifySystemJobFailures(now)).sent).toEqual([ + { queue: SYSTEM_QUEUE, kind: "problem" }, + ]); + for (const days of [1, 2, 5]) { + const later = await notifySystemJobFailures( + new Date(now.getTime() + days * 24 * HOUR_MS), + ); + expect(later.failing).toBe(1); + expect(later.sent).toEqual([]); + } + }); });