From 0aa5cec17d173b75465411f31fa45b6d7df935e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 27 Jul 2026 23:08:50 +0200 Subject: [PATCH] feat(jobs): bind every queue through createAndWork and ban the bare call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All 104 bindings across the five registrars now go through createAndWork, whose handler type is (jobs) => Promise. The forty-two inline handlers return what they did. A handler that falls off the end no longer compiles, and the healthlog/job-handler-outcome rule refuses a bare boss.work( anywhere under src/lib/jobs except the one module that owns the wrapper. The rule's header states its own limit: it catches the omission, not a handler that claims ok:true after a swallow. Retry policy, reviewed queue by queue because the rethrow changes what runs, not only what is visible: - Twenty-one cron passes carry cronIsTheRetry (retryLimit 0). They are the retention DELETEs, the OAuth-state prunes, the hourly maintenance sweeps and the nightly per-sample drain — every one of them a statement that will fail identically a second later. pg-boss defaults to two retries with no delay, so converting them without this would have run the same doomed statement three times a night. The next tick is the retry and it arrives with the whole gap as backoff. - The policy rides on the schedule, not on createQueue: create_queue() ends in ON CONFLICT DO NOTHING, so a queue-level default would be inert on every already-running instance. - Everything else keeps its retries deliberately. A TLS probe, an off-host upload, a restore drill and a provider poll all fail transiently, and the reminder ticks only fail at job level when the pass itself could not run. --- eslint-plugins/healthlog/index.js | 4 + .../healthlog/job-handler-outcome.js | 107 +++++ eslint.config.mjs | 7 + .../reminder/register-integration-sync.ts | 157 ++++++-- src/lib/jobs/reminder/register-maintenance.ts | 346 +++++++++++++--- src/lib/jobs/reminder/register-rollup.ts | 185 ++++++++- src/lib/jobs/reminder/register-status.ts | 371 ++++++++++++------ src/lib/jobs/reminder/registrar-shared.ts | 26 ++ 8 files changed, 975 insertions(+), 228 deletions(-) create mode 100644 eslint-plugins/healthlog/job-handler-outcome.js diff --git a/eslint-plugins/healthlog/index.js b/eslint-plugins/healthlog/index.js index d1529a8bc..8147b1540 100644 --- a/eslint-plugins/healthlog/index.js +++ b/eslint-plugins/healthlog/index.js @@ -8,6 +8,8 @@ * - `no-raw-palette-color` — ban raw Tailwind palette utilities in app UI. * - `spacing-scale` — no pt-/pb- overrides on gap-based Card slots, * and no off-scale `5` step on a bg-card shell. + * - `job-handler-outcome` — pg-boss bindings route through createAndWork, + * whose handler returns a JobOutcome. */ "use strict"; @@ -17,6 +19,7 @@ const safeFetchRequired = require("./safe-fetch-required.js"); const apiFetchRequired = require("./api-fetch-required.js"); const noRawPaletteColor = require("./no-raw-palette-color.js"); const spacingScale = require("./spacing-scale.js"); +const jobHandlerOutcome = require("./job-handler-outcome.js"); module.exports = { rules: { @@ -30,5 +33,6 @@ module.exports = { // named rule (now error-level too; the staged warn phase ended with // the semantic sweep). See the rule header. "no-dracula-utility": noRawPaletteColor, + "job-handler-outcome": jobHandlerOutcome, }, }; diff --git a/eslint-plugins/healthlog/job-handler-outcome.js b/eslint-plugins/healthlog/job-handler-outcome.js new file mode 100644 index 000000000..9dfdbafc4 --- /dev/null +++ b/eslint-plugins/healthlog/job-handler-outcome.js @@ -0,0 +1,107 @@ +/** + * @fileoverview ESLint rule — a pg-boss queue binding must go through + * `createAndWork`, never a bare `boss.work(`. + * + * `boss.work(queue, options, handler)` completes a job whenever the handler's + * promise resolves, whatever it resolves to. Every binding in this tree used + * to be written that way, so a handler that caught an error, logged a warning + * and returned looked exactly like one that did its work. `createAndWork` + * (`src/lib/jobs/reminder/registrar-shared.ts`) types its handler as + * `(jobs) => Promise` and wraps it in `runJob`, which fails the + * pg-boss job on `ok: false`. Routing every binding through it is what makes + * "the handler returned nothing" a compile error rather than a silent pass. + * + * WHAT THIS RULE PROVES, AND WHAT IT DOES NOT + * + * It proves that no queue in this tree is bound by a path that accepts a + * void handler. That is all it proves. It does NOT detect a handler that + * returns `{ ok: true }` after swallowing an exception — that compiles, it + * passes this rule, and it is exactly as invisible as the defect the outcome + * type was written for. The rule converts an omission (forgetting to signal + * failure) into a commission (claiming a success that did not happen). The + * second is a smaller class and a reviewable one; it is not zero. + * + * Two further evasions are known and unhandled: aliasing the method + * (`const w = boss.work.bind(boss)`, `const { work } = boss`) and calling + * `work` on an object not named `boss`. Both are syntactic dodges a reviewer + * would see in the diff; neither is defended against here. + * + * Scope: `src/lib/jobs/`, excluding the registrar module that owns the one + * legal `boss.work` call and excluding test files, which mock `boss.work` + * and assert against it directly. + * + * @see src/lib/jobs/run-job.ts + * @see src/lib/jobs/job-outcome.ts + * @see src/lib/jobs/reminder/registrar-shared.ts + */ + +"use strict"; + +// The one module allowed to call `boss.work` — `createAndWork` lives here +// and is the wrapper every other binding has to use. +const EXEMPT_FILES = ["src/lib/jobs/reminder/registrar-shared.ts"]; + +// The worker tree. A `boss.work` outside it would not be a queue binding. +const ENFORCED_ROOTS = ["src/lib/jobs/"]; + +function toPosix(filename) { + return filename.replace(/\\/g, "/"); +} + +function isTestFile(posix) { + return ( + /\.test\.[cm]?[jt]sx?$/.test(posix) || + /\.spec\.[cm]?[jt]sx?$/.test(posix) || + posix.includes("/__tests__/") || + posix.includes("/__mocks__/") + ); +} + +function isEnforced(filename) { + const posix = toPosix(filename); + if (!ENFORCED_ROOTS.some((root) => posix.includes(root))) return false; + if (EXEMPT_FILES.some((f) => posix.includes(f))) return false; + if (isTestFile(posix)) return false; + return true; +} + +/** @type {import("eslint").Rule.RuleModule} */ +const jobHandlerOutcomeRule = { + meta: { + type: "problem", + docs: { + description: + "pg-boss queue bindings must go through createAndWork so the handler returns a JobOutcome.", + }, + schema: [], + messages: { + bareBossWork: + 'boss.work() completes a job whenever the handler resolves, so a swallowed failure reads as success. Bind the queue with createAndWork(boss, QUEUE, options, handler) from "./registrar-shared" — its handler type is (jobs) => Promise and runJob fails the job on ok:false.', + }, + }, + create(context) { + const filename = context.filename ?? context.getFilename?.(); + if (!filename || !isEnforced(filename)) { + return {}; + } + + return { + CallExpression(node) { + const callee = node.callee; + if ( + callee.type !== "MemberExpression" || + callee.computed || + callee.property.type !== "Identifier" || + callee.property.name !== "work" || + callee.object.type !== "Identifier" || + callee.object.name !== "boss" + ) { + return; + } + context.report({ node, messageId: "bareBossWork" }); + }, + }; + }, +}; + +module.exports = jobHandlerOutcomeRule; diff --git a/eslint.config.mjs b/eslint.config.mjs index 96ad3e1fd..82fa8cfae 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -75,6 +75,13 @@ const eslintConfig = defineConfig([ // trip it. Density lives on the Card itself (`gap-2 py-3 md:py-4`) or on // a swept `p-4 md:p-6` shell. See the rule header. "healthlog/spacing-scale": "error", + // Every pg-boss queue binding under src/lib/jobs goes through + // `createAndWork`, whose handler type is `(jobs) => Promise`. + // A bare `boss.work(` accepts a handler that resolves to nothing, which + // is how a swallowed failure came to read as a completed job. The rule + // catches the omission only — a handler that returns `{ ok: true }` + // after swallowing still passes, and its header says so. + "healthlog/job-handler-outcome": "error", }, }, // v1.28.17 — every recharts-rendering component funnels through the diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index 85a5e1bf2..54e753948 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -9,8 +9,8 @@ * * 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 queue-wiring - * guards (`withings-queues`, `whoop-queues`, `fitbit-queues`, + * and a `createAndWork(boss, QUEUE, …, handler)` binding drains it. The + * queue-wiring guards (`withings-queues`, `whoop-queues`, `fitbit-queues`, * `nightscout-queues`, `sleep-timeline-queues`) read THIS module. */ import { PgBoss } from "pg-boss"; @@ -78,8 +78,11 @@ import { type IntegrationBackfillAdmissionPayload, } from "@/lib/jobs/integration-backfill-admission"; import { workerLog } from "./shared"; +import { jobDone } from "@/lib/jobs/job-outcome"; import { createAndSchedule, + createAndWork, + cronIsTheRetry, type QueuePolicyTable, type ScheduleEntry, } from "./registrar-shared"; @@ -333,7 +336,16 @@ const schedules: ScheduleEntry[] = [ // v1.32.23 — hourly ECG catch-net (:41) so watch-only accounts pull their // strips without a webhook or a scale. [WITHINGS_ECG_SYNC_QUEUE, WITHINGS_ECG_SYNC_CRON], - [WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, WITHINGS_OAUTH_STATE_CLEANUP_CRON], + // The five OAuth-state / handoff prunes carry `cronIsTheRetry`: their + // handlers used to warn and carry on, they now fail the job, and a DELETE + // that could not run will not run any better a second later. The daily tick + // is the retry. The sync crons above keep the default retries — a provider + // outage is transient and worth retrying inside the hour. + [ + WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, + WITHINGS_OAUTH_STATE_CLEANUP_CRON, + cronIsTheRetry, + ], // v1.11.0 — WHOOP poll-fallback crons. Recovery/sleep/workout catch // dropped webhooks; cycle is the sole driver (no webhook). Staggered off // the Withings crons so the hourly ticks don't pile up on one boss poll. @@ -342,11 +354,19 @@ const schedules: ScheduleEntry[] = [ [WHOOP_WORKOUT_SYNC_QUEUE, WHOOP_WORKOUT_SYNC_CRON], [WHOOP_CYCLE_SYNC_QUEUE, WHOOP_CYCLE_SYNC_CRON], // v1.11.0 — daily 03:22 Europe/Berlin prune for expired WHOOP OAuth states. - [WHOOP_OAUTH_STATE_CLEANUP_QUEUE, WHOOP_OAUTH_STATE_CLEANUP_CRON], + [ + WHOOP_OAUTH_STATE_CLEANUP_QUEUE, + WHOOP_OAUTH_STATE_CLEANUP_CRON, + cronIsTheRetry, + ], // v1.12.0 — hourly Fitbit poll (:08, staggered off WHOOP/Withings) + the // daily 03:24 Europe/Berlin prune for expired Fitbit OAuth states. [FITBIT_SYNC_QUEUE, FITBIT_SYNC_CRON], - [FITBIT_OAUTH_STATE_CLEANUP_QUEUE, FITBIT_OAUTH_STATE_CLEANUP_CRON], + [ + FITBIT_OAUTH_STATE_CLEANUP_QUEUE, + FITBIT_OAUTH_STATE_CLEANUP_CRON, + cronIsTheRetry, + ], // v1.26.0 — hourly Google Health poll (:18, staggered off WHOOP/Fitbit/ // Nightscout/Polar/Oura) + the daily 03:26 Europe/Berlin prune for expired // Google Health OAuth states. @@ -354,6 +374,7 @@ const schedules: ScheduleEntry[] = [ [ GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE, GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_CRON, + cronIsTheRetry, ], // v1.17.0 — hourly Nightscout CGM poll (:11, staggered off the other sync // ticks). @@ -364,7 +385,11 @@ const schedules: ScheduleEntry[] = [ // v1.28.x — hourly Strava (:17) OAuth poll. [STRAVA_SYNC_QUEUE, STRAVA_SYNC_CRON], // v1.30.x — daily 03:28 Europe/Berlin prune for expired native OIDC handoffs. - [OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, OIDC_NATIVE_HANDOFF_CLEANUP_CRON], + [ + OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, + OIDC_NATIVE_HANDOFF_CLEANUP_CRON, + cronIsTheRetry, + ], ]; /** @@ -450,19 +475,30 @@ export async function registerIntegrationSyncQueues( // reaches this shared durable queue before doing full-history work. The // common group is enforced in PostgreSQL across worker processes/nodes. - await boss.work( + await createAndWork( + boss, INTEGRATION_BACKFILL_ADMISSION_QUEUE, { localConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY, groupConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY, }, async (jobs) => { + // The drain has no per-job error isolation: a runner that throws fails + // the job and the admission retry policy applies. So reaching the end + // means every admitted job did its work, and the counts each runner + // already returns are what the pass reports. + let importedTotal = 0; + let removedTotal = 0; + let deletedTotal = 0; + let markersTotal = 0; + let linkedTotal = 0; for (const job of jobs) { const { kind, data } = job.data; const { userId } = data; switch (kind) { case "whoop-backfill": { const { imported } = await runWhoopBackfillForUser(userId); + importedTotal += imported; workerLog( "info", `[whoop-backfill] user=${userId} imported=${imported}`, @@ -471,6 +507,7 @@ export async function registerIntegrationSyncQueues( } case "fitbit-backfill": { const { imported } = await runFitbitBackfillForUser(userId); + importedTotal += imported; workerLog( "info", `[fitbit-backfill] user=${userId} imported=${imported}`, @@ -479,6 +516,7 @@ export async function registerIntegrationSyncQueues( } case "google-health-backfill": { const { imported } = await runGoogleHealthBackfillForUser(userId); + importedTotal += imported; workerLog( "info", `[google-health-backfill] user=${userId} imported=${imported}`, @@ -488,6 +526,7 @@ export async function registerIntegrationSyncQueues( case "google-health-sleep-repair": { const { imported } = await runGoogleHealthSleepRepairForUser(userId); + importedTotal += imported; workerLog( "info", `[google-health-sleep-repair] user=${userId} imported=${imported}`, @@ -497,6 +536,8 @@ export async function registerIntegrationSyncQueues( case "fitbit-sleep-repair": { const { imported, removed } = await runFitbitSleepRepairForUser(userId); + importedTotal += imported; + removedTotal += removed; workerLog( "info", `[fitbit-sleep-repair] user=${userId} imported=${imported} removed=${removed}`, @@ -513,6 +554,8 @@ export async function registerIntegrationSyncQueues( userId, data.provider, ); + deletedTotal += deleted; + importedTotal += imported; workerLog( "info", `[sleep-timeline-backfill] user=${userId} provider=${data.provider} deleted=${deleted} imported=${imported}`, @@ -522,6 +565,8 @@ export async function registerIntegrationSyncQueues( case "lab-biomarker-backfill": { const { markers, linked } = await runLabBiomarkerBackfillForUser(userId); + markersTotal += markers; + linkedTotal += linked; workerLog( "info", `[lab-biomarker-backfill] user=${userId} markers=${markers} linked=${linked}`, @@ -530,6 +575,7 @@ export async function registerIntegrationSyncQueues( } case "strava-backfill": { const { imported } = await runStravaBackfillForUser(userId); + importedTotal += imported; workerLog( "info", `[strava-backfill] user=${userId} imported=${imported}`, @@ -538,30 +584,43 @@ export async function registerIntegrationSyncQueues( } } } + return jobDone({ + jobs: jobs.length, + imported: importedTotal, + removed: removedTotal, + deleted: deletedTotal, + markers: markersTotal, + linked: linkedTotal, + }); }, ); - await boss.work( + await createAndWork( + boss, WITHINGS_SYNC_QUEUE, { localConcurrency: 1 }, handleWithingsFallbackSync, ); - await boss.work( + await createAndWork( + boss, WITHINGS_ACTIVITY_QUEUE, { localConcurrency: 1 }, handleWithingsActivitySync, ); - await boss.work( + await createAndWork( + boss, WITHINGS_SLEEP_QUEUE, { localConcurrency: 1 }, handleWithingsSleepSync, ); - await boss.work( + await createAndWork( + boss, WITHINGS_ECG_SYNC_QUEUE, { localConcurrency: 1 }, handleWithingsEcgSync, ); - await boss.work( + await createAndWork( + boss, WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, { localConcurrency: 1 }, handleWithingsOAuthStateCleanup, @@ -569,29 +628,34 @@ export async function registerIntegrationSyncQueues( // v1.11.0 — WHOOP per-resource sync handlers. Webhook-driven per-user + // cron full-iteration. Serial concurrency so a backfill-heavy tick never // crowds the request pool and stays inside WHOOP's 100 req/min app cap. - await boss.work( + await createAndWork( + boss, WHOOP_RECOVERY_SYNC_QUEUE, { localConcurrency: 1 }, handleWhoopRecoverySync, ); - await boss.work( + await createAndWork( + boss, WHOOP_SLEEP_SYNC_QUEUE, { localConcurrency: 1 }, handleWhoopSleepSync, ); - await boss.work( + await createAndWork( + boss, WHOOP_WORKOUT_SYNC_QUEUE, { localConcurrency: 1 }, handleWhoopWorkoutSync, ); - await boss.work( + await createAndWork( + boss, WHOOP_CYCLE_SYNC_QUEUE, { localConcurrency: 1 }, handleWhoopCycleSync, ); // v1.11.0 — self-converging WHOOP backfill. This source queue retains the // per-connection singleton and forwards admitted work to the shared drain. - await boss.work( + await createAndWork( + boss, WHOOP_BACKFILL_QUEUE, { localConcurrency: WHOOP_BACKFILL_CONCURRENCY }, async (jobs) => { @@ -601,29 +665,34 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); - await boss.work( + await createAndWork( + boss, WHOOP_OAUTH_STATE_CLEANUP_QUEUE, { localConcurrency: 1 }, handleWhoopOAuthStateCleanup, ); // v1.30.x — daily sweep for the native OIDC handoff ledger. - await boss.work( + await createAndWork( + boss, OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, { localConcurrency: 1 }, handleOidcNativeHandoffCleanup, ); // v1.12.0 — Fitbit poll-sync (cron full-iteration; no webhook). Serial // concurrency so a backfill-heavy tick never crowds the request pool. - await boss.work( + await createAndWork( + boss, FITBIT_SYNC_QUEUE, { localConcurrency: 1 }, handleFitbitSync, ); // v1.12.0 — self-converging Fitbit backfill. This source queue retains the // per-connection singleton and forwards admitted work to the shared drain. - await boss.work( + await createAndWork( + boss, FITBIT_BACKFILL_QUEUE, { localConcurrency: FITBIT_BACKFILL_CONCURRENCY }, async (jobs) => { @@ -633,12 +702,14 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); // One-shot Fitbit sleep duplicate repair. This source queue retains the // per-connection singleton and forwards the watermark-safe bounded sleep // re-read to the shared drain. - await boss.work( + await createAndWork( + boss, FITBIT_SLEEP_REPAIR_QUEUE, { localConcurrency: FITBIT_SLEEP_REPAIR_CONCURRENCY }, async (jobs) => { @@ -648,16 +719,19 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); - await boss.work( + await createAndWork( + boss, FITBIT_OAUTH_STATE_CLEANUP_QUEUE, { localConcurrency: 1 }, handleFitbitOAuthStateCleanup, ); // v1.26.0 — Google Health poll-sync (cron full-iteration; no webhook). Serial // concurrency so a backfill-heavy tick never crowds the request pool. - await boss.work( + await createAndWork( + boss, GOOGLE_HEALTH_SYNC_QUEUE, { localConcurrency: 1 }, handleGoogleHealthSync, @@ -665,7 +739,8 @@ export async function registerIntegrationSyncQueues( // v1.26.0 — self-converging Google Health backfill. This source queue // retains the per-connection singleton and forwards admitted work to the // shared drain. - await boss.work( + await createAndWork( + boss, GOOGLE_HEALTH_BACKFILL_QUEUE, { localConcurrency: GOOGLE_HEALTH_BACKFILL_CONCURRENCY }, async (jobs) => { @@ -675,12 +750,14 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); // v1.28.x — one-shot Google Health sleep duplicate repair. This source // queue retains the per-connection singleton and forwards the watermark-safe // full sleep-history re-read to the shared drain. - await boss.work( + await createAndWork( + boss, GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE, { localConcurrency: GOOGLE_HEALTH_SLEEP_REPAIR_CONCURRENCY }, async (jobs) => { @@ -690,9 +767,11 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); - await boss.work( + await createAndWork( + boss, GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE, { localConcurrency: 1 }, handleGoogleHealthOAuthStateCleanup, @@ -700,7 +779,8 @@ export async function registerIntegrationSyncQueues( // v1.17.1 — one-shot sleep-timeline backfill. This source queue retains the // per-(user, provider) singleton and forwards admitted work to the shared // drain. - await boss.work( + await createAndWork( + boss, SLEEP_TIMELINE_BACKFILL_QUEUE, { localConcurrency: SLEEP_TIMELINE_BACKFILL_CONCURRENCY }, async (jobs) => { @@ -710,11 +790,13 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); // v1.18.1 — one-shot lab-biomarker backfill. This source queue retains the // per-user singleton and forwards admitted work to the shared drain. - await boss.work( + await createAndWork( + boss, LAB_BIOMARKER_BACKFILL_QUEUE, { localConcurrency: LAB_BIOMARKER_BACKFILL_CONCURRENCY }, async (jobs) => { @@ -724,12 +806,14 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); // v1.17.0 — Nightscout CGM poll-cohort sync. The hourly cron tick (no // `userId`) walks every configured instance; one user's unreachable host is // warned, not fatal. - await boss.work( + await createAndWork( + boss, NIGHTSCOUT_SYNC_QUEUE, { localConcurrency: 1 }, handleNightscoutSync, @@ -737,12 +821,14 @@ export async function registerIntegrationSyncQueues( // v1.17.0 (F4) — Polar + Oura OAuth poll-cohort sync. The hourly cron tick // (no `userId`) re-walks every connected user; one user's revoked grant is // warned, not fatal. - await boss.work( + await createAndWork( + boss, POLAR_SYNC_QUEUE, { localConcurrency: 1 }, handlePolarSync, ); - await boss.work( + await createAndWork( + boss, OURA_SYNC_QUEUE, { localConcurrency: 1 }, handleOuraSync, @@ -750,14 +836,16 @@ export async function registerIntegrationSyncQueues( // v1.28.x — Strava OAuth poll-cohort sync (poll-only; webhook deferred). The // hourly cron tick (no `userId`) re-walks every connected user; one user's // revoked grant is warned, not fatal. - await boss.work( + await createAndWork( + boss, STRAVA_SYNC_QUEUE, { localConcurrency: 1 }, handleStravaSync, ); // v1.28.x — self-converging Strava backfill. This source queue retains the // per-connection singleton and forwards admitted work to the shared drain. - await boss.work( + await createAndWork( + boss, STRAVA_BACKFILL_QUEUE, { localConcurrency: STRAVA_BACKFILL_CONCURRENCY }, async (jobs) => { @@ -767,6 +855,7 @@ export async function registerIntegrationSyncQueues( data: job.data, }); } + return jobDone({ jobs: jobs.length }); }, ); return allQueues; diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 921f1595e..1f4f327ad 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -10,7 +10,7 @@ * * 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 * `measurement-tombstone-cleanup`, `withings-oauth-state-cleanup`, * `intake-slot-dedup-queue`, `tls-pin-monitor-queue`, and `geo-backfill` * guards read THIS module. @@ -131,9 +131,12 @@ import { type EnvironmentFetchPayload, } from "@/lib/jobs/environment-fetch"; import { recordError } from "@/lib/jobs/worker-status"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { workerLog, BOOT_BACKFILL_STAGGER_SECONDS } from "./shared"; import { createAndSchedule, + createAndWork, + cronIsTheRetry, type QueuePolicyTable, type ScheduleEntry, } from "./registrar-shared"; @@ -411,24 +414,35 @@ const allQueues = [ DOCUMENT_SUMMARY_CATCHUP_QUEUE, ]; +// `cronIsTheRetry` (retryLimit 0) marks the passes whose handler now returns +// `jobFailed` where it used to warn and carry on, AND whose failure repeats: +// a bulk DELETE or UPDATE that timed out will time out again on an immediate +// retry, so pg-boss's default two would run the same doomed statement three +// times a night. The queues without it are the ones whose failure is +// transient — the TLS probe, the off-host upload, the restore drill, the +// per-user environment fetch — and those still want the default retries. const schedules: ScheduleEntry[] = [ // v1.25 (W-ENV) — daily 02:10 Europe/Berlin discovery tick (empty payload) // that fans out one per-user environment fetch per opted-in account. [ENVIRONMENT_FETCH_QUEUE, ENVIRONMENT_FETCH_CRON], [DATA_BACKUP_QUEUE, DATA_BACKUP_CRON], - [RATE_LIMIT_CLEANUP_QUEUE, RATE_LIMIT_CLEANUP_CRON], - [IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON], + [RATE_LIMIT_CLEANUP_QUEUE, RATE_LIMIT_CLEANUP_CRON, cronIsTheRetry], + [IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON, cronIsTheRetry], [AUDIT_LOG_CLEANUP_QUEUE, AUDIT_LOG_CLEANUP_CRON], - [STEP_UP_ELEVATION_CLEANUP_QUEUE, STEP_UP_ELEVATION_CLEANUP_CRON], + [ + STEP_UP_ELEVATION_CLEANUP_QUEUE, + STEP_UP_ELEVATION_CLEANUP_CRON, + cronIsTheRetry, + ], [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON], [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON], - [HOST_METRIC_QUEUE, HOST_METRIC_CRON], - [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON], + [HOST_METRIC_QUEUE, HOST_METRIC_CRON, cronIsTheRetry], + [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON, cronIsTheRetry], // v1.4.37 — hourly geo backfill. The helper is idempotent + capped // at 5 000 rows per pass; running it at :40 every hour catches the // long tail of audit rows that landed with the offline MMDB // missing or the online provider unreachable. - [GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON], + [GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON, cronIsTheRetry], // v1.12.2 ios-coord — every-6-hour TLS leaf SPKI probe (:07 off the // hourly sync crons). Surfaces a pinned-leaf rotation well inside the // ≥11-day re-pin window the iOS release owner needs. @@ -443,14 +457,18 @@ const schedules: ScheduleEntry[] = [ // entities. Flips IN_USE rows whose 30-day clock has blown to // EXPIRED at 03:30 Europe/Berlin (in the existing 02:xx–03:xx // maintenance window, right after idempotency-cleanup). - [MEDICATION_INVENTORY_EXPIRE_QUEUE, MEDICATION_INVENTORY_EXPIRE_CRON], + [ + MEDICATION_INVENTORY_EXPIRE_QUEUE, + MEDICATION_INVENTORY_EXPIRE_CRON, + cronIsTheRetry, + ], // v1.4.46 — hourly auto-skip for medication intakes the user never // marked. Cron `5 * * * *` slots off the top-of-the-hour // reminder-check (:00) so the hourly ticks don't pile up on the same // boss poll. - [INTAKE_AUTO_SKIP_QUEUE, INTAKE_AUTO_SKIP_CRON], + [INTAKE_AUTO_SKIP_QUEUE, INTAKE_AUTO_SKIP_CRON, cronIsTheRetry], // v0.5.4 ios-coord — daily mood-reminder dispatch-ledger retention sweep. - [MOOD_REMINDER_CLEANUP_QUEUE, MOOD_REMINDER_CLEANUP_CRON], + [MOOD_REMINDER_CLEANUP_QUEUE, MOOD_REMINDER_CLEANUP_CRON, cronIsTheRetry], // v1.15.19 — daily 03:28 Europe/Berlin duplicate dose-slot dedup // discovery. The empty cron payload (no `userId`) is the handler's // signal to run the discovery fan-out instead of a per-user pass, so @@ -458,20 +476,24 @@ const schedules: ScheduleEntry[] = [ // a day instead of waiting for the next worker reboot. [INTAKE_SLOT_DEDUP_QUEUE, INTAKE_SLOT_DEDUP_CRON], // v1.4.49 — daily 03:35 Europe/Berlin prune for push_attempts. - [PUSH_ATTEMPT_CLEANUP_QUEUE, PUSH_ATTEMPT_CLEANUP_CRON], + [PUSH_ATTEMPT_CLEANUP_QUEUE, PUSH_ATTEMPT_CLEANUP_CRON, cronIsTheRetry], // v1.31.0 — daily 04:10 Europe/Berlin prune for arrival_reactions. - [ARRIVAL_REACTION_CLEANUP_QUEUE, ARRIVAL_REACTION_CLEANUP_CRON], + [ + ARRIVAL_REACTION_CLEANUP_QUEUE, + ARRIVAL_REACTION_CLEANUP_CRON, + cronIsTheRetry, + ], // v1.7.0 — daily 03:40 Europe/Berlin prune for expired measurement // tombstones. [MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE, MEASUREMENT_TOMBSTONE_CLEANUP_CRON], // v1.18.7 — daily 03:50 Europe/Berlin prune for stale Coach history. - [COACH_MESSAGE_CLEANUP_QUEUE, COACH_MESSAGE_CLEANUP_CRON], + [COACH_MESSAGE_CLEANUP_QUEUE, COACH_MESSAGE_CLEANUP_CRON, cronIsTheRetry], // v1.23 — daily 03:55 Europe/Berlin note-encryption backfill discovery. // The empty cron payload (no `userId`) is the handler's signal to fan out // one per-user job per account still holding a plaintext note. [NOTE_ENCRYPTION_BACKFILL_QUEUE, NOTE_ENCRYPTION_BACKFILL_CRON], // v1.24 — daily 04:00 Europe/Berlin prune for dead MCP connector tokens. - [MCP_TOKEN_CLEANUP_QUEUE, MCP_TOKEN_CLEANUP_CRON], + [MCP_TOKEN_CLEANUP_QUEUE, MCP_TOKEN_CLEANUP_CRON, cronIsTheRetry], // v1.25 — daily 04:05 Europe/Berlin medication-note encryption backfill // discovery. The empty cron payload (no `userId`) is the handler's signal to // fan out one per-user job per account still holding a plaintext medication @@ -479,14 +501,14 @@ const schedules: ScheduleEntry[] = [ [MED_NOTES_ENCRYPTION_BACKFILL_QUEUE, MED_NOTES_ENCRYPTION_BACKFILL_CRON], // Document vault — daily 04:10 Europe/Berlin purge for tombstoned // documents past the 30-day undo grace. - [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON], + [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON, cronIsTheRetry], // v1.32.1 (issue #588) — every-15-minute orphan-ImportJob sweep. Re-runs // the same reconcile the boot path uses, so a stuck "unpacking" row // whose worker crashed/restarted without the boot-time pass catching it // (heartbeat not yet stale, pg-boss job not yet expired) still converges // to a visible `failed` state within two ticks instead of staying stuck // forever on an otherwise-healthy worker. - [IMPORT_JOB_RECONCILE_QUEUE, IMPORT_JOB_RECONCILE_CRON], + [IMPORT_JOB_RECONCILE_QUEUE, IMPORT_JOB_RECONCILE_CRON, cronIsTheRetry], ]; /** @@ -584,61 +606,76 @@ export async function registerMaintenanceQueues( ): Promise { await createAndSchedule(boss, allQueues, schedules, queuePolicies); - await boss.work( + await createAndWork( + boss, DATA_BACKUP_QUEUE, { localConcurrency: 1 }, handleDataBackup, ); - await boss.work( + await createAndWork( + boss, RATE_LIMIT_CLEANUP_QUEUE, { localConcurrency: 1 }, handleRateLimitCleanup, ); - await boss.work( + await createAndWork( + boss, IDEMPOTENCY_CLEANUP_QUEUE, { localConcurrency: 1 }, handleIdempotencyCleanup, ); - await boss.work( + await createAndWork( + boss, AUDIT_LOG_CLEANUP_QUEUE, { localConcurrency: 1 }, handleAuditLogCleanup, ); - await boss.work( + await createAndWork( + boss, STEP_UP_ELEVATION_CLEANUP_QUEUE, { localConcurrency: 1 }, handleStepUpElevationCleanup, ); - await boss.work( + await createAndWork( + boss, OFFHOST_BACKUP_QUEUE, { localConcurrency: 1 }, handleOffhostBackup, ); - // prettier-ignore - await boss.work(RESTORE_DRILL_QUEUE, { localConcurrency: 1 }, handleRestoreDrill); - await boss.work( + await createAndWork( + boss, + RESTORE_DRILL_QUEUE, + { localConcurrency: 1 }, + handleRestoreDrill, + ); + await createAndWork( + boss, HOST_METRIC_QUEUE, { localConcurrency: 1 }, handleHostMetricSample, ); - await boss.work( + await createAndWork( + boss, FEEDBACK_AGGREGATOR_QUEUE, { localConcurrency: 1 }, handleFeedbackAggregator, ); - await boss.work( + await createAndWork( + boss, GEO_BACKFILL_QUEUE, { localConcurrency: 1 }, handleGeoBackfill, ); // v1.12.2 ios-coord — TLS leaf SPKI-change monitor. Single-flight: one // short outbound TLS handshake per tick, no benefit to overlapping ticks. - await boss.work( + await createAndWork( + boss, TLS_PIN_MONITOR_QUEUE, { localConcurrency: 1 }, handleTlsPinMonitor, ); - await boss.work( + await createAndWork( + boss, MOOD_REMINDER_CLEANUP_QUEUE, { localConcurrency: 1 }, handleMoodReminderCleanup, @@ -647,7 +684,8 @@ export async function registerMaintenanceQueues( // matches every other cleanup queue; two ticks racing on the same // DELETE statement is wasted work and the second tick's payload // would be a no-op anyway. - await boss.work( + await createAndWork( + boss, PUSH_ATTEMPT_CLEANUP_QUEUE, { localConcurrency: 1 }, handlePushAttemptCleanup, @@ -655,14 +693,16 @@ export async function registerMaintenanceQueues( // v1.31.0 — daily prune of the data-arrival reaction markers. Single-flight // like every other cleanup queue: two ticks racing the same DELETE is // wasted work. - await boss.work( + await createAndWork( + boss, ARRIVAL_REACTION_CLEANUP_QUEUE, { localConcurrency: 1 }, handleArrivalReactionCleanup, ); // v1.7.0 — daily prune of expired measurement tombstones. Single-flight // like every other cleanup queue. - await boss.work( + await createAndWork( + boss, MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE, { localConcurrency: 1 }, handleMeasurementTombstoneCleanup, @@ -670,14 +710,16 @@ export async function registerMaintenanceQueues( // v1.18.7 — daily prune of the encrypted Coach conversation history. // Single-flight like every other cleanup queue; two ticks racing on the // same DELETE is wasted work and the second is a no-op. - await boss.work( + await createAndWork( + boss, COACH_MESSAGE_CLEANUP_QUEUE, { localConcurrency: 1 }, handleCoachMessageCleanup, ); // MCP Phase 3 (M2) — daily prune of dead MCP connector access tokens + // long-revoked connection anchors. Single-flight like every other cleanup. - await boss.work( + await createAndWork( + boss, MCP_TOKEN_CLEANUP_QUEUE, { localConcurrency: 1 }, handleMcpTokenCleanup, @@ -685,17 +727,20 @@ export async function registerMaintenanceQueues( // Document vault — daily purge of tombstoned documents past the 30-day // undo grace. Single-flight like every other cleanup queue; the underlying // deleteMany is idempotent so a duplicate tick is a no-op. - await boss.work( + await createAndWork( + boss, DOCUMENT_PURGE_QUEUE, { localConcurrency: 1 }, handleDocumentPurge, ); - await boss.work( + await createAndWork( + boss, PR_DETECTION_QUEUE, { localConcurrency: PR_DETECTION_CONCURRENCY }, handlePrDetection, ); - await boss.work( + await createAndWork( + boss, MEDICATION_INVENTORY_EXPIRE_QUEUE, { localConcurrency: 1 }, handleMedicationInventoryExpire, @@ -704,7 +749,8 @@ export async function registerMaintenanceQueues( // Single-flight: two ticks racing against the same row pile is wasted // work, and the underlying `updateMany` is the canonical idempotent // shape anyway. - await boss.work( + await createAndWork( + boss, INTAKE_AUTO_SKIP_QUEUE, { localConcurrency: 1 }, handleIntakeAutoSkip, @@ -712,48 +758,64 @@ export async function registerMaintenanceQueues( // v1.4.34 — Apple Health export.zip ingest worker. localConcurrency // caps at 1 because the parse loop is CPU-bound and a concurrent // second 1 GB import would race the first for RSS. - await boss.work( + await createAndWork( + boss, APPLE_HEALTH_IMPORT_V2_QUEUE, { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY }, async (jobs) => { // pg-boss v12 work callbacks always receive an array (batched // worker mode); for our concurrency-1 + batchSize-1 case we just // process each job sequentially. + // + // Each import is independent work, so a failed one does not abort the + // rest of the batch; the first failure is what the job reports. + let failure: JobOutcome | null = null; for (const job of jobs) { - await handleAppleHealthImport(job); + const outcome = await handleAppleHealthImport(job); + if (!outcome.ok && failure === null) failure = outcome; } + return failure ?? jobDone({ jobs: jobs.length }); }, ); // Drain pre-revision-2 backlog by moving each claimed legacy job onto the // isolated v2 queue. This preserves the ImportJob status id without letting // the current parser execute under a revision-1 mirror. - await boss.work( + await createAndWork( + boss, APPLE_HEALTH_IMPORT_LEGACY_QUEUE, { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY }, async (jobs) => { for (const job of jobs) { await migrateLegacyAppleHealthImport(job); } + return jobDone({ jobs: jobs.length }); }, ); // v1.32.1 (issue #588) — periodic orphan-ImportJob sweep. Single-flight: // the underlying `updateMany` is idempotent and two ticks racing the same // read-then-write pass would just be wasted work. - await boss.work( + await createAndWork( + boss, IMPORT_JOB_RECONCILE_QUEUE, { localConcurrency: 1 }, handleImportJobReconcileTick, ); - await boss.work( + await createAndWork( + boss, MEDICATION_INTAKE_IMPORT_QUEUE, { localConcurrency: MEDICATION_INTAKE_IMPORT_CONCURRENCY, includeMetadata: true, }, async (jobs) => { + // Independent per-delivery imports: run them all, report the first + // failure rather than losing it behind a later success. + let failure: JobOutcome | null = null; for (const job of jobs) { - await handleMedicationIntakeImport(job); + const outcome = await handleMedicationIntakeImport(job); + if (!outcome.ok && failure === null) failure = outcome; } + return failure ?? jobDone({ jobs: jobs.length }); }, ); @@ -768,10 +830,21 @@ export async function registerMaintenanceQueues( // above) whose payload omits `userId`. That tick runs the SAME discovery // fan-out the boot pass uses, so duplicate slots created between deploys // collapse within a day instead of waiting for the next worker reboot. - await boss.work>( + await createAndWork>( + boss, INTAKE_SLOT_DEDUP_QUEUE, { localConcurrency: INTAKE_SLOT_DEDUP_CONCURRENCY }, async (jobs) => { + let users = 0; + let discoveryEnqueued = 0; + let discoverySkipped = 0; + // The discovery helper returns its error instead of throwing, so a + // failed fan-out would otherwise leave no trace on the job at all. + let discoveryFailed = 0; + let slotsCollapsed = 0; + let rowsSoftDeleted = 0; + let rowsNormalised = 0; + let daysRecomputed = 0; for (const job of jobs) { const { userId } = job.data; if (!userId) { @@ -779,6 +852,9 @@ export async function registerMaintenanceQueues( // still holding duplicate-slot candidates (singletonKey-coalesced, // identical to the boot pass). const result = await enqueueBootTimeIntakeSlotDedup(); + discoveryEnqueued += result.enqueued; + discoverySkipped += result.skipped; + if (result.error) discoveryFailed++; workerLog( "info", `[intake-slot-dedup] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`, @@ -787,6 +863,11 @@ export async function registerMaintenanceQueues( } try { const summary = await dedupeUserIntakeSlots(userId); + users++; + slotsCollapsed += summary.slotsCollapsed; + rowsSoftDeleted += summary.rowsSoftDeleted; + rowsNormalised += summary.rowsNormalised; + daysRecomputed += summary.daysRecomputed; workerLog( "info", `[intake-slot-dedup] user=${userId} slotsCollapsed=${summary.slotsCollapsed} rowsSoftDeleted=${summary.rowsSoftDeleted} rowsNormalised=${summary.rowsNormalised} daysRecomputed=${summary.daysRecomputed}`, @@ -797,6 +878,17 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ + jobs: jobs.length, + users, + discovery_enqueued: discoveryEnqueued, + discovery_skipped: discoverySkipped, + discovery_failed: discoveryFailed, + slots_collapsed: slotsCollapsed, + rows_soft_deleted: rowsSoftDeleted, + rows_normalised: rowsNormalised, + days_recomputed: daysRecomputed, + }); }, ); @@ -806,14 +898,26 @@ export async function registerMaintenanceQueues( // empty-userId payload is the daily discovery tick and re-fans the per-user // jobs (singletonKey-coalesced, identical to the boot pass). Serial // concurrency so the migration never crowds the request pool. - await boss.work>( + await createAndWork>( + boss, NOTE_ENCRYPTION_BACKFILL_QUEUE, { localConcurrency: NOTE_ENCRYPTION_BACKFILL_CONCURRENCY }, async (jobs) => { + let users = 0; + let discoveryEnqueued = 0; + let discoverySkipped = 0; + // The discovery helper returns its error instead of throwing, so a + // failed fan-out would otherwise leave no trace on the job at all. + let discoveryFailed = 0; + let measurements = 0; + let moodEntries = 0; for (const job of jobs) { const { userId } = job.data; if (!userId) { const result = await enqueueBootTimeNoteEncryptionBackfill(); + discoveryEnqueued += result.enqueued; + discoverySkipped += result.skipped; + if (result.error) discoveryFailed++; workerLog( "info", `[note-encryption-backfill] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`, @@ -823,6 +927,9 @@ export async function registerMaintenanceQueues( try { const { measurementsMigrated, moodEntriesMigrated } = await runNoteEncryptionBackfillForUser(userId); + users++; + measurements += measurementsMigrated; + moodEntries += moodEntriesMigrated; workerLog( "info", `[note-encryption-backfill] user=${userId} measurements=${measurementsMigrated} moodEntries=${moodEntriesMigrated}`, @@ -837,11 +944,21 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ + jobs: jobs.length, + users, + discovery_enqueued: discoveryEnqueued, + discovery_skipped: discoverySkipped, + discovery_failed: discoveryFailed, + measurements_migrated: measurements, + mood_entries_migrated: moodEntries, + }); }, ); // v1.23 — admin-triggered encryption-key rotation (on-demand, singleton). - await boss.work( + await createAndWork( + boss, ENCRYPTION_KEY_ROTATE_QUEUE, { localConcurrency: ENCRYPTION_KEY_ROTATE_CONCURRENCY }, handleEncryptionKeyRotate, @@ -854,14 +971,27 @@ export async function registerMaintenanceQueues( // payload is the daily discovery tick and re-fans the per-user jobs // (singletonKey-coalesced, identical to the boot pass). Serial concurrency // so the migration never crowds the request pool. - await boss.work>( + await createAndWork>( + boss, MED_NOTES_ENCRYPTION_BACKFILL_QUEUE, { localConcurrency: MED_NOTES_ENCRYPTION_BACKFILL_CONCURRENCY }, async (jobs) => { + let users = 0; + let discoveryEnqueued = 0; + let discoverySkipped = 0; + // The discovery helper returns its error instead of throwing, so a + // failed fan-out would otherwise leave no trace on the job at all. + let discoveryFailed = 0; + let sideEffects = 0; + let doseChanges = 0; + let inventoryItems = 0; for (const job of jobs) { const { userId } = job.data; if (!userId) { const result = await enqueueBootTimeMedNotesEncryptionBackfill(); + discoveryEnqueued += result.enqueued; + discoverySkipped += result.skipped; + if (result.error) discoveryFailed++; workerLog( "info", `[med-notes-encryption-backfill] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`, @@ -874,6 +1004,10 @@ export async function registerMaintenanceQueues( doseChangesMigrated, inventoryItemsMigrated, } = await runMedNotesEncryptionBackfillForUser(userId); + users++; + sideEffects += sideEffectsMigrated; + doseChanges += doseChangesMigrated; + inventoryItems += inventoryItemsMigrated; workerLog( "info", `[med-notes-encryption-backfill] user=${userId} sideEffects=${sideEffectsMigrated} doseChanges=${doseChangesMigrated} inventoryItems=${inventoryItemsMigrated}`, @@ -888,6 +1022,16 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ + jobs: jobs.length, + users, + discovery_enqueued: discoveryEnqueued, + discovery_skipped: discoverySkipped, + discovery_failed: discoveryFailed, + side_effects_migrated: sideEffects, + dose_changes_migrated: doseChanges, + inventory_items_migrated: inventoryItems, + }); }, ); @@ -895,16 +1039,25 @@ export async function registerMaintenanceQueues( // trigger endpoint sends one per-user job; this handler indexes that user's // not-yet-indexed documents (consent + budget gated, bounded + resumable). // Serial concurrency so the provider calls never crowd the request pool. - await boss.work( + await createAndWork( + boss, CONTENT_INDEX_BACKFILL_QUEUE, { localConcurrency: CONTENT_INDEX_BACKFILL_CONCURRENCY }, async (jobs) => { + let users = 0; + let skipped = 0; + let documentsIndexed = 0; for (const job of jobs) { const { userId } = job.data; - if (!userId) continue; + if (!userId) { + skipped++; + continue; + } try { const { indexed, reason } = await runContentIndexBackfillForUser(userId); + users++; + documentsIndexed += indexed; workerLog( "info", `[document-content-index-backfill] user=${userId} indexed=${indexed} reason=${reason}`, @@ -919,6 +1072,12 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ + jobs: jobs.length, + users, + skipped, + documents_indexed: documentsIndexed, + }); }, ); @@ -926,15 +1085,22 @@ export async function registerMaintenanceQueues( // (one job per stored document); provider-first (vision) when configured + // consented, else local text-layer extraction. Serial concurrency so the // provider calls + PDF parse never crowd the request pool. - await boss.work( + await createAndWork( + boss, DOCUMENT_INDEX_QUEUE, { localConcurrency: DOCUMENT_INDEX_CONCURRENCY }, async (jobs) => { + let documents = 0; + let skipped = 0; for (const job of jobs) { const { userId, documentId } = job.data; - if (!userId || !documentId) continue; + if (!userId || !documentId) { + skipped++; + continue; + } try { await runDocumentIndex(job.data); + documents++; } catch (err) { recordError(); workerLog( @@ -945,6 +1111,7 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ jobs: jobs.length, documents, skipped }); }, ); @@ -952,15 +1119,22 @@ export async function registerMaintenanceQueues( // upload (one job per stored document); decodes + downscales the original to // a small encrypted JPEG. Pure local compute; serial concurrency so the // canvas/pdfjs decode never crowds the request pool. - await boss.work( + await createAndWork( + boss, DOCUMENT_THUMBNAIL_QUEUE, { localConcurrency: DOCUMENT_THUMBNAIL_CONCURRENCY }, async (jobs) => { + let documents = 0; + let skipped = 0; for (const job of jobs) { const { userId, documentId } = job.data; - if (!userId || !documentId) continue; + if (!userId || !documentId) { + skipped++; + continue; + } try { await runDocumentThumbnail(job.data); + documents++; } catch (err) { recordError(); workerLog( @@ -971,6 +1145,7 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ jobs: jobs.length, documents, skipped }); }, ); @@ -979,15 +1154,22 @@ export async function registerMaintenanceQueues( // the `documentsAutoAiRead` opt-in is ON (egress consent + budget gated) and // persists the summary encrypted. Serial concurrency so the provider call // never crowds the request pool. - await boss.work( + await createAndWork( + boss, DOCUMENT_SUMMARY_QUEUE, { localConcurrency: DOCUMENT_SUMMARY_CONCURRENCY }, async (jobs) => { + let documents = 0; + let skipped = 0; for (const job of jobs) { const { userId, documentId } = job.data; - if (!userId || !documentId) continue; + if (!userId || !documentId) { + skipped++; + continue; + } try { await runDocumentSummaryJob(job.data); + documents++; } catch (err) { recordError(); workerLog( @@ -998,6 +1180,7 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ jobs: jobs.length, documents, skipped }); }, ); @@ -1007,15 +1190,24 @@ export async function registerMaintenanceQueues( // documents (the upload-time enqueue no-opped while the flag was OFF). It // only enqueues — the opt-in, egress consent and budget gates all still run // per document inside the summary job. Serial concurrency. - await boss.work( + await createAndWork( + boss, DOCUMENT_SUMMARY_CATCHUP_QUEUE, { localConcurrency: DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY }, async (jobs) => { + let users = 0; + let skipped = 0; + let summariesEnqueued = 0; for (const job of jobs) { const { userId } = job.data; - if (!userId) continue; + if (!userId) { + skipped++; + continue; + } try { - await runSummaryCatchUpForUser(userId); + const { enqueued } = await runSummaryCatchUpForUser(userId); + users++; + summariesEnqueued += enqueued; } catch (err) { recordError(); workerLog( @@ -1026,6 +1218,12 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ + jobs: jobs.length, + users, + skipped, + summaries_enqueued: summariesEnqueued, + }); }, ); @@ -1033,15 +1231,24 @@ export async function registerMaintenanceQueues( // discovery sends one per-user job; this handler fans out per-document // thumbnail jobs for that user's not-yet-thumbnailed documents. Serial // concurrency; cheap discovery + enqueue only. - await boss.work( + await createAndWork( + boss, DOCUMENT_THUMBNAIL_BACKFILL_QUEUE, { localConcurrency: DOCUMENT_THUMBNAIL_BACKFILL_CONCURRENCY }, async (jobs) => { + let users = 0; + let skipped = 0; + let thumbnailsEnqueued = 0; for (const job of jobs) { const { userId } = job.data; - if (!userId) continue; + if (!userId) { + skipped++; + continue; + } try { const { enqueued } = await runThumbnailBackfillForUser(userId); + users++; + thumbnailsEnqueued += enqueued; workerLog( "info", `[document-thumbnail-backfill] user=${userId} enqueued=${enqueued}`, @@ -1056,6 +1263,12 @@ export async function registerMaintenanceQueues( throw err; } } + return jobDone({ + jobs: jobs.length, + users, + skipped, + thumbnails_enqueued: thumbnailsEnqueued, + }); }, ); @@ -1063,13 +1276,20 @@ export async function registerMaintenanceQueues( // payload) fans out one per-user job per opted-in account; the queue also // serves the on-demand backfill payloads from the settings surface. Serial // concurrency so the staggered outbound fetches never crowd the request pool. - await boss.work( + await createAndWork( + boss, ENVIRONMENT_FETCH_QUEUE, { localConcurrency: ENVIRONMENT_FETCH_CONCURRENCY }, async (jobs) => { + // A discovery tick and a per-user backfill share this queue and are + // independent of each other, so the pass runs to the end and reports + // the first failure it saw. + let failure: JobOutcome | null = null; for (const job of jobs) { - await handleEnvironmentFetch(boss, job.data); + const outcome = await handleEnvironmentFetch(boss, job.data); + if (!outcome.ok && failure === null) failure = outcome; } + return failure ?? jobDone({ jobs: jobs.length }); }, ); diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index 1b9a0ae01..eca1bf4ff 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -10,7 +10,7 @@ * * 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 * `drain-cumulative-queue`, `mean-consolidation-queue`, and * `stress-strain-retention-queue` (dense-retention facet) guards read THIS * module. @@ -104,9 +104,12 @@ import { } from "./shared"; import { createAndSchedule, + createAndWork, + cronIsTheRetry, type QueuePolicyTable, type ScheduleEntry, } from "./registrar-shared"; +import { jobDone, jobFailed } from "@/lib/jobs/job-outcome"; // v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative rows. // Collapses each user × cumulative-type × calendar-day bucket into one // `stats:…` row so the list view stops painting hundreds of step chunks per @@ -170,7 +173,7 @@ const schedules: ScheduleEntry[] = [ // v1.4.37 W7c — nightly fold of per-sample APPLE_HEALTH cumulative // rows into one row per day per type. Slots between the // audit-log cleanup (03:15) and the feedback aggregator (04:00). - [DRAIN_CUMULATIVE_QUEUE, DRAIN_CUMULATIVE_CRON], + [DRAIN_CUMULATIVE_QUEUE, DRAIN_CUMULATIVE_CRON, cronIsTheRetry], ]; /** @@ -274,7 +277,8 @@ export async function registerRollupQueues( // the DAY bucket is already recomputed synchronously by the hook // itself. Concurrency-2 keeps two recomputes in flight without // crowding the dashboard request pool. - await boss.work( + await createAndWork( + boss, ROLLUP_RECOMPUTE_QUEUE, { localConcurrency: ROLLUP_RECOMPUTE_CONCURRENCY }, async (jobs) => { @@ -287,6 +291,10 @@ export async function registerRollupQueues( to: new Date(payload.to), }); } + // No per-bucket isolation here on purpose: a recompute that throws + // fails the job, so reaching this line means every bucket in the batch + // was folded. + return jobDone({ buckets_recomputed: jobs.length }); }, ); @@ -295,18 +303,25 @@ export async function registerRollupQueues( // full `recomputeUserRollups` against the default 5-year window // across every granularity. Serial concurrency so the populator // never crowds the dashboard request pool. - await boss.work( + await createAndWork( + boss, ROLLUP_FULL_BACKFILL_QUEUE, { localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY }, async (jobs) => { + let rowsUpsertedTotal = 0; for (const job of jobs) { const { userId } = job.data; const { rowsUpserted, durationMs } = await recomputeUserRollups(userId); + rowsUpsertedTotal += rowsUpserted; workerLog( "info", `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, ); } + return jobDone({ + users: jobs.length, + rows_upserted: rowsUpsertedTotal, + }); }, ); @@ -315,19 +330,29 @@ export async function registerRollupQueues( // step rows; this handler collapses them into one daily-total row per // calendar day and soft-deletes the originals. Serial concurrency so // the populator never crowds the dashboard request pool. - await boss.work( + await createAndWork( + boss, STEP_CONSOLIDATION_QUEUE, { localConcurrency: STEP_CONSOLIDATION_CONCURRENCY }, async (jobs) => { + let daysTotal = 0; + let legacyRowsTotal = 0; for (const job of jobs) { const { userId } = job.data; const { daysConsolidated, legacyRowsSoftDeleted } = await runStepConsolidationForUser(userId); + daysTotal += daysConsolidated; + legacyRowsTotal += legacyRowsSoftDeleted; workerLog( "info", `[step-consolidation] user=${userId} days=${daysConsolidated} legacyRowsSoftDeleted=${legacyRowsSoftDeleted}`, ); } + return jobDone({ + users: jobs.length, + days_consolidated: daysTotal, + legacy_rows_soft_deleted: legacyRowsTotal, + }); }, ); @@ -335,18 +360,39 @@ export async function registerRollupQueues( // GOOGLE_HEALTH / FITBIT daily-total step rows the pre-fix consolidation // swept and removes the shadow MANUAL totals it minted. Serial // concurrency so the repair never crowds the dashboard request pool. - await boss.work( + await createAndWork( + boss, STEP_CONSOLIDATION_REPAIR_QUEUE, { localConcurrency: STEP_CONSOLIDATION_REPAIR_CONCURRENCY }, async (jobs) => { + let rowsResurrected = 0; + let wedgeSkipped = 0; + let manualMintsRemoved = 0; + let daysRecomputed = 0; + let failures = 0; for (const job of jobs) { const { userId } = job.data; const summary = await runStepConsolidationRepairForUser(userId); + rowsResurrected += summary.rowsResurrected; + wedgeSkipped += summary.wedgeSkipped; + manualMintsRemoved += summary.manualMintsRemoved; + daysRecomputed += summary.daysRecomputed; + // The repair isolates its own per-day failures and reports them as a + // count, so they ride out as a fact rather than failing the job. + failures += summary.failures; workerLog( "info", `[step-consolidation-repair] user=${userId} resurrected=${summary.rowsResurrected} wedgeSkipped=${summary.wedgeSkipped} manualMintsRemoved=${summary.manualMintsRemoved} daysRecomputed=${summary.daysRecomputed} failures=${summary.failures}`, ); } + return jobDone({ + users: jobs.length, + rows_resurrected: rowsResurrected, + wedge_skipped: wedgeSkipped, + manual_mints_removed: manualMintsRemoved, + days_recomputed: daysRecomputed, + failures, + }); }, ); @@ -356,18 +402,28 @@ export async function registerRollupQueues( // it and re-runs detection silently so the honest re-derived best takes // its place. Serial concurrency so the repair never crowds the // dashboard request pool. - await boss.work( + await createAndWork( + boss, CUMULATIVE_PR_REDERIVE_QUEUE, { localConcurrency: CUMULATIVE_PR_REDERIVE_CONCURRENCY }, async (jobs) => { + let rowsDeleted = 0; + let rowsReinserted = 0; for (const job of jobs) { const { userId } = job.data; const summary = await runCumulativePrRederivationForUser(userId); + rowsDeleted += summary.rowsDeleted; + rowsReinserted += summary.rowsReinserted; workerLog( "info", `[cumulative-pr-rederive] user=${userId} rowsDeleted=${summary.rowsDeleted} rowsReinserted=${summary.rowsReinserted}`, ); } + return jobDone({ + users: jobs.length, + rows_deleted: rowsDeleted, + rows_reinserted: rowsReinserted, + }); }, ); @@ -376,19 +432,29 @@ export async function registerRollupQueues( // mean-type rows; this handler collapses each completed day to its // mean and soft-deletes the originals. Serial concurrency so the // populator never crowds the dashboard request pool. - await boss.work( + await createAndWork( + boss, MEAN_CONSOLIDATION_QUEUE, { localConcurrency: MEAN_CONSOLIDATION_CONCURRENCY }, async (jobs) => { + let daysTotal = 0; + let perSampleRowsTotal = 0; for (const job of jobs) { const { userId } = job.data; const { daysConsolidated, perSampleRowsSoftDeleted } = await runMeanConsolidationForUser(userId); + daysTotal += daysConsolidated; + perSampleRowsTotal += perSampleRowsSoftDeleted; workerLog( "info", `[mean-consolidation] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted}`, ); } + return jobDone({ + users: jobs.length, + days_consolidated: daysTotal, + per_sample_rows_soft_deleted: perSampleRowsTotal, + }); }, ); @@ -399,10 +465,14 @@ export async function registerRollupQueues( // soft-deletes the originals, keeping the in-window intra-day shape intact // for the Stress engine. Serial concurrency so the backfill never crowds // the request pool. - await boss.work( + await createAndWork( + boss, DENSE_INTRADAY_RETENTION_QUEUE, { localConcurrency: DENSE_INTRADAY_RETENTION_CONCURRENCY }, async (jobs) => { + let daysTotal = 0; + let perSampleRowsTotal = 0; + let derivedRestingRowsTotal = 0; for (const job of jobs) { const { userId } = job.data; try { @@ -411,6 +481,9 @@ export async function registerRollupQueues( perSampleRowsSoftDeleted, derivedRestingRowsUpserted, } = await runDenseIntradayRetentionForUser(userId); + daysTotal += daysConsolidated; + perSampleRowsTotal += perSampleRowsSoftDeleted; + derivedRestingRowsTotal += derivedRestingRowsUpserted; workerLog( "info", `[dense-intraday-retention] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted} derivedRestingRowsUpserted=${derivedRestingRowsUpserted}`, @@ -425,6 +498,12 @@ export async function registerRollupQueues( throw err; } } + return jobDone({ + users: jobs.length, + days_consolidated: daysTotal, + per_sample_rows_soft_deleted: perSampleRowsTotal, + derived_resting_rows_upserted: derivedRestingRowsTotal, + }); }, ); @@ -435,10 +514,15 @@ export async function registerRollupQueues( // Self-converging: a rebuilt day drops out of the discovery pairing, so // the pass runs once per install. Serial concurrency so the rebuild never // crowds the request pool. - await boss.work( + await createAndWork( + boss, DENSE_INTRADAY_HOURLY_REBUILD_QUEUE, { localConcurrency: DENSE_INTRADAY_HOURLY_REBUILD_CONCURRENCY }, async (jobs) => { + let daysRebuiltTotal = 0; + let hourlyRowsTotal = 0; + let dailyRowsRetiredTotal = 0; + let daysSkippedTotal = 0; for (const job of jobs) { const { userId } = job.data; try { @@ -448,6 +532,10 @@ export async function registerRollupQueues( dailyRowsRetired, daysSkippedNoTombstones, } = await runDenseIntradayHourlyRebuildForUser(userId); + daysRebuiltTotal += daysRebuilt; + hourlyRowsTotal += hourlyRowsUpserted; + dailyRowsRetiredTotal += dailyRowsRetired; + daysSkippedTotal += daysSkippedNoTombstones; workerLog( "info", `[dense-intraday-hourly-rebuild] user=${userId} daysRebuilt=${daysRebuilt} hourlyRowsUpserted=${hourlyRowsUpserted} dailyRowsRetired=${dailyRowsRetired} daysSkippedNoTombstones=${daysSkippedNoTombstones}`, @@ -462,6 +550,13 @@ export async function registerRollupQueues( throw err; } } + return jobDone({ + users: jobs.length, + days_rebuilt: daysRebuiltTotal, + hourly_rows_upserted: hourlyRowsTotal, + daily_rows_retired: dailyRowsRetiredTotal, + days_skipped_no_tombstones: daysSkippedTotal, + }); }, ); @@ -471,7 +566,8 @@ export async function registerRollupQueues( // No current read path consumes these buckets — they exist so a // future cross-granularity reader can ship without a backfill // step. Concurrency-2 mirrors the measurement-rollup worker. - await boss.work( + await createAndWork( + boss, MOOD_ROLLUP_RECOMPUTE_QUEUE, { localConcurrency: MOOD_ROLLUP_RECOMPUTE_CONCURRENCY }, async (jobs) => { @@ -483,6 +579,7 @@ export async function registerRollupQueues( to: new Date(payload.to), }); } + return jobDone({ buckets_recomputed: jobs.length }); }, ); @@ -491,10 +588,12 @@ export async function registerRollupQueues( // but zero rollup rows; this handler folds the full 5-year window // across every granularity. Concurrency-1 so the populator never // crowds the request pool. - await boss.work( + await createAndWork( + boss, MOOD_ROLLUP_FULL_BACKFILL_QUEUE, { localConcurrency: MOOD_ROLLUP_FULL_BACKFILL_CONCURRENCY }, async (jobs) => { + let rowsUpsertedTotal = 0; for (const job of jobs) { const { userId } = job.data; try { @@ -508,6 +607,7 @@ export async function registerRollupQueues( userId, { replace: true }, ); + rowsUpsertedTotal += rowsUpserted; workerLog( "info", `[mood-rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, @@ -522,6 +622,10 @@ export async function registerRollupQueues( throw err; } } + return jobDone({ + users: jobs.length, + rows_upserted: rowsUpsertedTotal, + }); }, ); @@ -530,15 +634,18 @@ export async function registerRollupQueues( // but zero rollup coverage; this handler folds the trailing 90-day // window per account. Concurrency-1 so the populator never crowds // the request pool. - await boss.work( + await createAndWork( + boss, MEDICATION_COMPLIANCE_BACKFILL_QUEUE, { localConcurrency: MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY }, async (jobs) => { + let rowsUpsertedTotal = 0; for (const job of jobs) { const { userId } = job.data; try { const { rowsUpserted, durationMs } = await recomputeUserMedicationCompliance(userId); + rowsUpsertedTotal += rowsUpserted; workerLog( "info", `[medication-compliance-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, @@ -553,6 +660,10 @@ export async function registerRollupQueues( throw err; } } + return jobDone({ + users: jobs.length, + rows_upserted: rowsUpsertedTotal, + }); }, ); @@ -563,10 +674,26 @@ export async function registerRollupQueues( // Concurrency-1 so the drain never crowds the dashboard request pool // and a long backfill on the maintainer's account (300 k+ measurement rows) // stays a single sequential walk. - await boss.work( + await createAndWork( + boss, DRAIN_CUMULATIVE_QUEUE, { localConcurrency: 1 }, async (jobs) => { + // The three drains below are independent passes sharing one nightly + // tick, so one failing pass must not stop the other two from running. + // The first failure is held back and reported once the tick is over — + // that keeps the existing order and still fails the job, which is what + // the swallowed errors never did. + let firstFailure: { reason: string; cause: unknown } | undefined; + let passesFailed = 0; + let bucketsCollapsed = 0; + let perSampleRowsDeleted = 0; + let meanDaysConsolidated = 0; + let meanRowsSoftDeleted = 0; + let denseDaysConsolidated = 0; + let denseRowsSoftDeleted = 0; + let denseSkipped = false; + for (const job of jobs) { try { const summary = await drainPerSampleCumulative(getWorkerPrisma(), { @@ -574,6 +701,8 @@ export async function registerRollupQueues( cutoffHours: DRAIN_CUMULATIVE_CUTOFF_HOURS, log: (line) => workerLog("info", line), }); + bucketsCollapsed += summary.totals.bucketsCollapsed; + perSampleRowsDeleted += summary.totals.perSampleRowsDeleted; workerLog( "info", `[drain-cumulative] triggeredAt=${job.data.triggeredAt} usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted}`, @@ -581,6 +710,8 @@ export async function registerRollupQueues( } catch (err) { recordError(); await reportWorkerError("drain-cumulative", err); + passesFailed++; + firstFailure ??= { reason: "cumulative drain failed", cause: err }; } // v1.8.5 — fold the daily-MEAN drain onto the same nightly tick as @@ -600,6 +731,8 @@ export async function registerRollupQueues( cutoffHours: MEAN_CONSOLIDATION_CUTOFF_HOURS, log: (line) => workerLog("info", line), }); + meanDaysConsolidated += meanSummary.totals.daysConsolidated; + meanRowsSoftDeleted += meanSummary.totals.perSampleRowsSoftDeleted; workerLog( "info", `[mean-consolidation] triggeredAt=${job.data.triggeredAt} usersScanned=${meanSummary.totals.usersScanned} daysConsolidated=${meanSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${meanSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${meanSummary.totals.dailyRowsUpserted} daysFailed=${meanSummary.totals.daysFailed}`, @@ -609,6 +742,8 @@ export async function registerRollupQueues( await reportWorkerError("mean-consolidation", err, { tick: "nightly", }); + passesFailed++; + firstFailure ??= { reason: "mean consolidation failed", cause: err }; } // v1.10.0 WX-E — fold the dense intra-day retention drain onto the @@ -624,6 +759,7 @@ export async function registerRollupQueues( // shipped. try { if (!DENSE_INTRADAY_RETENTION_ENABLED) { + denseSkipped = true; workerLog( "info", "[dense-intraday-retention] disabled via DENSE_INTRADAY_RETENTION_ENABLED — skipping the nightly walk (operator kill-switch)", @@ -636,6 +772,9 @@ export async function registerRollupQueues( log: (line) => workerLog("info", line), }, ); + denseDaysConsolidated += denseSummary.totals.daysConsolidated; + denseRowsSoftDeleted += + denseSummary.totals.perSampleRowsSoftDeleted; workerLog( "info", `[dense-intraday-retention] triggeredAt=${job.data.triggeredAt} usersScanned=${denseSummary.totals.usersScanned} daysConsolidated=${denseSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${denseSummary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${denseSummary.totals.hourlyRowsUpserted} dailyRowsRetired=${denseSummary.totals.dailyRowsRetired}`, @@ -646,8 +785,26 @@ export async function registerRollupQueues( await reportWorkerError("dense-intraday-retention", err, { tick: "nightly", }); + passesFailed++; + firstFailure ??= { reason: "dense retention failed", cause: err }; } } + + const did = { + passes_failed: passesFailed, + buckets_collapsed: bucketsCollapsed, + per_sample_rows_deleted: perSampleRowsDeleted, + mean_days_consolidated: meanDaysConsolidated, + mean_rows_soft_deleted: meanRowsSoftDeleted, + dense_days_consolidated: denseDaysConsolidated, + dense_rows_soft_deleted: denseRowsSoftDeleted, + dense_skipped: denseSkipped, + }; + + if (firstFailure) { + return jobFailed(firstFailure.reason, firstFailure.cause, did); + } + return jobDone(did); }, ); diff --git a/src/lib/jobs/reminder/register-status.ts b/src/lib/jobs/reminder/register-status.ts index 5e2efa814..9ebca0bc2 100644 --- a/src/lib/jobs/reminder/register-status.ts +++ b/src/lib/jobs/reminder/register-status.ts @@ -10,8 +10,8 @@ * * 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 status-wiring - * guards (`status-cron-candidates`, `insight-pregenerate`, + * and a `createAndWork(boss, QUEUE, …, handler)` binding drains it. The + * status-wiring guards (`status-cron-candidates`, `insight-pregenerate`, * `insight-status-generate`, `recovery-score-queue`, * `stress-strain-retention-queue`, `period-narrative-warm`, * `coach-memory-refresh-queue`) read THIS module. @@ -101,9 +101,11 @@ import { type InsightStatusGeneratePayload, } from "@/lib/jobs/insight-status-generate"; import { withBackgroundEvent } from "@/lib/logging/background"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { getWorkerPrisma, workerLog } from "./shared"; import { createAndSchedule, + createAndWork, insightRetryOptions, type QueuePolicyTable, type ScheduleEntry, @@ -363,37 +365,44 @@ export async function registerStatusQueues( ): Promise { await createAndSchedule(boss, allQueues, schedules, queuePolicies); - await boss.work( + await createAndWork( + boss, GENERAL_STATUS_QUEUE, { localConcurrency: 1 }, handleGeneralStatusGenerate, ); - await boss.work( + await createAndWork( + boss, BLOOD_PRESSURE_STATUS_QUEUE, { localConcurrency: 1 }, handleBloodPressureStatusGenerate, ); - await boss.work( + await createAndWork( + boss, WEIGHT_STATUS_QUEUE, { localConcurrency: 1 }, handleWeightStatusGenerate, ); - await boss.work( + await createAndWork( + boss, PULSE_STATUS_QUEUE, { localConcurrency: 1 }, handlePulseStatusGenerate, ); - await boss.work( + await createAndWork( + boss, BMI_STATUS_QUEUE, { localConcurrency: 1 }, handleBmiStatusGenerate, ); - await boss.work( + await createAndWork( + boss, MOOD_STATUS_QUEUE, { localConcurrency: 1 }, handleMoodStatusGenerate, ); - await boss.work( + await createAndWork( + boss, MEDICATION_COMPLIANCE_STATUS_QUEUE, { localConcurrency: 1 }, handleMedicationComplianceStatusGenerate, @@ -409,7 +418,8 @@ export async function registerStatusQueues( // (the cohort walk is itself sequential per user, and the content-hash // gate + per-user budget gate inside the handler cover the rare // double-tick overlap the old single slot serialised away). - await boss.work( + await createAndWork( + boss, INSIGHT_PREGENERATE_QUEUE, { localConcurrency: 2 }, handleInsightPregenerateJob, @@ -420,17 +430,23 @@ export async function registerStatusQueues( // night's samples into one job, and the handler's marker re-check makes a // rare double-fire a no-op, so one slot is enough and bounds the provider // concurrency the same way the nightly path does. - await boss.work( + await createAndWork( + boss, MORNING_DIGEST_REFRESH_QUEUE, { localConcurrency: 1 }, - async (jobs) => { - await withBackgroundEvent("job.morning_digest_refresh", async (evt) => { + async (jobs): Promise => { + return withBackgroundEvent("job.morning_digest_refresh", async (evt) => { + // The counts the batch already decides on: one refresh per job, and + // the finalised subset that reaches the morning push seam. + let refreshed = 0; + let finalised = 0; for (const job of jobs) { try { const result = await runMorningDigestRefresh( getWorkerPrisma(), job.data, ); + refreshed++; evt.addMeta( "morning_refresh", `${result.status}:${result.comprehensive ?? "none"}`, @@ -442,6 +458,7 @@ export async function registerStatusQueues( // there, so a user who never opted in or is outside the window is a // clean no-op. The fallback cron covers the no-sleep-arrived case. if (result.status === "finalised") { + finalised++; const pushResult = await maybeDispatchDailyBriefing( getWorkerPrisma(), job.data.userId, @@ -457,6 +474,7 @@ export async function registerStatusQueues( throw err; } } + return jobDone({ refreshed, finalised }); }); }, ); @@ -464,7 +482,8 @@ export async function registerStatusQueues( // read-only status route on a cold card. Low concurrency so a first // visit that cold-misses several cards can't saturate the Prisma pool // or fan out an unbounded number of concurrent provider calls. - await boss.work( + await createAndWork( + boss, INSIGHT_STATUS_GENERATE_QUEUE, { localConcurrency: INSIGHT_STATUS_GENERATE_CONCURRENCY }, handleInsightStatusGenerate, @@ -474,56 +493,84 @@ export async function registerStatusQueues( // eligible user and upserts one `COMPUTED RECOVERY_SCORE` row per scored // day (idempotent — a re-fire overwrites in place). Single-flight so two // ticks never double-walk the cohort. - await boss.work(RECOVERY_SCORE_QUEUE, { localConcurrency: 1 }, async () => { - try { - const summary = await runRecoveryScore(getWorkerPrisma()); - workerLog( - "info", - `[recovery-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, - ); - } catch (err) { - recordError(); - workerLog("error", "[recovery-score] pass failed", err); - throw err; - } - }); + await createAndWork( + boss, + RECOVERY_SCORE_QUEUE, + { localConcurrency: 1 }, + async (): Promise => { + try { + const summary = await runRecoveryScore(getWorkerPrisma()); + workerLog( + "info", + `[recovery-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, + ); + return jobDone({ + considered: summary.considered, + stored: summary.stored, + insufficient: summary.insufficient, + errored: summary.errored, + }); + } catch (err) { + recordError(); + workerLog("error", "[recovery-score] pass failed", err); + throw err; + } + }, + ); // v1.15.20 — proactive Coach nudge. Single-flight; the push-attempts // ledger caps a user at one nudge per rolling week, so an overlapping // tick would only waste reads. Deterministic triggers, no AI call. - await boss.work(COACH_NUDGE_QUEUE, { localConcurrency: 1 }, async () => { - await withBackgroundEvent("job.coach_nudge", async (evt) => { - try { - const summary = await runCoachNudgeTick(getWorkerPrisma(), new Date()); - evt.setBackground({ - task_name: "job.coach_nudge", - result: { + await createAndWork( + boss, + COACH_NUDGE_QUEUE, + { localConcurrency: 1 }, + async (): Promise => { + return withBackgroundEvent("job.coach_nudge", async (evt) => { + try { + const summary = await runCoachNudgeTick( + getWorkerPrisma(), + new Date(), + ); + evt.setBackground({ + task_name: "job.coach_nudge", + result: { + candidates_scanned: summary.candidatesScanned, + dispatched: summary.dispatched, + persisted: summary.persisted, + skipped_opted_out: summary.skippedOptedOut, + skipped_no_provider: summary.skippedNoProvider, + skipped_recent_nudge: summary.skippedRecentNudge, + skipped_recent_engagement: summary.skippedRecentEngagement, + skipped_no_trigger: summary.skippedNoTrigger, + skipped_no_channel: summary.skippedNoChannel, + failed: summary.failed, + }, + }); + // A per-user failure never fails this pass — it rides out as the + // `failed` count, exactly as it does in the wide event. + return jobDone({ candidates_scanned: summary.candidatesScanned, dispatched: summary.dispatched, persisted: summary.persisted, - skipped_opted_out: summary.skippedOptedOut, - skipped_no_provider: summary.skippedNoProvider, - skipped_recent_nudge: summary.skippedRecentNudge, - skipped_recent_engagement: summary.skippedRecentEngagement, - skipped_no_trigger: summary.skippedNoTrigger, - skipped_no_channel: summary.skippedNoChannel, failed: summary.failed, - }, - }); - } catch (err) { - evt.setError(err); - recordError(); - throw err; - } - }); - }); + }); + } catch (err) { + evt.setError(err); + recordError(); + throw err; + } + }); + }, + ); // v1.16.11 — medication low-stock pass. Single-flight; the // per-medication stamp makes a re-fire idempotent (already-notified // crossings skip), so an overlapping tick would only waste reads. - await boss.work( + await createAndWork( + boss, MEDICATION_LOW_STOCK_QUEUE, { localConcurrency: 1 }, - async () => { - await withBackgroundEvent("job.medication_low_stock", async (evt) => { + async (): Promise => { + return withBackgroundEvent("job.medication_low_stock", async (evt) => { try { const summary = await runMedicationLowStockTick( getWorkerPrisma(), @@ -544,6 +591,15 @@ export async function registerStatusQueues( failed: summary.failed, }, }); + // A per-medication failure never fails this pass — it rides out as + // the `failed` count. + return jobDone({ + users_scanned: summary.usersScanned, + medications_evaluated: summary.medicationsEvaluated, + notified: summary.notified, + rearmed: summary.rearmed, + failed: summary.failed, + }); } catch (err) { evt.setError(err); recordError(); @@ -556,7 +612,8 @@ export async function registerStatusQueues( // to claim a durable unique row, so two slots would only race each other for // a claim exactly one of them can win. It makes no provider call — a // module-graph test pins that — so one slot costs nothing in throughput. - await boss.work( + await createAndWork( + boss, DATA_ARRIVAL_QUEUE, { localConcurrency: 1 }, handleDataArrival, @@ -564,7 +621,8 @@ export async function registerStatusQueues( // v1.31.0 — the arrival reaction line. Single-flight: this is the spine's // ONE provider call, and a second concurrent slot would only race the same // durable `generatedAt` claim exactly one of them can win. - await boss.work( + await createAndWork( + boss, REACTION_LINE_QUEUE, { localConcurrency: 1 }, handleReactionLine, @@ -573,7 +631,8 @@ export async function registerStatusQueues( // path the spine deliberately does not walk, so it stays off the request pool // and cannot fan out concurrent completions when several workouts land in one // sync. - await boss.work( + await createAndWork( + boss, WORKOUT_INSIGHT_GENERATE_QUEUE, { localConcurrency: WORKOUT_INSIGHT_GENERATE_CONCURRENCY }, handleWorkoutInsightGenerate, @@ -583,84 +642,131 @@ export async function registerStatusQueues( // is suppressed), and the tick only loads a digest for opted-in users at // their exact local fallback hour. No provider call — it reads the cached // digest and dispatches the deterministic line + cached lead. - await boss.work(DAILY_BRIEFING_QUEUE, { localConcurrency: 1 }, async () => { - await withBackgroundEvent("job.daily_briefing", async (evt) => { - try { - const summary = await runDailyBriefingTick( - getWorkerPrisma(), - new Date(), - ); - evt.setBackground({ - task_name: "job.daily_briefing", - result: { + await createAndWork( + boss, + DAILY_BRIEFING_QUEUE, + { localConcurrency: 1 }, + async (): Promise => { + return withBackgroundEvent("job.daily_briefing", async (evt) => { + try { + const summary = await runDailyBriefingTick( + getWorkerPrisma(), + new Date(), + ); + evt.setBackground({ + task_name: "job.daily_briefing", + result: { + candidates_scanned: summary.candidatesScanned, + in_slot: summary.inSlot, + sent: summary.sent, + suppressed_frequency: summary.suppressedFrequency, + no_digest: summary.noDigest, + opted_out: summary.optedOut, + module_off: summary.moduleOff, + no_channel: summary.noChannel, + outside_window: summary.outsideWindow, + failed: summary.failed, + }, + }); + // Most ticks land outside every user's slot; that is the tick doing + // its job, so a zero `sent` is a success with zero counts. + return jobDone({ candidates_scanned: summary.candidatesScanned, in_slot: summary.inSlot, sent: summary.sent, - suppressed_frequency: summary.suppressedFrequency, - no_digest: summary.noDigest, - opted_out: summary.optedOut, - module_off: summary.moduleOff, - no_channel: summary.noChannel, - outside_window: summary.outsideWindow, failed: summary.failed, - }, - }); - } catch (err) { - evt.setError(err); - recordError(); - throw err; - } - }); - }); + }); + } catch (err) { + evt.setError(err); + recordError(); + throw err; + } + }); + }, + ); // v1.10.0 — computed scores (WX-E). Nightly Stress-score (HRV-derived // proxy) compute + store. Single-flight so two ticks never double-walk // the cohort. The runner iterates every eligible user and upserts one // `COMPUTED STRESS_SCORE` row per scored day (idempotent — a re-fire // overwrites in place). - await boss.work(STRESS_SCORE_QUEUE, { localConcurrency: 1 }, async () => { - try { - const summary = await runStressScore(getWorkerPrisma()); - workerLog( - "info", - `[stress-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, - ); - } catch (err) { - recordError(); - workerLog("error", "[stress-score] pass failed", err); - throw err; - } - }); + await createAndWork( + boss, + STRESS_SCORE_QUEUE, + { localConcurrency: 1 }, + async (): Promise => { + try { + const summary = await runStressScore(getWorkerPrisma()); + workerLog( + "info", + `[stress-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, + ); + return jobDone({ + considered: summary.considered, + stored: summary.stored, + insufficient: summary.insufficient, + errored: summary.errored, + }); + } catch (err) { + recordError(); + workerLog("error", "[stress-score] pass failed", err); + throw err; + } + }, + ); // v1.10.0 — computed scores (WX-E). Nightly Strain-score (Banister TRIMP // cardio-load) compute + store. Single-flight; upserts one `COMPUTED // STRAIN_SCORE` row per scored day (idempotent). - await boss.work(STRAIN_SCORE_QUEUE, { localConcurrency: 1 }, async () => { - try { - const summary = await runStrainScore(getWorkerPrisma()); - workerLog( - "info", - `[strain-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, - ); - } catch (err) { - recordError(); - workerLog("error", "[strain-score] pass failed", err); - throw err; - } - }); + await createAndWork( + boss, + STRAIN_SCORE_QUEUE, + { localConcurrency: 1 }, + async (): Promise => { + try { + const summary = await runStrainScore(getWorkerPrisma()); + workerLog( + "info", + `[strain-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, + ); + return jobDone({ + considered: summary.considered, + stored: summary.stored, + insufficient: summary.insufficient, + errored: summary.errored, + }); + } catch (err) { + recordError(); + workerLog("error", "[strain-score] pass failed", err); + throw err; + } + }, + ); // v1.11.0 — period-narrative warm. A scheduled tick (no `userId`) runs the // boundary-gated nightly fan-out; a `userId` payload runs a single-user warm // enqueued by the read-only GET on a cold/stale read. Single-flight so two // ticks never double-walk the cohort; the per-user budget gate covers the // fan-out and the enqueue `singletonKey` covers the single-user path. - await boss.work( + await createAndWork( + boss, PERIOD_NARRATIVE_QUEUE, { localConcurrency: 1 }, - async (jobs) => { + async (jobs): Promise => { + // Two modes on one queue, so the outcome counts them separately: a + // non-boundary night is a scheduled pass that generated nothing, which + // is a success with zero counts, not an idle queue. + let singleUser = 0; + let scheduled = 0; + let generated = 0; + let failed = 0; for (const job of jobs) { try { if (job.data?.userId) { await warmOneNarrative(job.data); + singleUser++; } else { const summary = await runPeriodNarrativeWarm(getWorkerPrisma()); + scheduled++; + generated += summary.generated; + failed += summary.failed; workerLog( "info", `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`, @@ -674,26 +780,45 @@ export async function registerStatusQueues( throw err; } } + return jobDone({ + single_user: singleUser, + scheduled, + generated, + failed, + }); }, ); // v1.11.1 — combined Coach memory refresh: rolling conversation summary + // durable fact extraction for one long conversation. localConcurrency 1 so a // burst of long-conversation turns can't fan out concurrent provider calls; // each step is budget-gated inside runStatusCompletion and fault-isolated. - await boss.work( + await createAndWork( + boss, COACH_MEMORY_REFRESH_QUEUE, { localConcurrency: 1 }, - async (jobs) => { + async (jobs): Promise => { + // An incomplete payload is skipped rather than failed, so it has to be + // countable — otherwise a queue full of them reads as a clean run. + let refreshed = 0; + let skippedIncomplete = 0; for (const job of jobs) { - if (!job.data?.conversationId || !job.data?.userId) continue; + if (!job.data?.conversationId || !job.data?.userId) { + skippedIncomplete++; + continue; + } try { await runCoachMemoryRefresh(job.data); + refreshed++; } catch (err) { recordError(); workerLog("error", "[coach-memory-refresh] failed", err); throw err; } } + return jobDone({ + refreshed, + skipped_incomplete: skippedIncomplete, + }); }, ); @@ -701,11 +826,12 @@ export async function registerStatusQueues( // reminder to `due` and minting plan-review reminders are both idempotent // across re-fires (an already-due reminder is not re-flipped; a plan whose // reviewDate was cleared is not re-minted). No provider call, no push. - await boss.work( + await createAndWork( + boss, COACH_REMINDER_SWEEP_QUEUE, { localConcurrency: 1 }, - async () => { - await withBackgroundEvent("job.coach_reminder_sweep", async (evt) => { + async (): Promise => { + return withBackgroundEvent("job.coach_reminder_sweep", async (evt) => { try { const summary = await runCoachReminderSweep( getWorkerPrisma(), @@ -719,6 +845,11 @@ export async function registerStatusQueues( errored: summary.errored, }, }); + return jobDone({ + reminders_due: summary.remindersDue, + plan_reviews_minted: summary.planReviewsMinted, + errored: summary.errored, + }); } catch (err) { evt.setError(err); recordError(); @@ -732,11 +863,12 @@ export async function registerStatusQueues( // reminders the sweep minted and writes the grounded before/after read-back // into the plan's encrypted outcome column, flipping it to `reviewed`. // Idempotent (a reviewed plan is not re-processed). No provider call, no push. - await boss.work( + await createAndWork( + boss, COACH_PLAN_REVIEW_QUEUE, { localConcurrency: 1 }, - async () => { - await withBackgroundEvent("job.coach_plan_review", async (evt) => { + async (): Promise => { + return withBackgroundEvent("job.coach_plan_review", async (evt) => { try { const summary = await runCoachPlanReviewTick( getWorkerPrisma(), @@ -750,6 +882,11 @@ export async function registerStatusQueues( errored: summary.errored, }, }); + return jobDone({ + reviewed: summary.reviewed, + insufficient: summary.insufficient, + errored: summary.errored, + }); } catch (err) { evt.setError(err); recordError(); diff --git a/src/lib/jobs/reminder/registrar-shared.ts b/src/lib/jobs/reminder/registrar-shared.ts index 178c580e6..c05f1ea87 100644 --- a/src/lib/jobs/reminder/registrar-shared.ts +++ b/src/lib/jobs/reminder/registrar-shared.ts @@ -167,6 +167,32 @@ export const insightRetryOptions = { retryBackoff: true, } as const; +/** + * Retry policy for a cron-driven pass whose failure mode is deterministic. + * + * Handlers now return a `JobOutcome`, and the retention cleanups that used to + * warn-and-return fail their job instead. That is the point — but it walks + * into pg-boss's queue default of `retryLimit: 2` with no delay and no + * backoff, so a bulk DELETE that timed out against a large trailing edge + * would run twice more, immediately, against the same rows. Three doomed + * sequential scans a night is a worse outcome than the silence it replaced. + * + * For this class the next cron tick IS the retry, and it arrives with the + * whole gap as backoff. So the schedule declares zero retries: the failure is + * recorded once, reaches the operator once, and the pass tries again on its + * own cadence. + * + * It rides on the SCHEDULE rather than on `createQueue`, deliberately. + * pg-boss's `create_queue()` ends in ON CONFLICT DO NOTHING, so a queue-level + * default would be inert on every already-running instance — the same trap + * `reconcileQueuePolicies` above exists to work around. Send options apply to + * every job the cron mints, new database or old. + * + * Use it only where the failure genuinely repeats. A transient failure (a TLS + * probe, an S3 fetch, a provider call) wants the default retries, not this. + */ +export const cronIsTheRetry = { retryLimit: 0 } as const; + /** * Create every queue in `queues`, then schedule every cron in `schedules`. * Centralised so each registrar provisions before it schedules in the exact