diff --git a/src/__tests__/job-retention-floor-guard.test.ts b/src/__tests__/job-retention-floor-guard.test.ts new file mode 100644 index 000000000..52aeedb38 --- /dev/null +++ b/src/__tests__/job-retention-floor-guard.test.ts @@ -0,0 +1,105 @@ +/** + * No queue may forget its failed jobs before the failure readers look. + * + * `job-failures.ts` reads pg-boss's own `state = 'failed'` rows over a 72-hour + * window: the admin status card, the per-account "last run failed" hint and + * the repeated-failure count behind the `job.run.failed` alert all rest on + * those rows still existing. pg-boss deletes a terminal row + * `deleteAfterSeconds` after it finished. v1.42 shortens that option on the + * pure-volume queues, and a value under the reader window would make a + * failing queue read as healthy, with nothing anywhere saying so. + * + * The rule: a `deleteAfterSeconds` in the source is either produced by + * `failureReaderRetention(hours)` (which throws under the floor) or written as + * a plain literal at or above `FAILED_ROW_AVAILABILITY_FLOOR_HOURS`. Anything + * else, a named constant included, fails here, because a constant's value is + * not checkable from the call site and the helper exists to make it so. + * + * Mutation check, run when this guard was written: a planted + * `deleteAfterSeconds: 48 * 60 * 60` in a job file fails the second case, and + * `deleteAfterSeconds: VOLUME_SECONDS` fails it too; the matcher self-test + * below proves the scan sees both shapes, so an empty sweep is not mistaken + * for a clean one. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; + +import { walkSourceFiles } from "./helpers/source-files"; +import { FAILED_ROW_AVAILABILITY_FLOOR_HOURS } from "@/lib/jobs/job-failures"; + +const SRC = join(process.cwd(), "src"); +const FLOOR_SECONDS = FAILED_ROW_AVAILABILITY_FLOOR_HOURS * 3600; + +/** The one file allowed to spell the option out: the helper itself. */ +const HELPER_FILE = "lib/jobs/job-failures.ts"; + +const OPTION = /\bdeleteAfterSeconds\s*:\s*([^,}\n]+)/g; + +/** Evaluate a sum of products of integer literals; null for anything else. */ +function literalSeconds(expression: string): number | null { + const cleaned = expression.trim().replace(/_/g, ""); + if (!/^\d+(\s*[*+]\s*\d+)*$/.test(cleaned)) return null; + return cleaned + .split("+") + .map((term) => + term + .split("*") + .map((factor) => Number(factor.trim())) + .reduce((a, b) => a * b, 1), + ) + .reduce((a, b) => a + b, 0); +} + +interface Finding { + file: string; + expression: string; +} + +function scan(file: string, text: string): Finding[] { + const out: Finding[] = []; + for (const match of text.matchAll(OPTION)) { + out.push({ file, expression: match[1].trim() }); + } + return out; +} + +function violations(findings: readonly Finding[]): string[] { + return findings + .filter((f) => { + const seconds = literalSeconds(f.expression); + return seconds === null || seconds < FLOOR_SECONDS; + }) + .map((f) => `${f.file}: deleteAfterSeconds: ${f.expression}`); +} + +function sourceFindings(): Finding[] { + return walkSourceFiles(SRC, { floor: 3000 }) + .filter((p) => !p.startsWith("generated/")) + .filter((p) => !p.includes("__tests__")) + .filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx")) + .filter((p) => p !== HELPER_FILE) + .flatMap((p) => scan(p, readFileSync(join(SRC, p), "utf8"))); +} + +describe("job retention floor", () => { + it("the matcher sees both a short literal and a named constant", () => { + const planted = scan( + "planted.ts", + [ + "await boss.createQueue(Q, { deleteAfterSeconds: 48 * 60 * 60 });", + "await boss.createQueue(R, { deleteAfterSeconds: VOLUME_SECONDS });", + "await boss.createQueue(S, { deleteAfterSeconds: 7 * 24 * 3600 });", + ].join("\n"), + ); + expect(planted).toHaveLength(3); + expect(violations(planted)).toEqual([ + "planted.ts: deleteAfterSeconds: 48 * 60 * 60", + "planted.ts: deleteAfterSeconds: VOLUME_SECONDS", + ]); + }); + + it("every queue retention in the tree keeps failed rows past the reader window", () => { + expect(violations(sourceFindings())).toEqual([]); + }); +}); diff --git a/src/lib/jobs/__tests__/job-failures-retention.test.ts b/src/lib/jobs/__tests__/job-failures-retention.test.ts index 7a0674c9b..89d798894 100644 --- a/src/lib/jobs/__tests__/job-failures-retention.test.ts +++ b/src/lib/jobs/__tests__/job-failures-retention.test.ts @@ -3,8 +3,10 @@ import { join } from "node:path"; import { describe, expect, it } from "vitest"; import { + FAILED_ROW_AVAILABILITY_FLOOR_HOURS, JOB_FAILURE_WINDOW_HOURS, - PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS, + PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS, + failureReaderRetention, } from "@/lib/jobs/job-failures"; const PG_BOSS_ROOT = join(process.cwd(), "node_modules/pg-boss"); @@ -26,10 +28,28 @@ describe("pg-boss failed-row ledger retention", () => { ); expect(plans).toContain("CLOCK_FUNCTION_BODY = 'SELECT pg_catalog.now();'"); expect(JOB_FAILURE_WINDOW_HOURS).toBeLessThan( - PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS, + PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS, ); }); + it("keeps the per-queue floor above the reader window, with a day of slack", () => { + expect(JOB_FAILURE_WINDOW_HOURS).toBe(72); + expect(FAILED_ROW_AVAILABILITY_FLOOR_HOURS).toBe(96); + expect(FAILED_ROW_AVAILABILITY_FLOOR_HOURS).toBeLessThanOrEqual( + PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS, + ); + }); + + it("refuses a queue retention that would hide failures from the readers", () => { + expect(failureReaderRetention(96)).toEqual({ deleteAfterSeconds: 345_600 }); + expect(failureReaderRetention(7 * 24)).toEqual({ + deleteAfterSeconds: 604_800, + }); + expect(() => failureReaderRetention(95)).toThrow(RangeError); + expect(() => failureReaderRetention(24)).toThrow(RangeError); + expect(() => failureReaderRetention(Number.NaN)).toThrow(RangeError); + }); + it("pins the queue defaults the data-backup window had to escape", () => { const plans = readFileSync(join(PG_BOSS_ROOT, "dist/plans.js"), "utf8"); diff --git a/src/lib/jobs/__tests__/job-run-report.test.ts b/src/lib/jobs/__tests__/job-run-report.test.ts new file mode 100644 index 000000000..2ccd2dd19 --- /dev/null +++ b/src/lib/jobs/__tests__/job-run-report.test.ts @@ -0,0 +1,142 @@ +/** + * v1.42 — `reportJobRun`: one line per run with every candidate's outcome, + * and a level that follows the run as a whole. + * + * The nightly briefing warm read as "all-failed" on seven nights because the + * last account of each run happened to fail. These cases pin the verdict + * (what counts as attempted, what makes a run failed versus partial) and the + * two signals an operator and the repeated-failure alert key on. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const emitSignal = vi.fn(); +const annotate = vi.fn(); +const readConsecutiveRunFailures = vi.fn(async (_queue: string) => 3); + +vi.mock("@/lib/logging/signal", () => ({ + emitSignal: (...a: unknown[]) => emitSignal(...a), +})); +vi.mock("@/lib/logging/context", () => ({ + annotate: (...a: unknown[]) => annotate(...a), +})); +vi.mock("@/lib/jobs/job-failures", () => ({ + readConsecutiveRunFailures: (queue: string) => + readConsecutiveRunFailures(queue), +})); + +import { judgeJobRun, reportJobRun } from "../job-run-report"; + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe("judgeJobRun", () => { + it("leaves skipped and deferred candidates out of the attempted set", () => { + expect( + judgeJobRun([ + { key: "a", outcome: "skipped" }, + { key: "b", outcome: "deferred" }, + { key: "c", outcome: "failed" }, + ]), + ).toEqual({ + total: 3, + attempted: 1, + failed: 1, + allFailed: true, + partial: false, + }); + }); + + it("counts a dead credential as a failure", () => { + expect( + judgeJobRun([ + { key: "a", outcome: "auth_failed" }, + { key: "b", outcome: "failed" }, + ]).allFailed, + ).toBe(true); + }); + + it("reads one success among failures as partial, not failed", () => { + const verdict = judgeJobRun([ + { key: "a", outcome: "ok" }, + { key: "b", outcome: "failed" }, + ]); + expect(verdict.allFailed).toBe(false); + expect(verdict.partial).toBe(true); + }); + + it("reads a withheld result as partial: the provider answered", () => { + const verdict = judgeJobRun([{ key: "a", outcome: "screened" }]); + expect(verdict.allFailed).toBe(false); + expect(verdict.partial).toBe(true); + }); + + it("is neither failed nor partial when nothing was attempted", () => { + const verdict = judgeJobRun([{ key: "a", outcome: "skipped" }]); + expect(verdict.allFailed).toBe(false); + expect(verdict.partial).toBe(false); + }); +}); + +describe("reportJobRun", () => { + it("emits job.run.failed at error with the streak when every attempt failed", async () => { + const verdict = await reportJobRun({ + queue: "insight-pregenerate", + runId: "r1", + candidates: [ + { key: "a", outcome: "auth_failed", cause: "all-providers-failed" }, + { key: "b", outcome: "skipped", cause: "no-consent" }, + ], + }); + expect(verdict.allFailed).toBe(true); + expect(readConsecutiveRunFailures).toHaveBeenCalledWith( + "insight-pregenerate", + ); + expect(emitSignal).toHaveBeenCalledTimes(1); + expect(emitSignal.mock.calls[0][0]).toMatchObject({ + action: "job.run.failed", + level: "error", + meta: { + queue: "insight-pregenerate", + consecutiveFailures: 3, + failed: 1, + total: 1, + causes: ["auth_failed:all-providers-failed"], + }, + }); + }); + + it("emits job.run.partial at warn when only some failed", async () => { + await reportJobRun({ + queue: "q", + runId: "r", + candidates: [ + { key: "a", outcome: "ok" }, + { key: "b", outcome: "failed", cause: "timeout" }, + ], + }); + expect(readConsecutiveRunFailures).not.toHaveBeenCalled(); + expect(emitSignal.mock.calls[0][0]).toMatchObject({ + action: "job.run.partial", + level: "warn", + meta: { queue: "q", failed: 1, total: 2 }, + }); + }); + + it("emits nothing extra for a clean run, and lists every outcome on the run line", async () => { + await reportJobRun({ + queue: "q", + runId: "r", + candidates: [ + { key: "a", outcome: "ok" }, + { key: "b", outcome: "ok" }, + { key: "c", outcome: "skipped", cause: "no-provider" }, + ], + }); + expect(emitSignal).not.toHaveBeenCalled(); + const meta = annotate.mock.calls[0][0].meta; + expect(meta.outcomes).toEqual({ ok: 2, skipped: 1 }); + expect(meta.candidate_outcomes).toHaveLength(3); + expect(meta.all_failed).toBe(false); + }); +}); diff --git a/src/lib/jobs/__tests__/run-job.test.ts b/src/lib/jobs/__tests__/run-job.test.ts index 9b32e591f..555ef57cb 100644 --- a/src/lib/jobs/__tests__/run-job.test.ts +++ b/src/lib/jobs/__tests__/run-job.test.ts @@ -17,6 +17,19 @@ vi.mock("../report-worker-error", () => ({ reportWorkerError: (...a: unknown[]) => reportWorkerError(...a), })); +// v1.42 — the failed run's own line and the streak behind it. +const emitSignal = vi.fn(); +const readConsecutiveRunFailures = vi.fn(async (_queue: string) => 2); +vi.mock("@/lib/logging/signal", () => ({ + emitSignal: (...a: unknown[]) => emitSignal(...a), +})); +vi.mock("../job-failures", async () => ({ + ...(await vi.importActual( + "../job-failures", + )), + readConsecutiveRunFailures: (q: string) => readConsecutiveRunFailures(q), +})); + import { jobDone, jobFailed } from "../job-outcome"; import { JobFailure, runJob } from "../run-job"; @@ -171,3 +184,47 @@ describe("runJob observation (#1031)", () => { observe.mockRestore(); }); }); + +describe("runJob failure signal (v1.42)", () => { + it("emits job.run.failed at error with the queue's streak for a failed outcome", async () => { + const wrapped = runJob("environment-fetch", async () => + jobFailed("fetch_failed", new Error("open-meteo 503")), + ); + await expect(wrapped(noJobs)).rejects.toBeInstanceOf(JobFailure); + expect(readConsecutiveRunFailures).toHaveBeenCalledWith( + "environment-fetch", + ); + expect(emitSignal).toHaveBeenCalledTimes(1); + expect(emitSignal.mock.calls[0][0]).toMatchObject({ + action: "job.run.failed", + level: "error", + meta: { + queue: "environment-fetch", + reason: "fetch_failed", + consecutiveFailures: 2, + }, + }); + }); + + it("emits it for a handler that throws, and still rethrows the original error", async () => { + const original = new Error("boom"); + const wrapped = runJob("data-backup", async () => { + throw original; + }); + await expect(wrapped(noJobs)).rejects.toBe(original); + expect(emitSignal.mock.calls[0][0]).toMatchObject({ + action: "job.run.failed", + level: "error", + meta: { queue: "data-backup", reason: "handler_threw" }, + error: original, + }); + }); + + it("emits nothing for a successful run", async () => { + const wrapped = runJob("rate-limit-cleanup", async () => + jobDone({ deleted: 1 }), + ); + await wrapped(noJobs); + expect(emitSignal).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/jobs/job-failures.ts b/src/lib/jobs/job-failures.ts index 9332b041d..db7604b6d 100644 --- a/src/lib/jobs/job-failures.ts +++ b/src/lib/jobs/job-failures.ts @@ -28,22 +28,61 @@ import { prisma } from "@/lib/db"; import { redactSecrets } from "@/lib/logging/redact"; /** - * pg-boss 12.34.0 stores terminal rows until - * `completed_on + deletion_seconds` (`dist/plans.js`). Its queue default is - * 604800 seconds, while `retention_seconds` defaults to 1209600 seconds; job - * insertion copies those queue values and maintenance applies the terminal-row - * deletion clock. The deployed queue-table audit found those same values on - * every queue, and no job submitter overrides either option. + * How long pg-boss keeps a terminal row, and why the readers below can rely + * on it. * - * Seven days is therefore the verified availability floor for failed rows. - * The reader stays at 72 hours, safely inside that floor, and pg-boss remains - * the single failure ledger. + * pg-boss 12.34.0 stores terminal rows until `completed_on + + * deletion_seconds` (`dist/plans.js`). The queue default is 604800 seconds + * (seven days), `retention_seconds` defaults to 1209600 seconds, and job + * insertion copies the queue's values onto each row. A queue created without + * a `deleteAfterSeconds` option therefore keeps its failed rows for seven + * days. + * + * That is no longer true for every queue. v1.42 shortens the retention of + * the pure-volume queues (a sampler that ticks every five minutes, the rollup + * recompute fan-out) so their completed rows stop bloating `job_common`. A + * failed row on such a queue lives only as long as that queue's own + * retention, and a reader that looks back 72 hours over a queue that forgets + * after 48 would report "nothing failed" about failures it can no longer see. + * + * So the availability is per queue now, and it has a floor: no queue may + * keep its terminal rows for less than `FAILED_ROW_AVAILABILITY_FLOOR_HOURS`, + * which is the 72-hour reader window plus a day of slack for a run that + * fails just before a maintenance pass. Every queue option that shortens + * retention goes through `failureReaderRetention`, which refuses a value + * under the floor, and `src/__tests__/job-retention-floor-guard.test.ts` + * refuses a `deleteAfterSeconds` anywhere that is neither a call to it nor a + * literal at or above the floor. pg-boss remains the single failure ledger. */ -export const PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS = 7 * 24; +export const PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS = 7 * 24; /** How far back a failure still counts as news. */ export const JOB_FAILURE_WINDOW_HOURS = 72; +/** + * The shortest terminal-row retention any queue may have: the reader window + * plus a day. See the block above. + */ +export const FAILED_ROW_AVAILABILITY_FLOOR_HOURS = + JOB_FAILURE_WINDOW_HOURS + 24; + +/** + * The `deleteAfterSeconds` queue option for a queue that keeps its terminal + * rows for `hours`. Throws below the floor, so a retention change that would + * blind the failure readers fails at boot and in the test suite rather than + * silently three days later. + */ +export function failureReaderRetention(hours: number): { + deleteAfterSeconds: number; +} { + if (!Number.isFinite(hours) || hours < FAILED_ROW_AVAILABILITY_FLOOR_HOURS) { + throw new RangeError( + `queue retention of ${hours} h is below the ${FAILED_ROW_AVAILABILITY_FLOOR_HOURS} h the failure readers need`, + ); + } + return { deleteAfterSeconds: Math.round(hours * 3600) }; +} + /** Queues named in one report — enough to see a pattern, bounded for the wire. */ const MAX_QUEUES = 20; @@ -373,3 +412,52 @@ export async function readQueueRunningSince( return null; } } + +/** Newest terminal rows `readConsecutiveRunFailures` looks at. */ +const CONSECUTIVE_LOOKBACK_ROWS = 50; + +/** + * How many runs of `queue` in a row ended in failure, newest first, counting + * the run that is failing right now (so the result is at least 1). + * + * A run counts as failed when pg-boss failed it (`state = 'failed'`), or when + * it completed but said in its own facts that every candidate it attempted + * failed (`did.all_failed`, written through `reportJobRun`). A fan-out pass + * like the nightly briefing warm completes even when nothing worked, because + * failing the queue would retry the whole cohort over one account; its + * all-failed nights are still a streak an operator wants to hear about. + * + * Completed runs that do not report `all_failed` at all (a queue's on-demand + * single-account jobs share it with the nightly pass) are not evidence either + * way and are skipped. The streak ends at the first completed run that says + * it did not all fail. + * + * Fails soft to 1: when the queue schema cannot be read, this run is still a + * failure, and one is the only count that is known. + */ +export async function readConsecutiveRunFailures( + queue: string, +): Promise { + try { + const rows = await prisma.$queryRaw< + Array<{ state: string; all_failed: string | null }> + >` + SELECT state, output->'did'->>'all_failed' AS all_failed + FROM pgboss.job + WHERE name = ${queue} + AND state IN ('completed', 'failed') + AND completed_on IS NOT NULL + AND (state = 'failed' OR output->'did'->>'all_failed' IS NOT NULL) + ORDER BY completed_on DESC + LIMIT ${CONSECUTIVE_LOOKBACK_ROWS} + `; + let streak = 1; + for (const row of rows) { + if (row.state === "failed" || row.all_failed === "true") streak += 1; + else break; + } + return streak; + } catch { + return 1; + } +} diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 33e994a32..61f9733b5 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -47,6 +47,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "ai_runs_deleted", "ai_runs_timed_out", "ai_runs_worker_unavailable", + // v1.42 — every candidate a fan-out pass attempted failed (`reportJobRun`). + "all_failed", // An admission delivery that found its import still running from an // earlier delivery of the same job and did nothing. "already_running", diff --git a/src/lib/jobs/job-run-report.ts b/src/lib/jobs/job-run-report.ts index f17d02795..4dadbd0f4 100644 --- a/src/lib/jobs/job-run-report.ts +++ b/src/lib/jobs/job-run-report.ts @@ -3,16 +3,31 @@ * * A pass over many accounts used to annotate its wide event once per * candidate, and the last write won: a run where six accounts worked and one - * failed read as whatever the last account did. `reportJobRun` takes the - * whole candidate list instead, so the run's line carries every outcome and - * the level follows the worst of them. + * failed read as whatever the last account did. The nightly briefing warm read + * as `all-failed` on seven nights in a row when it had in fact completed every + * night, because the last account of each run happened to fail. `reportJobRun` + * takes the whole candidate list instead, so the run's line carries every + * outcome and the level follows the run as a whole: * - * Contract stub: the signature is fixed for its callers (the insight - * pre-generation, the tombstone purge, the measurement maintenance). Until - * the implementation lands it annotates the run's counts the way the callers - * do today. + * - every attempted candidate failed → `job.run.failed` at `error`, with how + * many runs of the queue in a row have now failed (`consecutiveFailures`, + * the number the repeated-failure alert keys on); + * - some failed, or a result was withheld → `job.run.partial` at `warn`; + * - otherwise nothing extra: the run's own line says what it did. + * + * "Attempted" leaves out `skipped` and `deferred` candidates: an account + * without consent or one the job budget did not reach is not a failure of the + * run. `screened` (the model answered and the safety screen withheld it) is + * attempted but is not a provider failure, so a run of only screened + * candidates is partial, not failed. + * + * The verdict comes back to the caller, which writes `all_failed` into its + * `jobDone` facts. That fact is what lets `readConsecutiveRunFailures` count a + * completed-but-all-failed night as part of a streak. */ import { annotate } from "@/lib/logging/context"; +import { emitSignal } from "@/lib/logging/signal"; +import { readConsecutiveRunFailures } from "@/lib/jobs/job-failures"; /** How one candidate of a run ended. */ export type JobRunOutcome = @@ -26,20 +41,118 @@ export interface JobRunCandidate { cause?: string; } -export function reportJobRun(input: { +export interface JobRunVerdict { + /** Candidates the run looked at. */ + total: number; + /** Candidates it actually attempted (not skipped, not deferred). */ + attempted: number; + /** `failed` + `auth_failed`. */ + failed: number; + /** Every attempted candidate failed (and at least one was attempted). */ + allFailed: boolean; + /** Something failed or was withheld, but not everything. */ + partial: boolean; +} + +/** Candidates listed on the run's line; the counts always cover all of them. */ +const MAX_LISTED_CANDIDATES = 50; + +const FAILED: ReadonlySet = new Set(["failed", "auth_failed"]); +const NOT_ATTEMPTED: ReadonlySet = new Set([ + "skipped", + "deferred", +]); + +/** The verdict alone, without logging. Pure, for callers and tests. */ +export function judgeJobRun( + candidates: readonly JobRunCandidate[], +): JobRunVerdict { + const total = candidates.length; + const attempted = candidates.filter( + (c) => !NOT_ATTEMPTED.has(c.outcome), + ).length; + const failed = candidates.filter((c) => FAILED.has(c.outcome)).length; + const screened = candidates.filter((c) => c.outcome === "screened").length; + const allFailed = attempted > 0 && failed === attempted; + return { + total, + attempted, + failed, + allFailed, + partial: !allFailed && (failed > 0 || screened > 0), + }; +} + +export async function reportJobRun(input: { queue: string; runId: string; candidates: readonly JobRunCandidate[]; -}): void { - const failed = input.candidates.filter( - (c) => c.outcome === "failed" || c.outcome === "auth_failed", - ).length; +}): Promise { + const verdict = judgeJobRun(input.candidates); + const byOutcome: Record = {}; + for (const c of input.candidates) { + byOutcome[c.outcome] = (byOutcome[c.outcome] ?? 0) + 1; + } + annotate({ meta: { queue: input.queue, runId: input.runId, - candidates: input.candidates.length, - failed, + candidates: verdict.total, + attempted: verdict.attempted, + failed: verdict.failed, + all_failed: verdict.allFailed, + outcomes: byOutcome, + candidate_outcomes: input.candidates + .slice(0, MAX_LISTED_CANDIDATES) + .map((c) => ({ + key: c.key, + outcome: c.outcome, + ...(c.cause ? { cause: c.cause } : {}), + })), }, }); + + if (verdict.allFailed) { + const consecutiveFailures = await readConsecutiveRunFailures(input.queue); + emitSignal({ + action: "job.run.failed", + level: "error", + meta: { + queue: input.queue, + runId: input.runId, + consecutiveFailures, + failed: verdict.failed, + total: verdict.attempted, + causes: distinctCauses(input.candidates), + }, + }); + } else if (verdict.partial) { + emitSignal({ + action: "job.run.partial", + level: "warn", + meta: { + queue: input.queue, + runId: input.runId, + failed: verdict.failed, + total: verdict.attempted, + outcomes: byOutcome, + causes: distinctCauses(input.candidates), + }, + }); + } + return verdict; +} + +/** The distinct causes of the run's failed or withheld candidates, bounded. */ +function distinctCauses(candidates: readonly JobRunCandidate[]): string[] { + const causes = new Set(); + for (const c of candidates) { + if (c.outcome === "ok" || NOT_ATTEMPTED.has(c.outcome) || !c.cause) { + continue; + } + causes.add(`${c.outcome}:${c.cause}`); + if (causes.size >= 10) break; + } + return [...causes]; } diff --git a/src/lib/jobs/run-job.ts b/src/lib/jobs/run-job.ts index 9a673bc8c..114c26e26 100644 --- a/src/lib/jobs/run-job.ts +++ b/src/lib/jobs/run-job.ts @@ -29,6 +29,8 @@ import { } from "./job-outcome"; import { reportWorkerError } from "./report-worker-error"; import { observeJob } from "./job-observer"; +import { readConsecutiveRunFailures } from "./job-failures"; +import { emitSignal } from "@/lib/logging/signal"; /** A converted handler: it says what it did instead of just finishing. */ export type JobHandler = (jobs: Job[]) => Promise; @@ -67,14 +69,34 @@ function factMeta(did: JobFacts | undefined): Record { return meta; } +/** + * The failed run's own line: `job.run.failed` at `error`, naming the queue and + * how many runs of it in a row have now failed. Every queue reaches it, so + * the repeated-failure alert keys on one action and one number instead of on + * the handler's own task name, which differs per queue, or on the stderr line + * `reportWorkerError` writes, which is not structured. + */ +async function signalRunFailed( + queue: string, + reason: string, + error: unknown, +): Promise { + const consecutiveFailures = await readConsecutiveRunFailures(queue); + emitSignal({ + action: "job.run.failed", + level: "error", + meta: { queue, reason, consecutiveFailures }, + error, + }); +} + /** * Wrap a handler for `boss.work`. Resolves with the bounded, redacted * persistence representation on `ok: true`; reports and rethrows on * `ok: false`. * - * A handler that throws is left alone — the throw already fails the job, and - * re-wrapping it would only bury the original stack. `runJob` exists for the - * case the throw never happens. + * A handler that throws keeps its own error — re-wrapping it would only bury + * the original stack — but its run still gets the `job.run.failed` line. */ export function runJob( queue: string, @@ -83,11 +105,17 @@ export function runJob( return async (jobs: J[]): Promise => { // Observed, so a long run, an expiry and a cut-off leave a line in the // log naming the queue and how far it got (see `job-observer.ts`). - const outcome = await observeJob( - queue, - jobs as ReadonlyArray<{ id?: string; expireInSeconds?: number }>, - () => handler(jobs), - ); + let outcome: JobOutcome; + try { + outcome = await observeJob( + queue, + jobs as ReadonlyArray<{ id?: string; expireInSeconds?: number }>, + () => handler(jobs), + ); + } catch (error) { + await signalRunFailed(queue, "handler_threw", error); + throw error; + } if (outcome.ok) return serializeJobOutcome(outcome); const failure = new JobFailure(queue, outcome.reason, outcome.cause); @@ -102,6 +130,7 @@ export function runJob( ...factMeta(outcome.did), ...(causeMessage === undefined ? {} : { cause: causeMessage }), }); + await signalRunFailed(queue, outcome.reason, failure); throw failure; }; diff --git a/src/lib/logging/__tests__/signal.test.ts b/src/lib/logging/__tests__/signal.test.ts new file mode 100644 index 000000000..a3c01ba3f --- /dev/null +++ b/src/lib/logging/__tests__/signal.test.ts @@ -0,0 +1,101 @@ +/** + * v1.42 — the logging pieces the error-log discipline rests on. + * + * - `emitSignal` writes its own line with a stable action and level, joined + * to the surrounding event by request and trace id, and leaves that event + * alone. + * - `setError` keeps `error` for real failures and records a refusal the + * request was meant to get (any 4xx) at `warn`. Expired-token 401s made up + * most of the error lines and hid the real ones. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const emitted: Array> = []; +vi.mock("../transports", () => ({ + emitIfSampled: (event: Record) => emitted.push(event), +})); + +import { WideEventBuilder } from "../event-builder"; +import { eventStorage } from "../context"; +import { emitSignal, logCaught } from "../signal"; + +beforeEach(() => { + emitted.length = 0; +}); + +describe("emitSignal", () => { + it("writes one line with its own action and level, joined to the parent by id", () => { + const parent = new WideEventBuilder("http"); + parent.setAction({ name: "insights.generate" }); + eventStorage.run(parent, () => + emitSignal({ + action: "ai.chain.link_failed", + level: "error", + meta: { provider: "admin-openai", status: 500 }, + }), + ); + expect(emitted).toHaveLength(1); + const line = emitted[0] as { + action: { name: string }; + level: string; + request_id: string; + trace_id: string; + meta: Record; + }; + expect(line.action.name).toBe("ai.chain.link_failed"); + expect(line.level).toBe("error"); + expect(line.request_id).toBe(parent.getRequestId()); + expect(line.trace_id).toBe(parent.getTraceId()); + expect(line.meta).toMatchObject({ provider: "admin-openai", status: 500 }); + // The parent keeps its own action and level. + expect(parent.toJSON().action?.name).toBe("insights.generate"); + expect(parent.getLevel()).toBe("info"); + }); + + it("logCaught carries the error's type and a bounded, redacted message at warn", () => { + logCaught( + "jobs.enqueue.failed", + new Error(`boom Bearer sk-abcdefghijklmnopqrstuvwxyz ${"x".repeat(500)}`), + { queue: "insight-pregenerate" }, + ); + const line = emitted[0] as { level: string; meta: Record }; + expect(line.level).toBe("warn"); + expect(line.meta.error_type).toBe("Error"); + expect(line.meta.error_message.length).toBeLessThanOrEqual(240); + expect(line.meta.error_message).not.toContain("sk-abcdefghijklmnop"); + }); + + it("never throws, even when emitting fails", () => { + expect(() => + emitSignal({ + action: "x.y.z", + level: "warn", + meta: { + get bad(): never { + throw new Error("getter"); + }, + }, + }), + ).not.toThrow(); + }); +}); + +describe("setError level", () => { + it("records a 4xx refusal at warn", () => { + const evt = new WideEventBuilder("http"); + evt.setError( + Object.assign(new Error("Token expired"), { statusCode: 401 }), + ); + expect(evt.getLevel()).toBe("warn"); + expect(evt.toJSON().error?.message).toBe("Token expired"); + }); + + it("keeps a real failure at error", () => { + const evt = new WideEventBuilder("http"); + evt.setError(new Error("database unreachable")); + expect(evt.getLevel()).toBe("error"); + const five = new WideEventBuilder("http"); + five.setError(Object.assign(new Error("bad gateway"), { statusCode: 502 })); + expect(five.getLevel()).toBe("error"); + }); +}); diff --git a/src/lib/logging/event-builder.ts b/src/lib/logging/event-builder.ts index 1bcc0596c..09afd9819 100644 --- a/src/lib/logging/event-builder.ts +++ b/src/lib/logging/event-builder.ts @@ -167,7 +167,18 @@ export class WideEventBuilder { } setError(err: unknown): this { - this.elevateLevel("error"); + // v1.42 — a refusal the request was meant to get (an expired session, + // a missing permission, a validation error: anything that carries a 4xx + // status) is a `warn`, not an `error`. Every 401 from an expired token + // used to land at `error`, about nine in ten of all error lines, so a + // query or an alert on `level=error` could not see the real failures + // between them. The error details are recorded either way. + const status = (err as { statusCode?: unknown } | null)?.statusCode; + this.elevateLevel( + typeof status === "number" && status >= 400 && status < 500 + ? "warn" + : "error", + ); if (err instanceof Error) { this.event.error = { type: err.constructor.name, diff --git a/src/lib/logging/signal.ts b/src/lib/logging/signal.ts new file mode 100644 index 000000000..93e39b456 --- /dev/null +++ b/src/lib/logging/signal.ts @@ -0,0 +1,99 @@ +import { WideEventBuilder } from "./event-builder"; +import { getEvent } from "./context"; +import { emitIfSampled } from "./transports"; + +/** + * A failure that deserves its own log line, under its own action name. + * + * A wide event carries one action, and the last `annotate()` to name one wins. + * That is the right shape for "what did this request do", and the wrong one + * for "what went wrong on the way": a provider hop that failed before the + * next one answered, a job candidate that failed among six that worked, a + * Coach turn that ran out of budget, a caught error the code chose to carry + * on past. Each of those used to be either a meta key on someone else's line + * at level `info`, or nothing at all, so a query for `level=error` could not + * find it and an alert rule had no stable name to key on. + * + * `emitSignal` writes one extra line with the action, the level and a small + * pinned-shape meta bag. It carries the surrounding event's request and trace + * ids, so the signal and the line it happened inside join on one id; the + * surrounding event itself is not touched (its own level still describes its + * own outcome). + * + * Levels: `error` for a real failure nobody downstream recovered from, `warn` + * for one the code recovered from (a fallback answered, a degraded result was + * served) or a refusal by design that an operator should still be able to + * count. Nothing here is for `info`: an expected outcome belongs on the + * request's own line. + * + * Never throws. A log line must not break the work it describes. + */ +export interface SignalInput { + /** Stable `..` name, e.g. `ai.chain.link_failed`. */ + action: string; + level: "warn" | "error"; + /** Scalars and small arrays only; redacted on the way in like any meta. */ + meta?: Record; + /** The caught value, when there is one. Only its type and message leave. */ + error?: unknown; +} + +/** Longest error message a signal carries. */ +const MAX_MESSAGE_CHARS = 240; + +export function emitSignal(input: SignalInput): void { + try { + const event = new WideEventBuilder("background"); + const parent = getEvent(); + if (parent) { + event.setRequestId(parent.getRequestId()); + event.setTraceId(parent.getTraceId()); + } + event.setBackground({ task_name: input.action }); + event.setAction({ name: input.action }); + for (const [key, value] of Object.entries(input.meta ?? {})) { + event.addMeta(key, value); + } + if (input.error !== undefined) { + const err = input.error; + event.addMeta( + "error_type", + err instanceof Error ? err.name || "Error" : typeof err, + ); + const message = err instanceof Error ? err.message : String(err); + event.addMeta("error_message", message.slice(0, MAX_MESSAGE_CHARS)); + } + event.elevateLevel(input.level); + event.finish(); + emitIfSampled(event.toJSON()); + } catch { + // A log line must never break the work it describes. + } +} + +/** + * The common case of a `catch` that carries on: say what failed, at `warn`, + * under a stable action, with the caught error attached. Use it where the + * code deliberately degrades (an enqueue that the next tick repeats, a + * best-effort side write) instead of dropping the error or folding it into an + * `info` annotation. + */ +export function logCaught( + action: string, + error: unknown, + meta?: Record, +): void { + emitSignal({ action, level: "warn", error, meta }); +} + +/** + * `logCaught` shaped for a promise's `.catch`: `.catch(caughtAs("x.y.failed"))` + * in place of `.catch(() => {})`. The promise still settles, so a best-effort + * side write stays best-effort, and its failure leaves a `warn` line. + */ +export function caughtAs( + action: string, + meta?: Record, +): (error: unknown) => void { + return (error: unknown) => logCaught(action, error, meta); +} diff --git a/src/lib/logging/types.ts b/src/lib/logging/types.ts index 727f5440f..8a5119cdb 100644 --- a/src/lib/logging/types.ts +++ b/src/lib/logging/types.ts @@ -115,6 +115,12 @@ export interface WideEvent { duration_ms: number; status?: number; error?: string; + /** + * v1.42 — an expected non-2xx the caller recovers from by design (a + * filter dialect probe), recorded without `error` so a query for + * failed calls does not find it. + */ + note?: string; }>; // Hintergrundaufgabe (nur bei kind === "background") diff --git a/tests/integration/job-failures.test.ts b/tests/integration/job-failures.test.ts index a4c108986..981353dd0 100644 --- a/tests/integration/job-failures.test.ts +++ b/tests/integration/job-failures.test.ts @@ -17,6 +17,7 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { readActiveJobsStartedBefore, + readConsecutiveRunFailures, readFailingQueues, readQueueFailureForUser, } from "@/lib/jobs/job-failures"; @@ -189,6 +190,46 @@ describe("readActiveJobsStartedBefore (#1031)", () => { }); }); +describe("readConsecutiveRunFailures (v1.42)", () => { + const STREAK_QUEUE = "insight-pregenerate"; + + /** Complete one job with the facts a handler returned. */ + async function completeOneJob(did: Record) { + const jobId = await boss.send(STREAK_QUEUE, {}, { retryLimit: 0 }); + const [job] = await boss.fetch(STREAK_QUEUE); + if (!job || job.id !== jobId) throw new Error("nothing to fetch"); + await boss.complete(STREAK_QUEUE, job.id, { ok: true, did }); + return jobId!; + } + + /** Pin the finishing order so the test does not race the clock. */ + async function finishedMinutesAgo(jobId: string, minutes: number) { + await getPrismaClient().$executeRawUnsafe( + `UPDATE pgboss.job SET completed_on = now() - make_interval(mins => ${minutes}) WHERE id = '${jobId}'`, + ); + } + + it("counts failed runs and all-failed nights, skips runs that do not say, and stops at a good night", async () => { + await boss.createQueue(STREAK_QUEUE); + // Oldest first: a good night, then a pg-boss failure, an on-demand job + // that reports nothing about the run, and an all-failed night. + await finishedMinutesAgo(await completeOneJob({ all_failed: false }), 40); + await finishedMinutesAgo( + await failOneJob(STREAK_QUEUE, {}, "worker died"), + 30, + ); + await finishedMinutesAgo(await completeOneJob({ forced: 1 }), 20); + await finishedMinutesAgo(await completeOneJob({ all_failed: true }), 10); + + // Two earlier failures plus the run failing now. + await expect(readConsecutiveRunFailures(STREAK_QUEUE)).resolves.toBe(3); + }); + + it("answers 1 for a queue with no history: the run failing now", async () => { + await expect(readConsecutiveRunFailures("never-ran")).resolves.toBe(1); + }); +}); + describe("no queue schema at all", () => { it("answers null rather than throwing, so the status page still renders", async () => { const prisma = getPrismaClient();