diff --git a/src/__tests__/ai-job-expiry-guard.test.ts b/src/__tests__/ai-job-expiry-guard.test.ts index 5e17d801c..7babbafe3 100644 --- a/src/__tests__/ai-job-expiry-guard.test.ts +++ b/src/__tests__/ai-job-expiry-guard.test.ts @@ -62,6 +62,7 @@ import { STATUS_MODEL_CALLS, } from "@/lib/ai/ai-budgets"; import { AI_RESPONSE_TIMEOUT_MAX_SECONDS } from "@/lib/ai/effective-timeout"; +import { MODEL_RUN_SITES } from "./helpers/model-run-sites"; const SRC = join(process.cwd(), "src"); const ENTRY_POINTS = [ @@ -74,40 +75,6 @@ const ENTRY_POINTS = [ "executeSupplementLabelExtraction", ]; -/** Each importer of an entry point, and the queues whose jobs reach it. */ -const MODEL_RUN_SITES: Readonly> = { - "lib/insights/status-card-generation.ts": [ - "insight-status-generate", - "insights-general-status", - "insight-pregenerate", - ], - "lib/insights/derived/derived-assessment-ai.ts": ["insight-status-generate"], - "lib/insights/status-batch.ts": [ - "insights-general-status", - "insight-pregenerate", - ], - "lib/insights/narrative/period-narrative-generate.ts": [ - "period-narrative-warm", - ], - "lib/ai/coach/conversation-summary.ts": ["coach-memory-refresh"], - "lib/ai/coach/facts.ts": ["coach-memory-refresh"], - "lib/ai/coach/plans.ts": ["coach-memory-refresh"], - "lib/jobs/workout-insight-generate.ts": ["workout-insight-generate"], - "lib/jobs/insight-pregenerate.ts": ["insight-pregenerate"], - "lib/jobs/morning-digest-refresh.ts": ["morning-digest-refresh"], - "lib/documents/extract.ts": ["document-ai-run", "document-index"], - "lib/documents/index-document.ts": ["document-index", "document-ai-run"], - "lib/documents/ai-runs/index-run.ts": ["document-ai-run"], - "lib/documents/ai-runs/summary-run.ts": ["document-ai-run"], - "lib/documents/ai-runs/suggest-run.ts": ["document-ai-run"], - "lib/labs/ocr-extract.ts": ["document-ai-run"], - "lib/jobs/document-ai-run.ts": ["document-ai-run"], - "lib/jobs/document-content-index-backfill.ts": [ - "document-content-index-backfill", - ], - "lib/jobs/document-summary.ts": ["document-summary"], -}; - /** * Importers no queue reaches: a request waits on them, and the browser's * abort follows the published setting (`clientAbortMs`). diff --git a/src/__tests__/ai-job-retry-bound-guard.test.ts b/src/__tests__/ai-job-retry-bound-guard.test.ts new file mode 100644 index 000000000..ec7ac7f84 --- /dev/null +++ b/src/__tests__/ai-job-retry-bound-guard.test.ts @@ -0,0 +1,257 @@ +/** + * Structural guard (#1279): every queue whose jobs reach a model retries a + * bounded number of times, with a backoff, never in a tight loop. + * + * pg-boss gives a send that names no retry options the queue's defaults: + * two retries, no delay, no backoff. A job that reaches a model and fails is + * then run again at once, twice, against a provider that just failed. So + * every place a job for such a queue is created says what it wants: each + * `send` names `retryLimit` (at most `MAX_RETRY_LIMIT`), and unless that is + * 0, `retryBackoff: true` and a `retryDelay` of at least `MIN_RETRY_DELAY_S` + * seconds; each cron carries the same in its schedule options. + * + * The queues are the ones `MODEL_RUN_SITES` attributes to a model-run entry + * point (complete by `ai-job-expiry-guard.test.ts`), plus the queues whose + * jobs call a provider instance directly (`DIRECT_CALL_QUEUES`), which this + * guard keeps complete: every file under `lib/jobs` that calls + * `generateCompletion` must be named there. + * + * Its limit: a `send` whose options come from a variable rather than an + * object literal at the call is reported, not resolved; write the literal. + * The bound on what one failure costs a record is elsewhere (the status + * note's growing window, the chain's bench for background work); this guard + * holds the queue layer. + * + * Mutation check: drop `retryBackoff` from a send in the set, raise a + * `retryLimit` above three, remove a cron's schedule options, or call + * `generateCompletion` from a new job file, and this goes red. The "plants" + * case runs the checker over sources shaped like the tree before #1279. + */ +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { + readSourceText, + stripComments, + walkSourceFiles, +} from "./helpers/source-files"; +import { MODEL_RUN_SITES } from "./helpers/model-run-sites"; +import { allJobDefinitions } from "@/lib/jobs/registry"; + +const SRC = join(process.cwd(), "src"); + +/** The most retries a model-bound job may ask for. */ +const MAX_RETRY_LIMIT = 3; + +/** The least first retry delay, in seconds, when a job retries at all. */ +const MIN_RETRY_DELAY_S = 30; + +/** Job files that call a provider instance directly, and their queue. */ +const DIRECT_CALL_QUEUES: Readonly> = { + "lib/jobs/reaction-line.ts": "reaction-line-generate", + "lib/jobs/coach-nudge-ai.ts": "coach-nudge", +}; + +function modelQueues(): string[] { + return [ + ...new Set([ + ...Object.values(MODEL_RUN_SITES).flat(), + ...Object.values(DIRECT_CALL_QUEUES), + ]), + ].sort(); +} + +interface SourceFile { + rel: string; + text: string; +} + +function sources(): SourceFile[] { + 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.endsWith(".ts")) + .map((rel) => ({ + rel, + text: stripComments(readSourceText(join(SRC, rel))), + })); +} + +/** `export const NAME = "queue"` declarations: constant name → queue. */ +function queueConstants(files: readonly SourceFile[]): Map { + const out = new Map(); + for (const { text } of files) { + for (const m of text.matchAll( + /export const ([A-Z][A-Z0-9_]*_QUEUE)\s*=\s*"([^"]+)"/g, + )) { + out.set(m[1], m[2]); + } + } + return out; +} + +/** The text of the call whose `(` sits at `open`, parentheses balanced. */ +function callText(text: string, open: number): string { + let depth = 0; + for (let i = open; i < text.length; i += 1) { + if (text[i] === "(") depth += 1; + else if (text[i] === ")") { + depth -= 1; + if (depth === 0) return text.slice(open, i + 1); + } + } + return text.slice(open); +} + +/** What is wrong with one set of retry options, or `null`. */ +function retryProblem(options: { + retryLimit?: unknown; + retryDelay?: unknown; + retryBackoff?: unknown; +}): string | null { + const limit = options.retryLimit; + if (typeof limit !== "number") return "names no retryLimit"; + if (limit > MAX_RETRY_LIMIT) + return `retryLimit ${limit} above ${MAX_RETRY_LIMIT}`; + if (limit === 0) return null; + if (options.retryBackoff !== true) return "retries without retryBackoff"; + const delay = options.retryDelay; + if (typeof delay !== "number" || delay < MIN_RETRY_DELAY_S) { + return `retryDelay below ${MIN_RETRY_DELAY_S}s`; + } + return null; +} + +/** The retry options an object literal in `call` spells out. */ +function literalRetryOptions(call: string) { + const num = (key: string) => { + const m = new RegExp(`\\b${key}:\\s*(\\d+)`).exec(call); + return m ? Number(m[1]) : undefined; + }; + const backoff = /\bretryBackoff:\s*(true|false)/.exec(call); + return { + retryLimit: num("retryLimit"), + retryDelay: num("retryDelay"), + retryBackoff: backoff ? backoff[1] === "true" : undefined, + }; +} + +/** Every send to a model-bound queue whose retry options fall short. */ +function sendFindings(files: readonly SourceFile[], queues: readonly string[]) { + const constants = queueConstants(files); + const wanted = new Set(queues); + const problems: string[] = []; + let sends = 0; + for (const { rel, text } of files) { + for (const m of text.matchAll(/\.send(?:After)?\(\s*([A-Z][A-Z0-9_]*)/g)) { + const queue = constants.get(m[1]); + if (!queue || !wanted.has(queue)) continue; + sends += 1; + const open = text.indexOf("(", m.index); + const problem = retryProblem(literalRetryOptions(callText(text, open))); + if (problem) problems.push(`${rel}: send to ${queue} ${problem}`); + } + } + return { problems, sends }; +} + +/** Every cron of a model-bound queue whose schedule options fall short. */ +function cronFindings( + jobs: ReadonlyArray<{ + queue: string; + cron?: string; + scheduleOptions?: Readonly>; + }>, + queues: readonly string[], +) { + const wanted = new Set(queues); + const problems: string[] = []; + for (const job of jobs) { + if (!job.cron || !wanted.has(job.queue)) continue; + const problem = retryProblem(job.scheduleOptions ?? {}); + if (problem) problems.push(`cron ${job.queue} ${problem}`); + } + return problems; +} + +/** Job files that call `generateCompletion` and are not named above. */ +function unnamedDirectCallers(files: readonly SourceFile[]): string[] { + return files + .filter((f) => f.rel.startsWith("lib/jobs/")) + .filter((f) => /\.generateCompletion(?:Stream)?\(/.test(f.text)) + .map((f) => f.rel) + .filter((rel) => !(rel in DIRECT_CALL_QUEUES)) + .sort(); +} + +describe("model-bound queues retry a bounded number of times, backed off", () => { + const files = sources(); + const queues = modelQueues(); + + it("knows the model-bound queues (an empty set is a failure)", () => { + expect(queues.length).toBeGreaterThanOrEqual(12); + const declared = new Set(allJobDefinitions().map((j) => j.queue)); + expect(queues.filter((q) => !declared.has(q))).toEqual([]); + }); + + it("names every job file that calls a provider directly", () => { + expect(unnamedDirectCallers(files)).toEqual([]); + }); + + it("every send to one of them names a bounded, backed-off retry", () => { + const { problems, sends } = sendFindings(files, queues); + expect(sends).toBeGreaterThanOrEqual(12); + expect(problems).toEqual([]); + }); + + it("every cron of one of them carries the same in its schedule options", () => { + expect(cronFindings(allJobDefinitions(), queues)).toEqual([]); + }); + + it("plants: sends and crons shaped like the tree before #1279 are caught", () => { + const planted: SourceFile[] = [ + { + rel: "lib/jobs/planted-queue.ts", + text: 'export const PLANTED_MEMORY_QUEUE = "planted-memory";\nexport const PLANTED_CARD_QUEUE = "planted-card";', + }, + { + rel: "lib/jobs/planted-send.ts", + text: [ + "await boss.send(PLANTED_MEMORY_QUEUE, payload, {", + " singletonKey: `quiet:${id}`,", + " startAfter: 600,", + "});", + "await boss.send(PLANTED_CARD_QUEUE, payload, { retryLimit: 5, retryDelay: 0 });", + ].join("\n"), + }, + { + rel: "lib/jobs/planted-direct.ts", + text: "const r = await provider.generateCompletion(params);", + }, + ]; + expect( + sendFindings(planted, ["planted-memory", "planted-card"]).problems, + ).toEqual([ + "lib/jobs/planted-send.ts: send to planted-memory names no retryLimit", + "lib/jobs/planted-send.ts: send to planted-card retryLimit 5 above 3", + ]); + expect( + cronFindings( + [{ queue: "coach-nudge", cron: "15 5 * * *" }], + ["coach-nudge"], + ), + ).toEqual(["cron coach-nudge names no retryLimit"]); + expect(unnamedDirectCallers(planted)).toEqual([ + "lib/jobs/planted-direct.ts", + ]); + expect(retryProblem({ retryLimit: 2, retryDelay: 60 })).toBe( + "retries without retryBackoff", + ); + expect( + retryProblem({ retryLimit: 3, retryDelay: 5, retryBackoff: true }), + ).toBe("retryDelay below 30s"); + expect(retryProblem({ retryLimit: 0 })).toBeNull(); + }); +}); diff --git a/src/__tests__/helpers/model-run-sites.ts b/src/__tests__/helpers/model-run-sites.ts new file mode 100644 index 000000000..b0d64535b --- /dev/null +++ b/src/__tests__/helpers/model-run-sites.ts @@ -0,0 +1,38 @@ +/** + * Each file that imports a background model-run entry point, and the queues + * whose jobs reach it. Shared by the guards that hold those queues to their + * expiry (`ai-job-expiry-guard.test.ts`) and to bounded, backed-off retries + * (`ai-job-retry-bound-guard.test.ts`); the first proves the table complete. + */ +export const MODEL_RUN_SITES: Readonly> = { + "lib/insights/status-card-generation.ts": [ + "insight-status-generate", + "insights-general-status", + "insight-pregenerate", + ], + "lib/insights/derived/derived-assessment-ai.ts": ["insight-status-generate"], + "lib/insights/status-batch.ts": [ + "insights-general-status", + "insight-pregenerate", + ], + "lib/insights/narrative/period-narrative-generate.ts": [ + "period-narrative-warm", + ], + "lib/ai/coach/conversation-summary.ts": ["coach-memory-refresh"], + "lib/ai/coach/facts.ts": ["coach-memory-refresh"], + "lib/ai/coach/plans.ts": ["coach-memory-refresh"], + "lib/jobs/workout-insight-generate.ts": ["workout-insight-generate"], + "lib/jobs/insight-pregenerate.ts": ["insight-pregenerate"], + "lib/jobs/morning-digest-refresh.ts": ["morning-digest-refresh"], + "lib/documents/extract.ts": ["document-ai-run", "document-index"], + "lib/documents/index-document.ts": ["document-index", "document-ai-run"], + "lib/documents/ai-runs/index-run.ts": ["document-ai-run"], + "lib/documents/ai-runs/summary-run.ts": ["document-ai-run"], + "lib/documents/ai-runs/suggest-run.ts": ["document-ai-run"], + "lib/labs/ocr-extract.ts": ["document-ai-run"], + "lib/jobs/document-ai-run.ts": ["document-ai-run"], + "lib/jobs/document-content-index-backfill.ts": [ + "document-content-index-backfill", + ], + "lib/jobs/document-summary.ts": ["document-summary"], +}; diff --git a/src/lib/ai/coach/__tests__/coach-memory-shared.test.ts b/src/lib/ai/coach/__tests__/coach-memory-shared.test.ts index 3c251954c..14def70ec 100644 --- a/src/lib/ai/coach/__tests__/coach-memory-shared.test.ts +++ b/src/lib/ai/coach/__tests__/coach-memory-shared.test.ts @@ -34,6 +34,10 @@ describe("enqueueCoachMemoryRefresh", () => { singletonSeconds: COACH_MEMORY_QUIET_MS / 1000, startAfter: COACH_MEMORY_QUIET_MS / 1000, expireInSeconds: COACH_MEMORY_REFRESH_EXPIRE_SECONDS, + // #1279 — bounded, backed-off retries instead of pg-boss's two at once. + retryLimit: 2, + retryDelay: 60, + retryBackoff: true, }); expect(COACH_MEMORY_QUIET_MS).toBe(30 * 60_000); }); diff --git a/src/lib/ai/coach/coach-memory-shared.ts b/src/lib/ai/coach/coach-memory-shared.ts index bfbb80393..b5b9eb597 100644 --- a/src/lib/ai/coach/coach-memory-shared.ts +++ b/src/lib/ai/coach/coach-memory-shared.ts @@ -85,6 +85,11 @@ export async function enqueueCoachMemoryRefresh( singletonSeconds: quietSeconds, startAfter: Math.max(1, Math.round(delayMs / 1000)), expireInSeconds: COACH_MEMORY_REFRESH_EXPIRE_SECONDS, + // #1279 — pg-boss's queue default retried a failed pass twice at + // once; the same two retries now wait a minute, then longer. + retryLimit: 2, + retryDelay: 60, + retryBackoff: true, }, ); annotate({ diff --git a/src/lib/jobs/registry/status.ts b/src/lib/jobs/registry/status.ts index 87df05d45..e9e651947 100644 --- a/src/lib/jobs/registry/status.ts +++ b/src/lib/jobs/registry/status.ts @@ -227,9 +227,12 @@ const jobs: readonly JobDefinition[] = [ work: { localConcurrency: 1 }, load: async () => (await handlers()).handleCoachMemoryRefresh, }), + // #1279 — the nudge tick reaches a model; pg-boss's queue default retried + // a failed tick twice at once. The same two retries now back off. defineJob({ queue: COACH_NUDGE_QUEUE, cron: COACH_NUDGE_CRON, + scheduleOptions: { retryLimit: 2, retryDelay: 60, retryBackoff: true }, work: { localConcurrency: 1 }, load: async () => (await handlers()).handleCoachNudge, }), diff --git a/tests/integration/__golden__/queue-registration.json b/tests/integration/__golden__/queue-registration.json index 47d73a4e2..d85705a21 100644 --- a/tests/integration/__golden__/queue-registration.json +++ b/tests/integration/__golden__/queue-registration.json @@ -2035,6 +2035,9 @@ "kind": "cron", "name": "coach-nudge", "options": { + "retryBackoff": true, + "retryDelay": 60, + "retryLimit": 2, "tz": "Europe/Berlin" }, "timezone": "Europe/Berlin"