fix(jobs): report every caught failure in background jobs or say why not

A catch that drops what it caught is where the silent nightly failures
started. Every `catch` and `.catch()` under src/lib/jobs now rethrows,
fails the job, writes to a ledger an operator reads, counts into the run's
failure facts, hands the error back to its caller, or logs at warn under a
stable action. The enqueue helpers that swallowed a failed send (briefing,
status, narrative, morning digest, document index / summary / thumbnail,
workout insight, boot-time sweeps), the Coach nudge's provider fallback,
the workout and reaction-line cleanups, the advisory-lock release, the
backup attempt bookkeeping and the cycle reminder's dedupe reads now leave
a warn line instead of nothing or an info annotation.

`error-level-discipline-guard` holds the rule. The catches that stay
silent are listed with their reason (the pgboss readers' documented null on
a web-only deployment, the key-ring trial, the restore drill's per-account
result, an already-deleted Telegram message); a new silent catch fails, an
entry that no longer matches fails, and each entry is shown to be
load-bearing by removing it.

A Coach turn cut short by its token, time or round budget writes
`coach.turn.budget_stop` at warn.
This commit is contained in:
Marc-André Bombeck committed 2026-10-08 09:06:17 +02:00
1 parent 27342ee734
commit 511eae8241
24 files changed
+492 -44

No files matched your search

@@ -0,0 +1,316 @@
/**
* A caught error in a background job is either reported or explained.
*
* The failures that cost the most in this codebase were never thrown at
* anyone: a nightly pass that caught its error, annotated an `info` line and
* carried on, completed in pg-boss every night while doing nothing. The
* briefing warm read "all-failed" for a week before anyone could tell which
* account and why; the environment fetch threw on every run for the life of
* the module. A `catch` that drops what it caught is where that starts.
*
* Scope: every `catch` clause and every `.catch(handler)` under
* `src/lib/jobs/` (tests excluded). A catch passes when its body does one of:
*
* - rethrows, rejects, or hands the error to a stream callback;
* - fails the job (`jobFailed`);
* - logs at warn or error: `logCaught` / `caughtAs`, `emitSignal`,
* `reportWorkerError`,
* `addWarning`, `setError`, `elevateLevel`, `workerLog("error", …)`,
* `console.error` / `console.warn`;
* - writes the failure to a ledger an operator surface reads
* (`recordSyncFailure`, `failAiRun`, `markFailed`,
* `recordThumbnailFailure`, `recordBriefingFailure`);
* - counts it into a failure tally (`failed += 1`, `errored++`,
* `failedSubjects.set(…)`) that the job reports in its outcome facts, so a
* systemic failure shows as a count and a partial run as
* `job.run.partial`;
* - hands it back to its caller in the result (`error: err…`), which is
* how every boot-time discovery enqueue reports to the registrar that
* logs it at error.
*
* Anything else is listed in `ALLOWED` below with the reason it may stay
* silent, keyed by file and enclosing function, with how many such catches
* that function holds. The reasons are the review: "best effort" is not one;
* "the pgboss schema is absent on a web-only deployment and null is the
* documented answer" is.
*
* The list cannot rot quietly in either direction. A new silent catch fails
* the second case. An entry whose catches were fixed or removed fails the
* third case (stale entry). And the fourth case removes each entry in turn
* and asserts the sweep goes red without it, so every entry is shown to be
* load-bearing, and the sweep is shown to find what it claims to find.
*/
import { readFileSync } from "node:fs";
import { join } from "node:path";
import ts from "typescript";
import { describe, expect, it } from "vitest";
import { walkSourceFiles } from "./helpers/source-files";
const SRC = join(process.cwd(), "src");
const SCOPE = "lib/jobs/";
const REPORTS =
/\bthrow\b|\breject\(|\bcallback\(\s*err|\bjobFailed\(|\blogCaught\(|\bcaughtAs\(|\bemitSignal\(|\breportWorkerError\(|\.addWarning\(|\.setError\(|\belevateLevel\(|\bworkerLog\(\s*["']error|\bconsole\.(error|warn)\(|\brecordSyncFailure\(|\bfailAiRun\(|\bmarkFailed\(|\brecordThumbnailFailure\(|\brecordBriefingFailure\(|\berror:\s*err\b/;
/** A failure tally: `summary.failed += 1`, `errored++`, `failures += 1`. */
const COUNTS =
/\b[\w.]*(?:failed|failures|errored|Failed|Failures)\w*\s*(?:\+=|\+\+|\.set\(|\.push\()/;
interface Finding {
/** `<file under src>::<enclosing function>` */
key: string;
line: number;
body: string;
}
function scan(file: string, text: string): Finding[] {
const sf = ts.createSourceFile(file, text, ts.ScriptTarget.Latest, true);
const out: Finding[] = [];
const visit = (node: ts.Node, fn: string): void => {
let name = fn;
if (
(ts.isFunctionDeclaration(node) || ts.isMethodDeclaration(node)) &&
node.name
) {
name = node.name.getText(sf);
} else if (
ts.isVariableDeclaration(node) &&
node.initializer &&
(ts.isArrowFunction(node.initializer) ||
ts.isFunctionExpression(node.initializer))
) {
name = node.name.getText(sf);
}
let body: string | null = null;
if (ts.isCatchClause(node)) body = node.block.getText(sf);
if (
ts.isCallExpression(node) &&
ts.isPropertyAccessExpression(node.expression) &&
node.expression.name.getText(sf) === "catch" &&
node.arguments[0]
) {
body = node.arguments[0].getText(sf);
}
if (body !== null && !REPORTS.test(body) && !COUNTS.test(body)) {
out.push({
key: `${file}::${name}`,
line: sf.getLineAndCharacterOfPosition(node.getStart(sf)).line + 1,
body,
});
}
ts.forEachChild(node, (child) => visit(child, name));
};
visit(sf, "<module>");
return out;
}
/**
* Catches allowed to stay silent, by file and function: how many, and why.
* Keep the reason checkable against the code it names.
*/
const ALLOWED: Readonly<Record<string, { count: number; reason: string }>> = {
"lib/jobs/apple-health-import-worker.ts::reconcileOrphanImportJobs": {
count: 1,
reason:
"A failed live-state lookup reads as live on purpose: the heartbeat is fresh, flipping a running import would kill it, and the next boot re-evaluates once the heartbeat goes stale.",
},
"lib/jobs/apple-health-import-worker.ts::safeUnlink": {
count: 1,
reason:
"Deleting a staged file that a previous step or sweep already removed; ENOENT is the expected case and nothing is lost either way.",
},
"lib/jobs/backup-restore.ts::writeProgress": {
count: 1,
reason:
"A progress tick for the status poll; the next tick rewrites it and the job's own outcome is recorded by `finish`.",
},
"lib/jobs/backup-restore.ts::runBackupRestoreJob": {
count: 1,
reason:
"Waits out the in-flight progress write before the final state is written; that write's failure is the progress tick's, see writeProgress.",
},
"lib/jobs/backup-restore.ts::sweepInterruptedRestores": {
count: 1,
reason:
"Completing the dead worker's superseded pg-boss delivery; if it fails the delivery expires on its own, and the restore row it belonged to is already failed and counted.",
},
"lib/jobs/boss-instance.ts::startProducerAttempts": {
count: 2,
reason:
"The start error is kept and thrown after the last attempt; stopping a boss that never started can only fail for the same reason.",
},
"lib/jobs/boss-instance.ts::superviseProducerReconnect": {
count: 1,
reason:
"Stopping the producer whose connection already failed, before reconnecting; the reconnect loop reports its own failures.",
},
"lib/jobs/free-text-encryption-backfill.ts::scrubContactAuditDetails": {
count: 1,
reason:
"A parser: details that are not this action's JSON are left alone by contract (null means 'nothing to scrub'), not an error.",
},
"lib/jobs/geo-backfill.ts::runGeoBackfill": {
count: 1,
reason:
"The row was deleted by the retention sweep between read and update; counted as still unresolved, and the next pass no longer sees it.",
},
"lib/jobs/host-metric-sampler.ts::captureHostMetric": {
count: 1,
reason:
"Disk counters exist only on Linux; elsewhere the sample records null for them by design.",
},
"lib/jobs/host-metric-sampler.ts::readLinuxDiskStats": {
count: 1,
reason: "No /proc/diskstats outside Linux; see captureHostMetric.",
},
"lib/jobs/job-failures.ts::readActiveJobsStartedBefore": {
count: 1,
reason:
"No pgboss schema on a web-only deployment; null is the documented 'no queue to ask', kept apart from an empty list.",
},
"lib/jobs/job-failures.ts::readConsecutiveRunFailures": {
count: 1,
reason:
"Called from a failure path that is already logging at error; when the queue cannot be read, 1 (this run) is the only known count.",
},
"lib/jobs/job-failures.ts::readFailingQueues": {
count: 1,
reason: "See readActiveJobsStartedBefore: null means no queue schema.",
},
"lib/jobs/job-failures.ts::readLastQueueRun": {
count: 1,
reason: "See readActiveJobsStartedBefore: null means no queue schema.",
},
"lib/jobs/job-failures.ts::readQueueFailureForUser": {
count: 1,
reason: "See readActiveJobsStartedBefore: null means no queue schema.",
},
"lib/jobs/job-failures.ts::readQueueRunningSince": {
count: 1,
reason: "See readActiveJobsStartedBefore: null means no queue schema.",
},
"lib/jobs/job-observer.ts::emitJobLine": {
count: 1,
reason:
"The catch around writing a log line itself; there is nowhere further to log to, and a log line must not break the job.",
},
"lib/jobs/offhost-backup.ts::getS3Client": {
count: 2,
reason:
"headObject answers 'absent' for a missing object (its contract), and the lifecycle read maps NoSuchLifecycleConfiguration to 'missing' and anything else to 'unknown', which the status card shows.",
},
"lib/jobs/offhost-backup.ts::loadOffhostConfigSafe": {
count: 1,
reason:
"A malformed key reads as 'not configured' for the status card; the backup pass itself calls the throwing loader and fails loudly.",
},
"lib/jobs/offhost-backup.ts::probeOffhostLifecycle": {
count: 1,
reason:
"The lifecycle probe answers 'unknown', which the admin card renders as such; it is a display read, not a pass.",
},
"lib/jobs/offhost-backup.ts::runOffhostRoundtripTest": {
count: 2,
reason:
"The admin's connection test returns its error in the result it shows (ok: false, error), and deleting its own probe object is cleanup.",
},
"lib/jobs/offhost-backup.ts::withEachKey": {
count: 1,
reason:
"Trying each key of the ring in turn; the last error is thrown when no key opens the object.",
},
"lib/jobs/reminder/medication-reminder-check.ts::cleanupScheduledTelegramDeletions":
{
count: 1,
reason:
"Deleting a reminder message the person usually already deleted; the scheduled row is removed either way, and logging every expected 400 would bury the real ones.",
},
"lib/jobs/reminder/polar-sync.ts::syncUserPolarLegs": {
count: 2,
reason:
"Both legs run, the first error is kept and thrown after the second leg.",
},
"lib/jobs/restore-drill.ts::runRestoreDrill": {
count: 1,
reason:
"The per-account drill failure is recorded on the account's result (ok: false, error) and counted into the run's restore_drill_accounts_failed fact.",
},
};
function sweep(): Finding[] {
return walkSourceFiles(join(SRC, SCOPE), { floor: 100 })
.filter((p) => !p.includes("__tests__") && !p.endsWith(".test.ts"))
.flatMap((p) =>
scan(`${SCOPE}${p}`, readFileSync(join(SRC, SCOPE, p), "utf8")),
);
}
function unallowed(
findings: readonly Finding[],
allowed: Readonly<Record<string, { count: number }>>,
): string[] {
const byKey = new Map<string, Finding[]>();
for (const f of findings) {
byKey.set(f.key, [...(byKey.get(f.key) ?? []), f]);
}
const out: string[] = [];
for (const [key, list] of byKey) {
const allowance = allowed[key]?.count ?? 0;
if (list.length > allowance) {
for (const f of list.slice(allowance)) {
out.push(
`${key} (line ${f.line}): ${f.body.replace(/\s+/g, " ").slice(0, 120)}`,
);
}
}
}
return out.sort();
}
describe("error-level discipline in background jobs", () => {
it("the scanner flags a silent catch and passes a reported one", () => {
const planted = scan(
"lib/jobs/planted.ts",
[
"async function quiet() { try { await a(); } catch { /* best effort */ } }",
"async function info() { try { await a(); } catch (e) { annotate({ action: { name: 'x.failed' } }); } }",
"async function chained() { await a().catch(() => {}); }",
"async function loud() { try { await a(); } catch (e) { logCaught('x.y.failed', e); } }",
"async function counted() { try { await a(); } catch { summary.failed += 1; } }",
"async function rethrown() { try { await a(); } catch (e) { throw e; } }",
].join("\n"),
);
expect(planted.map((f) => f.key)).toEqual([
"lib/jobs/planted.ts::quiet",
"lib/jobs/planted.ts::info",
"lib/jobs/planted.ts::chained",
]);
});
it("every silent catch in src/lib/jobs is fixed or explained", () => {
expect(unallowed(sweep(), ALLOWED)).toEqual([]);
});
it("no allow-list entry outlives the catches it explains", () => {
const counts = new Map<string, number>();
for (const f of sweep()) counts.set(f.key, (counts.get(f.key) ?? 0) + 1);
const stale = Object.entries(ALLOWED)
.filter(([key, { count }]) => (counts.get(key) ?? 0) < count)
.map(
([key, { count }]) =>
`${key}: allows ${count}, finds ${counts.get(key) ?? 0}`,
);
expect(stale).toEqual([]);
});
it("every allow-list entry is load-bearing: removing it turns the sweep red", () => {
const findings = sweep();
const notLoadBearing = Object.keys(ALLOWED).filter((key) => {
const without = { ...ALLOWED };
delete (without as Record<string, unknown>)[key];
return unallowed(findings, without).length === 0;
});
expect(notLoadBearing).toEqual([]);
});
});
@@ -18,6 +18,11 @@ vi.mock("@/lib/ai/provider-runner", () => ({
runRawCompletionWithFallback: (args: unknown) =>
runRawCompletionWithFallback(args),
}));
// v1.42 — the budget stop's own warn line.
const emitSignal = vi.fn();
vi.mock("@/lib/logging/signal", () => ({
emitSignal: (...a: unknown[]) => emitSignal(...a),
}));
const runDialogTool = vi.fn();
vi.mock("@/lib/ai/coach/tools/dialog-tools", async (importOriginal) => ({
...(await importOriginal<object>()),
@@ -106,6 +111,7 @@ function lastParams() {
}
beforeEach(() => {
emitSignal.mockReset();
executeCoachTool.mockReset();
runRawCompletionWithFallback.mockReset();
runDialogTool.mockReset();
@@ -461,3 +467,31 @@ describe("reasoning, provider state and the trail", () => {
});
});
});
describe("budget stop line (v1.42)", () => {
it("writes coach.turn.budget_stop at warn, once, when the budget ends the turn", async () => {
runRawCompletionWithFallback
.mockResolvedValueOnce(round("", [table("bp")]))
.mockResolvedValueOnce(round("Answer from what I have."));
await runCoachToolLoop({
...baseArgs,
budget: createTurnBudget({
payer: "operator",
initialInputTokens: 10,
limits: { tokens: 500, wallMs: 1e9, maxRounds: 12 },
}),
});
expect(emitSignal).toHaveBeenCalledTimes(1);
expect(emitSignal.mock.calls[0][0]).toMatchObject({
action: "coach.turn.budget_stop",
level: "warn",
meta: { reason: "budget", round: 2, payer: "operator" },
});
});
it("writes nothing for a turn that answered on its own", async () => {
runRawCompletionWithFallback.mockResolvedValueOnce(round("Plain answer."));
await runCoachToolLoop({ ...baseArgs, budget: roomy() });
expect(emitSignal).not.toHaveBeenCalled();
});
});
+19
View File
@@ -33,6 +33,7 @@
import { coachPromptCacheKey } from "@/lib/ai/coach/prompt-cache-key";
import type { CoachHistoryReach } from "@/lib/ai/coach/history-reach";
import { annotate } from "@/lib/logging/context";
import { emitSignal } from "@/lib/logging/signal";
import type { Locale } from "@/lib/i18n/config";
import { runRawCompletionWithFallback } from "@/lib/ai/provider-runner";
import type { ProviderChainResolved } from "@/lib/ai/provider-runner";
@@ -295,6 +296,7 @@ export async function runCoachToolLoop(args: {
let planProposal: CoachPlanProposal | undefined;
let callCount = 0;
let stopReason: CoachStopReason | null = null;
let stopSignalled = false;
let progressStalled = false;
for (let round = 1; ; round += 1) {
@@ -308,6 +310,23 @@ export async function runCoachToolLoop(args: {
}
const isFinal = stopReason !== null;
if (stopReason) {
// v1.42 — a turn cut short by its budget is not a failure, but an
// operator who sees answers getting thin needs to count it. It used to
// be one `stop` meta key on the turn's info line.
if (!stopSignalled && stopReason !== "no_progress") {
stopSignalled = true;
emitSignal({
action: "coach.turn.budget_stop",
level: "warn",
meta: {
reason: stopReason,
round,
spent: budget.spent(),
elapsedMs: budget.elapsedMs(),
payer: budget.payer,
},
});
}
const label = stopActivityLabel(locale, stopReason);
const id = activity.start({
phase: "stop",
+10 -3
View File
@@ -54,6 +54,7 @@ import {
sweepStaleImportStaging,
type StagingInUse,
} from "@/lib/import/apple-health-staging";
import { caughtAs, logCaught } from "@/lib/logging/signal";
/**
* Queue + cron for the periodic orphan-ImportJob sweep. v1.32.1
@@ -100,10 +101,16 @@ export async function handleImportJobReconcileTick(
// deploy), the person's export does not stay in /tmp past this.
// A queued or running import keeps its files whatever their age;
// when that cannot be read, nothing is swept this tick.
const inUse = await stagedImportFilesInUse().catch(() => null);
const inUse = await stagedImportFilesInUse().catch((err: unknown) => {
logCaught("apple_health.import.staging_probe_failed", err);
return null;
});
const swept = inUse
? await sweepStaleImportStaging(undefined, undefined, inUse).catch(
() => 0,
(err: unknown) => {
logCaught("apple_health.import.staging_sweep_failed", err);
return 0;
},
)
: 0;
evt.addMeta("import_staging_swept", swept);
@@ -463,7 +470,7 @@ export async function handleAppleHealthImport(
onExportDate: async (exportedAt) => {
await prisma.importJob
.update({ where: { id: importJobId }, data: { exportedAt } })
.catch(() => {});
.catch(caughtAs("apple_health.import.export_date_failed"));
},
})),
ecg: {
+9 -4
View File
@@ -28,6 +28,7 @@
* least every second run.
*/
import type { PrismaClient } from "@/generated/prisma/client";
import { logCaught } from "@/lib/logging/signal";
/** The pass's queue name, which is also the marker's key. */
export type BackupPass = "data-backup" | "data-backup-offhost";
@@ -46,7 +47,8 @@ export async function markBackupAttemptStarted(
update: { startedAt: at },
create: { userId, pass, startedAt: at },
});
} catch {
} catch (err) {
logCaught("backup.attempt.mark_failed", err, { phase: "started" });
// Bookkeeping only; an account deleted meanwhile has no row to write.
}
}
@@ -62,7 +64,8 @@ export async function markBackupAttemptFinished(
where: { userId, pass },
data: { finishedAt: at },
});
} catch {
} catch (err) {
logCaught("backup.attempt.mark_failed", err, { phase: "finished" });
// Bookkeeping only.
}
}
@@ -87,7 +90,8 @@ export async function readInterruptedBackupAttempts(
)
.map((row) => [row.userId, row.startedAt]),
);
} catch {
} catch (err) {
logCaught("backup.attempt.read_failed", err);
return new Map();
}
}
@@ -131,7 +135,8 @@ export async function readAccountsToTakeLast(
select: { userId: true, startedAt: true, finishedAt: true },
});
return heldBackAccounts(rows);
} catch {
} catch (err) {
logCaught("backup.attempt.read_failed", err);
return new Map();
}
}
+5 -2
View File
@@ -77,6 +77,7 @@ import {
RESTORE_FAILURE_CAUSES,
type RestoreFailureCause,
} from "@/lib/export/restore-failure-cause";
import { logCaught } from "@/lib/logging/signal";
export const BACKUP_RESTORE_QUEUE = "backup-restore";
@@ -478,7 +479,8 @@ export async function admitBackupRestore(input: {
data: { pgBossJobId: bossJobId },
});
return { admitted: true, jobId };
} catch {
} catch (err) {
logCaught("backup.restore.enqueue_failed", err, { jobId });
// Written but never queued: fail it here, or it holds the account's slot
// until the unclaimed window runs out.
await prisma.backupRestoreJob.update({
@@ -818,7 +820,8 @@ export async function enqueueBackupRestoreSweepAtBoot(
expireInSeconds: BACKUP_RESTORE_EXPIRE_SECONDS,
},
);
} catch {
} catch (err) {
logCaught("backup.restore.sweep_enqueue_failed", err);
// The admission check fails an abandoned job the next time the account
// is restored, so a missed sweep costs a retry, not a stuck account.
}
+7 -4
View File
@@ -49,6 +49,7 @@ import { singleUserTurn } from "@/lib/ai/types";
import { screenCoachReply } from "@/lib/ai/coach/outbound-guard";
import { extractAssessmentSummary } from "@/lib/insights/status-shared";
import { annotate } from "@/lib/logging/context";
import { caughtAs, logCaught } from "@/lib/logging/signal";
/** Per-call upstream timeout — kept inside the spec's ≤8–10 s window. */
export const COACH_NUDGE_AI_CALL_TIMEOUT_MS = 9_000;
@@ -246,13 +247,14 @@ export const composeNudgeWithAI: ComposeNudgeWithAI = async (params) => {
signal: AbortSignal.timeout(COACH_NUDGE_AI_CALL_TIMEOUT_MS + 1_000),
}),
);
} catch {
} catch (err) {
// Timeout / network / provider error → refund what wasn't spent and
// fall back to the template.
logCaught("coach.nudge.ai_failed", err);
await reconcileSpend(params.userId, reservation.reserved, 0, dateKey, 0, {
servedBy: null,
reservedOwner: reservation.owner,
}).catch(() => {});
}).catch(caughtAs("coach.nudge.reconcile_failed"));
annotate({ action: { name: "coach.nudge.ai.fallback" } });
return null;
}
@@ -264,7 +266,7 @@ export const composeNudgeWithAI: ComposeNudgeWithAI = async (params) => {
dateKey,
result.cachedInputTokens ?? 0,
{ servedBy: chain[0].providerType, reservedOwner: reservation.owner },
).catch(() => {});
).catch(caughtAs("coach.nudge.reconcile_failed"));
const body = sanitiseAiBody(result.content, params.locale);
if (!body) {
@@ -274,7 +276,8 @@ export const composeNudgeWithAI: ComposeNudgeWithAI = async (params) => {
annotate({ action: { name: "coach.nudge.ai.composed" } });
return { title: params.template.title, body };
} catch {
} catch (err) {
logCaught("coach.nudge.compose_failed", err);
// Defensive: any unexpected failure keeps the template.
return null;
}
+7 -2
View File
@@ -49,6 +49,7 @@ import type { EventType } from "@/lib/notifications/types";
import { isCycleReminderClientManaged } from "@/lib/validations/notification-prefs";
import { getEvent } from "@/lib/logging/context";
import { DEFAULT_TIMEZONE } from "@/lib/tz/format";
import { logCaught } from "@/lib/logging/signal";
/**
* Local-time hour (0–23) at which the daily cycle reminder fires. 09:00 is
@@ -266,7 +267,10 @@ async function alreadyNotifiedToday(
select: { id: true },
});
return row !== null;
} catch {
} catch (err) {
// Fail toward "not yet notified": a read failure must not silence the
// reminder for the day, but it has to be visible.
logCaught("cycle.reminder.dedupe_read_failed", err);
return false;
}
}
@@ -548,7 +552,8 @@ async function hasObservedPeriodNearPrediction(
delta >= -PERIOD_CONFIRM_GRACE_DAYS && delta <= PERIOD_SOON_LEAD_DAYS
);
});
} catch {
} catch (err) {
logCaught("cycle.reminder.period_read_failed", err);
return false;
}
}
+3 -1
View File
@@ -44,6 +44,7 @@ import { isArrivalKind, type DataArrival } from "@/lib/arrivals/types";
import { enqueueWorkoutInsight } from "@/lib/jobs/workout-insight-generate-shared";
import { getWorkerPrisma, workerLog } from "./reminder/shared";
import { logCaught } from "@/lib/logging/signal";
export { DATA_ARRIVAL_QUEUE };
@@ -274,7 +275,8 @@ export async function runDataArrival(
// process, and it costs nothing when it is not.
try {
invalidateUserDashboardSnapshot(arrival.userId);
} catch {
} catch (err) {
logCaught("daily.arrival.cache_evict_failed", err);
// A cache miss is never worth failing a job over.
}
+2
View File
@@ -89,6 +89,7 @@ import {
OCR_EXTRACT_MODEL_CALLS,
} from "@/lib/labs/ocr-run";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const DOCUMENT_AI_RUN_QUEUE = "document-ai-run";
@@ -133,6 +134,7 @@ export async function enqueueDocumentAiRun(
action: { name: "ai_runs.enqueue_failed" },
meta: { reason: err instanceof Error ? err.name : "unknown" },
});
logCaught("ai_runs.enqueue_failed", err);
return false;
}
}
+5 -1
View File
@@ -16,6 +16,7 @@ import { maybeAutoStageLabFacts } from "@/lib/documents/auto-stage-labs";
import { indexDocumentContent } from "@/lib/documents/index-document";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { annotate } from "@/lib/logging/context";
import { caughtAs, logCaught } from "@/lib/logging/signal";
export const DOCUMENT_INDEX_QUEUE = "document-index";
@@ -63,7 +64,9 @@ export async function runDocumentIndex(
// A deferred import stages nothing: lab staging is an AI read, and the
// person asked for those to wait.
if (outcome.indexed && !localOnly) {
await maybeAutoStageLabFacts(userId, documentId).catch(() => {});
await maybeAutoStageLabFacts(userId, documentId).catch(
caughtAs("documents.auto_index.lab_staging_failed"),
);
}
}
@@ -101,6 +104,7 @@ export async function enqueueDocumentIndex(
action: { name: "documents.autoIndex.enqueueFailed" },
meta: { documentId, reason: err instanceof Error ? err.name : "unknown" },
});
logCaught("documents.auto_index.enqueue_failed", err);
return { enqueued: false };
}
}
+2
View File
@@ -39,6 +39,7 @@ import { prisma } from "@/lib/db";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { enqueueDocumentSummary } from "@/lib/jobs/document-summary";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const DOCUMENT_SUMMARY_CATCHUP_QUEUE = "document-summary-catchup";
@@ -184,6 +185,7 @@ export async function enqueueSummaryCatchUp(
action: { name: "documents.autoRead.catchUpEnqueueFailed" },
meta: { reason: err instanceof Error ? err.name : "unknown" },
});
logCaught("documents.auto_read.catch_up_enqueue_failed", err);
return { enqueued: false };
}
}
+4 -1
View File
@@ -48,6 +48,7 @@ import { prisma } from "@/lib/db";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { annotate } from "@/lib/logging/context";
import type { DocumentSummaryState } from "@/generated/prisma/client";
import { logCaught } from "@/lib/logging/signal";
export const DOCUMENT_SUMMARY_QUEUE = "document-summary";
@@ -311,7 +312,8 @@ export async function runDocumentSummaryJob(
persisted: written.count > 0,
},
});
} catch {
} catch (err) {
logCaught("documents.summary.auto_failed", err);
// Refund the reservation on a provider miss; the document keeps no summary
// (the on-demand route remains the manual fallback). Never rethrow — a
// transient provider error must not retry-loop or fail the queue.
@@ -394,6 +396,7 @@ export async function enqueueDocumentSummary(
action: { name: "documents.summary.enqueueFailed" },
meta: { documentId, reason: err instanceof Error ? err.name : "unknown" },
});
logCaught("documents.summary.enqueue_failed", err);
return { enqueued: false };
}
}
+3 -1
View File
@@ -29,6 +29,7 @@ import { nativeCanvasSupported } from "@/lib/documents/native-canvas-support";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { enqueueDocumentThumbnail } from "@/lib/jobs/document-thumbnail";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const DOCUMENT_THUMBNAIL_BACKFILL_QUEUE = "document-thumbnail-backfill";
@@ -211,7 +212,8 @@ export async function enqueueBootTimeThumbnailBackfill(): Promise<{
},
);
if (jobId) enqueued += 1;
} catch {
} catch (err) {
logCaught("documents.thumbnail.backfill_enqueue_failed", err);
// A transient send failure is a no-op — the next boot re-discovers the
// still-missing thumbnails.
}
+2
View File
@@ -33,6 +33,7 @@ import { generateThumbnail } from "@/lib/documents/thumbnail";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { withBackgroundEvent } from "@/lib/logging/background";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const DOCUMENT_THUMBNAIL_QUEUE = "document-thumbnail";
@@ -213,6 +214,7 @@ export async function enqueueDocumentThumbnail(
action: { name: "documents.thumbnail.enqueueFailed" },
meta: { documentId, reason: err instanceof Error ? err.name : "unknown" },
});
logCaught("documents.thumbnail.enqueue_failed", err);
return { enqueued: false };
}
}
+3 -1
View File
@@ -23,6 +23,7 @@ import { envFlag } from "@/lib/env";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { offlineGeoReady } from "@/lib/geo";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const GEOLITE2_FETCH_QUEUE = "geolite2-fetch";
@@ -71,7 +72,8 @@ export async function enqueueGeolite2FetchBootDiscovery(): Promise<{
meta: { enqueued },
});
return { enqueued };
} catch {
} catch (err) {
logCaught("geolite2.fetch.enqueue_failed", err);
// A transient send failure is a no-op — the monthly cron and the next boot
// both re-attempt.
return { enqueued: 0 };
@@ -20,6 +20,7 @@ import {
providerWorkAuthorityForRecord,
type ProviderWorkAuthority,
} from "@/lib/sharing/provider-work-authority";
import { logCaught } from "@/lib/logging/signal";
export const INSIGHT_STATUS_GENERATE_QUEUE = "insight-status-generate";
@@ -165,7 +166,10 @@ export async function enqueueStatusGeneration(
action: { name: "insights.status.generate.enqueued" },
meta: { metric: payload.metric },
});
} catch {
} catch (err) {
logCaught("insights.status.enqueue_failed", err, {
metric: payload.metric,
});
// Enqueue is best-effort; a failure here just means the card stays
// in `preparing` until the next poll / nightly cron warms the cache.
}
+3 -2
View File
@@ -19,6 +19,7 @@
* One extra connection per running locked job.
*/
import { Client } from "pg";
import { caughtAs } from "@/lib/logging/signal";
/** What a guarded run came to. */
export type GuardedRun<T> = { ran: true; result: T } | { ran: false };
@@ -50,9 +51,9 @@ export async function withJobLock<T>(
} finally {
await client
.query("SELECT pg_advisory_unlock(hashtextextended($1, 0))", [key])
.catch(() => {});
.catch(caughtAs("jobs.lock.release_failed"));
}
} finally {
await client.end().catch(() => {});
await client.end().catch(caughtAs("jobs.lock.release_failed"));
}
}
@@ -15,6 +15,7 @@
*/
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const MORNING_DIGEST_REFRESH_QUEUE = "morning-digest-refresh";
@@ -76,7 +77,8 @@ export async function enqueueMorningDigestRefresh(
action: { name: "daily.morning_refresh.enqueued" },
meta: { local_date: payload.localDate },
});
} catch {
} catch (err) {
logCaught("daily.morning_refresh.enqueue_failed", err);
// Best-effort: a failed enqueue leaves the next sleep landing + the nightly
// cron as the catch-net; the digest stays honestly provisional meanwhile.
}
+3 -1
View File
@@ -45,6 +45,7 @@ import {
import { reportWorkerError } from "@/lib/jobs/report-worker-error";
import { getWorkerPrisma } from "@/lib/jobs/reminder/shared";
import { withBackgroundEvent } from "@/lib/logging/background";
import { logCaught } from "@/lib/logging/signal";
export const OFFHOST_PURGE_QUEUE = "offhost-backup-purge";
/** 03:40, after the nightly upload (02:30) has had its hour. */
@@ -104,7 +105,8 @@ export async function kickOffhostPurge(): Promise<void> {
retryDelay: 300,
},
);
} catch {
} catch (err) {
logCaught("backup.offhost_purge.kick_failed", err);
// The cron is the backstop.
}
}
+5 -1
View File
@@ -20,6 +20,7 @@ import {
} from "@/lib/sharing/provider-work-authority";
import type { NarrativePeriod } from "@/lib/insights/narrative/period-narrative";
import type { Locale } from "@/lib/i18n/config";
import { logCaught } from "@/lib/logging/signal";
export const PERIOD_NARRATIVE_QUEUE = "period-narrative-warm";
@@ -100,7 +101,10 @@ export async function enqueueNarrativeWarm(payload: {
action: { name: "insights.narrative.warm.enqueued" },
meta: { period: payload.period, locale: payload.locale },
});
} catch {
} catch (err) {
logCaught("insights.narrative.enqueue_failed", err, {
period: payload.period,
});
// Best-effort — a failure just means the narrative stays as-is until the
// next read or the nightly cron warms it.
}
+20 -9
View File
@@ -78,6 +78,7 @@ import {
} from "@/lib/arrivals/reaction-line-shared";
import { workerLog } from "./reminder/shared";
import { caughtAs } from "@/lib/logging/signal";
export { REACTION_LINE_QUEUE };
@@ -567,7 +568,7 @@ export async function runReactionLine(
const hasReservedTokens = row.generationReservedTokens != null;
const hasReservationDate = row.generationBudgetDateKey != null;
if (hasReservedTokens !== hasReservationDate) {
await releaseClaim().catch(() => {});
await releaseClaim().catch(caughtAs("reaction_line.cleanup_failed"));
return { status: "skipped", reason: "invalid_reservation_state" };
}
@@ -600,7 +601,7 @@ export async function runReactionLine(
if (err instanceof ReactionClaimLostError) {
return { status: "skipped", reason: "claim_lost" };
}
await releaseClaim().catch(() => {});
await releaseClaim().catch(caughtAs("reaction_line.cleanup_failed"));
throw err;
}
}
@@ -620,7 +621,7 @@ export async function runReactionLine(
} catch (err) {
// No provider was touched, so releasing ownership is retry-safe. The
// durable reservation remains attached and is reused by the retry.
await releaseClaim().catch(() => {});
await releaseClaim().catch(caughtAs("reaction_line.cleanup_failed"));
throw err;
}
@@ -690,8 +691,10 @@ export async function runReactionLine(
reservation.dateKey,
0,
{ servedBy: chain[0].providerType, reservedOwner: reservation.owner },
).catch(() => {});
await finishTerminalAttempt().catch(() => {});
).catch(caughtAs("reaction_line.cleanup_failed"));
await finishTerminalAttempt().catch(
caughtAs("reaction_line.cleanup_failed"),
);
workerLog("error", "[reaction-line] generation failed", err);
return { status: "skipped", reason: "provider_failed" };
}
@@ -708,14 +711,18 @@ export async function runReactionLine(
} catch (err) {
// Do not publish a line whose spend was not durably reconciled. The
// invocation timestamp remains terminal, preventing a second provider call.
await finishTerminalAttempt().catch(() => {});
await finishTerminalAttempt().catch(
caughtAs("reaction_line.cleanup_failed"),
);
workerLog("error", "[reaction-line] spend reconciliation failed", err);
return { status: "skipped", reason: "spend_reconciliation_failed" };
}
const line = sanitiseReactionLine(result.content, locale);
if (!line) {
await finishTerminalAttempt().catch(() => {});
await finishTerminalAttempt().catch(
caughtAs("reaction_line.cleanup_failed"),
);
return { status: "skipped", reason: "unusable_output" };
}
// Prompt instructions are not an enforcement boundary. Grade every numeric
@@ -729,7 +736,9 @@ export async function runReactionLine(
(evidenceLedger.length === 0 && /\d/.test(line)) ||
findUnverifiedCoachNumbersInLedger(line, evidenceLedger, locale).length > 0;
if (hasUngroundedNumber) {
await finishTerminalAttempt().catch(() => {});
await finishTerminalAttempt().catch(
caughtAs("reaction_line.cleanup_failed"),
);
return { status: "skipped", reason: "ungrounded_output" };
}
@@ -740,7 +749,9 @@ export async function runReactionLine(
chain[0].providerType,
]);
if (late) {
await finishTerminalAttempt().catch(() => {});
await finishTerminalAttempt().catch(
caughtAs("reaction_line.cleanup_failed"),
);
return { status: "skipped", reason: late.reason };
}
@@ -14,6 +14,7 @@
*/
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const WORKOUT_INSIGHT_GENERATE_QUEUE = "workout-insight-generate";
@@ -57,11 +58,12 @@ export async function enqueueWorkoutInsight(
retryBackoff: true,
});
return { enqueued: true };
} catch {
} catch (err) {
annotate({
action: { name: "workouts.insight.enqueue_failed" },
meta: { workoutId: payload.workoutId },
});
logCaught("workouts.insight.enqueue_failed", err);
return { enqueued: false };
}
}
+19 -8
View File
@@ -79,6 +79,7 @@ import {
WORKOUT_INSIGHT_GENERATE_QUEUE,
type WorkoutInsightGeneratePayload,
} from "./workout-insight-generate-shared";
import { caughtAs, logCaught } from "@/lib/logging/signal";
export { WORKOUT_INSIGHT_GENERATE_CONCURRENCY, WORKOUT_INSIGHT_GENERATE_QUEUE };
export type { WorkoutInsightGeneratePayload };
@@ -418,12 +419,12 @@ export async function runWorkoutInsightGenerate(
})().catch(async (err) => {
// Nothing could have reached a provider. Delete the reservation so a
// backed-off retry can safely reclaim both ownership and daily capacity.
await releaseClaim().catch(() => {});
await releaseClaim().catch(caughtAs("workouts.insight.cleanup_failed"));
throw err;
});
if (prepared.status === "unchanged") {
await releaseClaim().catch(() => {});
await releaseClaim().catch(caughtAs("workouts.insight.cleanup_failed"));
annotate({
action: { name: "workouts.insight.skipped" },
meta: { workoutId, reason: "unchanged" },
@@ -472,13 +473,18 @@ export async function runWorkoutInsightGenerate(
temperature: AI_BUDGETS.workoutInsight.temperature,
maxTokens: AI_BUDGETS.workoutInsight.maxTokens,
});
} catch {
await finishTerminalAttempt().catch(() => {});
} catch (err) {
logCaught("workouts.insight.provider_failed", err, { workoutId });
await finishTerminalAttempt().catch(
caughtAs("workouts.insight.cleanup_failed"),
);
return { status: "skipped", reason: "provider_uncertain" };
}
if (outcome.kind !== "ok") {
await finishTerminalAttempt().catch(() => {});
await finishTerminalAttempt().catch(
caughtAs("workouts.insight.cleanup_failed"),
);
annotate({
action: { name: "workouts.insight.skipped" },
meta: { workoutId, reason: outcome.kind },
@@ -488,7 +494,9 @@ export async function runWorkoutInsightGenerate(
const screened = finalizeStatusSummary(outcome.content, prepared.locale);
if (!screened.ok || !screened.text) {
await finishTerminalAttempt().catch(() => {});
await finishTerminalAttempt().catch(
caughtAs("workouts.insight.cleanup_failed"),
);
annotate({
action: { name: "workouts.insight.outbound_blocked" },
meta: { workoutId, reason: screened.ok ? "empty" : screened.reason },
@@ -531,8 +539,11 @@ export async function runWorkoutInsightGenerate(
throw new Error("Workout insight generation claim was lost");
}
});
} catch {
await finishTerminalAttempt().catch(() => {});
} catch (err) {
logCaught("workouts.insight.persist_failed", err, { workoutId });
await finishTerminalAttempt().catch(
caughtAs("workouts.insight.cleanup_failed"),
);
return { status: "skipped", reason: "persistence_uncertain" };
}