From 67aa3338dae0adf1de9c42c4d7cc70212ed045a9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 27 Jul 2026 22:56:59 +0200 Subject: [PATCH] feat(jobs): return an outcome from every named queue handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sixty-two handler functions across twenty-four modules now resolve to a JobOutcome instead of undefined. The counts they already wrote to their wide event ride out in `did`, so a run that deleted nothing and a run that could not run are no longer the same shape. Three rules decided each conversion: - a catch that swallowed becomes jobFailed, keeping its warning line; - a catch that already rethrew keeps its throw, because the throw already fails the job and rewrapping it buries the stack; - a fan-out pass is judged on the pass. One account's revoked grant belongs on that account's integration ledger, not in a queue-wide failure that retries the whole cohort, so per-leg failures ride out as a users_failed count — visible for the first time. The retention cleanups are the behaviour change worth naming: every one of them used to warn and return, so a purge that had not run for weeks looked like a purge with nothing to do. They fail now. Their registrar sets retryLimit accordingly in the commit that converts the bindings. JobFailure carries the thrown error's stack; without it every one of the 101 queues would report the same runJob frame. --- src/lib/jobs/apple-health-import-worker.ts | 34 ++++- src/lib/jobs/data-arrival.ts | 13 +- src/lib/jobs/document-purge.ts | 11 +- src/lib/jobs/encryption-key-rotate.ts | 14 +- src/lib/jobs/environment-fetch.ts | 18 ++- src/lib/jobs/job-outcome.ts | 7 + src/lib/jobs/medication-intake-import.ts | 15 +- src/lib/jobs/reaction-line.ts | 13 +- src/lib/jobs/reminder/backup-handlers.ts | 46 +++++- src/lib/jobs/reminder/cleanup-handlers.ts | 138 +++++++++++++----- src/lib/jobs/reminder/fitbit-sync.ts | 37 ++++- src/lib/jobs/reminder/google-health-sync.ts | 35 ++++- src/lib/jobs/reminder/insights-handlers.ts | 88 ++++++++--- .../jobs/reminder/medication-maintenance.ts | 19 ++- .../reminder/medication-reminder-check.ts | 26 +++- src/lib/jobs/reminder/mood-cycle-checks.ts | 57 +++++++- src/lib/jobs/reminder/ops-handlers.ts | 97 ++++++++---- src/lib/jobs/reminder/poll-cohort.ts | 27 +++- src/lib/jobs/reminder/register-reminders.ts | 29 +++- src/lib/jobs/reminder/whoop-sync.ts | 42 +++++- src/lib/jobs/reminder/withings-sync.ts | 90 ++++++++++-- src/lib/jobs/restore-drill.ts | 27 +++- src/lib/jobs/run-job.ts | 6 + src/lib/jobs/workout-insight-generate.ts | 13 +- 24 files changed, 729 insertions(+), 173 deletions(-) diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 9bf197017..b71dfa0ab 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -31,6 +31,7 @@ import type { Job } from "pg-boss"; import { extractExportXml } from "@/lib/import/unzip-export-xml"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { streamParseExportXml, type ImportJobProgress, @@ -66,23 +67,26 @@ export const IMPORT_JOB_RECONCILE_CRON = "*/15 * * * *"; * the same `reconcileOrphanImportJobs()` the boot path calls — the * heartbeat + live pg-boss-job check already make it safe to re-run * on a healthy worker (a live import's fresh heartbeat and `active` - * pg-boss state both keep its row untouched). Errors are logged, not - * rethrown — a reconcile miss must never spam pg-boss's retry/backoff - * machinery for a background sweep this cheap to just re-run in 15 - * minutes. + * pg-boss state both keep its row untouched). A sweep that throws now + * reports a failed outcome instead of only leaving a warning behind: + * this pass exists to end a silent stall, and a stall in the pass + * itself was the one condition nothing surfaced. The next tick is 15 + * minutes away regardless, so nothing rests on the retry. */ export async function handleImportJobReconcileTick( jobs: Job[], -): Promise { +): Promise { void jobs; - await withBackgroundEvent( + return withBackgroundEvent( "job.apple_health_import_reconcile", async (evt) => { try { await reconcileOrphanImportJobs(); } catch (err) { evt.addWarning(`apple-health-import-reconcile failed: ${err}`); + return jobFailed("apple health import reconcile failed", err); } + return jobDone(); }, ); } @@ -230,7 +234,7 @@ async function writeProgress( */ export async function handleAppleHealthImport( job: Job, -): Promise { +): Promise { const { userId, uploadPath, uploadBytes, triggeredByAdminId } = job.data; const prisma = getWorkerPrisma(); @@ -280,7 +284,7 @@ export async function handleAppleHealthImport( ` — row is already terminal (${importJob.status}); the staged upload` + " was consumed by the first run and a re-run could only mask its outcome", ); - return; + return jobDone({ skipped: "already_terminal" }); } // Extracted-XML path, hoisted so the failure path can clean it up — @@ -364,6 +368,12 @@ export async function handleAppleHealthImport( // the end, scoped to the user's full measurement span, so // post-import reads of the analytics + comprehensive surfaces // hit the warm rollup table on first paint. + // + // The miss rides out as a fact rather than as a failed job: the + // import is committed and its mirror row already reads `done`, so + // failing here would report a completed import as failed and + // contradict the row the status poll surfaces to the operator. + let rollupFailed = false; try { const span = await prisma.measurement.aggregate({ where: { userId }, @@ -382,6 +392,7 @@ export async function handleAppleHealthImport( } catch (rollupErr) { // Rollup failure is non-fatal — the next read falls through to // live aggregation. Log but don't poison the import. + rollupFailed = true; console.warn( `[apple-health-import] Rollup recompute failed for user ${userId}`, rollupErr, @@ -392,6 +403,13 @@ export async function handleAppleHealthImport( // periodically swept on the host. safeUnlink(unzip.xmlPath); safeUnlink(uploadPath); + + return jobDone({ + records_read: result.totals.recordsRead, + rows_upserted: result.totals.rowsUpserted, + duration_ms: result.totals.durationMs, + rollup_failed: rollupFailed, + }); } catch (err) { // v1.28.33 (issue #486) — a missing staging file is an operational // condition, not a parse failure: `/tmp` is wiped on a container diff --git a/src/lib/jobs/data-arrival.ts b/src/lib/jobs/data-arrival.ts index d51132bd9..cb4318ac6 100644 --- a/src/lib/jobs/data-arrival.ts +++ b/src/lib/jobs/data-arrival.ts @@ -30,6 +30,7 @@ import type { Job } from "pg-boss"; import { invalidateUserDashboardSnapshot } from "@/lib/cache/invalidate"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { annotate } from "@/lib/logging/context"; import { withBackgroundEvent } from "@/lib/logging/background"; @@ -248,14 +249,17 @@ export async function runDataArrival( export async function handleDataArrival( jobs: Job[], -): Promise { - await withBackgroundEvent("job.data_arrival", async (evt) => { +): Promise { + return withBackgroundEvent("job.data_arrival", async (evt) => { + let processed = 0; + let skipped = 0; for (const job of jobs) { const arrival = job.data; try { const outcome = await runDataArrival(getWorkerPrisma(), arrival); if (outcome.status === "skipped") { + skipped++; annotate({ action: { name: `arrival.${arrival.kind}.skipped` }, meta: { reason: outcome.reason, source: arrival.source }, @@ -264,6 +268,7 @@ export async function handleDataArrival( continue; } + processed++; annotate({ action: { name: `arrival.${arrival.kind}.processed` }, meta: { @@ -285,5 +290,9 @@ export async function handleDataArrival( throw err; } } + // Both statuses are successes — a refused arrival is a decision the spine + // made, not work it failed to do. The counts are what make the split + // between them readable at all. + return jobDone({ total: jobs.length, processed, skipped }); }); } diff --git a/src/lib/jobs/document-purge.ts b/src/lib/jobs/document-purge.ts index 8f136be07..eebb5eb7e 100644 --- a/src/lib/jobs/document-purge.ts +++ b/src/lib/jobs/document-purge.ts @@ -11,6 +11,7 @@ */ import { type Job } from "pg-boss"; import { withBackgroundEvent } from "@/lib/logging/background"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { getWorkerPrisma } from "@/lib/jobs/reminder/shared"; import type { PrismaClient } from "@/generated/prisma/client"; @@ -43,16 +44,22 @@ export async function purgeExpiredDocumentTombstones( return count; } -export async function handleDocumentPurge(jobs: Job[]) { +export async function handleDocumentPurge( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.document_tombstone_purge", async (evt) => { + return withBackgroundEvent("job.document_tombstone_purge", async (evt) => { const prisma = getWorkerPrisma(); try { const purged = await purgeExpiredDocumentTombstones(prisma); evt.setAction({ name: "documents.vault.purge" }); evt.addMeta("document_purge_deleted", purged); + // A night with no expired tombstone purges nothing and is still a run + // that did its work — the zero is the fact, not an absence of one. + return jobDone({ document_purge_deleted: purged }); } catch (err) { evt.addWarning(`document-tombstone-purge failed: ${err}`); + return jobFailed("document tombstone purge failed", err); } }); } diff --git a/src/lib/jobs/encryption-key-rotate.ts b/src/lib/jobs/encryption-key-rotate.ts index be6652341..a33588426 100644 --- a/src/lib/jobs/encryption-key-rotate.ts +++ b/src/lib/jobs/encryption-key-rotate.ts @@ -24,6 +24,7 @@ */ import { type Job } from "pg-boss"; import { withBackgroundEvent } from "@/lib/logging/background"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { auditLog } from "@/lib/auth/audit"; import { rotateCorpus, @@ -60,8 +61,8 @@ export async function runEncryptionKeyRotation(): Promise<{ export async function handleEncryptionKeyRotate( jobs: Job[], -) { - await withBackgroundEvent("job.encryption_key_rotate", async (evt) => { +): Promise { + return withBackgroundEvent("job.encryption_key_rotate", async (evt) => { const requestedBy = jobs[0]?.data?.requestedByUserId ?? null; try { const result = await runEncryptionKeyRotation(); @@ -78,6 +79,15 @@ export async function handleEncryptionKeyRotate( errors: result.totalErrors, }, }); + // Per-row errors are the fail-closed skip of a row under a key the + // deployment no longer configures. The pass still ran, so they ride out + // as a count rather than failing (and retrying) the whole corpus. + return jobDone({ + rotate_active_key_id: result.activeKeyId, + rotate_scanned: result.totalScanned, + rotate_rotated: result.totalRotated, + rotate_errors: result.totalErrors, + }); } catch (err) { evt.addWarning(`encryption-key-rotate failed: ${err}`); await auditLog("admin.encryption.rotate.failed", { diff --git a/src/lib/jobs/environment-fetch.ts b/src/lib/jobs/environment-fetch.ts index b21d8379d..84c8dd7b2 100644 --- a/src/lib/jobs/environment-fetch.ts +++ b/src/lib/jobs/environment-fetch.ts @@ -33,6 +33,7 @@ import { utcDayKey, } from "@/lib/environment/service"; import { recordError } from "@/lib/jobs/worker-status"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { workerLog } from "./reminder/shared"; export const ENVIRONMENT_FETCH_QUEUE = "environment-fetch"; @@ -152,14 +153,14 @@ export async function enqueueEnvironmentFetchDiscovery( export async function handleEnvironmentFetch( boss: PgBoss, payload: EnvironmentFetchPayload, -): Promise { +): Promise { if (!payload.userId) { const { enqueued, skipped } = await enqueueEnvironmentFetchDiscovery(boss); workerLog( "info", `[environment-fetch] discovery enqueued=${enqueued} skipped=${skipped}`, ); - return; + return jobDone({ users_enqueued: enqueued, users_skipped: skipped }); } const userId = payload.userId; @@ -168,7 +169,7 @@ export async function handleEnvironmentFetch( "info", `[environment-fetch] user=${userId} module disabled — skipping`, ); - return; + return jobDone({ skipped: "environment module disabled" }); } const { startDate, endDate } = resolveRange(payload); @@ -179,7 +180,11 @@ export async function handleEnvironmentFetch( "error", `[environment-fetch] user=${userId} range ${startDate}..${endDate} (${span}d) exceeds cap — skipping`, ); - return; + // A refused range is not an opt-out: the job was asked to fill days and + // filled none, so it says so instead of completing quietly on a stderr line. + return jobFailed("environment backfill range exceeds cap", undefined, { + span_days: span, + }); } try { @@ -192,6 +197,11 @@ export async function handleEnvironmentFetch( "info", `[environment-fetch] user=${userId} ${startDate}..${endDate} stored=${result.stored} skipped=${result.skipped} fetches=${result.fetches}`, ); + return jobDone({ + days_stored: result.stored, + days_skipped: result.skipped, + fetches: result.fetches, + }); } catch (err) { recordError(); workerLog("error", `[environment-fetch] user=${userId} failed`, err); diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index d90b90e05..34b4c8b8e 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -17,6 +17,13 @@ * signal a failure) into commission (claiming a success that did not * happen), which is a smaller and much more reviewable class of defect. * + * A fan-out pass is judged on the pass, not on its legs. A nightly sweep + * over every connected account is `ok: true` when the sweep ran, even if one + * account's grant was revoked: that belongs on the account's own integration + * ledger, and failing the queue would retry the whole cohort over one user's + * expired token. The per-leg failures ride out as a count in `did`, which is + * what makes them visible at all — today they are visible nowhere. + * * `did` is the small pinned-shape bag of facts about the run — counts, * identifiers, booleans. It rides into the failure report and, once the * failure ledger lands, into the operator surface. Keep it to scalars that diff --git a/src/lib/jobs/medication-intake-import.ts b/src/lib/jobs/medication-intake-import.ts index 2d0a365e9..2bf370f41 100644 --- a/src/lib/jobs/medication-intake-import.ts +++ b/src/lib/jobs/medication-intake-import.ts @@ -3,6 +3,7 @@ import { Prisma, type PrismaClient } from "@/generated/prisma/client"; import { invalidateUserMedications } from "@/lib/cache/invalidate"; import { prisma, toJson } from "@/lib/db"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { consumeImportedIntakesBatch } from "@/lib/medications/inventory/consumption"; import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync"; import { @@ -711,9 +712,19 @@ async function recordWorkerFailure( export async function handleMedicationIntakeImport( job: Job, -): Promise { +): Promise { try { - await processMedicationIntakeImportJob(job.data.jobId); + const result = await processMedicationIntakeImportJob(job.data.jobId); + // A null result is a delivery that found the row already terminal or gone. + // There was nothing left to import, which is an absence of work rather + // than a failure to do it. + return result === null + ? jobDone({ finalized: false }) + : jobDone({ + finalized: true, + imported: result.imported, + skipped: result.skipped, + }); } catch (error) { await recordWorkerFailure(job, error); throw error; diff --git a/src/lib/jobs/reaction-line.ts b/src/lib/jobs/reaction-line.ts index 413c8fcc0..db75ed241 100644 --- a/src/lib/jobs/reaction-line.ts +++ b/src/lib/jobs/reaction-line.ts @@ -29,6 +29,7 @@ import type { Job } from "pg-boss"; import { prisma } from "@/lib/db"; import { randomUUID } from "node:crypto"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { annotate } from "@/lib/logging/context"; import { withBackgroundEvent } from "@/lib/logging/background"; import { resolveProviderChain } from "@/lib/ai/provider"; @@ -677,13 +678,16 @@ export async function runReactionLine( export async function handleReactionLine( jobs: Job[], -): Promise { - await withBackgroundEvent("job.reaction_line", async (evt) => { +): Promise { + return withBackgroundEvent("job.reaction_line", async (evt) => { + let generated = 0; + let skipped = 0; for (const job of jobs) { try { const outcome = await runReactionLine(job.data); if (outcome.status === "skipped") { + skipped++; annotate({ action: { name: "arrival.reaction_line.skipped" }, meta: { reason: outcome.reason, kind: job.data.kind }, @@ -692,6 +696,7 @@ export async function handleReactionLine( continue; } + generated++; annotate({ action: { name: "arrival.reaction_line.generated" }, meta: { kind: job.data.kind, local_date: job.data.localDate }, @@ -705,5 +710,9 @@ export async function handleReactionLine( throw err; } } + // A skip is a refusal the surface is designed around, not a miss: the + // deterministic lead already stands without a line. The pass did its work + // whenever it ran to the end. + return jobDone({ total: jobs.length, generated, skipped }); }); } diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 1106159ed..6d23f8a44 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -8,8 +8,12 @@ import { type Job } from "pg-boss"; import { recordError } from "@/lib/jobs/worker-status"; import { buildFullBackupPayload } from "@/lib/export/full-backup-payload"; import { encrypt } from "@/lib/crypto"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; -import { runOffhostBackup } from "@/lib/jobs/offhost-backup"; +import { + OffhostBackupNotConfiguredError, + runOffhostBackup, +} from "@/lib/jobs/offhost-backup"; import { getWorkerPrisma } from "./shared"; export interface DataBackupPayload { @@ -20,9 +24,11 @@ export interface OffhostBackupPayload { triggeredAt: string; } -export async function handleOffhostBackup(jobs: Job[]) { +export async function handleOffhostBackup( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.offhost_backup", async (evt) => { + return withBackgroundEvent("job.offhost_backup", async (evt) => { const p = getWorkerPrisma(); try { const report = await runOffhostBackup(p); @@ -39,16 +45,34 @@ export async function handleOffhostBackup(jobs: Job[]) { JSON.stringify(report.failures.slice(0, 10)), ); } + // Per-user upload failures ride out as `offhost_backup_failed`: the run + // itself uploaded what it could, and failing the queue over one user's + // object would re-upload the whole cohort on every retry. + return jobDone({ + offhost_backup_uploaded: report.uploaded, + offhost_backup_failed: report.failed, + offhost_backup_total_users: report.totalUsers, + offhost_backup_endpoint: report.config.endpoint, + offhost_backup_bucket: report.config.bucket, + }); } catch (err) { - // Not configured ⇒ skip silently with a warning, not an error. + // Not configured ⇒ skip silently with a warning, not an error: most + // self-hosts never set the S3 credentials, and a nightly failed job on + // every one of them buries the runs that genuinely could not upload. evt.addWarning(`offhost-backup skipped/failed: ${err}`); + if (err instanceof OffhostBackupNotConfiguredError) { + return jobDone({ offhost_backup_configured: false }); + } + return jobFailed("offhost backup failed", err); } }); } -export async function handleDataBackup(jobs: Job[]) { +export async function handleDataBackup( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.data_backup", async (evt) => { + return withBackgroundEvent("job.data_backup", async (evt) => { const prisma = getWorkerPrisma(); try { const users = await prisma.user.findMany({ @@ -56,6 +80,7 @@ export async function handleDataBackup(jobs: Job[]) { }); let backed = 0; + let usersFailed = 0; for (const user of users) { try { const { payload } = await buildFullBackupPayload(prisma, user.id, { @@ -82,6 +107,10 @@ export async function handleDataBackup(jobs: Job[]) { }); backed++; } catch (err) { + // One user's payload failing is that user's problem, not the pass's: + // it rides out as a count so the weekly run is not retried for the + // whole cohort. + usersFailed++; evt.addWarning(`Failed for user ${user.id}: ${err}`); } } @@ -90,6 +119,11 @@ export async function handleDataBackup(jobs: Job[]) { task_name: "job.data_backup", result: { backed, total: users.length }, }); + return jobDone({ + backed, + total: users.length, + users_failed: usersFailed, + }); } catch (err) { evt.setError(err); recordError(); diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts index a4191a3c0..145ac3e30 100644 --- a/src/lib/jobs/reminder/cleanup-handlers.ts +++ b/src/lib/jobs/reminder/cleanup-handlers.ts @@ -2,10 +2,27 @@ * Retention and maintenance cleanups: rate-limit rows, idempotency keys, audit logs, OAuth states (Withings / WHOOP), mood-reminder events, push attempts, and measurement tombstones. * * Extracted from reminder-worker.ts, which owns the queue names, cron - * schedules, and boss.work registrations. + * schedules, and the queue bindings. + * + * Every pass in here used to catch its own failure, write a warning nobody + * reads, and return — so a retention purge that had not run for weeks was + * indistinguishable from one with nothing to delete. They now return + * `jobFailed`, which fails the pg-boss job and puts the queue on the operator's + * failing-jobs surface. + * + * That change has a price, and the registrars pay it: these are bulk DELETEs + * over a retention horizon, and their failure mode is deterministic. A + * statement that timed out against a large trailing edge times out again + * immediately. Under pg-boss's default policy (retryLimit 2, no delay, no + * backoff) each converted queue would run the same doomed statement three + * times in a row every night. So every cron-driven cleanup registered from + * this module is scheduled with `retryLimit: 0`: the failure is recorded once + * and the next night's tick is the retry, which is the right backoff for + * retention work that has a whole day of slack. */ import { type Job } from "pg-boss"; import { withBackgroundEvent } from "@/lib/logging/background"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { cleanupExpiredWhoopConnectTickets, cleanupExpiredWhoopOAuthStates, @@ -78,9 +95,9 @@ export interface MoodReminderCleanupPayload { export async function handleMoodReminderCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.mood_reminder_cleanup", async (evt) => { + return withBackgroundEvent("job.mood_reminder_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const cutoff = new Date(); @@ -90,8 +107,10 @@ export async function handleMoodReminderCleanup( where: { date: { lt: cutoffIso } }, }); evt.addMeta("mood_reminder_cleanup_deleted", deleted.count); + return jobDone({ deleted: deleted.count }); } catch (err) { evt.addWarning(`mood-reminder-cleanup failed: ${err}`); + return jobFailed("mood-reminder cleanup failed", err); } }); } @@ -118,9 +137,9 @@ export interface PushAttemptCleanupPayload { export async function handlePushAttemptCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.push_attempt_cleanup", async (evt) => { + return withBackgroundEvent("job.push_attempt_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const cutoff = new Date(); @@ -129,8 +148,10 @@ export async function handlePushAttemptCleanup( where: { createdAt: { lt: cutoff } }, }); evt.addMeta("push_attempt_cleanup_deleted", deleted.count); + return jobDone({ deleted: deleted.count }); } catch (err) { evt.addWarning(`push-attempt-cleanup failed: ${err}`); + return jobFailed("push-attempt cleanup failed", err); } }); } @@ -152,9 +173,9 @@ export interface ArrivalReactionCleanupPayload { export async function handleArrivalReactionCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.arrival_reaction_cleanup", async (evt) => { + return withBackgroundEvent("job.arrival_reaction_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const cutoff = new Date(); @@ -163,8 +184,10 @@ export async function handleArrivalReactionCleanup( where: { createdAt: { lt: cutoff } }, }); evt.addMeta("arrival_reaction_cleanup_deleted", deleted.count); + return jobDone({ deleted: deleted.count }); } catch (err) { evt.addWarning(`arrival-reaction-cleanup failed: ${err}`); + return jobFailed("arrival-reaction cleanup failed", err); } }); } @@ -175,9 +198,9 @@ export interface MeasurementTombstoneCleanupPayload { export async function handleMeasurementTombstoneCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent( + return withBackgroundEvent( "job.measurement_tombstone_cleanup", async (evt) => { const p = getWorkerPrisma(); @@ -208,6 +231,12 @@ export async function handleMeasurementTombstoneCleanup( "tombstone-cleanup stopped at the batch cap; a backlog remains for the next run", ); } + return jobDone({ + measurements_pruned: measurements.deleted, + mood_pruned: mood.deleted, + intakes_pruned: intakes.deleted, + drained, + }); } catch (err) { // Rethrow. A retention purge that fails silently every night looks // exactly like one that has nothing to do; pg-boss recording a failed @@ -221,32 +250,36 @@ export async function handleMeasurementTombstoneCleanup( export async function handleRateLimitCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.rate_limit_cleanup", async (evt) => { + return withBackgroundEvent("job.rate_limit_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const result = await p.$executeRaw` DELETE FROM rate_limits WHERE reset_at < NOW() `; evt.addMeta("rate_limit_cleanup_deleted", result); + return jobDone({ deleted: result }); } catch (err) { evt.addWarning(`rate-limit-cleanup failed: ${err}`); + return jobFailed("rate-limit cleanup failed", err); } }); } export async function handleIdempotencyCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.idempotency_cleanup", async (evt) => { + return withBackgroundEvent("job.idempotency_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const deleted = await cleanupExpiredIdempotencyKeys(p); evt.addMeta("idempotency_cleanup_deleted", deleted); + return jobDone({ deleted }); } catch (err) { evt.addWarning(`idempotency-cleanup failed: ${err}`); + return jobFailed("idempotency-key cleanup failed", err); } }); } @@ -266,26 +299,28 @@ export async function handleIdempotencyCleanup( */ export async function handleStepUpElevationCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.step_up_elevation_cleanup", async (evt) => { + return withBackgroundEvent("job.step_up_elevation_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const { count } = await p.stepUpElevation.deleteMany({ where: { expiresAt: { lt: new Date() } }, }); evt.addMeta("step_up_elevation_cleanup_deleted", count); + return jobDone({ deleted: count }); } catch (err) { evt.addWarning(`step-up-elevation-cleanup failed: ${err}`); + return jobFailed("step-up elevation cleanup failed", err); } }); } export async function handleAuditLogCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.audit_log_cleanup", async (evt) => { + return withBackgroundEvent("job.audit_log_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const outcome = await cleanupOldAuditLogs(p); @@ -296,6 +331,7 @@ export async function handleAuditLogCleanup( "audit-log-cleanup stopped at the batch cap; a backlog remains for the next run", ); } + return jobDone({ deleted: outcome.deleted, drained: outcome.drained }); } catch (err) { // Rethrow — see the tombstone handler. Storage-limitation retention that // fails quietly is retention that is not happening. @@ -307,15 +343,17 @@ export async function handleAuditLogCleanup( export async function handleCoachMessageCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.coach_message_cleanup", async (evt) => { + return withBackgroundEvent("job.coach_message_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const deleted = await cleanupOldCoachMessages(p); evt.addMeta("coach_message_cleanup_deleted", deleted); + return jobDone({ deleted }); } catch (err) { evt.addWarning(`coach-message-cleanup failed: ${err}`); + return jobFailed("coach-message cleanup failed", err); } }); } @@ -326,9 +364,9 @@ export interface McpTokenCleanupPayload { export async function handleMcpTokenCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.mcp_token_cleanup", async (evt) => { + return withBackgroundEvent("job.mcp_token_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const result = await cleanupExpiredMcpTokens(p); @@ -340,27 +378,40 @@ export async function handleMcpTokenCleanup( "mcp_token_cleanup_connections_deleted", result.connectionsDeleted, ); + return jobDone({ + access_tokens_deleted: result.accessTokensDeleted, + connections_deleted: result.connectionsDeleted, + }); } catch (err) { evt.addWarning(`mcp-token-cleanup failed: ${err}`); + return jobFailed("mcp-token cleanup failed", err); } }); } export async function handleWithingsOAuthStateCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.withings_oauth_state_cleanup", async (evt) => { - const p = getWorkerPrisma(); - try { - const deleted = await cleanupExpiredWithingsOAuthStates(p); - evt.addMeta("withings_oauth_state_cleanup_deleted", deleted); - } catch (err) { - // The OAuth flow tolerates a stale row sticking around for an - // extra day — log + carry on so the boss queue doesn't retry-loop. - evt.addWarning(`withings-oauth-state-cleanup failed: ${err}`); - } - }); + return withBackgroundEvent( + "job.withings_oauth_state_cleanup", + async (evt) => { + const p = getWorkerPrisma(); + try { + const deleted = await cleanupExpiredWithingsOAuthStates(p); + evt.addMeta("withings_oauth_state_cleanup_deleted", deleted); + return jobDone({ deleted }); + } catch (err) { + // The OAuth flow tolerates a stale row for an extra day, which used + // to be the argument for swallowing this so the queue would not + // retry-loop. Tolerable is not the same as invisible: the failure is + // now reported, and the retry-loop worry is answered where it belongs, + // by the `retryLimit: 0` this queue is scheduled with. + evt.addWarning(`withings-oauth-state-cleanup failed: ${err}`); + return jobFailed("withings oauth-state cleanup failed", err); + } + }, + ); } export interface OidcNativeHandoffCleanupPayload { @@ -369,17 +420,21 @@ export interface OidcNativeHandoffCleanupPayload { export async function handleOidcNativeHandoffCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.oidc_native_handoff_cleanup", async (evt) => { + return withBackgroundEvent("job.oidc_native_handoff_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const deleted = await cleanupExpiredOidcNativeHandoffs(p); evt.addMeta("oidc_native_handoff_cleanup_deleted", deleted); + return jobDone({ deleted }); } catch (err) { - // The handoff flow tolerates a stale row for an extra day (expiry is - // enforced at read) — log + carry on so the boss queue doesn't retry-loop. + // Expiry is enforced at read, so a stale row for an extra day is + // harmless — but see the withings twin above: harmless is not a + // reason to hide the failure. `retryLimit: 0` on the schedule is what + // keeps a doomed statement from running three times a night. evt.addWarning(`oidc-native-handoff-cleanup failed: ${err}`); + return jobFailed("oidc native-handoff cleanup failed", err); } }); } @@ -390,17 +445,22 @@ export interface WhoopOAuthStateCleanupPayload { export async function handleWhoopOAuthStateCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.whoop_oauth_state_cleanup", async (evt) => { + return withBackgroundEvent("job.whoop_oauth_state_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const deleted = await cleanupExpiredWhoopOAuthStates(p); evt.addMeta("whoop_oauth_state_cleanup_deleted", deleted); const ticketsDeleted = await cleanupExpiredWhoopConnectTickets(p); evt.addMeta("whoop_connect_ticket_cleanup_deleted", ticketsDeleted); + return jobDone({ + states_deleted: deleted, + connect_tickets_deleted: ticketsDeleted, + }); } catch (err) { evt.addWarning(`whoop-oauth-state-cleanup failed: ${err}`); + return jobFailed("whoop oauth-state cleanup failed", err); } }); } diff --git a/src/lib/jobs/reminder/fitbit-sync.ts b/src/lib/jobs/reminder/fitbit-sync.ts index d9a45b895..e55886ff5 100644 --- a/src/lib/jobs/reminder/fitbit-sync.ts +++ b/src/lib/jobs/reminder/fitbit-sync.ts @@ -7,6 +7,7 @@ import { type Job } from "pg-boss"; import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError } from "@/lib/jobs/worker-status"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { runFitbitPollCohort } from "@/lib/fitbit/sync"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; @@ -23,8 +24,10 @@ export interface FitbitSyncPayload { userId?: string; } -export async function handleFitbitSync(jobs: Job[]) { - await withBackgroundEvent("job.fitbit_sync", async (evt) => { +export async function handleFitbitSync( + jobs: Job[], +): Promise { + return withBackgroundEvent("job.fitbit_sync", async (evt) => { const prisma = getWorkerPrisma(); try { const targets: Array<{ userId: string }> = []; @@ -37,16 +40,26 @@ export async function handleFitbitSync(jobs: Job[]) { }); targets.push(...connections); } - if (targets.length === 0) return; + if (targets.length === 0) { + return jobDone({ + total: 0, + users_synced: 0, + users_failed: 0, + measurements_imported: 0, + }); + } // Fan the cohort out with bounded concurrency + per-user error isolation: // one slow Google response can't stall the whole cohort, and a single // user's failure is warned without aborting the pass. + let usersFailed = 0; const { usersSynced, measurementsImported } = await runFitbitPollCohort( targets.map((t) => t.userId), { - onUserError: (userId, err) => - evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`), + onUserError: (userId, err) => { + usersFailed++; + evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`); + }, // v1.18.1 — a fresh reading landed; resolve the user's Vorsorge // reminders eventfully. Fire-and-forget; the cron is the net. onUserSynced: (userId, imported) => { @@ -67,6 +80,14 @@ export async function handleFitbitSync(jobs: Job[]) { measurements_imported: measurementsImported, }, }); + // The pass is the unit of success: a revoked grant belongs on that user's + // integration ledger, not in a queue failure that retries every account. + return jobDone({ + total: targets.length, + users_synced: usersSynced, + users_failed: usersFailed, + measurements_imported: measurementsImported, + }); } catch (err) { evt.setError(err); recordError(); @@ -81,15 +102,17 @@ export interface FitbitOAuthStateCleanupPayload { export async function handleFitbitOAuthStateCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.fitbit_oauth_state_cleanup", async (evt) => { + return withBackgroundEvent("job.fitbit_oauth_state_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const deleted = await cleanupExpiredFitbitOAuthStates(p); evt.addMeta("fitbit_oauth_state_cleanup_deleted", deleted); + return jobDone({ deleted }); } catch (err) { evt.addWarning(`fitbit-oauth-state-cleanup failed: ${err}`); + return jobFailed("fitbit oauth state cleanup failed", err); } }); } diff --git a/src/lib/jobs/reminder/google-health-sync.ts b/src/lib/jobs/reminder/google-health-sync.ts index 0006f5ca8..d6791814b 100644 --- a/src/lib/jobs/reminder/google-health-sync.ts +++ b/src/lib/jobs/reminder/google-health-sync.ts @@ -7,6 +7,7 @@ import { type Job } from "pg-boss"; import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError } from "@/lib/jobs/worker-status"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { runGoogleHealthPollCohort } from "@/lib/google-health/sync"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; @@ -26,8 +27,8 @@ export interface GoogleHealthSyncPayload { export async function handleGoogleHealthSync( jobs: Job[], -) { - await withBackgroundEvent("job.google_health_sync", async (evt) => { +): Promise { + return withBackgroundEvent("job.google_health_sync", async (evt) => { const prisma = getWorkerPrisma(); try { const targets: Array<{ userId: string }> = []; @@ -40,19 +41,29 @@ export async function handleGoogleHealthSync( }); targets.push(...connections); } - if (targets.length === 0) return; + if (targets.length === 0) { + return jobDone({ + total: 0, + users_synced: 0, + users_failed: 0, + measurements_imported: 0, + }); + } // Fan the cohort out with bounded concurrency + per-user error isolation: // one slow Google response can't stall the whole cohort, and a single // user's failure is warned without aborting the pass. + let usersFailed = 0; const { usersSynced, measurementsImported } = await runGoogleHealthPollCohort( targets.map((t) => t.userId), { - onUserError: (userId, err) => + onUserError: (userId, err) => { + usersFailed++; evt.addWarning( `job.google_health_sync failed for user ${userId}: ${err}`, - ), + ); + }, // A fresh reading landed; resolve the user's Vorsorge reminders // eventfully. Fire-and-forget; the cron is the net. onUserSynced: (userId, imported) => { @@ -73,6 +84,14 @@ export async function handleGoogleHealthSync( measurements_imported: measurementsImported, }, }); + // The pass is the unit of success: a revoked grant belongs on that user's + // integration ledger, not in a queue failure that retries every account. + return jobDone({ + total: targets.length, + users_synced: usersSynced, + users_failed: usersFailed, + measurements_imported: measurementsImported, + }); } catch (err) { evt.setError(err); recordError(); @@ -87,17 +106,19 @@ export interface GoogleHealthOAuthStateCleanupPayload { export async function handleGoogleHealthOAuthStateCleanup( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent( + return withBackgroundEvent( "job.google_health_oauth_state_cleanup", async (evt) => { const p = getWorkerPrisma(); try { const deleted = await cleanupExpiredGoogleHealthOAuthStates(p); evt.addMeta("google_health_oauth_state_cleanup_deleted", deleted); + return jobDone({ deleted }); } catch (err) { evt.addWarning(`google-health-oauth-state-cleanup failed: ${err}`); + return jobFailed("google health oauth state cleanup failed", err); } }, ); diff --git a/src/lib/jobs/reminder/insights-handlers.ts b/src/lib/jobs/reminder/insights-handlers.ts index 2c7117714..fd275f43c 100644 --- a/src/lib/jobs/reminder/insights-handlers.ts +++ b/src/lib/jobs/reminder/insights-handlers.ts @@ -19,6 +19,7 @@ import { type InsightStatusGeneratePayload, runInsightStatusGenerate, } from "@/lib/jobs/insight-status-generate"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status"; import { generateWeightStatusForUser } from "@/lib/insights/weight-status"; @@ -69,6 +70,10 @@ export interface MedicationComplianceStatusPayload { * `status-cron-candidates.ts` for the full division of nightly labour). * The generators normalise `locale` themselves (de stays de, everything * else gets English prose). + * + * The pass is what succeeds or fails. One user whose generation threw is that + * user's cold card, reported as the `failed` count; failing the job would + * re-run the whole cohort over it. */ export async function runStatusCronGenerate( taskName: string, @@ -76,13 +81,15 @@ export async function runStatusCronGenerate( userId: string, options: { locale: string | null; force: boolean }, ) => Promise, -): Promise { - await withBackgroundEvent(taskName, async (evt) => { +): Promise { + return withBackgroundEvent(taskName, async (evt) => { const prisma = getWorkerPrisma(); try { const users = await findStatusCronCandidates(prisma); - if (users.length === 0) return; + if (users.length === 0) { + return jobDone({ total: 0, generated: 0, failed: 0 }); + } let generated = 0; let failed = 0; @@ -103,6 +110,7 @@ export async function runStatusCronGenerate( task_name: taskName, result: { generated, failed, total: users.length }, }); + return jobDone({ total: users.length, generated, failed }); } catch (err) { evt.setError(err); recordError(); @@ -129,13 +137,15 @@ export async function runStatusCronGenerate( * inside the batch itself). Net effect: the nightly ladder pays one call per * user instead of seven, with no queue/registry churn. */ -async function runStatusBatchCron(taskName: string): Promise { - await withBackgroundEvent(taskName, async (evt) => { +async function runStatusBatchCron(taskName: string): Promise { + return withBackgroundEvent(taskName, async (evt) => { const prisma = getWorkerPrisma(); try { recordInsightsRun(); const users = await findStatusCronCandidates(prisma); - if (users.length === 0) return; + if (users.length === 0) { + return jobDone({ total: 0, generated: 0, served: 0, failed: 0 }); + } let generated = 0; let served = 0; @@ -167,6 +177,10 @@ async function runStatusBatchCron(taskName: string): Promise { task_name: taskName, result: { generated, served, failed, total: users.length }, }); + // A user whose batch call threw already fell back to the single-card + // path inside the batch, and the six later per-metric crons cover what + // is still cold. The batch pass ran, so it reports the count. + return jobDone({ total: users.length, generated, served, failed }); } catch (err) { evt.setError(err); recordError(); @@ -176,14 +190,16 @@ async function runStatusBatchCron(taskName: string): Promise { }); } -export function handleGeneralStatusGenerate(jobs: Job[]) { +export function handleGeneralStatusGenerate( + jobs: Job[], +): Promise { void jobs; return runStatusBatchCron("job.insights.batch"); } export function handleBloodPressureStatusGenerate( jobs: Job[], -) { +): Promise { void jobs; return runStatusCronGenerate( "job.insights.blood_pressure", @@ -191,7 +207,9 @@ export function handleBloodPressureStatusGenerate( ); } -export function handleWeightStatusGenerate(jobs: Job[]) { +export function handleWeightStatusGenerate( + jobs: Job[], +): Promise { void jobs; return runStatusCronGenerate( "job.insights.weight", @@ -199,7 +217,9 @@ export function handleWeightStatusGenerate(jobs: Job[]) { ); } -export function handlePulseStatusGenerate(jobs: Job[]) { +export function handlePulseStatusGenerate( + jobs: Job[], +): Promise { void jobs; return runStatusCronGenerate( "job.insights.pulse", @@ -207,19 +227,23 @@ export function handlePulseStatusGenerate(jobs: Job[]) { ); } -export function handleBmiStatusGenerate(jobs: Job[]) { +export function handleBmiStatusGenerate( + jobs: Job[], +): Promise { void jobs; return runStatusCronGenerate("job.insights.bmi", generateBmiStatusForUser); } -export function handleMoodStatusGenerate(jobs: Job[]) { +export function handleMoodStatusGenerate( + jobs: Job[], +): Promise { void jobs; return runStatusCronGenerate("job.insights.mood", generateMoodStatusForUser); } export function handleMedicationComplianceStatusGenerate( jobs: Job[], -) { +): Promise { void jobs; return runStatusCronGenerate( "job.insights.medication_compliance", @@ -229,8 +253,8 @@ export function handleMedicationComplianceStatusGenerate( export async function handleInsightPregenerateJob( jobs: Job[], -) { - await withBackgroundEvent("job.insight_pregenerate", async (evt) => { +): Promise { + return withBackgroundEvent("job.insight_pregenerate", async (evt) => { // v1.8.7.1 — a forced single-user warm carries `{ userId, force }`; // the scheduled tick carries neither. Route each job individually so a // batch that mixes a cron tick with on-demand warms (it never does in @@ -265,13 +289,30 @@ export async function handleInsightPregenerateJob( } } - if (scheduled.length === 0) return; + // A batch of forced warms only is a complete run: every warm either landed + // or threw out of the loop above. + if (scheduled.length === 0) { + return jobDone({ forced: forced.length, scheduled: 0 }); + } try { const summary = await runInsightPregenerate(getWorkerPrisma()); evt.setBackground({ task_name: "job.insight_pregenerate", result: { ...summary }, }); + return jobDone({ + forced: forced.length, + scheduled: scheduled.length, + total: summary.total, + generated: summary.generated, + cached: summary.cached, + unchanged: summary.unchanged, + skipped: summary.skipped, + failed: summary.failed, + budget_blocked: summary.budgetBlocked, + assessments_warmed: summary.assessmentsWarmed, + metric_assessments_warmed: summary.metricAssessmentsWarmed, + }); } catch (err) { evt.addWarning( `insight-pregenerate failed: ${ @@ -296,12 +337,18 @@ export async function handleInsightPregenerateJob( */ export async function handleInsightStatusGenerate( jobs: Job[], -) { - await withBackgroundEvent("job.insight_status_generate", async (evt) => { +): Promise { + return withBackgroundEvent("job.insight_status_generate", async (evt) => { + let generated = 0; + let malformed = 0; for (const job of jobs) { - if (!job.data?.userId || !job.data?.metric) continue; + if (!job.data?.userId || !job.data?.metric) { + malformed++; + continue; + } try { await runInsightStatusGenerate(job.data); + generated++; evt.addMeta( "status_generated", `${job.data.metric}:${job.data.locale}`, @@ -321,5 +368,8 @@ export async function handleInsightStatusGenerate( throw err; } } + // Every job in the batch either generated a card or carried no metric to + // generate; a generation that failed left through the rethrow above. + return jobDone({ generated, malformed }); }); } diff --git a/src/lib/jobs/reminder/medication-maintenance.ts b/src/lib/jobs/reminder/medication-maintenance.ts index 326cb8cd3..b906eb8f6 100644 --- a/src/lib/jobs/reminder/medication-maintenance.ts +++ b/src/lib/jobs/reminder/medication-maintenance.ts @@ -10,6 +10,7 @@ import { type IntakeAutoSkipPayload, runIntakeAutoSkipPass, } from "@/lib/jobs/intake-auto-skip"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { expireStaleInUseItems } from "@/lib/medications/inventory/service"; import { getWorkerPrisma } from "./shared"; @@ -21,18 +22,21 @@ import { getWorkerPrisma } from "./shared"; */ export async function handleMedicationInventoryExpire( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.medication_inventory_expire", async (evt) => { + return withBackgroundEvent("job.medication_inventory_expire", async (evt) => { try { const count = await expireStaleInUseItems({ nowMs: Date.now() }); evt.addMeta("inventory_expired_count", count); + // A pass that found no lapsed pen did its work; zero is a result. + return jobDone({ inventory_expired_count: count }); } catch (err) { evt.addWarning( `medication-inventory-expire failed: ${ err instanceof Error ? err.message : String(err) }`, ); + return jobFailed("inventory expire pass failed", err); } }); } @@ -46,9 +50,11 @@ export async function handleMedicationInventoryExpire( * unit test can drive it with an in-memory fake Prisma; this wrapper * threads the worker's pg-boss + background-event plumbing. */ -export async function handleIntakeAutoSkip(jobs: Job[]) { +export async function handleIntakeAutoSkip( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.intake_auto_skip", async (evt) => { + return withBackgroundEvent("job.intake_auto_skip", async (evt) => { const prisma = getWorkerPrisma(); try { const result = await runIntakeAutoSkipPass(prisma, { @@ -56,12 +62,17 @@ export async function handleIntakeAutoSkip(jobs: Job[]) { }); evt.addMeta("intake_auto_skip_count", result.skippedCount); evt.addMeta("intake_auto_skip_cutoff", result.cutoff.toISOString()); + return jobDone({ + intake_auto_skip_count: result.skippedCount, + intake_auto_skip_cutoff: result.cutoff.toISOString(), + }); } catch (err) { evt.addWarning( `intake-auto-skip failed: ${ err instanceof Error ? err.message : String(err) }`, ); + return jobFailed("intake auto-skip pass failed", err); } }); } diff --git a/src/lib/jobs/reminder/medication-reminder-check.ts b/src/lib/jobs/reminder/medication-reminder-check.ts index ff64ab72d..a21f07939 100644 --- a/src/lib/jobs/reminder/medication-reminder-check.ts +++ b/src/lib/jobs/reminder/medication-reminder-check.ts @@ -8,6 +8,7 @@ import { type Job } from "pg-boss"; import { recordError, recordReminderCheck } from "@/lib/jobs/worker-status"; import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups"; import { decrypt } from "@/lib/crypto"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { dispatchNotification } from "@/lib/notifications/dispatcher"; import { medicationDoseTag } from "@/lib/notifications/dose-tag"; @@ -137,9 +138,11 @@ export async function cleanupScheduledTelegramDeletions(): Promise { * Uses phase-based logic (GREEN/YELLOW/ORANGE/RED) to send one notification * per phase transition rather than every 15 minutes. */ -export async function handleReminderCheck(jobs: Job[]) { +export async function handleReminderCheck( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.medication_reminder", async (evt) => { + return withBackgroundEvent("job.medication_reminder", async (evt) => { const prisma = getWorkerPrisma(); try { recordReminderCheck(); @@ -176,6 +179,12 @@ export async function handleReminderCheck(jobs: Job[]) { }, }); + // The per-slot facts the tick already emits as wide-event meta, counted + // so the run's outcome carries them as scalars. + let missedDosesMinted = 0; + let notificationsDispatched = 0; + let notificationsFailed = 0; + for (const med of medications) { const userTz = med.user.timezone || "Europe/Berlin"; const { start: todayStart, end: todayEnd } = getUserTodayBounds( @@ -440,6 +449,7 @@ export async function handleReminderCheck(jobs: Job[]) { }, }); + missedDosesMinted++; evt.addMeta("missed_dose", `${med.name}:${slotTime}`); // v1.4.39 W-MED — refresh the compliance rollup so the @@ -521,7 +531,12 @@ export async function handleReminderCheck(jobs: Job[]) { url: `/medications/${med.id}`, }, }); + notificationsDispatched++; } catch (notifErr) { + // One user's channel failing must not fail the tick — the + // dedup anchor is already stamped, so the count is the only + // trace this slot's notice never left the host. + notificationsFailed++; evt.addWarning( `Notification dispatch failed for ${currentPhase} phase ${med.name}: ${notifErr}`, ); @@ -530,6 +545,13 @@ export async function handleReminderCheck(jobs: Job[]) { } } } + + return jobDone({ + medications: medications.length, + missed_doses_minted: missedDosesMinted, + notifications_dispatched: notificationsDispatched, + notifications_failed: notificationsFailed, + }); } catch (err) { evt.setError(err); recordError(); diff --git a/src/lib/jobs/reminder/mood-cycle-checks.ts b/src/lib/jobs/reminder/mood-cycle-checks.ts index afa03a835..53b783c9d 100644 --- a/src/lib/jobs/reminder/mood-cycle-checks.ts +++ b/src/lib/jobs/reminder/mood-cycle-checks.ts @@ -6,6 +6,7 @@ */ import { type Job } from "pg-boss"; import { recordError } from "@/lib/jobs/worker-status"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { runMoodReminderTick } from "@/lib/jobs/mood-reminder"; import { runCycleReminderTick } from "@/lib/jobs/cycle-reminder"; @@ -30,9 +31,9 @@ export interface MeasurementReminderPayload { export async function handleMoodReminderCheck( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.mood_reminder", async (evt) => { + return withBackgroundEvent("job.mood_reminder", async (evt) => { const prisma = getWorkerPrisma(); try { const summary = await runMoodReminderTick(prisma, new Date()); @@ -47,6 +48,16 @@ export async function handleMoodReminderCheck( skipped_outside_window: summary.skippedOutsideWindow, }, }); + // A tick that found nobody in their local reminder hour still ran the + // sweep, so it is a success with zero counts. + return jobDone({ + candidates_scanned: summary.candidatesScanned, + in_window: summary.inWindow, + dispatched: summary.dispatched, + skipped_already_logged: summary.skippedAlreadyLogged, + skipped_already_dispatched: summary.skippedAlreadyDispatched, + skipped_outside_window: summary.skippedOutsideWindow, + }); } catch (err) { evt.setError(err); recordError(); @@ -63,9 +74,9 @@ export async function handleMoodReminderCheck( */ export async function handleCycleReminderCheck( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.cycle_reminder", async (evt) => { + return withBackgroundEvent("job.cycle_reminder", async (evt) => { const prisma = getWorkerPrisma(); try { const summary = await runCycleReminderTick(prisma, new Date()); @@ -82,6 +93,16 @@ export async function handleCycleReminderCheck( skipped_outside_window: summary.skippedOutsideWindow, }, }); + return jobDone({ + candidates_scanned: summary.candidatesScanned, + in_window: summary.inWindow, + dispatched_period_soon: summary.dispatchedPeriodSoon, + dispatched_period_confirm: summary.dispatchedPeriodConfirm, + suppressed_client_managed: summary.suppressedClientManaged, + suppressed_discreet: summary.suppressedDiscreet, + skipped_already_notified: summary.skippedAlreadyNotified, + skipped_outside_window: summary.skippedOutsideWindow, + }); } catch (err) { evt.setError(err); recordError(); @@ -97,9 +118,9 @@ export async function handleCycleReminderCheck( */ export async function handleMeasurementReminderCheck( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.measurement_reminder", async (evt) => { + return withBackgroundEvent("job.measurement_reminder", async (evt) => { const prisma = getWorkerPrisma(); try { const summary = await runMeasurementReminderTick(prisma, new Date()); @@ -116,6 +137,19 @@ export async function handleMeasurementReminderCheck( failed: summary.failed, }, }); + // `failed` counts per-reminder dispatch failures the tick isolated. The + // sweep itself ran, so those ride out as a count rather than failing the + // job and retrying the whole cohort over one user's channel. + return jobDone({ + candidates_scanned: summary.candidatesScanned, + in_window: summary.inWindow, + dispatched: summary.dispatched, + auto_resolved: summary.autoResolved, + skipped_not_due: summary.skippedNotDue, + skipped_outside_window: summary.skippedOutsideWindow, + skipped_no_channel: summary.skippedNoChannel, + failed: summary.failed, + }); } catch (err) { evt.setError(err); recordError(); @@ -133,8 +167,8 @@ export async function handleMeasurementReminderCheck( */ export async function handleReminderSatisfy( jobs: Job[], -) { - await withBackgroundEvent("job.reminder_satisfy", async (evt) => { +): Promise { + return withBackgroundEvent("job.reminder_satisfy", async (evt) => { const prisma = getWorkerPrisma(); const now = new Date(); // Distinct users in this batch — a multi-source ingest spike can enqueue @@ -160,6 +194,13 @@ export async function handleReminderSatisfy( satisfied, }, }); + // A batch whose jobs carried no usable userId resolved nothing and is + // still a success with zero counts. + return jobDone({ + users: userIds.length, + candidates_scanned: candidates, + satisfied, + }); } catch (err) { evt.setError(err); recordError(); diff --git a/src/lib/jobs/reminder/ops-handlers.ts b/src/lib/jobs/reminder/ops-handlers.ts index a03154056..b483243c5 100644 --- a/src/lib/jobs/reminder/ops-handlers.ts +++ b/src/lib/jobs/reminder/ops-handlers.ts @@ -8,6 +8,7 @@ import { type Job } from "pg-boss"; import { runGeoBackfill } from "@/lib/jobs/geo-backfill"; import { runTlsPinMonitor } from "@/lib/jobs/tls-pin-monitor"; import { type PrDetectionPayload } from "@/lib/jobs/pr-detection"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { runHostMetricTick } from "@/lib/jobs/host-metric-sampler"; import { aggregateRecommendationFeedback } from "@/lib/jobs/feedback-aggregator"; @@ -34,40 +35,45 @@ export interface TlsPinMonitorPayload { export async function handleHostMetricSample( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.host_metric_sample", async (evt) => { + return withBackgroundEvent("job.host_metric_sample", async (evt) => { const p = getWorkerPrisma(); try { const { pruned } = await runHostMetricTick(p); evt.addMeta("host_metric_pruned", pruned); + return jobDone({ host_metric_pruned: pruned }); } catch (err) { - // The chart degrades gracefully when samples are missing — log - // and move on rather than poisoning the boss queue with retries. + // The chart degrades gracefully when samples are missing, so the + // warning stays — but a tick that took no sample did not do its work. evt.addWarning(`host-metric-sample failed: ${err}`); + return jobFailed("host metric sample failed", err); } }); } export async function handleFeedbackAggregator( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.feedback_aggregator", async (evt) => { + return withBackgroundEvent("job.feedback_aggregator", async (evt) => { const p = getWorkerPrisma(); try { const summary = await aggregateRecommendationFeedback(p); + const totalRows = summary.buckets.reduce((acc, b) => acc + b.total, 0); evt.addMeta("feedback_buckets", summary.buckets.length); - evt.addMeta( - "feedback_total_rows", - summary.buckets.reduce((acc, b) => acc + b.total, 0), - ); + evt.addMeta("feedback_total_rows", totalRows); evt.addMeta("feedback_window_days", summary.windowDays); + return jobDone({ + feedback_buckets: summary.buckets.length, + feedback_total_rows: totalRows, + feedback_window_days: summary.windowDays, + }); } catch (err) { - // The admin dashboard tolerates a stale summary — log and move - // on rather than poisoning the boss queue with retries that - // would block the next cleanup window. + // The admin dashboard tolerates a stale summary, so the warning stays — + // but a pass that aggregated nothing did not do its work. evt.addWarning(`feedback-aggregator failed: ${err}`); + return jobFailed("feedback aggregation failed", err); } }); } @@ -91,18 +97,21 @@ export async function handleFeedbackAggregator( */ let geoBackfillRunning = false; -export async function handleGeoBackfill(jobs: Job[]) { +export async function handleGeoBackfill( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.geo_backfill", async (evt) => { + return withBackgroundEvent("job.geo_backfill", async (evt) => { if (geoBackfillRunning) { // Earlier pass still in flight; skip this tick. Idempotent + the // next tick after the in-flight pass completes will pick up - // anything we miss here. + // anything we miss here. Deliberately ceding to the in-flight pass + // is not a failure, so the tick reports the skip and succeeds. evt.addWarning( "geo-backfill skipped — earlier pass still in flight inside the same worker process", ); evt.addMeta("geo_backfill_skipped", true); - return; + return jobDone({ geo_backfill_skipped: true }); } geoBackfillRunning = true; const p = getWorkerPrisma(); @@ -112,38 +121,60 @@ export async function handleGeoBackfill(jobs: Job[]) { evt.addMeta("geo_backfill_located", summary.located); evt.addMeta("geo_backfill_carrier_resolved", summary.carrierResolved); evt.addMeta("geo_backfill_still_unresolved", summary.stillUnresolved); + return jobDone({ + geo_backfill_scanned: summary.scanned, + geo_backfill_located: summary.located, + geo_backfill_carrier_resolved: summary.carrierResolved, + geo_backfill_still_unresolved: summary.stillUnresolved, + }); } catch (err) { - // The admin sign-in overview tolerates a stale Standort cell — - // log and move on so a one-off resolver hiccup does not poison - // the queue and block the next pass. + // The admin sign-in overview tolerates a stale Standort cell, so the + // warning stays — but a pass that resolved nothing did not do its work. evt.addWarning(`geo-backfill failed: ${err}`); + return jobFailed("geo backfill pass failed", err); } finally { geoBackfillRunning = false; } }); } -export async function handleTlsPinMonitor(jobs: Job[]) { +export async function handleTlsPinMonitor( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.tls_pin_monitor", async (evt) => { + return withBackgroundEvent("job.tls_pin_monitor", async (evt) => { const p = getWorkerPrisma(); try { const summary = await runTlsPinMonitor(p); evt.addMeta("tls_pin_monitor_outcome", summary.outcome); evt.addMeta("tls_pin_monitor_host", summary.host); evt.addMeta("tls_pin_monitor_known_count", summary.knownPinCount); + // No configured target leaves `host` null; the empty string keeps the + // fact shape scalar, and `outcome: "skipped"` already names that case. + return jobDone({ + tls_pin_monitor_outcome: summary.outcome, + tls_pin_monitor_host: summary.host ?? "", + tls_pin_monitor_known_count: summary.knownPinCount, + }); } catch (err) { // runTlsPinMonitor swallows probe failures internally; anything that - // escapes is unexpected. Log and move on so a one-off failure does not - // poison the queue and block the next tick. + // escapes is unexpected, and a tick that never compared a pin is + // exactly the silence this monitor exists to break. evt.addWarning(`tls-pin-monitor failed: ${err}`); + return jobFailed("tls pin probe failed", err); } }); } export async function handlePrDetection( jobs: Job[], -) { +): Promise { + // Each job in the batch opens its own wide event; the outcome is reported + // once for the batch, so the per-job facts accumulate out here. + let usersScanned = 0; + let insertedAcrossJobs = 0; + let tiesAcrossJobs = 0; + let usersFailed = 0; for (const job of jobs) { await withBackgroundEvent("job.pr_detection", async (evt) => { const p = getWorkerPrisma(); @@ -170,6 +201,10 @@ export async function handlePrDetection( insertedTotal += result.inserted; tiesTotal += result.ties; } catch (err) { + // One user's detection failing is that user's problem, not the + // pass's: it rides out as a count so the cron is not retried for + // the whole cohort. + usersFailed++; evt.addWarning(`pr-detection failed for user ${userId}: ${err}`); } } @@ -178,6 +213,18 @@ export async function handlePrDetection( evt.addMeta("pr_detection_ties", tiesTotal); evt.addMeta("pr_detection_silent", silent); evt.addMeta("pr_detection_mode", payloadUserId ? "ingest" : "cron"); + + usersScanned += userIds.length; + insertedAcrossJobs += insertedTotal; + tiesAcrossJobs += tiesTotal; }); } + + return jobDone({ + pr_detection_jobs: jobs.length, + pr_detection_users: usersScanned, + pr_detection_inserted: insertedAcrossJobs, + pr_detection_ties: tiesAcrossJobs, + pr_detection_users_failed: usersFailed, + }); } diff --git a/src/lib/jobs/reminder/poll-cohort.ts b/src/lib/jobs/reminder/poll-cohort.ts index a24a12ff3..154f61af0 100644 --- a/src/lib/jobs/reminder/poll-cohort.ts +++ b/src/lib/jobs/reminder/poll-cohort.ts @@ -21,6 +21,7 @@ import { type Job } from "pg-boss"; import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError } from "@/lib/jobs/worker-status"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { isSyncFailureRecorded } from "@/lib/integrations/status"; @@ -58,6 +59,13 @@ export interface PollCohortConfig { * handler resolves its targets from the batch payloads (single-user enqueue) or * the cohort query (cron tick), then re-syncs each sequentially with per-user * error isolation so one revoked grant / outage never starves the cohort. + * + * Outcome semantics for a fan-out pass: the JOB succeeded if the pass ran. + * One user's revoked grant is that user's problem — it belongs on their + * integration ledger, not in a queue-wide failure that retries the whole + * cohort — so per-user failures ride out as the `users_failed` count. Only a + * failure of the pass itself (cohort discovery, the wide-event scope) fails + * the job, and that path already throws. */ export function makePollCohortHandler({ taskName, @@ -65,8 +73,10 @@ export function makePollCohortHandler({ syncUser, recordFailure, }: PollCohortConfig) { - return async function handlePollCohort(jobs: Job[]) { - await withBackgroundEvent(taskName, async (evt) => { + return async function handlePollCohort( + jobs: Job[], + ): Promise { + return withBackgroundEvent(taskName, async (evt) => { const prisma = getWorkerPrisma(); try { const targets: Array<{ userId: string }> = []; @@ -77,9 +87,12 @@ export function makePollCohortHandler({ const ids = await findCohort(prisma); targets.push(...ids.map((userId) => ({ userId }))); } - if (targets.length === 0) return; + if (targets.length === 0) { + return jobDone({ total: 0, users_synced: 0, users_failed: 0 }); + } let usersSynced = 0; + let usersFailed = 0; let measurementsImported = 0; for (const { userId } of targets) { try { @@ -106,6 +119,7 @@ export function makePollCohortHandler({ ); }); } + usersFailed++; evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } } @@ -114,10 +128,17 @@ export function makePollCohortHandler({ task_name: taskName, result: { users_synced: usersSynced, + users_failed: usersFailed, total: targets.length, measurements_imported: measurementsImported, }, }); + return jobDone({ + total: targets.length, + users_synced: usersSynced, + users_failed: usersFailed, + measurements_imported: measurementsImported, + }); } catch (err) { evt.setError(err); recordError(); diff --git a/src/lib/jobs/reminder/register-reminders.ts b/src/lib/jobs/reminder/register-reminders.ts index 66a42de2a..581a81358 100644 --- a/src/lib/jobs/reminder/register-reminders.ts +++ b/src/lib/jobs/reminder/register-reminders.ts @@ -7,9 +7,15 @@ * * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`, - * and a `boss.work(QUEUE, …, handler)` binding drains it. The + * and a `createAndWork(boss, QUEUE, …, handler)` binding drains it. The * `cycle-reminder-queue` and `measurement-reminder-queue` guards read THIS * module. + * + * Retry policy: every queue here keeps the pg-boss default. These ticks are + * fan-out passes whose per-user failures never fail the job, so a job-level + * failure means the pass itself could not run — a transient the default two + * retries are the right answer to. The satisfy queue carries its own + * `retryLimit` at send time (see `reminder-satisfy.ts`). */ import { PgBoss } from "pg-boss"; import { @@ -17,7 +23,11 @@ import { REMINDER_SATISFY_CONCURRENCY, type ReminderSatisfyPayload, } from "@/lib/jobs/reminder-satisfy"; -import { createAndSchedule, type ScheduleEntry } from "./registrar-shared"; +import { + createAndSchedule, + createAndWork, + type ScheduleEntry, +} from "./registrar-shared"; import { ReminderCheckPayload, handleReminderCheck, @@ -121,7 +131,8 @@ export async function registerReminderQueues( ): Promise { await createAndSchedule(boss, allQueues, schedules); - await boss.work( + await createAndWork( + boss, QUEUE_NAME, { localConcurrency: 1 }, handleReminderCheck, @@ -130,7 +141,8 @@ export async function registerReminderQueues( // two reminder ticks from interleaving against the same user row; // the dedup ledger would still save us, but skipping the race here // avoids spurious P2002 errors in the wide-event log. - await boss.work( + await createAndWork( + boss, MOOD_REMINDER_QUEUE, { localConcurrency: 1 }, handleMoodReminderCheck, @@ -138,7 +150,8 @@ export async function registerReminderQueues( // v1.15 — single-flight cycle-reminder worker. localConcurrency=1 keeps // two ticks from racing the fire-and-forget `push_attempts` ledger that // anchors the per-day idempotency, exactly like the mood-reminder worker. - await boss.work( + await createAndWork( + boss, CYCLE_REMINDER_QUEUE, { localConcurrency: 1 }, handleCycleReminderCheck, @@ -147,7 +160,8 @@ export async function registerReminderQueues( // localConcurrency=1 keeps two ticks from racing the `nextDueAt` // advance that anchors the per-cycle idempotency, exactly like the // mood / cycle reminder workers. - await boss.work( + await createAndWork( + boss, MEASUREMENT_REMINDER_QUEUE, { localConcurrency: 1 }, handleMeasurementReminderCheck, @@ -155,7 +169,8 @@ export async function registerReminderQueues( // v1.18.1 — eventful Vorsorge satisfaction. Resolves a user's reminders // against their just-landed measurement / lab. Read-heavy on the user's // own data; the same small concurrency budget as PR detection. - await boss.work( + await createAndWork( + boss, REMINDER_SATISFY_QUEUE, { localConcurrency: REMINDER_SATISFY_CONCURRENCY }, handleReminderSatisfy, diff --git a/src/lib/jobs/reminder/whoop-sync.ts b/src/lib/jobs/reminder/whoop-sync.ts index 10f07df56..830ab4ebc 100644 --- a/src/lib/jobs/reminder/whoop-sync.ts +++ b/src/lib/jobs/reminder/whoop-sync.ts @@ -7,6 +7,7 @@ import { type Job } from "pg-boss"; import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError } from "@/lib/jobs/worker-status"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { syncUserRecovery, @@ -66,6 +67,10 @@ type ByIdSync = (userId: string, resourceId: string) => Promise; * - A cron tick (no `userId` on any job) walks every connection's collection. * * One user's parked-at-reauth state never starves the rest of the cohort. + * + * The pass is what the outcome judges: a per-user failure is that user's + * problem, recorded on their integration ledger and counted as `users_failed`, + * not a queue-wide failure that would retry the whole cohort. */ export async function runWhoopResourceSync( taskName: string, @@ -73,8 +78,8 @@ export async function runWhoopResourceSync( jobs: Job[], syncFn: CollectionSync, byIdFn?: ByIdSync, -): Promise { - await withBackgroundEvent(taskName, async (evt) => { +): Promise { + return withBackgroundEvent(taskName, async (evt) => { const prisma = getWorkerPrisma(); try { // (userId, optional resourceId) work items from the webhook payloads. @@ -94,9 +99,17 @@ export async function runWhoopResourceSync( }); targets.push(...connections); } - if (targets.length === 0) return; + if (targets.length === 0) { + return jobDone({ + total: 0, + users_synced: 0, + users_failed: 0, + measurements_imported: 0, + }); + } let usersSynced = 0; + let usersFailed = 0; let measurementsImported = 0; for (const { userId, resourceId } of targets) { try { @@ -123,6 +136,7 @@ export async function runWhoopResourceSync( if (!isSyncFailureRecorded(err)) { await recordWhoopSyncFailure(userId, err, leg); } + usersFailed++; evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } } @@ -135,6 +149,12 @@ export async function runWhoopResourceSync( measurements_imported: measurementsImported, }, }); + return jobDone({ + total: targets.length, + users_synced: usersSynced, + users_failed: usersFailed, + measurements_imported: measurementsImported, + }); } catch (err) { evt.setError(err); recordError(); @@ -143,7 +163,9 @@ export async function runWhoopResourceSync( }); } -export function handleWhoopRecoverySync(jobs: Job[]) { +export function handleWhoopRecoverySync( + jobs: Job[], +): Promise { return runWhoopResourceSync( "job.whoop_recovery_sync", "recovery", @@ -153,7 +175,9 @@ export function handleWhoopRecoverySync(jobs: Job[]) { ); } -export function handleWhoopSleepSync(jobs: Job[]) { +export function handleWhoopSleepSync( + jobs: Job[], +): Promise { return runWhoopResourceSync( "job.whoop_sleep_sync", "sleep", @@ -163,7 +187,9 @@ export function handleWhoopSleepSync(jobs: Job[]) { ); } -export function handleWhoopWorkoutSync(jobs: Job[]) { +export function handleWhoopWorkoutSync( + jobs: Job[], +): Promise { return runWhoopResourceSync( "job.whoop_workout_sync", "workout", @@ -173,7 +199,9 @@ export function handleWhoopWorkoutSync(jobs: Job[]) { ); } -export function handleWhoopCycleSync(jobs: Job[]) { +export function handleWhoopCycleSync( + jobs: Job[], +): Promise { // Cycle has no webhook (poll-only), so no fetch-by-id path. return runWhoopResourceSync( "job.whoop_cycle_sync", diff --git a/src/lib/jobs/reminder/withings-sync.ts b/src/lib/jobs/reminder/withings-sync.ts index 96be0b354..153283b58 100644 --- a/src/lib/jobs/reminder/withings-sync.ts +++ b/src/lib/jobs/reminder/withings-sync.ts @@ -7,6 +7,7 @@ import { type Job } from "pg-boss"; import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError, recordWithingsSync } from "@/lib/jobs/worker-status"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { retryDueWithingsWebhookSubscriptions, @@ -50,15 +51,20 @@ export interface WithingsSleepSyncPayload { */ export async function handleWithingsFallbackSync( jobs: Job[], -) { +): Promise { void jobs; - await withBackgroundEvent("job.withings_sync", async (evt) => { + return withBackgroundEvent("job.withings_sync", async (evt) => { const prisma = getWorkerPrisma(); try { recordWithingsSync(); + // The subscription repair is an auxiliary leg, not the pass this queue is + // named for. Failing the job on it would skip the fallback sync that every + // user relies on when a webhook goes missing, so it rides out as a fact. + let subscriptionRepairFailed = false; try { await retryDueWithingsWebhookSubscriptions(); } catch { + subscriptionRepairFailed = true; evt.addWarning( "Withings subscription repair failed; continuing fallback sync", ); @@ -68,10 +74,17 @@ export async function handleWithingsFallbackSync( }); if (connections.length === 0) { - return; + return jobDone({ + total: 0, + users_synced: 0, + users_failed: 0, + measurements_imported: 0, + subscription_repair_failed: subscriptionRepairFailed, + }); } let usersSynced = 0; + let usersFailed = 0; let measurementsImported = 0; for (const connection of connections) { @@ -87,6 +100,7 @@ export async function handleWithingsFallbackSync( }); } } catch { + usersFailed++; evt.addWarning(`Fallback sync failed for user ${connection.userId}`); } } @@ -99,6 +113,13 @@ export async function handleWithingsFallbackSync( measurements_imported: measurementsImported, }, }); + return jobDone({ + total: connections.length, + users_synced: usersSynced, + users_failed: usersFailed, + measurements_imported: measurementsImported, + subscription_repair_failed: subscriptionRepairFailed, + }); } catch (err) { evt.setError(err); recordError(); @@ -127,8 +148,8 @@ export async function handleWithingsFallbackSync( */ export async function handleWithingsEcgSync( jobs: Job[], -) { - await withBackgroundEvent("job.withings_ecg_sync", async (evt) => { +): Promise { + return withBackgroundEvent("job.withings_ecg_sync", async (evt) => { const prisma = getWorkerPrisma(); try { const targets: Array<{ @@ -152,9 +173,17 @@ export async function handleWithingsEcgSync( }); targets.push(...connections); } - if (targets.length === 0) return; + if (targets.length === 0) { + return jobDone({ + total: 0, + users_synced: 0, + users_failed: 0, + recordings_imported: 0, + }); + } let usersSynced = 0; + let usersFailed = 0; let recordingsImported = 0; for (const { userId, startdate, enddate } of targets) { try { @@ -165,6 +194,7 @@ export async function handleWithingsEcgSync( recordingsImported += await syncUserEcg(userId, options); usersSynced++; } catch (err) { + usersFailed++; evt.addWarning(`Withings ECG sync failed for user ${userId}: ${err}`); } } @@ -177,6 +207,12 @@ export async function handleWithingsEcgSync( recordings_imported: recordingsImported, }, }); + return jobDone({ + total: targets.length, + users_synced: usersSynced, + users_failed: usersFailed, + recordings_imported: recordingsImported, + }); } catch (err) { evt.setError(err); recordError(); @@ -203,8 +239,8 @@ export async function handleWithingsEcgSync( */ export async function handleWithingsActivitySync( jobs: Job[], -) { - await withBackgroundEvent("job.withings_activity_sync", async (evt) => { +): Promise { + return withBackgroundEvent("job.withings_activity_sync", async (evt) => { const prisma = getWorkerPrisma(); try { const targets: Array<{ userId: string }> = []; @@ -220,9 +256,17 @@ export async function handleWithingsActivitySync( }); targets.push(...connections); } - if (targets.length === 0) return; + if (targets.length === 0) { + return jobDone({ + total: 0, + users_synced: 0, + users_failed: 0, + measurements_imported: 0, + }); + } let usersSynced = 0; + let usersFailed = 0; let measurementsImported = 0; for (const { userId } of targets) { try { @@ -235,6 +279,7 @@ export async function handleWithingsActivitySync( }); } } catch { + usersFailed++; evt.addWarning(`Withings activity sync failed for user ${userId}`); } } @@ -247,6 +292,12 @@ export async function handleWithingsActivitySync( measurements_imported: measurementsImported, }, }); + return jobDone({ + total: targets.length, + users_synced: usersSynced, + users_failed: usersFailed, + measurements_imported: measurementsImported, + }); } catch (err) { evt.setError(err); recordError(); @@ -262,8 +313,8 @@ export async function handleWithingsActivitySync( */ export async function handleWithingsSleepSync( jobs: Job[], -) { - await withBackgroundEvent("job.withings_sleep_sync", async (evt) => { +): Promise { + return withBackgroundEvent("job.withings_sleep_sync", async (evt) => { const prisma = getWorkerPrisma(); try { const targets: Array<{ userId: string }> = []; @@ -278,9 +329,17 @@ export async function handleWithingsSleepSync( }); targets.push(...connections); } - if (targets.length === 0) return; + if (targets.length === 0) { + return jobDone({ + total: 0, + users_synced: 0, + users_failed: 0, + measurements_imported: 0, + }); + } let usersSynced = 0; + let usersFailed = 0; let measurementsImported = 0; for (const { userId } of targets) { try { @@ -293,6 +352,7 @@ export async function handleWithingsSleepSync( }); } } catch { + usersFailed++; evt.addWarning(`Withings sleep sync failed for user ${userId}`); } } @@ -305,6 +365,12 @@ export async function handleWithingsSleepSync( measurements_imported: measurementsImported, }, }); + return jobDone({ + total: targets.length, + users_synced: usersSynced, + users_failed: usersFailed, + measurements_imported: measurementsImported, + }); } catch (err) { evt.setError(err); recordError(); diff --git a/src/lib/jobs/restore-drill.ts b/src/lib/jobs/restore-drill.ts index c37b6fc73..5497174ce 100644 --- a/src/lib/jobs/restore-drill.ts +++ b/src/lib/jobs/restore-drill.ts @@ -30,6 +30,7 @@ import { type S3Like, } from "@/lib/jobs/offhost-backup"; import { reportWorkerError } from "@/lib/jobs/report-worker-error"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; export const RESTORE_DRILL_QUEUE = "data-restore-drill"; @@ -149,9 +150,11 @@ export async function runRestoreDrill( * page); every other failure goes through `reportWorkerError` so the * operator hears about a rotting backup chain. */ -export async function handleRestoreDrill(jobs: Job[]): Promise { +export async function handleRestoreDrill( + jobs: Job[], +): Promise { void jobs; - await withBackgroundEvent("job.restore_drill", async (evt) => { + return withBackgroundEvent("job.restore_drill", async (evt) => { try { const report = await runRestoreDrill(); evt.addMeta("restore_drill_object_key", report.objectKey); @@ -181,13 +184,31 @@ export async function handleRestoreDrill(jobs: Job[]): Promise { { objectKey: report.objectKey, ageDays: report.ageDays }, ); } + // A stale chain is a verdict about the uploader, not a failure of the + // drill: the artefact was fetched, decrypted and parsed, and the page + // above already reached the operator. Failing the job here would retry a + // reading that cannot change until the next upload lands. + return jobDone({ + restore_drill_object_key: report.objectKey, + restore_drill_age_days: report.ageDays, + restore_drill_ciphertext_bytes: report.ciphertextBytes, + restore_drill_plaintext_bytes: report.plaintextBytes, + restore_drill_measurements: report.recordCounts.measurements, + restore_drill_medications: report.recordCounts.medications, + restore_drill_intake_events: report.recordCounts.intakeEvents, + restore_drill_mood_entries: report.recordCounts.moodEntries, + restore_drill_stale: report.stale, + }); } catch (err) { if (err instanceof OffhostBackupNotConfiguredError) { evt.addWarning(`restore-drill skipped: ${err.message}`); - return; + // Most self-hosters never set the S3 vars. Nothing to verify is not a + // failed verification, and a monthly failed job would be noise. + return jobDone({ skipped: "offhost backup not configured" }); } evt.addWarning(`restore-drill failed: ${err}`); await reportWorkerError(RESTORE_DRILL_QUEUE, err); + return jobFailed("restore drill failed", err); } }); } diff --git a/src/lib/jobs/run-job.ts b/src/lib/jobs/run-job.ts index 27fcd268a..05a180084 100644 --- a/src/lib/jobs/run-job.ts +++ b/src/lib/jobs/run-job.ts @@ -45,6 +45,12 @@ export class JobFailure extends Error { this.name = "JobFailure"; this.queue = queue; this.reason = reason; + // Carry the thrown error's stack. Without this the only stack the + // operator ever sees points at `runJob`, which is the same for all 101 + // queues and says nothing about where the work actually broke. + if (cause instanceof Error && cause.stack) { + this.stack = `${this.name}: ${this.message}\ncaused by ${cause.stack}`; + } } } diff --git a/src/lib/jobs/workout-insight-generate.ts b/src/lib/jobs/workout-insight-generate.ts index 4a690d088..a5854f285 100644 --- a/src/lib/jobs/workout-insight-generate.ts +++ b/src/lib/jobs/workout-insight-generate.ts @@ -48,6 +48,7 @@ import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-work import { defaultLocale, locales, type Locale } from "@/lib/i18n/config"; import { finalizeStatusSummary } from "@/lib/insights/status-shared"; import { runStatusCompletion } from "@/lib/insights/status-provider"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { annotate } from "@/lib/logging/context"; import { withBackgroundEvent } from "@/lib/logging/background"; import { resolveModuleMap } from "@/lib/modules/gate"; @@ -524,10 +525,14 @@ export async function runWorkoutInsightGenerate( export async function handleWorkoutInsightGenerate( jobs: Job[], -): Promise { - await withBackgroundEvent("job.workout_insight_generate", async (evt) => { +): Promise { + return withBackgroundEvent("job.workout_insight_generate", async (evt) => { + let generated = 0; + let skipped = 0; for (const job of jobs) { const outcome = await runWorkoutInsightGenerate(job.data); + if (outcome.status === "skipped") skipped++; + else generated++; evt.addMeta( "workout_insight", outcome.status === "skipped" @@ -535,5 +540,9 @@ export async function handleWorkoutInsightGenerate( : `generated:${outcome.providerType}`, ); } + // Every gate in front of the provider returns `skipped`, so a pass that + // wrote no paragraph at all is still a pass that ran. Only a transient + // fault escapes `runWorkoutInsightGenerate`, and that one throws. + return jobDone({ total: jobs.length, generated, skipped }); }); }