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

@@ -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(

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(

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 });
 });

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(

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(

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(

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);

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 },

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<symbol, unknown>)[STARTUP_KEY];
+  delete (globalThis as Record<string, unknown>)[BOSS_START_KEY];
+});
+
+afterEach(() => {
+  vi.useRealTimers();
+  setGlobalBoss(null);
+  delete (globalThis as Record<symbol, unknown>)[STARTUP_KEY];
+  delete (globalThis as Record<string, unknown>)[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<string, unknown>)[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();
+  });
+});

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;

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<string, unknown>)[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<PgBoss | null> {
+  const bound = getGlobalBoss();
+  if (bound) return bound;
+  if (!bossStartPending()) return null;
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    await new Promise<void>((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<unknown>,
+): 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";
+}

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<void> {
-  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()}`),
+    }),
+  );
 }
 
 /**

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<void> {
-  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()}`,
+    }),
+  );
 }
 
 /**

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" },