diff --git a/messages/de.json b/messages/de.json index 0fe8a7223..05c1c847b 100644 --- a/messages/de.json +++ b/messages/de.json @@ -9243,6 +9243,13 @@ "titleFailing": "Sicherungen sind gefährdet", "titleReminder": "Sicherungen sind weiterhin gefährdet", "titleRecovered": "Sicherungen laufen wieder normal" + }, + "systemJobAlert": { + "titleFailing": "Hintergrundjob schlägt fehl: {queue}", + "titleReminder": "Hintergrundjob schlägt weiter fehl: {queue}", + "titleRecovered": "Hintergrundjob läuft wieder: {queue}", + "bodyFailing": "Der Hintergrundjob {queue} ist {failures}-mal in Folge fehlgeschlagen und steht auf der Admin-Statusseite. Letzter Fehler: {error}", + "bodyRecovered": "Der Hintergrundjob {queue} ist wieder erfolgreich gelaufen." } }, "i18n": { diff --git a/messages/en.json b/messages/en.json index 3f99d31fe..c1593952e 100644 --- a/messages/en.json +++ b/messages/en.json @@ -9243,6 +9243,13 @@ "titleFailing": "Backups are at risk", "titleReminder": "Backups are still at risk", "titleRecovered": "Backups are back to normal" + }, + "systemJobAlert": { + "titleFailing": "Background job failing: {queue}", + "titleReminder": "Background job still failing: {queue}", + "titleRecovered": "Background job working again: {queue}", + "bodyFailing": "The background job {queue} has failed {failures} times in a row and is listed on the admin status page. Last error: {error}", + "bodyRecovered": "The background job {queue} has run successfully again." } }, "i18n": { diff --git a/messages/es.json b/messages/es.json index 4677c684f..67997a3c0 100644 --- a/messages/es.json +++ b/messages/es.json @@ -9243,6 +9243,13 @@ "titleFailing": "Las copias de seguridad están en riesgo", "titleReminder": "Las copias de seguridad siguen en riesgo", "titleRecovered": "Las copias de seguridad vuelven a la normalidad" + }, + "systemJobAlert": { + "titleFailing": "Tarea en segundo plano con fallos: {queue}", + "titleReminder": "La tarea en segundo plano sigue fallando: {queue}", + "titleRecovered": "La tarea en segundo plano vuelve a funcionar: {queue}", + "bodyFailing": "La tarea en segundo plano {queue} ha fallado {failures} veces seguidas y aparece en la página de estado de administración. Último error: {error}", + "bodyRecovered": "La tarea en segundo plano {queue} se ha vuelto a ejecutar correctamente." } }, "i18n": { diff --git a/messages/fr.json b/messages/fr.json index 76a13377a..e6a3ba82f 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -9243,6 +9243,13 @@ "titleFailing": "Les sauvegardes sont menacées", "titleReminder": "Les sauvegardes sont toujours menacées", "titleRecovered": "Les sauvegardes sont revenues à la normale" + }, + "systemJobAlert": { + "titleFailing": "Tâche d'arrière-plan en échec : {queue}", + "titleReminder": "Tâche d'arrière-plan toujours en échec : {queue}", + "titleRecovered": "Tâche d'arrière-plan rétablie : {queue}", + "bodyFailing": "La tâche d'arrière-plan {queue} a échoué {failures} fois de suite ; vous la trouverez sur la page d'état de l'administration. Dernière erreur : {error}", + "bodyRecovered": "La tâche d'arrière-plan {queue} s'est de nouveau exécutée avec succès." } }, "i18n": { diff --git a/messages/it.json b/messages/it.json index 0585ad838..c82cf8855 100644 --- a/messages/it.json +++ b/messages/it.json @@ -9243,6 +9243,13 @@ "titleFailing": "I backup sono a rischio", "titleReminder": "I backup sono ancora a rischio", "titleRecovered": "I backup sono tornati alla normalità" + }, + "systemJobAlert": { + "titleFailing": "Attività in background non riuscita: {queue}", + "titleReminder": "Attività in background ancora non riuscita: {queue}", + "titleRecovered": "Attività in background di nuovo funzionante: {queue}", + "bodyFailing": "L'attività in background {queue} non è riuscita {failures} volte di seguito ed è elencata nella pagina di stato dell'amministrazione. Ultimo errore: {error}", + "bodyRecovered": "L'attività in background {queue} è stata di nuovo eseguita correttamente." } }, "i18n": { diff --git a/messages/ko.json b/messages/ko.json index 2a2b86dd3..62a8b12cf 100644 --- a/messages/ko.json +++ b/messages/ko.json @@ -9243,6 +9243,13 @@ "titleFailing": "백업이 위험해요", "titleReminder": "백업이 여전히 위험해요", "titleRecovered": "백업이 다시 정상이에요" + }, + "systemJobAlert": { + "titleFailing": "백그라운드 작업이 실패하고 있어요: {queue}", + "titleReminder": "백그라운드 작업이 계속 실패하고 있어요: {queue}", + "titleRecovered": "백그라운드 작업이 다시 작동해요: {queue}", + "bodyFailing": "백그라운드 작업 {queue}이(가) {failures}번 연속 실패했어요. 관리자 상태 페이지에서 확인할 수 있어요. 마지막 오류: {error}", + "bodyRecovered": "백그라운드 작업 {queue}이(가) 다시 정상적으로 실행됐어요." } }, "i18n": { diff --git a/messages/nl.json b/messages/nl.json index e93adec7c..3315d44cb 100644 --- a/messages/nl.json +++ b/messages/nl.json @@ -9243,6 +9243,13 @@ "titleFailing": "Back-ups lopen gevaar", "titleReminder": "Back-ups lopen nog steeds gevaar", "titleRecovered": "Back-ups zijn weer in orde" + }, + "systemJobAlert": { + "titleFailing": "Achtergrondtaak mislukt: {queue}", + "titleReminder": "Achtergrondtaak mislukt nog steeds: {queue}", + "titleRecovered": "Achtergrondtaak werkt weer: {queue}", + "bodyFailing": "De achtergrondtaak {queue} is {failures} keer achter elkaar mislukt en staat op de beheerstatuspagina. Laatste fout: {error}", + "bodyRecovered": "De achtergrondtaak {queue} is weer met succes uitgevoerd." } }, "i18n": { diff --git a/messages/pl.json b/messages/pl.json index 2055df78e..27c0711d7 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -9243,6 +9243,13 @@ "titleFailing": "Kopie zapasowe są zagrożone", "titleReminder": "Kopie zapasowe są nadal zagrożone", "titleRecovered": "Kopie zapasowe znów działają normalnie" + }, + "systemJobAlert": { + "titleFailing": "Zadanie w tle kończy się błędem: {queue}", + "titleReminder": "Zadanie w tle nadal kończy się błędem: {queue}", + "titleRecovered": "Zadanie w tle znów działa: {queue}", + "bodyFailing": "Zadanie w tle {queue} zakończyło się błędem {failures} razy z rzędu i jest widoczne na stronie stanu w panelu administratora. Ostatni błąd: {error}", + "bodyRecovered": "Zadanie w tle {queue} znów wykonało się poprawnie." } }, "i18n": { diff --git a/messages/pt-BR.json b/messages/pt-BR.json index ae8919d20..d8ff915ca 100644 --- a/messages/pt-BR.json +++ b/messages/pt-BR.json @@ -9243,6 +9243,13 @@ "titleFailing": "Os backups estão em risco", "titleReminder": "Os backups continuam em risco", "titleRecovered": "Os backups voltaram ao normal" + }, + "systemJobAlert": { + "titleFailing": "Tarefa em segundo plano falhando: {queue}", + "titleReminder": "Tarefa em segundo plano ainda falhando: {queue}", + "titleRecovered": "Tarefa em segundo plano funcionando de novo: {queue}", + "bodyFailing": "A tarefa em segundo plano {queue} falhou {failures} vezes seguidas e aparece na página de status da administração. Último erro: {error}", + "bodyRecovered": "A tarefa em segundo plano {queue} voltou a ser executada com sucesso." } }, "i18n": { diff --git a/prisma/migrations/0490_system_job_alert_notices/migration.sql b/prisma/migrations/0490_system_job_alert_notices/migration.sql new file mode 100644 index 000000000..fa78e55e2 --- /dev/null +++ b/prisma/migrations/0490_system_job_alert_notices/migration.sql @@ -0,0 +1,24 @@ +-- System job alert: what the admins were last told about a background job of +-- the instance that keeps failing. +-- +-- `system_job_alert_notices` holds, per pg-boss queue, the state the admins +-- were last notified about and when. It is the whole memory of the hourly +-- evaluation (`src/lib/jobs/system-job-alert.ts`): a queue that has failed +-- three runs in a row notifies once, is reminded at most once a day while it +-- keeps failing, and its first successful run after that is announced once. +-- A row exists only for a queue something was announced about. +-- +-- One new table, nothing else touched, so the migration takes no lock on an +-- existing table and running it again changes nothing. +-- +-- Reversibility: dropping the table loses only the memory of what was +-- announced; the next evaluation announces a still-failing queue once more. + +CREATE TABLE IF NOT EXISTS "system_job_alert_notices" ( + "queue" TEXT NOT NULL, + "notified_state" TEXT, + "notified_at" TIMESTAMP(3), + CONSTRAINT "system_job_alert_notices_pkey" PRIMARY KEY ("queue"), + CONSTRAINT "system_job_alert_notices_state_check" + CHECK ("notified_state" IS NULL OR "notified_state" IN ('ok', 'failing')) +); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 648b62a75..0e566b0a6 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -5958,6 +5958,23 @@ model RecoveryReadinessNotice { @@map("recovery_readiness_notices") } +/// Per background queue of the instance, the state the admins were last +/// notified about. The hourly system job evaluation compares a queue's run +/// history against it: three failed runs in a row notify once, a queue still +/// failing is reminded at most once a day, and its first successful run after +/// that is announced once. The update is a compare-and-set on both columns, so +/// two workers never send the same notice. Written and read by +/// `notifySystemJobFailures`. +model SystemJobAlertNotice { + /// The pg-boss queue name (`rollup-full-backfill`, …). + queue String @id + /// `ok` or `failing`; null once a queue's history has aged out. + notifiedState String? @map("notified_state") + notifiedAt DateTime? @map("notified_at") + + @@map("system_job_alert_notices") +} + /// The recovery kit an admin confirmed: every key a restore needs, by kind, /// id and fingerprint, never the key itself. The kit is due again as soon as /// a key it needs is not in this set (a rotation, a new backup key). Written diff --git a/src/__tests__/error-level-discipline-guard.test.ts b/src/__tests__/error-level-discipline-guard.test.ts index edbc87410..c3432c8be 100644 --- a/src/__tests__/error-level-discipline-guard.test.ts +++ b/src/__tests__/error-level-discipline-guard.test.ts @@ -163,6 +163,11 @@ const ALLOWED: Readonly> = { count: 1, reason: "No /proc/diskstats outside Linux; see captureHostMetric.", }, + "lib/jobs/system-job-alert.ts::readTerminalRuns": { + count: 1, + reason: + "No pgboss schema on a web-only deployment; null means no queue to ask, and the evaluation then judges nothing.", + }, "lib/jobs/job-failures.ts::readActiveJobsStartedBefore": { count: 1, reason: diff --git a/src/lib/data-wipe/wipe-plan.ts b/src/lib/data-wipe/wipe-plan.ts index 9edc727b8..6900918ed 100644 --- a/src/lib/data-wipe/wipe-plan.ts +++ b/src/lib/data-wipe/wipe-plan.ts @@ -358,6 +358,8 @@ export const INSTANCE_SCOPED: Readonly> = { "the monthly restore drill's last verdict for the instance's off-host bucket; says nothing about any account", RecoveryReadinessNotice: "per backup signal, what the admins were last told; a fact about the instance's backups, not about any account", + SystemJobAlertNotice: + "per background queue, what the admins were last told about its failures; a fact about the instance's jobs, not about any account", RecoveryKitConfirmation: "the instance's recovery kit confirmation (key ids and fingerprints); the confirming admin is a plain id, not a relation", DatabaseBackupRun: diff --git a/src/lib/jobs/__tests__/system-job-alert.test.ts b/src/lib/jobs/__tests__/system-job-alert.test.ts new file mode 100644 index 000000000..9719308d4 --- /dev/null +++ b/src/lib/jobs/__tests__/system-job-alert.test.ts @@ -0,0 +1,87 @@ +/** + * The system job alert's verdict on one queue's ledger (#1298). The notice + * flow itself runs against a real pg-boss in + * `tests/integration/system-job-alert.test.ts`. + */ +import { describe, expect, it } from "vitest"; + +import { + judgeQueueRuns, + SYSTEM_JOB_FAILURE_STREAK, + watchedQueues, + type TerminalRun, +} from "@/lib/jobs/system-job-alert"; + +const at = new Date("2026-10-29T10:00:00.000Z"); + +function run(overrides: Partial = {}): TerminalRun { + return { + state: "failed", + retryCount: 0, + allFailed: false, + completedOn: at, + error: "boom", + expireSeconds: null, + ...overrides, + }; +} + +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 a completed run that says every candidate failed", () => { + const allFailed = run({ state: "completed", allFailed: true }); + expect(judgeQueueRuns([allFailed, allFailed, allFailed]).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 }), + ]); + expect(verdict.lastError).toMatch( + /^job timed out: did not finish within its 1 h limit/, + ); + }); +}); + +describe("watchedQueues", () => { + it("watches the system queues and none with an account-facing surface", () => { + const queues = watchedQueues(); + expect(queues).toContain("rollup-full-backfill"); + expect(queues).toContain("system-job-alert"); + expect(queues).not.toContain("google-health-sync"); + }); +}); diff --git a/src/lib/jobs/job-failures.ts b/src/lib/jobs/job-failures.ts index 3b4b8f004..c9bf7a82b 100644 --- a/src/lib/jobs/job-failures.ts +++ b/src/lib/jobs/job-failures.ts @@ -145,7 +145,7 @@ function explainTimeout(message: string, expireSeconds: number | null): string { * serialises to `{ message, stack, … }`; a thrown non-Error can be any JSON at * all, so the raw text is the fallback rather than an invented placeholder. */ -function presentError( +export function presentError( raw: string | null, expireSeconds: number | null = null, ): string { diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 5cbc2d758..40bc1d872 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -307,6 +307,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "summary_pending_healed", "suppressed_client_managed", "suppressed_discreet", + // The hourly system job alert: queues failing now, notices sent, admins. + "system_job_admins", + "system_job_notices", + "system_jobs_failing", "thumbnails_enqueued", "tls_pin_monitor_known_count", "tls_pin_monitor_outcome", diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index fb14c254a..3c16627a1 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -386,6 +386,9 @@ export const QUEUE_RUNTIME: Readonly> = { "recovery-readiness": short( "A few indexed reads, one row per account from two backup ledgers, and at most one notice per signal.", ), + "system-job-alert": short( + "The newest fifty terminal runs of each system queue over pg-boss's completion index, and at most one notice per queue.", + ), "integration-watchdog-hourly": short( "A few indexed reads per account with a connection and no provider call; a 14-day history count only for a silent source.", ), diff --git a/src/lib/jobs/registry/status.ts b/src/lib/jobs/registry/status.ts index e9e651947..53505e081 100644 --- a/src/lib/jobs/registry/status.ts +++ b/src/lib/jobs/registry/status.ts @@ -4,7 +4,8 @@ * the morning digest refresh, the data-arrival spine with its reaction line * and workout insight, the nightly scores, the period-narrative warm, the * Coach memory refresh, nudge, reminder sweep and plan review, the low-stock - * pass and the daily-briefing fallback. + * pass, the daily-briefing fallback, and the hourly alert to the admins + * about a system queue that keeps failing. * * Most queues here are keyless cron ticks, so they carry no policy. The * LLM-bound warm queues (pre-generate, period narrative, Coach memory) pass @@ -77,6 +78,12 @@ import { WORKOUT_INSIGHT_GENERATE_QUEUE, } from "@/lib/jobs/workout-insight-generate-shared"; +import { cronIsTheRetry } from "@/lib/jobs/reminder/registrar-shared"; +import { + SYSTEM_JOB_ALERT_CRON, + SYSTEM_JOB_ALERT_QUEUE, +} from "@/lib/jobs/system-job-alert-queue"; + import { defineJob, type JobDefinition, type JobGroup } from "./define-job"; const GENERAL_STATUS_QUEUE = "insights-general-status"; @@ -261,6 +268,18 @@ const jobs: readonly JobDefinition[] = [ work: { localConcurrency: 1 }, load: async () => (await handlers()).handleDailyBriefing, }), + // Hourly: tell the admins about a system queue that keeps failing, the way + // the recovery readiness evaluation tells them about a backup. Single-flight + // and no retries; the next tick is the retry, and the notice claim keeps + // two evaluations from sending the same notice. + defineJob({ + queue: SYSTEM_JOB_ALERT_QUEUE, + cron: SYSTEM_JOB_ALERT_CRON, + scheduleOptions: cronIsTheRetry, + work: { localConcurrency: 1 }, + load: async () => + (await import("@/lib/jobs/system-job-alert")).handleSystemJobAlert, + }), ]; export const STATUS_GROUP: JobGroup = { diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index ea14eac89..2b03a5e05 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -67,6 +67,7 @@ const subsystemSurface = { "coach-plan-review": { audience: "system" }, "medication-low-stock": { audience: "system" }, "daily-briefing": { audience: "system" }, + "system-job-alert": { audience: "system" }, // register-rollup.ts "rollup-recompute": { audience: "system" }, diff --git a/src/lib/jobs/system-job-alert-queue.ts b/src/lib/jobs/system-job-alert-queue.ts new file mode 100644 index 000000000..2671edd0e --- /dev/null +++ b/src/lib/jobs/system-job-alert-queue.ts @@ -0,0 +1,12 @@ +/** + * The queue facts of `system-job-alert.ts`: its name and schedule. They live + * apart from the job so the worker registry can read them without loading it. + */ +export const SYSTEM_JOB_ALERT_QUEUE = "system-job-alert"; + +/** + * Hourly at :43, a minute no other schedule uses. A notice goes out once per + * change whatever the cadence, and an hour is soon enough to hear that a + * background job keeps failing. + */ +export const SYSTEM_JOB_ALERT_CRON = "43 * * * *"; diff --git a/src/lib/jobs/system-job-alert.ts b/src/lib/jobs/system-job-alert.ts new file mode 100644 index 000000000..d6174eec7 --- /dev/null +++ b/src/lib/jobs/system-job-alert.ts @@ -0,0 +1,303 @@ +/** + * 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 + * reminder at most once a day while it keeps failing, and once when it runs + * successfully again. Nothing else. + * + * Why (#1298). A full rollup backfill that failed on every boot showed only + * on the admin page's card of failing background jobs. From the account's + * side the only symptom was empty dashboard cards, and nothing pointed an + * operator at the card. The recovery readiness evaluation already pushes a + * stopped backup; this does the same for the queues whose failure has no + * 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 memory is one row per announced queue (`SystemJobAlertNotice`), claimed + * with a compare-and-set before the notice is sent, exactly as the readiness + * notices are (`readiness-notify.ts`, whose decision this reuses), so two + * workers send a notice once and a crash between the claim and the send loses + * a notice rather than repeating it. Delivery is the ordinary localised path, + * as the other operator alerts use: a `SYSTEM_ALERT` to every admin, in their + * locale, on the channels they enabled. + */ +import type { Job } from "pg-boss"; + +import { ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account"; +import { prisma } from "@/lib/db"; +import { presentError } from "@/lib/jobs/job-failures"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { SUBSYSTEM_SURFACE } from "@/lib/jobs/subsystem-surface"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { dispatchLocalisedNotification } from "@/lib/notifications/dispatch-localised"; +import { + decideNotice, + type NoticeKind, + type StoredNotice, +} from "@/lib/recovery/readiness-notify"; + +/** Failed runs in a row, retries included, before the admins are told. */ +export const SYSTEM_JOB_FAILURE_STREAK = 3; + +/** Newest terminal runs read per queue. */ +const LOOKBACK_ROWS = 50; + +/** 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. */ +export interface TerminalRun { + state: "completed" | "failed"; + /** Retries pg-boss ran before this run settled. */ + retryCount: number; + /** The run completed but reported that every candidate failed. */ + allFailed: boolean; + completedOn: Date; + error: string | null; + expireSeconds: number | null; +} + +/** 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. + */ + state: "failing" | "ok" | "pending" | "inactive"; + /** Failed attempts in a row, newest first. */ + failures: number; + /** The newest failure's message, redacted and truncated. */ + lastError: string; +} + +/** Pure: judge one queue's newest terminal runs. */ +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; + 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 state = + failures >= SYSTEM_JOB_FAILURE_STREAK + ? "failing" + : succeeded + ? "ok" + : "pending"; + return { state, failures, lastError: lastError ?? "" }; +} + +/** The queues this alert watches: every one whose failure has no other surface. */ +export function watchedQueues(): string[] { + return Object.entries(SUBSYSTEM_SURFACE) + .filter(([, surface]) => surface.audience === "system") + .map(([queue]) => queue); +} + +/** + * 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. + */ +export async function readTerminalRuns( + queues: readonly string[], +): Promise | null> { + try { + const rows = await prisma.$queryRaw< + Array<{ + name: 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 + 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 + `; + const byQueue = new Map(); + for (const row of rows) { + const runs = byQueue.get(row.name) ?? []; + runs.push({ + state: row.state === "failed" ? "failed" : "completed", + retryCount: Number(row.retry_count ?? 0), + allFailed: row.all_failed === "true", + completedOn: row.completed_on, + error: row.error, + expireSeconds: row.expire_seconds, + }); + byQueue.set(row.name, runs); + } + return byQueue; + } catch { + return null; + } +} + +function titleKey(kind: NoticeKind): string { + if (kind === "recovered") + return "notifications.systemJobAlert.titleRecovered"; + if (kind === "reminder") return "notifications.systemJobAlert.titleReminder"; + return "notifications.systemJobAlert.titleFailing"; +} + +export interface SystemJobAlertOutcome { + /** Queues at or past the threshold now. */ + failing: number; + /** Notices claimed and sent, by queue. */ + sent: Array<{ queue: string; kind: NoticeKind }>; + /** Admins each notice went to. */ + admins: number; +} + +/** + * Judge every watched queue against what the admins were last told, claim + * every notice that is due, and send it. + */ +export async function notifySystemJobFailures( + now: Date = new Date(), +): Promise { + const outcome: SystemJobAlertOutcome = { failing: 0, sent: [], admins: 0 }; + const runs = await readTerminalRuns(watchedQueues()); + if (runs === null) return outcome; + + // account-deletion: one row per queue of the instance; no account in it. + const rows = await prisma.systemJobAlertNotice.findMany(); + const byQueue = new Map(rows.map((row) => [row.queue, row])); + const admins = await prisma.user.findMany({ + where: { role: "ADMIN", ...ACTIVE_ACCOUNT }, + select: { id: true }, + }); + outcome.admins = admins.length; + + for (const queue of watchedQueues()) { + const verdict = judgeQueueRuns(runs.get(queue) ?? []); + if (verdict.state === "failing") outcome.failing += 1; + // Too few failures to call it a streak and no success to call it over. + if (verdict.state === "pending") continue; + + const row = byQueue.get(queue); + const stored: StoredNotice | null = row + ? { + notifiedState: + row.notifiedState === "failing" || row.notifiedState === "ok" + ? row.notifiedState + : null, + notifiedAt: row.notifiedAt, + } + : null; + const decision = decideNotice(stored, verdict.state, now); + // 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; + + // Claim: only the evaluation that moves the row from what it read sends. + const claimed = row + ? // account-deletion: the one queue row read above; no account in it. + ( + await prisma.systemJobAlertNotice.updateMany({ + where: { + queue, + notifiedState: row.notifiedState, + notifiedAt: row.notifiedAt, + }, + data: { + notifiedState: decision.next.notifiedState, + notifiedAt: decision.next.notifiedAt, + }, + }) + ).count + : ( + await prisma.systemJobAlertNotice.createMany({ + data: [ + { + queue, + notifiedState: decision.next.notifiedState, + notifiedAt: decision.next.notifiedAt, + }, + ], + skipDuplicates: true, + }) + ).count; + if (claimed !== 1 || decision.send === null) continue; + + for (const admin of admins) { + await dispatchLocalisedNotification({ + userId: admin.id, + eventType: "SYSTEM_ALERT", + titleKey: titleKey(decision.send), + messageKey: + decision.send === "recovered" + ? "notifications.systemJobAlert.bodyRecovered" + : "notifications.systemJobAlert.bodyFailing", + params: { + queue, + failures: verdict.failures, + error: verdict.lastError, + }, + metadata: { + source: "system-job-alert", + url: ADMIN_STATUS_PATH, + queue, + kind: decision.send, + }, + }); + } + outcome.sent.push({ queue, kind: decision.send }); + } + return outcome; +} + +/** The hourly evaluation. Single-flight; the next tick is the retry. */ +export async function handleSystemJobAlert( + jobs: Job[], +): Promise { + void jobs; + return withBackgroundEvent("job.system_job_alert", async (evt) => { + try { + const outcome = await notifySystemJobFailures(); + const did = { + system_jobs_failing: outcome.failing, + system_job_notices: outcome.sent.length, + system_job_admins: outcome.admins, + }; + for (const [key, value] of Object.entries(did)) evt.addMeta(key, value); + return jobDone(did); + } catch (err) { + evt.addWarning(`system-job-alert failed: ${err}`); + return jobFailed("system job alert evaluation failed", err); + } + }); +} diff --git a/tests/integration/__golden__/queue-registration.json b/tests/integration/__golden__/queue-registration.json index 5521dd6b6..016678537 100644 --- a/tests/integration/__golden__/queue-registration.json +++ b/tests/integration/__golden__/queue-registration.json @@ -1,5 +1,5 @@ { - "registeredQueueCount": 125, + "registeredQueueCount": 126, "registeredQueues": [ "account-deletion-purge", "achievement-unlock-sweep", @@ -112,6 +112,7 @@ "strava-sync", "stress-score-compute", "supplement-nutrient-recompute", + "system-job-alert", "tls-pin-monitor", "whoop-backfill", "whoop-cycle-sync", @@ -1793,6 +1794,21 @@ "retry_delay_max": null, "retry_limit": 2 }, + { + "dead_letter": null, + "deletion_seconds": 604800, + "expire_seconds": 900, + "heartbeat_seconds": null, + "name": "system-job-alert", + "notify": false, + "partition": false, + "policy": "standard", + "retention_seconds": 1209600, + "retry_backoff": false, + "retry_delay": 0, + "retry_delay_max": null, + "retry_limit": 2 + }, { "dead_letter": null, "deletion_seconds": 604800, @@ -2825,6 +2841,18 @@ }, "timezone": "Europe/Berlin" }, + { + "cron": "43 * * * *", + "data": {}, + "key": "", + "kind": "cron", + "name": "system-job-alert", + "options": { + "retryLimit": 0, + "tz": "Europe/Berlin" + }, + "timezone": "Europe/Berlin" + }, { "cron": "7 */6 * * *", "data": {}, @@ -3682,6 +3710,13 @@ "localConcurrency": 1 } }, + { + "queue": "system-job-alert", + "options": { + "localConcurrency": 1, + "pollingIntervalSeconds": 30 + } + }, { "queue": "tls-pin-monitor", "options": { diff --git a/tests/integration/system-job-alert.test.ts b/tests/integration/system-job-alert.test.ts new file mode 100644 index 000000000..bb267bf76 --- /dev/null +++ b/tests/integration/system-job-alert.test.ts @@ -0,0 +1,178 @@ +/** + * The system job alert against a ledger pg-boss itself wrote (#1298). + * + * 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. + * + * 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 + * against the table pg-boss writes rather than one the test wrote. + */ +import { PgBoss } from "pg-boss"; +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; + +const dispatched = vi.hoisted(() => ({ + calls: [] as Array<{ + userId: string; + titleKey: string; + messageKey: string; + params?: Record; + eventType?: string; + }>, +})); + +vi.mock("@/lib/notifications/dispatch-localised", () => ({ + dispatchLocalisedNotification: vi.fn( + async (opts: (typeof dispatched.calls)[number]) => { + dispatched.calls.push(opts); + }, + ), +})); + +import { + notifySystemJobFailures, + SYSTEM_JOB_FAILURE_STREAK, +} from "@/lib/jobs/system-job-alert"; + +import { makeUser } from "../kit/factories"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const SYSTEM_QUEUE = "rollup-full-backfill"; +const ACCOUNT_QUEUE = "google-health-sync"; +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 }); + 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 }); + } +} + +beforeAll(async () => { + boss = new PgBoss({ + connectionString: process.env.DATABASE_URL, + supervise: false, + schedule: false, + }); + await boss.start(); + await boss.createQueue(SYSTEM_QUEUE); + await boss.createQueue(ACCOUNT_QUEUE); +}, 60_000); + +afterAll(async () => { + await boss?.stop({ graceful: false, close: true }); + await getPrismaClient().$executeRawUnsafe( + `DROP SCHEMA IF EXISTS pgboss CASCADE`, + ); +}); + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + await getPrismaClient().$executeRawUnsafe(`DELETE FROM pgboss.job`); + dispatched.calls.length = 0; +}); + +describe("notifySystemJobFailures", () => { + it("tells every admin once, reminds a day later, and announces the recovery once", async () => { + const admin = await makeUser({ role: "ADMIN" }); + const second = await makeUser({ role: "ADMIN" }); + await makeUser({ role: "USER" }); + for (let i = 0; i < SYSTEM_JOB_FAILURE_STREAK; i++) { + await runOnce(SYSTEM_QUEUE, "fail"); + } + const now = new Date(); + + const first = await notifySystemJobFailures(now); + expect(first.failing).toBe(1); + expect(first.sent).toEqual([{ queue: SYSTEM_QUEUE, kind: "problem" }]); + expect(dispatched.calls.map((c) => c.userId).sort()).toEqual( + [admin.id, second.id].sort(), + ); + const call = dispatched.calls[0]; + expect(call.eventType).toBe("SYSTEM_ALERT"); + expect(call.titleKey).toBe("notifications.systemJobAlert.titleFailing"); + expect(call.params).toMatchObject({ + queue: SYSTEM_QUEUE, + failures: SYSTEM_JOB_FAILURE_STREAK, + }); + expect(String(call.params?.error)).toContain("statement timeout"); + + // The next hour: still failing, already told. + dispatched.calls.length = 0; + const again = await notifySystemJobFailures( + new Date(now.getTime() + HOUR_MS), + ); + expect(again.sent).toEqual([]); + expect(dispatched.calls).toEqual([]); + + // A day later: one reminder. + const reminder = await notifySystemJobFailures( + new Date(now.getTime() + 25 * HOUR_MS), + ); + expect(reminder.sent).toEqual([{ queue: SYSTEM_QUEUE, kind: "reminder" }]); + + // It runs again: one recovery notice, then silence. + await runOnce(SYSTEM_QUEUE, "complete"); + dispatched.calls.length = 0; + const recovered = await notifySystemJobFailures( + new Date(now.getTime() + 26 * HOUR_MS), + ); + expect(recovered.sent).toEqual([ + { queue: SYSTEM_QUEUE, kind: "recovered" }, + ]); + expect(dispatched.calls[0].titleKey).toBe( + "notifications.systemJobAlert.titleRecovered", + ); + const quiet = await notifySystemJobFailures( + new Date(now.getTime() + 27 * HOUR_MS), + ); + expect(quiet.sent).toEqual([]); + }); + + it("stays quiet below the streak and for queues with an account-facing surface", async () => { + await makeUser({ role: "ADMIN" }); + for (let i = 0; i < SYSTEM_JOB_FAILURE_STREAK - 1; i++) { + await runOnce(SYSTEM_QUEUE, "fail"); + } + for (let i = 0; i < SYSTEM_JOB_FAILURE_STREAK + 2; i++) { + await runOnce(ACCOUNT_QUEUE, "fail"); + } + const outcome = await notifySystemJobFailures(); + expect(outcome.failing).toBe(0); + expect(outcome.sent).toEqual([]); + expect(dispatched.calls).toEqual([]); + }); + + it("counts a success between failures as the end of the streak", async () => { + await makeUser({ role: "ADMIN" }); + await runOnce(SYSTEM_QUEUE, "fail"); + await runOnce(SYSTEM_QUEUE, "fail"); + await runOnce(SYSTEM_QUEUE, "complete"); + await runOnce(SYSTEM_QUEUE, "fail"); + const outcome = await notifySystemJobFailures(); + expect(outcome.failing).toBe(0); + expect(outcome.sent).toEqual([]); + }); +});