fix(jobs): let a sliced import give way to a waiting heavy job and warn on long lane waits

A job sent back by the heavy lane now leaves word that the lane is
wanted. The holder, and only the holder, sees it through laneWanted();
the Google Health backfill asks it where its slice budget already
stops resumable walks, after the first page, ends the slice there and
queues the rest past the deferred job's return. Holders that cannot
stop part-way run to their end as before.

The way back to the queue now returns when the oldest deferred job was
first queued; a job that has stood in line for two hours is reported
once per holder as job.heavy_lane.waiting_long at warn, so a lane held
for hours reaches the alerts.

docs/integrations/google-health.md

@@ -245,8 +245,10 @@ reading:
     upper-bound filter Google refused.
   - `stopReason`: why it ended early. `time_budget` and `page_budget` are the
     normal end of a slice, `rate_limited` means Google would not take more
-    requests for now, `job_budget` that the job itself ran out of time.
-    `null` means the slice read everything it started.
+    requests for now, `job_budget` that the job itself ran out of time,
+    `lane_wanted` that the slice made room for another heavy job (a nightly
+    run, a backup, an upload) that was waiting; the next slice starts two
+    minutes later. `null` means the slice read everything it started.
   - `collectionsDone` of `collectionsTotal`: data types read to the end so
     far.
   - `collections`: one entry per data type the slice touched, with its

src/lib/google-health/__tests__/walk-resume-and-retry.test.ts

@@ -152,6 +152,22 @@ describe("a resumable walk", () => {
     expect(budget.stopReason).toBe("rate_limited");
   });
 
+  it("gives way to a waiting heavy job after its first page, keeping its place", async () => {
+    safeFetchMock.mockImplementation(async (url: string) =>
+      page(Number(new URL(url).searchParams.get("pageToken") ?? 0), true),
+    );
+    const budget = googleHealthSliceBudget(60_000, 500, Date.now, () => true);
+    const { outcome } = await runWithGoogleHealthStop(
+      () => false,
+      () => walk({ resumable: true }),
+      budget,
+    );
+    // Never before a page is on disk: a slice that ended empty would fail.
+    expect(safeFetchMock).toHaveBeenCalledTimes(1);
+    expect(outcome.truncated).toBe(true);
+    expect(budget.stopReason).toBe("lane_wanted");
+  });
+
   it("still fails on a rate limit outside a slice", async () => {
     safeFetchMock.mockResolvedValue(
       page(0, true, 429, { "retry-after": "3600" }),
@@ -178,6 +194,20 @@ describe("a walk that is not resumable", () => {
     expect(outcome.truncated).toBe(false);
   });
 
+  it("does not give way part-way, since it would start over", async () => {
+    safeFetchMock.mockImplementation(async (url: string) => {
+      const n = Number(new URL(url).searchParams.get("pageToken") ?? 0);
+      return page(n, n < 3);
+    });
+    const { outcome } = await runWithGoogleHealthStop(
+      () => false,
+      () => walk(),
+      googleHealthSliceBudget(60_000, 500, Date.now, () => true),
+    );
+    expect(safeFetchMock).toHaveBeenCalledTimes(4);
+    expect(outcome.truncated).toBe(false);
+  });
+
   it("still stops on the job's own budget", async () => {
     safeFetchMock.mockImplementation(async () => page(0, true));
     let asked = 0;

src/lib/google-health/client.ts

@@ -155,7 +155,7 @@ interface GoogleHealthWalkLimits {
 
 /** Why a backfill slice stopped walking. */
 export type GoogleHealthSliceStopReason =
-  "time_budget" | "page_budget" | "rate_limited" | "job_budget";
+  "time_budget" | "page_budget" | "rate_limited" | "job_budget" | "lane_wanted";
 
 /**
  * One backfill slice's share of the work: a span of wall-clock time and a
@@ -184,6 +184,14 @@ export interface GoogleHealthSliceBudget {
   skippedPages: number;
   stopReason: GoogleHealthSliceStopReason | null;
   now: () => number;
+  /**
+   * Whether another heavy job waits for the lane this slice holds
+   * (`laneWanted` in `jobs/heavy-lane.ts`). Asked where the slice's own
+   * limits are, so giving way keeps every walk's place, and only once the
+   * slice has read a page, so the slice it ends has something on disk and is
+   * continued rather than failed.
+   */
+  yieldWanted: () => boolean;
 }
 
 /** A fresh slice budget of `ms` milliseconds and `maxPages` pages. */
@@ -191,6 +199,7 @@ export function googleHealthSliceBudget(
   ms: number,
   maxPages: number,
   now: () => number = Date.now,
+  yieldWanted: () => boolean = () => false,
 ): GoogleHealthSliceBudget {
   return {
     ms,
@@ -200,6 +209,7 @@ export function googleHealthSliceBudget(
     skippedPages: 0,
     stopReason: null,
     now,
+    yieldWanted,
   };
 }
 
@@ -212,6 +222,8 @@ function sliceExhausted(slice: GoogleHealthSliceBudget): boolean {
   if (slice.now() >= slice.deadline) slice.stopReason = "time_budget";
   else if (slice.pages - slice.skippedPages >= slice.maxPages) {
     slice.stopReason = "page_budget";
+  } else if (slice.pages > 0 && slice.yieldWanted()) {
+    slice.stopReason = "lane_wanted";
   }
   return slice.stopReason !== null;
 }

src/lib/jobs/__tests__/google-health-backfill.test.ts

@@ -133,6 +133,10 @@ import {
   runGoogleHealthBackfillForUser,
 } from "../google-health-backfill";
 import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission";
+import {
+  LANE_DEFER_SECONDS,
+  LANE_YIELD_DELAY_SECONDS,
+} from "@/lib/jobs/heavy-lane";
 import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys";
 import {
   GOOGLE_HEALTH_DATA_TYPES,
@@ -472,6 +476,35 @@ describe("runGoogleHealthBackfillForUser — slices", () => {
     });
   });
 
+  it("queues the next slice past the waiting job's return after giving way", async () => {
+    resumeState.advances = ["fetchHeartRate"];
+    syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true });
+    const client = await import("@/lib/google-health/client");
+    const spy = vi
+      .spyOn(client, "googleHealthSliceBudget")
+      .mockImplementation(() => ({
+        ms: 300_000,
+        deadline: null,
+        maxPages: 500,
+        pages: 3,
+        skippedPages: 0,
+        stopReason: "lane_wanted",
+        now: Date.now,
+        yieldWanted: () => true,
+      }));
+    try {
+      await expect(runGoogleHealthBackfillForUser("u1")).resolves.toMatchObject(
+        { continued: true },
+      );
+    } finally {
+      spy.mockRestore();
+    }
+    expect(bossSend.mock.calls[0]![2]).toMatchObject({
+      startAfter: LANE_YIELD_DELAY_SECONDS,
+    });
+    expect(LANE_YIELD_DELAY_SECONDS).toBeGreaterThan(LANE_DEFER_SECONDS);
+  });
+
   it("waits longer before the next slice when this one ended on a rate limit", async () => {
     resumeState.advances = ["fetchHeartRate"];
     syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true });
@@ -488,6 +521,7 @@ describe("runGoogleHealthBackfillForUser — slices", () => {
         skippedPages: 0,
         stopReason: "rate_limited",
         now: Date.now,
+        yieldWanted: () => false,
       }));
     try {
       await runGoogleHealthBackfillForUser("u1");

src/lib/jobs/__tests__/heavy-lane.test.ts

@@ -8,8 +8,10 @@ import type { Job } from "pg-boss";
 
 vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
 vi.mock("@/lib/db", () => ({ prisma: {} }));
+vi.mock("@/lib/logging/signal", () => ({ emitSignal: vi.fn() }));
 
-import { heavyLane } from "../heavy-lane";
+import { emitSignal } from "@/lib/logging/signal";
+import { heavyLane, LANE_WAIT_WARN_MS, laneWanted } from "../heavy-lane";
 import { jobDone } from "../job-outcome";
 
 type TestJob = Job<{ force?: boolean }>;
@@ -36,7 +38,7 @@ function deferrals() {
   const back: Array<{ queue: string; ids: string[] }> = [];
   const defer = vi.fn(async (queue: string, jobs: readonly TestJob[]) => {
     back.push({ queue, ids: jobs.map((j) => j.id) });
-    return jobs.length;
+    return { deferred: jobs.length, oldestCreatedOn: null };
   });
   return { back, defer };
 }
@@ -160,3 +162,107 @@ describe("heavyLane", () => {
     ).toThrow(/not in HEAVY_QUEUES/);
   });
 });
+
+describe("laneWanted", () => {
+  it("tells the holder, and only the holder, that another queue is waiting", async () => {
+    const seen: Record<string, boolean[]> = { holder: [], beside: [] };
+    let release!: () => void;
+    const released = new Promise<void>((r) => (release = r));
+    const { defer } = deferrals();
+    expect(laneWanted()).toBe(false);
+    const held = heavyLane(
+      "integration-backfill-admission",
+      async () => {
+        seen.holder.push(laneWanted());
+        await released;
+        seen.holder.push(laneWanted());
+        return jobDone({ jobs: 1 });
+      },
+      undefined,
+      defer,
+    )([job("slice")]);
+    await tick();
+    // A job of the same queue going back is not a reason to give way.
+    await heavyLane(
+      "integration-backfill-admission",
+      vi.fn(),
+      undefined,
+      defer,
+    )([job("next-slice")]);
+    expect(laneWanted()).toBe(false);
+    await heavyLane(
+      "insights-general-status",
+      vi.fn(),
+      undefined,
+      defer,
+    )([job("nightly")]);
+    // A light job of a heavy queue running beside the lane is never asked, even now.
+    await heavyLane(
+      "insight-pregenerate",
+      async () => {
+        seen.beside.push(laneWanted());
+        return jobDone({ jobs: 1 });
+      },
+      () => false,
+      defer,
+    )([job("warm")]);
+    release();
+    await held;
+    expect(seen.holder).toEqual([false, true]);
+    expect(seen.beside).toEqual([false]);
+    // The next holder starts with a clean slate.
+    await heavyLane(
+      "data-backup",
+      async () => {
+        seen.holder.push(laneWanted());
+        return jobDone({ jobs: 1 });
+      },
+      undefined,
+      defer,
+    )([job("weekly")]);
+    expect(seen.holder).toEqual([false, true, false]);
+  });
+});
+
+describe("a long wait for the lane", () => {
+  it("is reported at warn once per holder when a job has waited past the limit", async () => {
+    vi.mocked(emitSignal).mockClear();
+    const now = Date.parse("2026-10-20T06:00:00Z");
+    let release!: () => void;
+    const released = new Promise<void>((r) => (release = r));
+    const queuedAt = (ms: number) => async () => ({
+      deferred: 1,
+      oldestCreatedOn: new Date(now - ms),
+    });
+    const held = heavyLane("integration-backfill-admission", async () => {
+      await released;
+      return jobDone({ jobs: 1 });
+    })([job("import")]);
+    await tick();
+    const wait = (ms: number) =>
+      heavyLane(
+        "insights-general-status",
+        vi.fn(),
+        undefined,
+        queuedAt(ms),
+        () => now,
+      )([job("nightly")]);
+
+    await wait(LANE_WAIT_WARN_MS - 60_000);
+    expect(emitSignal).not.toHaveBeenCalled();
+    await wait(LANE_WAIT_WARN_MS + 60_000);
+    await wait(LANE_WAIT_WARN_MS + 120_000);
+    expect(emitSignal).toHaveBeenCalledTimes(1);
+    expect(emitSignal).toHaveBeenCalledWith({
+      action: "job.heavy_lane.waiting_long",
+      level: "warn",
+      meta: {
+        queue: "insights-general-status",
+        holder: "integration-backfill-admission",
+        waited_minutes: 121,
+      },
+    });
+    release();
+    await held;
+  });
+});

src/lib/jobs/google-health-backfill.ts

@@ -38,9 +38,11 @@ import {
 import { GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock";
 import {
   googleHealthSliceBudget,
+  type GoogleHealthSliceStopReason,
   runWithGoogleHealthStop,
 } from "@/lib/google-health/client";
 import { reportJobProgress } from "@/lib/jobs/job-observer";
+import { LANE_YIELD_DELAY_SECONDS, laneWanted } from "@/lib/jobs/heavy-lane";
 import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys";
 
 export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill";
@@ -122,6 +124,21 @@ export function googleHealthBackfillRetryKey(
   return `${identity}|retry-${new Date(dueAt).toISOString()}`;
 }
 
+/**
+ * The pause before the next slice, by why this one stopped: longer after a
+ * rate limit, and past the waiting job's return after giving way to it, so
+ * that job takes the lane before this import does again.
+ */
+export function nextSliceDelaySeconds(
+  stopReason: GoogleHealthSliceStopReason | null,
+): number {
+  if (stopReason === "rate_limited") {
+    return GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS;
+  }
+  if (stopReason === "lane_wanted") return LANE_YIELD_DELAY_SECONDS;
+  return INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS;
+}
+
 /** Test seam: a slice's budget, when not the production one. */
 export interface GoogleHealthBackfillSliceOptions {
   sliceMs?: number;
@@ -240,6 +257,8 @@ export async function runGoogleHealthBackfillForUser(
   const budget = googleHealthSliceBudget(
     sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS,
     sliceOptions.slicePages ?? GOOGLE_HEALTH_BACKFILL_SLICE_PAGES,
+    Date.now,
+    laneWanted,
   );
   const startedAt = Date.now();
   const {
@@ -387,9 +406,7 @@ export async function runGoogleHealthBackfillForUser(
     const progressed = finished > 0 || advanced.size > 0 || newlyParked > 0;
     if (progressed) {
       const sent = await sendNextSlice(
-        budget.stopReason === "rate_limited"
-          ? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS
-          : INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
+        nextSliceDelaySeconds(budget.stopReason),
       );
       if (sent) {
         annotate({

src/lib/jobs/heavy-lane.ts

@@ -23,12 +23,24 @@
  * longer than its own expiry was failed, stopped counting as in use, and had
  * its staged file swept from under it. A deferred job keeps its id, its
  * retries and its staged upload, and its expiry starts when it runs.
+ *
+ * A deferred job also leaves word that the lane is wanted (`laneWanted`). A
+ * holder that runs in slices asks it at the boundaries where it can already
+ * stop and keep its place, ends its slice there, and queues the rest a little
+ * later than the job that waited (`LANE_YIELD_DELAY_SECONDS`), so a long
+ * import makes room for the nightly runs and for an upload instead of holding
+ * them back for hours. A holder that cannot stop part-way runs to its end; a
+ * job that has stood in line for `LANE_WAIT_WARN_MS` is reported at `warn`
+ * (`job.heavy_lane.waiting_long`), once per holder, so a lane that is held
+ * for hours shows in the alerts instead of only in `info` lines.
  */
+import { AsyncLocalStorage } from "node:async_hooks";
 import type { Job } from "pg-boss";
 
 import { prisma } from "@/lib/db";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { annotate } from "@/lib/logging/context";
+import { emitSignal } from "@/lib/logging/signal";
 
 export interface HeavyQueue {
   /** What makes the job heavy. */
@@ -73,21 +85,39 @@ export const HEAVY_QUEUES: Readonly<Record<string, HeavyQueue>> = {
 /** How far ahead a deferred job is put back in the queue. */
 export const LANE_DEFER_SECONDS = 60;
 
+/**
+ * How far ahead a holder that gave way queues the rest of its work: past the
+ * deferred job's own return, so the job it gave way to takes the lane first.
+ */
+export const LANE_YIELD_DELAY_SECONDS = 2 * LANE_DEFER_SECONDS;
+
+/** How long a job may stand in line before the wait is reported at `warn`. */
+export const LANE_WAIT_WARN_MS = 2 * 60 * 60 * 1000;
+
+/** What one trip back to the queue did. */
+export interface LaneDeferral {
+  /** Rows that went back. */
+  deferred: number;
+  /** When the oldest of them was first queued, if any went back. */
+  oldestCreatedOn: Date | null;
+}
+
 /**
  * Put `jobs` of `queue` back in the queue to run `seconds` from now. The same
  * reset pg-boss's own `restore()` makes, with `start_after` moved in the same
  * statement: a restore followed by a separate update leaves a moment in which
  * another fetch could claim the row, and the completion of this handler would
  * then settle that newer claim, since a restored job's next claim keeps its
- * attempt number. Answers how many rows went back.
+ * attempt number. Answers how many rows went back, and since when the oldest
+ * of them has been waiting.
  */
 export async function deferLaneJobs(
   queue: string,
   jobs: readonly Job<unknown>[],
   seconds: number = LANE_DEFER_SECONDS,
-): Promise<number> {
+): Promise<LaneDeferral> {
   const ids = jobs.map((job) => job.id);
-  return prisma.$executeRaw`
+  const rows = await prisma.$queryRaw<Array<{ created_on: Date }>>`
     UPDATE pgboss.job
     SET state = 'created',
         started_on = NULL,
@@ -97,42 +127,102 @@ export async function deferLaneJobs(
     WHERE name = ${queue}
       AND id = ANY(${ids}::uuid[])
       AND state = 'active'
+    RETURNING created_on
   `;
+  const oldest = rows.reduce<Date | null>(
+    (min, row) => (min === null || row.created_on < min ? row.created_on : min),
+    null,
+  );
+  return { deferred: rows.length, oldestCreatedOn: oldest };
 }
 
-let busy = false;
+/** The job holding the lane, and what has happened while it held it. */
+interface LaneHold {
+  queue: string;
+  /** The first other queue that found the lane taken during this hold. */
+  wantedBy: string | null;
+  /** Queues whose long wait has been reported during this hold. */
+  warned: Set<string>;
+}
+
+let hold: LaneHold | null = null;
+const holdScope = new AsyncLocalStorage<LaneHold>();
+
+/**
+ * Whether the caller holds the lane and a job of another queue has been sent
+ * back while it did. False outside the lane, so a job that runs beside the
+ * lane (a light job of a heavy queue, any other queue) never gives way.
+ */
+export function laneWanted(): boolean {
+  const current = holdScope.getStore();
+  return current !== undefined && current === hold && current.wantedBy !== null;
+}
+
+function reportLongWait(
+  current: LaneHold,
+  queue: string,
+  deferral: LaneDeferral,
+  now: number,
+): void {
+  if (!deferral.oldestCreatedOn || current.warned.has(queue)) return;
+  const waitedMs = now - deferral.oldestCreatedOn.getTime();
+  if (waitedMs < LANE_WAIT_WARN_MS) return;
+  current.warned.add(queue);
+  emitSignal({
+    action: "job.heavy_lane.waiting_long",
+    level: "warn",
+    meta: {
+      queue,
+      holder: current.queue,
+      waited_minutes: Math.floor(waitedMs / 60_000),
+    },
+  });
+}
 
 /**
  * Wrap `handler` so the jobs of `queue` run in the lane. `inLane` picks the
  * jobs that are heavy where a queue carries light ones too (a single
  * account's warm beside the nightly walk); a batch with none of them runs at
- * once. `defer` is the way back to the queue; tests replace it.
+ * once. `defer` is the way back to the queue; tests replace it, and `now`.
  */
 export function heavyLane<J extends Job<unknown>>(
   queue: string,
   handler: (jobs: J[]) => Promise<JobOutcome>,
   inLane: (job: J) => boolean = () => true,
-  defer: (queue: string, jobs: readonly J[]) => Promise<number> = deferLaneJobs,
+  defer: (
+    queue: string,
+    jobs: readonly J[],
+  ) => Promise<LaneDeferral> = deferLaneJobs,
+  now: () => number = Date.now,
 ): (jobs: J[]) => Promise<JobOutcome> {
   if (!HEAVY_QUEUES[queue]) {
     throw new Error(`heavyLane: ${queue} is not in HEAVY_QUEUES`);
   }
   return async (jobs) => {
     if (!jobs.some(inLane)) return handler(jobs);
-    if (busy) {
-      const deferred = await defer(queue, jobs);
+    const current = hold;
+    if (current) {
+      const deferral = await defer(queue, jobs);
+      if (queue !== current.queue) current.wantedBy ??= queue;
       annotate({
         action: { name: "job.heavy_lane.deferred" },
-        meta: { queue, jobs: jobs.length, deferred },
+        meta: {
+          queue,
+          holder: current.queue,
+          jobs: jobs.length,
+          deferred: deferral.deferred,
+        },
       });
-      return jobDone({ lane_deferred: deferred });
+      reportLongWait(current, queue, deferral, now());
+      return jobDone({ lane_deferred: deferral.deferred });
     }
-    busy = true;
+    const entered: LaneHold = { queue, wantedBy: null, warned: new Set() };
+    hold = entered;
     annotate({ action: { name: "job.heavy_lane.entered" }, meta: { queue } });
     try {
-      return await handler(jobs);
+      return await holdScope.run(entered, () => handler(jobs));
     } finally {
-      busy = false;
+      hold = null;
     }
   };
 }

tests/integration/heavy-lane-defer.test.ts

@@ -59,7 +59,13 @@ describe("deferLaneJobs", () => {
     expect(fetched?.id).toBe(id);
     expect((await row(id!))?.state).toBe("active");
 
-    expect(await deferLaneJobs(WAITER_QUEUE, [fetched!], 30)).toBe(1);
+    const deferral = await deferLaneJobs(WAITER_QUEUE, [fetched!], 30);
+    expect(deferral.deferred).toBe(1);
+    // When it was first queued, so a long wait can be told from a short one.
+    expect(deferral.oldestCreatedOn).toBeInstanceOf(Date);
+    expect(Date.now() - deferral.oldestCreatedOn!.getTime()).toBeLessThan(
+      60_000,
+    );
     expect(await row(id!)).toMatchObject({
       state: "created",
       retry_count: 0,
@@ -77,7 +83,10 @@ describe("deferLaneJobs", () => {
   it("leaves a job alone that is no longer active", async () => {
     const id = await boss.send(WAITER_QUEUE, { probe: 2 });
     const job = { id, name: WAITER_QUEUE, data: {} } as unknown as Job<object>;
-    expect(await deferLaneJobs(WAITER_QUEUE, [job], 30)).toBe(0);
+    expect(await deferLaneJobs(WAITER_QUEUE, [job], 30)).toEqual({
+      deferred: 0,
+      oldestCreatedOn: null,
+    });
     expect((await row(id!))?.future).toBe(false);
     await boss.deleteJob(WAITER_QUEUE, id!);
   });