From a6f4c386fedeab9bff9928628905f00232a2af4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 10:58:04 +0100 Subject: [PATCH] fix(jobs): judge a system job's failure streak per job, per account The alert added every attempt of a failed job to the streak, so one job with two retries read as three failures. It also judged a queue by its newest job across all accounts: on a queue that runs one job per account, one account failing every hour beside accounts that succeed never reached the streak, or flipped between failing and recovered. The streak now counts jobs, and is kept per account (the job's userId, or the whole queue for jobs without one). A queue is failing while any account has failed three jobs in a row, and recovered only when a failing account has itself run successfully since. A reminder needs a failure newer than the notice before it, so a one-off job that failed and does not run again is announced once. --- docs/self-hosting/scaling.md | 6 +- .../jobs/__tests__/system-job-alert.test.ts | 144 ++++++++++---- src/lib/jobs/system-job-alert.ts | 185 ++++++++++++------ tests/integration/system-job-alert.test.ts | 99 ++++++++-- 4 files changed, 323 insertions(+), 111 deletions(-) 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([]); + } + }); });