From b795b3b1f360ac2855a5b99a16660a828cb2398e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 21 Oct 2026 21:37:21 +0200 Subject: [PATCH] fix(jobs): enqueue during boot waits for the boss instead of dropping The server answers before pg-boss is bound, and a producer rebinds in the background after losing its connection. A request in either window read the boss as null and skipped its enqueue: OAuth callbacks lost the first sync and the history backfill, measurement and mood writes lost their rollup recompute, WHOOP webhook deliveries were dropped. whenBossReady() waits for the boss while a start is under way (boot or reconnect), bounded at 60 s and logged when it gives up, and answers null at once when nothing is coming. enqueueWhenBossReady() sends at once through a bound boss and otherwise defers the send to the background, so the request is not held for the boot. --- src/app/api/fitbit/callback/route.ts | 7 +- src/app/api/google-health/callback/route.ts | 7 +- src/app/api/nightscout/connect/route.ts | 7 +- src/app/api/oura/callback/route.ts | 7 +- src/app/api/polar/callback/route.ts | 7 +- src/app/api/strava/callback/route.ts | 7 +- src/app/api/whoop/callback/route.ts | 7 +- src/instrumentation.ts | 5 +- src/lib/jobs/__tests__/boss-ready.test.ts | 89 ++++++++++++++++++++ src/lib/jobs/boss-instance.ts | 4 +- src/lib/jobs/boss-ready.ts | 93 +++++++++++++++++++++ src/lib/rollups/measurement-rollups.ts | 72 ++++++++-------- src/lib/rollups/mood-rollups.ts | 21 ++--- src/lib/whoop/webhook-handler.ts | 16 ++-- 14 files changed, 266 insertions(+), 83 deletions(-) create mode 100644 src/lib/jobs/__tests__/boss-ready.test.ts create mode 100644 src/lib/jobs/boss-ready.ts diff --git a/src/app/api/fitbit/callback/route.ts b/src/app/api/fitbit/callback/route.ts index c553c0aff..a8284ae92 100644 --- a/src/app/api/fitbit/callback/route.ts +++ b/src/app/api/fitbit/callback/route.ts @@ -16,7 +16,7 @@ import { import { getUserFitbitCredentials } from "@/lib/fitbit/credentials"; import { FITBIT_OAUTH_STATE_COOKIE } from "@/lib/fitbit/oauth-state"; import { FITBIT_BACKFILL_QUEUE } from "@/lib/jobs/fitbit-backfill"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { markReconnected } from "@/lib/integrations/status"; import { Prisma } from "@/generated/prisma/client"; import { @@ -205,8 +205,7 @@ export const GET = apiHandler(async (request: NextRequest) => { // Enqueue the self-converging history backfill. Best-effort: the boot-time // discovery query (`backfillCompletedAt IS NULL`) is the safety net, so a // missing boss instance here doesn't strand the connection. - const boss = getGlobalBoss(); - if (boss) { + await enqueueWhenBossReady("fitbit-backfill", async (boss) => { await boss .send(FITBIT_BACKFILL_QUEUE, { userId: user.id, @@ -215,7 +214,7 @@ export const GET = apiHandler(async (request: NextRequest) => { .catch((err) => getEvent()?.addWarning(`fitbit-backfill enqueue failed: ${err}`), ); - } + }); const response = NextResponse.redirect( new URL( diff --git a/src/app/api/google-health/callback/route.ts b/src/app/api/google-health/callback/route.ts index 821c81530..4c0ac3d98 100644 --- a/src/app/api/google-health/callback/route.ts +++ b/src/app/api/google-health/callback/route.ts @@ -18,7 +18,7 @@ import { import { getUserGoogleHealthCredentials } from "@/lib/google-health/credentials"; import { GOOGLE_HEALTH_OAUTH_STATE_COOKIE } from "@/lib/google-health/oauth-state"; import { GOOGLE_HEALTH_BACKFILL_QUEUE } from "@/lib/jobs/google-health-backfill"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { markReconnected } from "@/lib/integrations/status"; import { Prisma } from "@/generated/prisma/client"; import { @@ -235,8 +235,7 @@ export const GET = apiHandler(async (request: NextRequest) => { // Enqueue the self-converging history backfill. Best-effort: the boot-time // discovery query (`backfillCompletedAt IS NULL`) is the safety net, so a // missing boss instance here doesn't strand the connection. - const boss = getGlobalBoss(); - if (boss) { + await enqueueWhenBossReady("google-health-backfill", async (boss) => { await boss .send(GOOGLE_HEALTH_BACKFILL_QUEUE, { userId: recordUserId, @@ -247,7 +246,7 @@ export const GET = apiHandler(async (request: NextRequest) => { `google-health-backfill enqueue failed: ${err}`, ), ); - } + }); const response = NextResponse.redirect( new URL( diff --git a/src/app/api/nightscout/connect/route.ts b/src/app/api/nightscout/connect/route.ts index a0d297e38..8dc4ae8a5 100644 --- a/src/app/api/nightscout/connect/route.ts +++ b/src/app/api/nightscout/connect/route.ts @@ -16,7 +16,7 @@ import { encrypt } from "@/lib/crypto"; import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; import { markReconnected } from "@/lib/integrations/status"; import { NIGHTSCOUT_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { fetchSgvEntries, NightscoutApiError, @@ -160,8 +160,7 @@ export const POST = apiHandler(async (request: NextRequest) => { // empty card until the next hourly tick. Best-effort by design: the hourly // cron remains the safety net, so a missing boss instance must never fail the // connect response the user just earned. - const boss = getGlobalBoss(); - if (boss) { + await enqueueWhenBossReady("nightscout-sync", async (boss) => { await boss .send(NIGHTSCOUT_SYNC_QUEUE, { userId: user.id }) .catch((err) => @@ -169,7 +168,7 @@ export const POST = apiHandler(async (request: NextRequest) => { `nightscout-sync connect enqueue failed: ${err}`, ), ); - } + }); return apiSuccess({ connected: true }); }); diff --git a/src/app/api/oura/callback/route.ts b/src/app/api/oura/callback/route.ts index ec67933d4..26b2e677e 100644 --- a/src/app/api/oura/callback/route.ts +++ b/src/app/api/oura/callback/route.ts @@ -25,7 +25,7 @@ import { runOAuthCallbackOnce } from "@/lib/oauth/callback-claim"; import { ouraCollectionsMissingFromGrant } from "@/lib/oura/scope"; import { markReconnected } from "@/lib/integrations/status"; import { OURA_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; import { OURA_BACKFILL_QUEUE, @@ -184,8 +184,7 @@ async function completeCallback( // missing. Reconnects ride the same path. Best-effort by design: the hourly // cron remains the safety net, so a missing boss instance must never strand // the redirect. - const boss = getGlobalBoss(); - if (boss) { + await enqueueWhenBossReady("oura-connect", async (boss) => { await boss .send(OURA_SYNC_QUEUE, { userId }) .catch((err) => @@ -206,7 +205,7 @@ async function completeCallback( `oura-backfill connect enqueue failed: ${err}`, ), ); - } + }); const response = NextResponse.redirect( new URL( diff --git a/src/app/api/polar/callback/route.ts b/src/app/api/polar/callback/route.ts index 6bb84b67e..84230b171 100644 --- a/src/app/api/polar/callback/route.ts +++ b/src/app/api/polar/callback/route.ts @@ -19,7 +19,7 @@ import { import { runOAuthCallbackOnce } from "@/lib/oauth/callback-claim"; import { markReconnected } from "@/lib/integrations/status"; import { POLAR_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { NextRequest, NextResponse } from "next/server"; /** @@ -156,14 +156,13 @@ async function completeCallback( // card until the next hourly tick. Best-effort by design: the hourly cron // remains the safety net, so a missing boss instance must never strand the // redirect. - const boss = getGlobalBoss(); - if (boss) { + await enqueueWhenBossReady("polar-connect", async (boss) => { await boss .send(POLAR_SYNC_QUEUE, { userId }) .catch((err) => getEvent()?.addWarning(`polar-sync connect enqueue failed: ${err}`), ); - } + }); const response = NextResponse.redirect( new URL( diff --git a/src/app/api/strava/callback/route.ts b/src/app/api/strava/callback/route.ts index 9e0c4c08e..e94544c3c 100644 --- a/src/app/api/strava/callback/route.ts +++ b/src/app/api/strava/callback/route.ts @@ -19,7 +19,7 @@ import { import { runOAuthCallbackOnce } from "@/lib/oauth/callback-claim"; import { markReconnected } from "@/lib/integrations/status"; import { STRAVA_BACKFILL_QUEUE } from "@/lib/jobs/strava-backfill"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { NextRequest, NextResponse } from "next/server"; /** @@ -161,8 +161,7 @@ async function completeCallback( // (`stravaBackfillCompletedAt IS NULL`) is the safety net, so a missing // boss instance here never strands the connection. Mirrors the WHOOP / // Fitbit callback enqueue. - const boss = getGlobalBoss(); - if (boss) { + await enqueueWhenBossReady("strava-backfill", async (boss) => { await boss .send(STRAVA_BACKFILL_QUEUE, { userId, @@ -171,7 +170,7 @@ async function completeCallback( .catch((err) => getEvent()?.addWarning(`strava-backfill enqueue failed: ${err}`), ); - } + }); const response = NextResponse.redirect( new URL( diff --git a/src/app/api/whoop/callback/route.ts b/src/app/api/whoop/callback/route.ts index fdf5fb1cd..9bf1cef17 100644 --- a/src/app/api/whoop/callback/route.ts +++ b/src/app/api/whoop/callback/route.ts @@ -15,7 +15,7 @@ import { getUserWhoopCredentials } from "@/lib/whoop/credentials"; import { WHOOP_OAUTH_STATE_COOKIE } from "@/lib/whoop/oauth-state"; import { buildReturnSchemeRedirect } from "@/lib/whoop/return-scheme"; import { WHOOP_BACKFILL_QUEUE } from "@/lib/jobs/whoop-backfill"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { markReconnected } from "@/lib/integrations/status"; import { Prisma } from "@/generated/prisma/client"; import { isP2002 } from "@/lib/prisma-errors"; @@ -257,8 +257,7 @@ export const GET = apiHandler(async (request: NextRequest) => { // Enqueue the self-converging history backfill. Best-effort: the boot-time // discovery query (`backfillCompletedAt IS NULL`) is the safety net, so a // missing boss instance here doesn't strand the connection. - const boss = getGlobalBoss(); - if (boss) { + await enqueueWhenBossReady("whoop-backfill", async (boss) => { await boss .send(WHOOP_BACKFILL_QUEUE, { userId, @@ -267,7 +266,7 @@ export const GET = apiHandler(async (request: NextRequest) => { .catch((err) => getEvent()?.addWarning(`whoop-backfill enqueue failed: ${err}`), ); - } + }); const response = outcomeRedirect(returnScheme, "connected"); response.cookies.delete(WHOOP_OAUTH_STATE_COOKIE); diff --git a/src/instrumentation.ts b/src/instrumentation.ts index f152e67ca..ea0e46748 100644 --- a/src/instrumentation.ts +++ b/src/instrumentation.ts @@ -84,8 +84,9 @@ export async function register() { // needs to start, provision its queues and run the boot discovery; the // probe now answers `starting` meanwhile. A request that enqueues while // the boss is not bound yet takes the same path as during a producer - // reconnect: a garnish skips its enqueue (the owning cron catches up), - // and a route that needs the worker answers 503 until it is up. + // reconnect: the enqueue waits for the boss in the background + // (`enqueueWhenBossReady`, bounded and logged when it gives up), and a + // route that needs the worker answers 503 until it is up. const [ { shouldRunWorker }, { markStartup, startupPhase, startupElapsedMs, STARTING_MAX_MS }, diff --git a/src/lib/jobs/__tests__/boss-ready.test.ts b/src/lib/jobs/__tests__/boss-ready.test.ts new file mode 100644 index 000000000..78d4a6332 --- /dev/null +++ b/src/lib/jobs/__tests__/boss-ready.test.ts @@ -0,0 +1,89 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { PgBoss } from "pg-boss"; + +import { markStartup } from "@/lib/boot/startup-state"; +import { setGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + BOSS_START_KEY, + enqueueWhenBossReady, + whenBossReady, +} from "@/lib/jobs/boss-ready"; + +const STARTUP_KEY = Symbol.for("healthlog.boot.startupState"); + +function fakeBoss(): PgBoss { + return { send: vi.fn(async () => "job-1") } as unknown as PgBoss; +} + +beforeEach(() => { + vi.useFakeTimers(); + setGlobalBoss(null); + delete (globalThis as Record)[STARTUP_KEY]; + delete (globalThis as Record)[BOSS_START_KEY]; +}); + +afterEach(() => { + vi.useRealTimers(); + setGlobalBoss(null); + delete (globalThis as Record)[STARTUP_KEY]; + delete (globalThis as Record)[BOSS_START_KEY]; + vi.restoreAllMocks(); +}); + +describe("enqueueWhenBossReady", () => { + it("sends at once through a bound boss", async () => { + const boss = fakeBoss(); + setGlobalBoss(boss); + const send = vi.fn(async (b: PgBoss) => b.send("q", {})); + await expect(enqueueWhenBossReady("q", send)).resolves.toBe("sent"); + expect(send).toHaveBeenCalledWith(boss); + }); + + it("skips without waiting when no start is under way", async () => { + const send = vi.fn(async () => undefined); + await expect(enqueueWhenBossReady("q", send)).resolves.toBe("skipped"); + expect(send).not.toHaveBeenCalled(); + }); + + it("sends once the boss is bound when it arrives during boot", async () => { + markStartup("starting"); + const send = vi.fn(async () => undefined); + await expect(enqueueWhenBossReady("q", send)).resolves.toBe("deferred"); + expect(send).not.toHaveBeenCalled(); + + const boss = fakeBoss(); + setGlobalBoss(boss); + await vi.advanceTimersByTimeAsync(300); + expect(send).toHaveBeenCalledWith(boss); + }); + + it("waits for a producer reconnect too", async () => { + (globalThis as Record)[BOSS_START_KEY] = Promise.resolve(); + const send = vi.fn(async () => undefined); + await expect(enqueueWhenBossReady("q", send)).resolves.toBe("deferred"); + setGlobalBoss(fakeBoss()); + await vi.advanceTimersByTimeAsync(300); + expect(send).toHaveBeenCalledTimes(1); + }); +}); + +describe("whenBossReady", () => { + it("gives up after the timeout and says so", async () => { + markStartup("starting"); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const pending = whenBossReady("rollup-recompute", 1_000); + await vi.advanceTimersByTimeAsync(1_100); + await expect(pending).resolves.toBeNull(); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("rollup-recompute was not enqueued"), + ); + }); + + it("stops waiting as soon as the start fails", async () => { + markStartup("starting"); + const pending = whenBossReady("q", 60_000); + markStartup("failed"); + await vi.advanceTimersByTimeAsync(300); + await expect(pending).resolves.toBeNull(); + }); +}); diff --git a/src/lib/jobs/boss-instance.ts b/src/lib/jobs/boss-instance.ts index 438e85818..c9f056acd 100644 --- a/src/lib/jobs/boss-instance.ts +++ b/src/lib/jobs/boss-instance.ts @@ -9,9 +9,11 @@ import { PgBoss } from "pg-boss"; import { getPgBossPoolMax, getPoolConnectionTimeoutMs } from "@/lib/db"; import { timeZoneSessionOptions } from "@/lib/db-session"; +// The slot a start is parked in; `whenBossReady` reads it to know one is +// under way. +import { BOSS_START_KEY } from "@/lib/jobs/boss-ready"; const BOSS_KEY = "__healthlog_pgboss__" as const; -const BOSS_START_KEY = "__healthlog_pgboss_start__" as const; const PRODUCER_START_MAX_ATTEMPTS = 5; const PRODUCER_RETRY_BASE_DELAY_MS = 250; const PRODUCER_RETRY_MAX_DELAY_MS = 2_000; diff --git a/src/lib/jobs/boss-ready.ts b/src/lib/jobs/boss-ready.ts new file mode 100644 index 000000000..1388fcfff --- /dev/null +++ b/src/lib/jobs/boss-ready.ts @@ -0,0 +1,93 @@ +/** + * Enqueue work that must not be lost while the process is still booting. + * + * The HTTP server answers before pg-boss is bound (`instrumentation.ts` starts + * the background role after the server is up), and a producer that lost its + * connection rebinds in the background. A request in either window used to + * read `getGlobalBoss()` as null and drop its enqueue: an OAuth callback + * skipped the first sync and the history backfill, a measurement write skipped + * its rollup recompute. Those waited for the next cron or boot discovery. + * + * `whenBossReady()` waits for the boss while a start is under way, bounded by + * `BOSS_READY_TIMEOUT_MS`, and logs one line when it gives up. With no start + * under way (a process that never runs a background role, or one whose start + * failed) it answers null at once: nothing is coming, so nothing is waited for. + * + * `enqueueWhenBossReady()` is what a producer calls. With the boss bound it + * sends at once and the caller sees the send's own result and errors, as + * before. While a start is under way it does not hold the request: the send + * waits in the background and runs once the boss is up. + */ +import type { PgBoss } from "pg-boss"; + +import { startupPhase } from "@/lib/boot/startup-state"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { trackBackgroundTask } from "@/lib/logging/background-tasks"; + +/** The global slot a producer (re)start is parked in while it runs. */ +export const BOSS_START_KEY = "__healthlog_pgboss_start__" as const; + +/** Longest an enqueue waits for the boss before it is given up. */ +export const BOSS_READY_TIMEOUT_MS = 60_000; + +const POLL_MS = 250; + +/** Is a boss start (boot, or a producer reconnect) under way right now? */ +export function bossStartPending(): boolean { + if ((globalThis as Record)[BOSS_START_KEY]) return true; + return startupPhase() === "starting"; +} + +/** + * The bound boss, waiting for it while a start is under way. Null when no + * start is under way or the wait ran out (logged with `label`). + */ +export async function whenBossReady( + label: string, + timeoutMs: number = BOSS_READY_TIMEOUT_MS, +): Promise { + const bound = getGlobalBoss(); + if (bound) return bound; + if (!bossStartPending()) return null; + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + await new Promise((resolve) => { + setTimeout(resolve, Math.min(POLL_MS, deadline - Date.now())).unref?.(); + }); + const boss = getGlobalBoss(); + if (boss) return boss; + if (!bossStartPending()) return null; + } + console.warn( + `[pg-boss] not ready after ${Math.round(timeoutMs / 1000)} s; ${label} was not enqueued`, + ); + return null; +} + +/** + * Send through the boss now when it is bound, or once it is up when a start + * is under way. Returns whether the send ran (or was deferred) rather than + * being skipped for want of a boss. + */ +export async function enqueueWhenBossReady( + label: string, + send: (boss: PgBoss) => Promise, +): Promise<"sent" | "deferred" | "skipped"> { + const bound = getGlobalBoss(); + if (bound) { + await send(bound); + return "sent"; + } + if (!bossStartPending()) return "skipped"; + trackBackgroundTask( + whenBossReady(label).then(async (boss) => { + if (!boss) return; + try { + await send(boss); + } catch (err) { + console.warn(`[pg-boss] deferred ${label} enqueue failed:`, err); + } + }), + ); + return "deferred"; +} diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index a5277687f..bf3343d18 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -36,6 +36,7 @@ */ import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { annotate } from "@/lib/logging/context"; import { readCanonicalRollupBuckets, @@ -521,11 +522,12 @@ function bucketRuns( } /** - * Best-effort pg-boss enqueue. Silent no-op when no boss is attached - * (route is running in a context without the worker — typically the - * test suite). The reader-side fall-through still produces fresh - * data on a cache miss, so missing the enqueue is at-most a one-read - * cost rather than a correctness bug. + * Best-effort pg-boss enqueue. A write that lands while the process is still + * bringing its boss up is sent once the boss is bound (`enqueueWhenBossReady`) + * instead of being dropped; with no boss coming at all (a context without + * the worker, typically the test suite) it is a silent no-op. The + * reader-side fall-through still produces fresh data on a cache miss, so a + * missed enqueue is at most a one-read cost rather than a correctness bug. */ export async function enqueueRollupRecompute(input: { userId: string; @@ -534,8 +536,6 @@ export async function enqueueRollupRecompute(input: { from: Date; to: Date; }): Promise { - const boss = getGlobalBoss(); - if (!boss) return; const payload: RollupRecomputePayload = { userId: input.userId, type: input.type, @@ -544,34 +544,36 @@ export async function enqueueRollupRecompute(input: { to: input.to.toISOString(), enqueuedAt: new Date().toISOString(), }; - await boss.send(ROLLUP_RECOMPUTE_QUEUE, payload, { - retryLimit: 3, - retryDelay: 30, - retryBackoff: true, - // Coalesce duplicate enqueues for the same bucket key — multiple - // measurements landing on the same day shouldn't each spawn a - // separate worker run. The singleton key uses bucket-start so - // two writes inside the same week share one queued job. - // - // The key only does that because the queue carries the `short` policy - // (see `reminder/register-rollup.ts`); under pg-boss's default `standard` - // policy nothing indexes `singleton_key` and this key coalesced nothing, - // which is how a large sync came to fan out thousands of jobs re-deriving - // byte-identical rows. `short` collapses sends only while an identical job - // is still QUEUED — safe here precisely because the handler re-reads the - // bucket's live rows when it starts, so the queued job still sees writes - // that land after it was enqueued. - // - // A job covering a run of several buckets (`recomputeBucketsForKeys`) - // carries its end in the key too: under the bare start it would collapse - // into a queued single-bucket job and the rest of the run would be lost. - singletonKey: - `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}` + - (bucketSpan(input.from, input.granularity).to.getTime() === - input.to.getTime() - ? "" - : `|${input.to.toISOString()}`), - }); + await enqueueWhenBossReady("rollup-recompute", (boss) => + boss.send(ROLLUP_RECOMPUTE_QUEUE, payload, { + retryLimit: 3, + retryDelay: 30, + retryBackoff: true, + // Coalesce duplicate enqueues for the same bucket key — multiple + // measurements landing on the same day shouldn't each spawn a + // separate worker run. The singleton key uses bucket-start so + // two writes inside the same week share one queued job. + // + // The key only does that because the queue carries the `short` policy + // (see `reminder/register-rollup.ts`); under pg-boss's default `standard` + // policy nothing indexes `singleton_key` and this key coalesced nothing, + // which is how a large sync came to fan out thousands of jobs re-deriving + // byte-identical rows. `short` collapses sends only while an identical job + // is still QUEUED — safe here precisely because the handler re-reads the + // bucket's live rows when it starts, so the queued job still sees writes + // that land after it was enqueued. + // + // A job covering a run of several buckets (`recomputeBucketsForKeys`) + // carries its end in the key too: under the bare start it would collapse + // into a queued single-bucket job and the rest of the run would be lost. + singletonKey: + `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}` + + (bucketSpan(input.from, input.granularity).to.getTime() === + input.to.getTime() + ? "" + : `|${input.to.toISOString()}`), + }), + ); } /** diff --git a/src/lib/rollups/mood-rollups.ts b/src/lib/rollups/mood-rollups.ts index 1a8627190..cd86d10c6 100644 --- a/src/lib/rollups/mood-rollups.ts +++ b/src/lib/rollups/mood-rollups.ts @@ -47,6 +47,7 @@ */ import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { annotate } from "@/lib/logging/context"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; @@ -349,8 +350,6 @@ export async function enqueueMoodRollupRecompute(input: { from: Date; to: Date; }): Promise { - const boss = getGlobalBoss(); - if (!boss) return; const payload: MoodRollupRecomputePayload = { userId: input.userId, granularity: input.granularity, @@ -358,14 +357,16 @@ export async function enqueueMoodRollupRecompute(input: { to: input.to.toISOString(), enqueuedAt: new Date().toISOString(), }; - await boss.send(MOOD_ROLLUP_RECOMPUTE_QUEUE, payload, { - retryLimit: 3, - retryDelay: 30, - retryBackoff: true, - // Coalesce duplicate enqueues for the same bucket key — two writes - // inside the same week share one queued job. - singletonKey: `${input.userId}|${input.granularity}|${input.from.toISOString()}`, - }); + await enqueueWhenBossReady("mood-rollup-recompute", (boss) => + boss.send(MOOD_ROLLUP_RECOMPUTE_QUEUE, payload, { + retryLimit: 3, + retryDelay: 30, + retryBackoff: true, + // Coalesce duplicate enqueues for the same bucket key — two writes + // inside the same week share one queued job. + singletonKey: `${input.userId}|${input.granularity}|${input.from.toISOString()}`, + }), + ); } /** diff --git a/src/lib/whoop/webhook-handler.ts b/src/lib/whoop/webhook-handler.ts index 5cdaff6ff..a815cabba 100644 --- a/src/lib/whoop/webhook-handler.ts +++ b/src/lib/whoop/webhook-handler.ts @@ -30,7 +30,7 @@ import { prisma } from "@/lib/db"; import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; import { getClientIp } from "@/lib/api-response"; import { annotate, getEvent } from "@/lib/logging/context"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import type { MeasurementType, Prisma } from "@/generated/prisma/client"; import { WHOOP_RECOVERY_FIELD_TAGS } from "./client"; @@ -259,15 +259,17 @@ export async function processWhoopNotification( // resource sync. The payload carries the resource id so the worker can do a // targeted fetch-by-id refresh — landing the exact record immediately rather // than waiting for the next overlap window to re-walk the collection. - const boss = getGlobalBoss(); - if (!boss) { + // A delivery during boot is sent once the boss is up rather than dropped. + const outcome = await enqueueWhenBossReady("whoop-webhook", (boss) => + boss.send(queue, { + userId: connection.userId, + resourceId: resourceId ?? undefined, + }), + ); + if (outcome === "skipped") { getEvent()?.addWarning("whoop-webhook: pg-boss not initialised"); return NextResponse.json({ status: "ok" }, { status: 200 }); } - await boss.send(queue, { - userId: connection.userId, - resourceId: resourceId ?? undefined, - }); annotate({ action: { name: "whoop.webhook.enqueued" },