fix(jobs): keep a full-history import from running beside its own retry

The shared admission queue sent its jobs without an expiry, so every one
took pg-boss's fifteen minutes. Against pg-boss 12.34, a handler that runs
past its expiry is not stopped: the job is failed, its abort signal fires,
the worker and group slots are freed, and the retry of the same job starts
in the same worker beside the run it replaced. A per-minute heart-rate
history is thousands of pages, so on a slow link one import turned into up
to four copies of itself, on the one queue meant to run a single import at
a time.

- The admission job now carries a sixteen-hour expiry, sized for the
  longest legitimate import and under pg-boss's 24-hour ceiling.
- Each run holds a session advisory lock on its identity (kind, provider,
  account) on a connection of its own. A delivery that finds it taken does
  nothing and says so in the log; the running import stamps its marker or
  is offered again at the next boot. Ending the connection or the process
  releases the lock.
- The Google Health backfill runs under the job's budget and abort signal.
  The page walk asks before each request, stops there, and reports the run
  truncated, so it throws without the completion stamp and is retried.
- The drain moves out of the registrar into its own module so it can be
  driven against a real pg-boss.

The integration test pins the pg-boss behaviour and runs the lane's
handler with a two-second expiry: without the lock the import ran twice.

Refs #1023

src/lib/google-health/client.ts

@@ -121,6 +121,28 @@ export async function runWithGoogleHealthClientOutcome<T>(
   return { result, outcome };
 }
 
+/**
+ * A stop condition for the walks below, set by a caller that runs under a
+ * time limit (the full-history backfill, which runs under its pg-boss job's
+ * budget and abort signal). A walk asks it before each request and, when it
+ * says stop, ends there and reports itself truncated, the same outcome as
+ * reaching the page ceiling: the resource is incomplete, the cycle is not
+ * stamped, and the reason shown is that only part of the history was
+ * processed. Outside such a scope the walks never stop early.
+ */
+const stopStorage = new AsyncLocalStorage<() => boolean>();
+
+function stopRequested(): boolean {
+  return stopStorage.getStore()?.() === true;
+}
+
+export async function runWithGoogleHealthStop<T>(
+  shouldStop: () => boolean,
+  fn: () => Promise<T>,
+): Promise<T> {
+  return stopStorage.run(shouldStop, fn);
+}
+
 export const GOOGLE_HEALTH_API_BASE = "https://health.googleapis.com/v4";
 const GOOGLE_OAUTH_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth";
 const GOOGLE_OAUTH_TOKEN_URL = "https://oauth2.googleapis.com/token";
@@ -769,8 +791,13 @@ export async function forEachDataPointPage(
     let fetched = 0;
     let pageToken: string | null | undefined;
     let pageCount = 0;
+    let stopped = false;
 
     do {
+      if (stopRequested()) {
+        stopped = true;
+        break;
+      }
       const params = new URLSearchParams({ pageSize: String(pageSize) });
       if (query.start) {
         const { field, bound } = incrementalFilter(
@@ -830,7 +857,7 @@ export async function forEachDataPointPage(
       await onPage(page);
     } while (pageToken && pageCount < maxPages);
 
-    notePagination(pageCount, fetched, Boolean(pageToken));
+    notePagination(pageCount, fetched, stopped || Boolean(pageToken));
   };
 
   // The fallback only exists for filtered daily-summary reads — every other
@@ -1055,6 +1082,10 @@ export async function fetchDailyRollUp(
     let totalPages = 0;
     let truncated = false;
     for (const chunk of chunkCivilRange(startCivil, endCivil, maxDays)) {
+      if (stopRequested()) {
+        truncated = true;
+        break;
+      }
       let pageToken: string | null | undefined;
       let pageCount = 0;
       do {

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

@@ -27,6 +27,12 @@ vi.mock("@/lib/logging/context", () => ({
   getEvent: () => null,
 }));
 
+const { safeFetchMock } = vi.hoisted(() => ({ safeFetchMock: vi.fn() }));
+vi.mock("@/lib/safe-fetch", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@/lib/safe-fetch")>();
+  return { ...actual, safeFetch: safeFetchMock };
+});
+
 vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: isReauthRequiredMock,
 }));
@@ -39,6 +45,11 @@ vi.mock("@/lib/google-health/sync", () => ({
 }));
 
 import { runGoogleHealthBackfillForUser } from "../google-health-backfill";
+import {
+  GOOGLE_HEALTH_DATA_TYPES,
+  forEachDataPointPage,
+  runWithGoogleHealthClientOutcome,
+} from "@/lib/google-health/client";
 
 beforeEach(() => {
   vi.clearAllMocks();
@@ -83,4 +94,35 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
     expect(syncUserGoogleHealthMock).not.toHaveBeenCalled();
     expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
   });
+
+  it("stops the walk when its budget runs out and throws without stamping", async () => {
+    // An endless heart-rate stream: every page names a next one.
+    safeFetchMock.mockImplementation(async () => ({
+      status: 200,
+      ok: true,
+      json: async () => ({ dataPoints: [{}], nextPageToken: "more" }),
+    }));
+    // The real walk under the sync mock, reporting what the real sync reports
+    // for a truncated resource.
+    syncUserGoogleHealthMock.mockImplementation(async () => {
+      const { outcome } = await runWithGoogleHealthClientOutcome(() =>
+        forEachDataPointPage(
+          GOOGLE_HEALTH_DATA_TYPES.heartRate,
+          "token",
+          "fetchHeartRate",
+          { maxPages: 10_000 },
+          () => {},
+        ),
+      );
+      return { imported: outcome.fetched, failed: outcome.truncated };
+    });
+    let asked = 0;
+    const shouldStop = () => ++asked > 3;
+
+    await expect(
+      runGoogleHealthBackfillForUser("u1", shouldStop),
+    ).rejects.toThrow(/incomplete/);
+    expect(safeFetchMock).toHaveBeenCalledTimes(3);
+    expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
+  });
 });

src/lib/jobs/__tests__/integration-backfill-admission.test.ts

@@ -6,6 +6,7 @@ import {
   INTEGRATION_BACKFILL_ADMISSION_GROUP,
   INTEGRATION_BACKFILL_ADMISSION_QUEUE,
   INTEGRATION_BACKFILL_BOOT_ORDER,
+  INTEGRATION_BACKFILL_EXPIRE_SECONDS,
   INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY,
   INTEGRATION_BACKFILL_STAGGER_SECONDS,
   bootStaggerSecondsFor,
@@ -109,12 +110,20 @@ describe("shared full-history backfill admission", () => {
       expect.objectContaining({
         singletonKey: "sleep-timeline-backfill|WHOOP|u1",
         group: INTEGRATION_BACKFILL_ADMISSION_GROUP,
+        expireInSeconds: INTEGRATION_BACKFILL_EXPIRE_SECONDS,
       }),
     );
     expect(first).toBe("job-1");
     expect(duplicate).toBeNull();
   });
 
+  it("gives the admission job an expiry sized for a long import, under pg-boss's 24-hour ceiling", () => {
+    // pg-boss's default is 900 s; a full-history heart-rate import runs for
+    // hours, and past the expiry pg-boss retries the job beside the run.
+    expect(INTEGRATION_BACKFILL_EXPIRE_SECONDS).toBe(16 * 60 * 60);
+    expect(INTEGRATION_BACKFILL_EXPIRE_SECONDS).toBeLessThan(24 * 60 * 60);
+  });
+
   it("keeps source enqueue singleton keys while applying stagger only to boot discovery", () => {
     expect(integrationBackfillSourceOptions("whoop-backfill|u1", 60)).toEqual(
       expect.objectContaining({

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

@@ -18,6 +18,7 @@ import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfil
 import { isReauthRequired } from "@/lib/integrations/status";
 import { GOOGLE_HEALTH_INTEGRATION_KEY } from "@/lib/google-health/sync-core";
 import { syncUserGoogleHealth } from "@/lib/google-health/sync";
+import { runWithGoogleHealthStop } from "@/lib/google-health/client";
 
 export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill";
 
@@ -47,9 +48,17 @@ export interface GoogleHealthBackfillPayload {
  * just burn the retry budget on the same no-op: return WITHOUT stamping — the
  * boot discovery re-enqueues the account on the next boot, and the stamp lands
  * once the user reconnects and a clean walk completes.
+ *
+ * `shouldStop` is the admission job's budget (`jobBudget`). A per-minute
+ * heart-rate history runs to thousands of pages, and the walk asks it before
+ * every page: once it says stop, the walk ends where it is, the cycle reports
+ * itself truncated, and this throws like any other incomplete run, without
+ * the stamp. The rows already written stay; the retry or the next boot walks
+ * again and the key-stable upserts overwrite them.
  */
 export async function runGoogleHealthBackfillForUser(
   userId: string,
+  shouldStop: () => boolean = () => false,
 ): Promise<{ imported: number }> {
   if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) {
     annotate({
@@ -61,9 +70,9 @@ export async function runGoogleHealthBackfillForUser(
     return { imported: 0 };
   }
 
-  const { imported, failed } = await syncUserGoogleHealth(userId, {
-    fullSync: true,
-  });
+  const { imported, failed } = await runWithGoogleHealthStop(shouldStop, () =>
+    syncUserGoogleHealth(userId, { fullSync: true }),
+  );
 
   if (failed) {
     // Surface through the pg-boss retry path (retryLimit 3, backoff). If the

src/lib/jobs/integration-backfill-admission.ts

@@ -49,6 +49,27 @@ const RETRY_OPTIONS = {
   retryBackoff: true,
 } as const;
 
+/**
+ * The admission job's expiry: sixteen hours, so a legitimately long import
+ * is never declared dead underneath itself.
+ *
+ * Without it every admission job took pg-boss's default of fifteen minutes.
+ * A full-history import of a per-minute heart-rate stream is thousands of
+ * pages (a page is 1 000 readings, about seventeen hours of a watch's
+ * history, so five years is some 2 600 pages), and at a few seconds a page on
+ * a slow link that is hours. Past the expiry pg-boss fails the job and retries
+ * it while the handler keeps running (`integration-backfill-lock.ts`), and the
+ * import that could not finish in fifteen minutes started over beside itself.
+ *
+ * The runs that honour the job's budget (`jobBudget`, three quarters of this)
+ * stop on their own at twelve hours, which is still some 8 000 pages at five
+ * seconds each, and are retried or re-offered at the next boot. The expiry is
+ * also what ends a run that hangs: the lane is freed for the next import, and
+ * the import lock keeps a retry from starting beside the stuck run. pg-boss
+ * refuses an expiry of 24 hours or more.
+ */
+export const INTEGRATION_BACKFILL_EXPIRE_SECONDS = 16 * 60 * 60;
+
 export function bootStaggerSecondsFor(kind: IntegrationBackfillKind): number {
   return (
     (INTEGRATION_BACKFILL_BOOT_ORDER.indexOf(kind) + 1) *
@@ -83,9 +104,25 @@ export async function enqueueIntegrationBackfillAdmission(
   boss: BossSender,
   payload: IntegrationBackfillAdmissionPayload,
 ): Promise<string | null> {
-  return boss.send(INTEGRATION_BACKFILL_ADMISSION_QUEUE, payload, {
+  return boss.send(
+    INTEGRATION_BACKFILL_ADMISSION_QUEUE,
+    payload,
+    integrationBackfillAdmissionSendOptions(payload),
+  );
+}
+
+/**
+ * The send options of one admission job. The options ride on the send, not on
+ * `createQueue`, whose `ON CONFLICT DO NOTHING` leaves an existing queue's
+ * defaults as they were.
+ */
+export function integrationBackfillAdmissionSendOptions(
+  payload: IntegrationBackfillAdmissionPayload,
+): SendOptions {
+  return {
     ...RETRY_OPTIONS,
+    expireInSeconds: INTEGRATION_BACKFILL_EXPIRE_SECONDS,
     singletonKey: integrationBackfillAdmissionSingletonKey(payload),
     group: INTEGRATION_BACKFILL_ADMISSION_GROUP,
-  });
+  };
 }

src/lib/jobs/integration-backfill-drain.ts

@@ -0,0 +1,188 @@
+/**
+ * The handler of the shared full-history admission queue.
+ *
+ * Every provider's backfill and repair reaches this one queue, and the worker
+ * runs it one job at a time across the whole instance
+ * (`register-integration-sync.ts`). Each job runs under two guards:
+ *
+ *   - The import lock (`withIntegrationBackfillLock`), keyed on the job's
+ *     identity. A retry that pg-boss starts after the job's expiry while the
+ *     first run is still going finds the lock taken and does nothing, so one
+ *     account's import of one kind never runs twice at once.
+ *   - The job's budget (`jobBudget`), handed to the runners that can stop
+ *     part-way. A run that stops on its budget throws without stamping its
+ *     completion marker, so it is retried, and after that the next boot's
+ *     discovery offers it again.
+ */
+import type { Job } from "pg-boss";
+
+import { runWhoopBackfillForUser } from "@/lib/jobs/whoop-backfill";
+import { runFitbitBackfillForUser } from "@/lib/jobs/fitbit-backfill";
+import { runGoogleHealthBackfillForUser } from "@/lib/jobs/google-health-backfill";
+import { runGoogleHealthSleepRepairForUser } from "@/lib/jobs/google-health-sleep-repair";
+import { runFitbitSleepRepairForUser } from "@/lib/jobs/fitbit-sleep-repair";
+import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill";
+import { runSleepTimelineBackfillForUser } from "@/lib/jobs/sleep-timeline-backfill";
+import { runLabBiomarkerBackfillForUser } from "@/lib/jobs/lab-biomarker-backfill";
+import {
+  integrationBackfillAdmissionSingletonKey,
+  type IntegrationBackfillAdmissionPayload,
+} from "@/lib/jobs/integration-backfill-admission";
+import { withIntegrationBackfillLock } from "@/lib/jobs/integration-backfill-lock";
+import { jobBudget } from "@/lib/jobs/job-budget";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { workerLog } from "@/lib/jobs/reminder/shared";
+import { withBackgroundEvent } from "@/lib/logging/background";
+
+interface RunCounts {
+  imported: number;
+  removed: number;
+  deleted: number;
+  markers: number;
+  linked: number;
+}
+
+async function runAdmitted(
+  payload: IntegrationBackfillAdmissionPayload,
+  shouldStop: () => boolean,
+  totals: RunCounts,
+): Promise<void> {
+  const { kind, data } = payload;
+  const { userId } = data;
+  switch (kind) {
+    case "whoop-backfill": {
+      const { imported } = await runWhoopBackfillForUser(userId);
+      totals.imported += imported;
+      workerLog("info", `[whoop-backfill] user=${userId} imported=${imported}`);
+      return;
+    }
+    case "fitbit-backfill": {
+      const { imported } = await runFitbitBackfillForUser(userId);
+      totals.imported += imported;
+      workerLog(
+        "info",
+        `[fitbit-backfill] user=${userId} imported=${imported}`,
+      );
+      return;
+    }
+    case "google-health-backfill": {
+      const { imported } = await runGoogleHealthBackfillForUser(
+        userId,
+        shouldStop,
+      );
+      totals.imported += imported;
+      workerLog(
+        "info",
+        `[google-health-backfill] user=${userId} imported=${imported}`,
+      );
+      return;
+    }
+    case "google-health-sleep-repair": {
+      const { imported } = await runGoogleHealthSleepRepairForUser(userId);
+      totals.imported += imported;
+      workerLog(
+        "info",
+        `[google-health-sleep-repair] user=${userId} imported=${imported}`,
+      );
+      return;
+    }
+    case "fitbit-sleep-repair": {
+      const { imported, removed } = await runFitbitSleepRepairForUser(userId);
+      totals.imported += imported;
+      totals.removed += removed;
+      workerLog(
+        "info",
+        `[fitbit-sleep-repair] user=${userId} imported=${imported} removed=${removed}`,
+      );
+      return;
+    }
+    case "sleep-timeline-backfill": {
+      if (!data.provider) {
+        throw new Error(
+          "sleep-timeline backfill admission requires a provider",
+        );
+      }
+      const { deleted, imported } = await runSleepTimelineBackfillForUser(
+        userId,
+        data.provider,
+      );
+      totals.deleted += deleted;
+      totals.imported += imported;
+      workerLog(
+        "info",
+        `[sleep-timeline-backfill] user=${userId} provider=${data.provider} deleted=${deleted} imported=${imported}`,
+      );
+      return;
+    }
+    case "lab-biomarker-backfill": {
+      const { markers, linked } = await runLabBiomarkerBackfillForUser(userId);
+      totals.markers += markers;
+      totals.linked += linked;
+      workerLog(
+        "info",
+        `[lab-biomarker-backfill] user=${userId} markers=${markers} linked=${linked}`,
+      );
+      return;
+    }
+    case "strava-backfill": {
+      const { imported } = await runStravaBackfillForUser(userId);
+      totals.imported += imported;
+      workerLog(
+        "info",
+        `[strava-backfill] user=${userId} imported=${imported}`,
+      );
+      return;
+    }
+  }
+}
+
+/**
+ * Drain a batch of admission jobs. There is 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 either did its work or found its
+ * import already running, and the counts each runner returns are what the
+ * pass reports.
+ */
+export async function drainIntegrationBackfillAdmission(
+  jobs: Job<IntegrationBackfillAdmissionPayload>[],
+): Promise<JobOutcome> {
+  const shouldStop = jobBudget(jobs);
+  const totals: RunCounts = {
+    imported: 0,
+    removed: 0,
+    deleted: 0,
+    markers: 0,
+    linked: 0,
+  };
+  let alreadyRunning = 0;
+  for (const job of jobs) {
+    const identity = integrationBackfillAdmissionSingletonKey(job.data);
+    const run = await withIntegrationBackfillLock(identity, () =>
+      runAdmitted(job.data, shouldStop, totals),
+    );
+    if (!run.ran) {
+      alreadyRunning += 1;
+      await withBackgroundEvent(
+        "integration_backfill.already_running",
+        async (event) => {
+          event.addMeta("identity", identity);
+          event.addMeta("job_id", job.id);
+          event.addMeta(
+            "reason",
+            "an earlier delivery of this import still holds its lock; this delivery does nothing",
+          );
+          event.elevateLevel("warn");
+        },
+      );
+    }
+  }
+  return jobDone({
+    jobs: jobs.length,
+    already_running: alreadyRunning,
+    imported: totals.imported,
+    removed: totals.removed,
+    deleted: totals.deleted,
+    markers: totals.markers,
+    linked: totals.linked,
+  });
+}

src/lib/jobs/integration-backfill-lock.ts

@@ -0,0 +1,70 @@
+/**
+ * One full-history import per account and kind at a time, across processes.
+ *
+ * The admission queue runs one import at a time (`groupConcurrency: 1`), but
+ * pg-boss only holds that line while the job is inside its expiry. Once the
+ * expiry passes, pg-boss fails the job, fires its abort signal, frees the
+ * worker slot and the group slot, and retries the job after its retry delay.
+ * It does not stop the handler. Measured against pg-boss 12.34
+ * (`tests/integration/integration-backfill-lane-expiry.test.ts`): a handler
+ * that ignores the signal was still running when the retry of the same job
+ * started beside it in the same worker, so the lane's "one at a time" did not
+ * hold for the one case it exists for, an import too long for its expiry.
+ *
+ * So each run holds a session-level advisory lock keyed on the job's identity
+ * (kind, provider where there is one, account) for as long as it runs. A
+ * delivery that finds the lock taken does no work: the run that holds it is
+ * still importing, and it either stamps its completion marker or leaves the
+ * account for the next boot's discovery. The lock lives on a connection of its
+ * own, outside the Prisma pool, because a session lock has to stay on one
+ * connection for the whole run and a pooled one is handed back between
+ * statements. Ending the connection releases the lock, and so does the
+ * process dying, so a crash never leaves an account locked.
+ *
+ * One extra connection per running import; the lane admits one import per
+ * process at a time, so that is one connection.
+ */
+import { Client } from "pg";
+
+/** The lock key for one admission job's identity. */
+function integrationBackfillLockKey(identity: string): string {
+  return `integration-backfill:${identity}`;
+}
+
+/** What a guarded run came to. */
+type GuardedRun<T> = { ran: true; result: T } | { ran: false };
+
+/**
+ * Run `run` while holding the import lock for `identity`. Resolves
+ * `{ ran: false }` at once, without calling `run`, when another run holds it.
+ * Rejects exactly as `run` does otherwise.
+ */
+export async function withIntegrationBackfillLock<T>(
+  identity: string,
+  run: () => Promise<T>,
+  connectionString: string | undefined = process.env.DATABASE_URL,
+): Promise<GuardedRun<T>> {
+  const client = new Client({ connectionString, keepAlive: true });
+  // An idle client whose server goes away emits `error`; unhandled, that
+  // takes the whole worker down. The run itself notices the database is
+  // gone through its own queries.
+  client.on("error", () => {});
+  await client.connect();
+  try {
+    const key = integrationBackfillLockKey(identity);
+    const { rows } = await client.query<{ held: boolean }>(
+      "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held",
+      [key],
+    );
+    if (!rows[0]?.held) return { ran: false };
+    try {
+      return { ran: true, result: await run() };
+    } finally {
+      await client
+        .query("SELECT pg_advisory_unlock(hashtextextended($1, 0))", [key])
+        .catch(() => {});
+    }
+  } finally {
+    await client.end().catch(() => {});
+  }
+}

src/lib/jobs/job-outcome.ts

@@ -43,6 +43,9 @@ export type JobFacts = Readonly<Record<string, JobFact>>;
  */
 export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "access_tokens_deleted",
+  // An admission delivery that found its import still running from an
+  // earlier delivery of the same job and did nothing.
+  "already_running",
   "appointment_addresses_cleared",
   "assessments_warmed",
   "auto_resolved",

src/lib/jobs/reminder/register-integration-sync.ts

@@ -17,56 +17,48 @@ import { PgBoss } from "pg-boss";
 import {
   WHOOP_BACKFILL_QUEUE,
   WHOOP_BACKFILL_CONCURRENCY,
-  runWhoopBackfillForUser,
   enqueueBootTimeWhoopBackfill,
   type WhoopBackfillPayload,
 } from "@/lib/jobs/whoop-backfill";
 import {
   FITBIT_BACKFILL_QUEUE,
   FITBIT_BACKFILL_CONCURRENCY,
-  runFitbitBackfillForUser,
   enqueueBootTimeFitbitBackfill,
   type FitbitBackfillPayload,
 } from "@/lib/jobs/fitbit-backfill";
 import {
   GOOGLE_HEALTH_BACKFILL_QUEUE,
   GOOGLE_HEALTH_BACKFILL_CONCURRENCY,
-  runGoogleHealthBackfillForUser,
   enqueueBootTimeGoogleHealthBackfill,
   type GoogleHealthBackfillPayload,
 } from "@/lib/jobs/google-health-backfill";
 import {
   GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE,
   GOOGLE_HEALTH_SLEEP_REPAIR_CONCURRENCY,
-  runGoogleHealthSleepRepairForUser,
   enqueueBootTimeGoogleHealthSleepRepair,
   type GoogleHealthSleepRepairPayload,
 } from "@/lib/jobs/google-health-sleep-repair";
 import {
   FITBIT_SLEEP_REPAIR_QUEUE,
   FITBIT_SLEEP_REPAIR_CONCURRENCY,
-  runFitbitSleepRepairForUser,
   enqueueBootTimeFitbitSleepRepair,
   type FitbitSleepRepairPayload,
 } from "@/lib/jobs/fitbit-sleep-repair";
 import {
   STRAVA_BACKFILL_QUEUE,
   STRAVA_BACKFILL_CONCURRENCY,
-  runStravaBackfillForUser,
   enqueueBootTimeStravaBackfill,
   type StravaBackfillPayload,
 } from "@/lib/jobs/strava-backfill";
 import {
   SLEEP_TIMELINE_BACKFILL_QUEUE,
   SLEEP_TIMELINE_BACKFILL_CONCURRENCY,
-  runSleepTimelineBackfillForUser,
   enqueueBootTimeSleepTimelineBackfill,
   type SleepTimelineBackfillPayload,
 } from "@/lib/jobs/sleep-timeline-backfill";
 import {
   LAB_BIOMARKER_BACKFILL_QUEUE,
   LAB_BIOMARKER_BACKFILL_CONCURRENCY,
-  runLabBiomarkerBackfillForUser,
   enqueueBootTimeLabBiomarkerBackfill,
   type LabBiomarkerBackfillPayload,
 } from "@/lib/jobs/lab-biomarker-backfill";
@@ -77,6 +69,7 @@ import {
   enqueueIntegrationBackfillAdmission,
   type IntegrationBackfillAdmissionPayload,
 } from "@/lib/jobs/integration-backfill-admission";
+import { drainIntegrationBackfillAdmission } from "@/lib/jobs/integration-backfill-drain";
 import { workerLog } from "./shared";
 import { jobDone } from "@/lib/jobs/job-outcome";
 import {
@@ -482,117 +475,7 @@ export async function registerIntegrationSyncQueues(
       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}`,
-            );
-            break;
-          }
-          case "fitbit-backfill": {
-            const { imported } = await runFitbitBackfillForUser(userId);
-            importedTotal += imported;
-            workerLog(
-              "info",
-              `[fitbit-backfill] user=${userId} imported=${imported}`,
-            );
-            break;
-          }
-          case "google-health-backfill": {
-            const { imported } = await runGoogleHealthBackfillForUser(userId);
-            importedTotal += imported;
-            workerLog(
-              "info",
-              `[google-health-backfill] user=${userId} imported=${imported}`,
-            );
-            break;
-          }
-          case "google-health-sleep-repair": {
-            const { imported } =
-              await runGoogleHealthSleepRepairForUser(userId);
-            importedTotal += imported;
-            workerLog(
-              "info",
-              `[google-health-sleep-repair] user=${userId} imported=${imported}`,
-            );
-            break;
-          }
-          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}`,
-            );
-            break;
-          }
-          case "sleep-timeline-backfill": {
-            if (!data.provider) {
-              throw new Error(
-                "sleep-timeline backfill admission requires a provider",
-              );
-            }
-            const { deleted, imported } = await runSleepTimelineBackfillForUser(
-              userId,
-              data.provider,
-            );
-            deletedTotal += deleted;
-            importedTotal += imported;
-            workerLog(
-              "info",
-              `[sleep-timeline-backfill] user=${userId} provider=${data.provider} deleted=${deleted} imported=${imported}`,
-            );
-            break;
-          }
-          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}`,
-            );
-            break;
-          }
-          case "strava-backfill": {
-            const { imported } = await runStravaBackfillForUser(userId);
-            importedTotal += imported;
-            workerLog(
-              "info",
-              `[strava-backfill] user=${userId} imported=${imported}`,
-            );
-            break;
-          }
-        }
-      }
-      return jobDone({
-        jobs: jobs.length,
-        imported: importedTotal,
-        removed: removedTotal,
-        deleted: deletedTotal,
-        markers: markersTotal,
-        linked: linkedTotal,
-      });
-    },
+    drainIntegrationBackfillAdmission,
   );
 
   await createAndWork<WithingsSyncPayload>(

tests/integration/integration-backfill-lane-expiry.test.ts

@@ -0,0 +1,169 @@
+/**
+ * The full-history admission lane against a real pg-boss on a real Postgres:
+ * what happens to an import that runs past its job's expiry.
+ *
+ * pg-boss 12.34, measured with the lane's worker shape (localConcurrency 1,
+ * groupConcurrency 1, exclusive policy, retries with backoff), a two-second
+ * expiry, a one-second retry delay and a handler that ignores the abort
+ * signal and runs for eight seconds:
+ *
+ *   +5ms     start #1 (job 1709c8a6…)   running=1
+ *   +2008ms  abort signal #1
+ *   +4051ms  start #2 (same job id)     running=2
+ *   +6054ms  abort signal #2
+ *   +8007ms  end #1                     running=1
+ *   +9593ms  start #3 (same job id)     running=2
+ *   +12053ms end #2
+ *   +17595ms end #3
+ *   +19166ms start #4 (same job id)     running=1
+ *
+ * At the expiry pg-boss fails the job, fires the signal, frees the worker and
+ * group slots, and retries the job after its delay; the handler keeps going.
+ * The retry then starts in the same worker beside the run it replaced, and
+ * the lane's one-at-a-time holds for everything except the import too long
+ * for its expiry. The first test keeps that as an executable statement about
+ * pg-boss; the second runs the lane's own handler under the same conditions.
+ */
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { PgBoss, type Job } from "pg-boss";
+
+const runner = vi.hoisted(() => ({
+  running: 0,
+  maxRunning: 0,
+  calls: 0,
+  holdMs: 8_000,
+}));
+
+vi.mock("@/lib/jobs/google-health-backfill", () => ({
+  // Ignores the budget and the signal on purpose: the case the lock covers
+  // is a run that does not stop when pg-boss gives up on it.
+  runGoogleHealthBackfillForUser: async () => {
+    runner.calls += 1;
+    runner.running += 1;
+    runner.maxRunning = Math.max(runner.maxRunning, runner.running);
+    await new Promise((resolve) => setTimeout(resolve, runner.holdMs));
+    runner.running -= 1;
+    return { imported: 1 };
+  },
+}));
+
+import {
+  integrationBackfillAdmissionSendOptions,
+  type IntegrationBackfillAdmissionPayload,
+} from "@/lib/jobs/integration-backfill-admission";
+import { drainIntegrationBackfillAdmission } from "@/lib/jobs/integration-backfill-drain";
+import { withIntegrationBackfillLock } from "@/lib/jobs/integration-backfill-lock";
+import { runJob } from "@/lib/jobs/run-job";
+
+let boss: PgBoss;
+const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`;
+
+/** Short enough to cross in a test, the lane's shape otherwise. */
+const EXPIRY_OVERRIDE = { expireInSeconds: 2, retryDelay: 1 } as const;
+
+const WORK_OPTIONS = {
+  localConcurrency: 1,
+  groupConcurrency: 1,
+  pollingIntervalSeconds: 0.5,
+} as const;
+
+const payload: IntegrationBackfillAdmissionPayload = {
+  kind: "google-health-backfill",
+  data: { userId: `lane-user-${suffix}`, enqueuedAt: new Date().toISOString() },
+};
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+beforeAll(async () => {
+  boss = new PgBoss({ connectionString: process.env.DATABASE_URL });
+  await boss.start();
+}, 120_000);
+
+afterAll(async () => {
+  await boss?.stop({ graceful: false });
+});
+
+describe("an import that outlives its expiry", () => {
+  it("pg-boss starts the retry of the same job beside the run it expired", async () => {
+    const queue = `lane-expiry-raw-${suffix}`;
+    await boss.createQueue(queue, { policy: "exclusive" });
+    const ids: string[] = [];
+    let running = 0;
+    let maxRunning = 0;
+    await boss.work(queue, WORK_OPTIONS, async ([job]: Job<object>[]) => {
+      ids.push(job!.id);
+      running += 1;
+      maxRunning = Math.max(maxRunning, running);
+      await sleep(8_000);
+      running -= 1;
+    });
+    await boss.send(queue, payload, {
+      ...integrationBackfillAdmissionSendOptions(payload),
+      ...EXPIRY_OVERRIDE,
+    });
+
+    await sleep(12_000);
+    await boss.offWork(queue);
+
+    expect(ids.length).toBeGreaterThanOrEqual(2);
+    expect(new Set(ids).size).toBe(1);
+    expect(maxRunning).toBeGreaterThanOrEqual(2);
+  }, 60_000);
+
+  it("the lane's handler runs the import once while its retries find it running", async () => {
+    const queue = `lane-expiry-drain-${suffix}`;
+    await boss.createQueue(queue, { policy: "exclusive" });
+    const handler = runJob(queue, drainIntegrationBackfillAdmission);
+    const deliveries: string[] = [];
+    await boss.work(
+      queue,
+      WORK_OPTIONS,
+      async (jobs: Job<IntegrationBackfillAdmissionPayload>[]) => {
+        deliveries.push(jobs[0]!.id);
+        return handler(jobs);
+      },
+    );
+    await boss.send(queue, payload, {
+      ...integrationBackfillAdmissionSendOptions(payload),
+      ...EXPIRY_OVERRIDE,
+    });
+
+    // The run holds for eight seconds; its retries arrive while it does.
+    await sleep(7_000);
+    await boss.offWork(queue);
+    await sleep(2_000);
+
+    expect(deliveries.length).toBeGreaterThanOrEqual(2);
+    expect(runner.calls).toBe(1);
+    expect(runner.maxRunning).toBe(1);
+  }, 60_000);
+});
+
+describe("withIntegrationBackfillLock", () => {
+  it("admits one run per identity, others alongside, and the next once it ends", async () => {
+    let release!: () => void;
+    const held = new Promise<void>((resolve) => (release = resolve));
+    const first = withIntegrationBackfillLock("kind|a", () => held);
+    await sleep(300);
+
+    const same = await withIntegrationBackfillLock("kind|a", async () => 1);
+    const other = await withIntegrationBackfillLock("kind|b", async () => 2);
+    expect(same).toEqual({ ran: false });
+    expect(other).toEqual({ ran: true, result: 2 });
+
+    release();
+    await first;
+    const after = await withIntegrationBackfillLock("kind|a", async () => 3);
+    expect(after).toEqual({ ran: true, result: 3 });
+  });
+
+  it("releases the lock when the run throws", async () => {
+    await expect(
+      withIntegrationBackfillLock("kind|c", async () => {
+        throw new Error("boom");
+      }),
+    ).rejects.toThrow("boom");
+    const after = await withIntegrationBackfillLock("kind|c", async () => 4);
+    expect(after).toEqual({ ran: true, result: 4 });
+  });
+});