Merge pull request #1066 from MBombeck/fix/backfill-lane-expiry

Keep a full-history import from running beside its own retry; tombstone purge steps around a restore

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/__tests__/measurement-tombstone-cleanup.test.ts

@@ -21,19 +21,43 @@ import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client";
 
 const DAY_MS = 86_400_000;
 
+/**
+ * A client whose interactive transaction hands the callback a client with the
+ * same delegates, and whose restore-lock probe answers `held`.
+ */
+function mockClient(
+  measurement: { findMany: unknown; deleteMany: unknown },
+  held: (userId: string) => boolean = () => true,
+) {
+  const lockCalls: string[] = [];
+  const tx = {
+    measurement,
+    $queryRaw: vi.fn(async (_strings: TemplateStringsArray, key: string) => {
+      const userId = key.replace(/^backup-restore:/, "");
+      lockCalls.push(userId);
+      return [{ held: held(userId) }];
+    }),
+  };
+  const prisma = {
+    measurement,
+    $transaction: vi.fn(async (fn: (client: typeof tx) => unknown) => fn(tx)),
+  } as unknown as Parameters<typeof cleanupExpiredMeasurementTombstones>[0];
+  return { prisma, lockCalls };
+}
+
 describe("cleanupExpiredMeasurementTombstones", () => {
   it("prunes only tombstones older than the retention horizon", async () => {
     const now = new Date("2026-06-01T00:00:00.000Z");
-    const findMany = vi
-      .fn()
-      .mockResolvedValueOnce([{ id: "a" }, { id: "b" }, { id: "c" }]);
+    const findMany = vi.fn().mockResolvedValueOnce([
+      { id: "a", userId: "u1" },
+      { id: "b", userId: "u1" },
+      { id: "c", userId: "u1" },
+    ]);
     const deleteMany = vi.fn().mockResolvedValue({ count: 3 });
-    const prisma = {
-      measurement: { findMany, deleteMany },
-    } as unknown as Parameters<typeof cleanupExpiredMeasurementTombstones>[0];
+    const { prisma } = mockClient({ findMany, deleteMany });
 
     const outcome = await cleanupExpiredMeasurementTombstones(prisma, now);
-    expect(outcome).toEqual({ deleted: 3, drained: true });
+    expect(outcome).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 });
 
     expect(findMany).toHaveBeenCalledTimes(1);
     const where = findMany.mock.calls[0][0].where as {
@@ -61,12 +85,11 @@ describe("cleanupExpiredMeasurementTombstones", () => {
     // to stop at the cap rather than loop forever.
     const fullBatch = Array.from({ length: PURGE_BATCH_SIZE }, (_, i) => ({
       id: `row-${i}`,
+      userId: "u1",
     }));
     const findMany = vi.fn().mockResolvedValue(fullBatch);
     const deleteMany = vi.fn().mockResolvedValue({ count: PURGE_BATCH_SIZE });
-    const prisma = {
-      measurement: { findMany, deleteMany },
-    } as unknown as Parameters<typeof cleanupExpiredMeasurementTombstones>[0];
+    const { prisma } = mockClient({ findMany, deleteMany });
 
     const outcome = await cleanupExpiredMeasurementTombstones(prisma, now);
 
@@ -81,6 +104,57 @@ describe("cleanupExpiredMeasurementTombstones", () => {
   });
 });
 
+describe("cleanupExpiredMeasurementTombstones — accounts under restore", () => {
+  it("deletes one statement per account per batch, each under the account's restore lock", async () => {
+    const findMany = vi.fn().mockResolvedValueOnce([
+      { id: "a1", userId: "a" },
+      { id: "b1", userId: "b" },
+      { id: "a2", userId: "a" },
+    ]);
+    const deleteMany = vi.fn(async ({ where }) => ({
+      count: (where.id.in as string[]).length,
+    }));
+    const { prisma, lockCalls } = mockClient({ findMany, deleteMany });
+
+    const outcome = await cleanupExpiredMeasurementTombstones(prisma);
+
+    expect(outcome).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 });
+    expect(lockCalls).toEqual(["a", "b"]);
+    expect(deleteMany.mock.calls.map((call) => call[0].where)).toEqual([
+      { id: { in: ["a1", "a2"] } },
+      { id: { in: ["b1"] } },
+    ]);
+  });
+
+  it("leaves an account under restore and keeps it out of the rest of the run", async () => {
+    const full = (userId: string) =>
+      Array.from({ length: PURGE_BATCH_SIZE }, (_, i) => ({
+        id: `${userId}-${i}`,
+        userId,
+      }));
+    const findMany = vi
+      .fn()
+      .mockResolvedValueOnce(full("restoring"))
+      .mockResolvedValueOnce([{ id: "other-1", userId: "other" }]);
+    const deleteMany = vi.fn(async ({ where }) => ({
+      count: (where.id.in as string[]).length,
+    }));
+    const { prisma } = mockClient(
+      { findMany, deleteMany },
+      (userId) => userId !== "restoring",
+    );
+
+    const outcome = await cleanupExpiredMeasurementTombstones(prisma);
+
+    expect(outcome).toEqual({ deleted: 1, drained: true, deferredAccounts: 1 });
+    expect(findMany.mock.calls[0][0].where).not.toHaveProperty("userId");
+    expect(findMany.mock.calls[1][0].where.userId).toEqual({
+      notIn: ["restoring"],
+    });
+    expect(deleteMany).toHaveBeenCalledTimes(1);
+  });
+});
+
 describe("reminder-worker — measurement-tombstone-cleanup schedule", () => {
   // v1.18.1 — the cleanup wiring moved out of the 2143-LOC reminder-worker
   // boot file into the maintenance registrar. The dead-queue guard follows it.

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",
@@ -68,6 +71,9 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "days_skipped",
   "days_skipped_no_tombstones",
   "days_stored",
+  // Tombstone purge: accounts left for the next run because a restore of
+  // the account was running.
+  "deferred_accounts",
   "deleted",
   "dense_days_consolidated",
   "dense_rows_soft_deleted",

src/lib/jobs/measurement-tombstone-cleanup.ts

@@ -18,9 +18,17 @@
  * (`NATIVE_REFRESH_TOKEN_DAYS` + margin) so it moves automatically if the
  * refresh-token lifetime changes — the two never drift.
  */
-import type { PrismaClient } from "@/generated/prisma/client";
+import type { Prisma, PrismaClient } from "@/generated/prisma/client";
 import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client";
-import { purgeInBatches, type PurgeOutcome } from "@/lib/jobs/purge-batch";
+import {
+  AccountRestoreInProgressError,
+  holdAccountAgainstRestore,
+} from "@/lib/export/restore-lock";
+import {
+  PURGE_BATCH_SIZE,
+  PURGE_MAX_BATCHES,
+  type PurgeOutcome,
+} from "@/lib/jobs/purge-batch";
 
 const DAY_MS = 86_400_000;
 
@@ -28,6 +36,103 @@ function tombstoneCutoff(now: Date): Date {
   return new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS);
 }
 
+/** A purge run, plus the accounts it left for the next run. */
+export interface TombstonePurgeOutcome extends PurgeOutcome {
+  /**
+   * Accounts whose tombstones were left in place because a restore of the
+   * account was running. The next night purges them.
+   */
+  deferredAccounts: number;
+}
+
+type TombstoneRow = { id: string; userId: string };
+
+interface TombstonePurgeOptions {
+  prisma: PrismaClient;
+  /** Up to `take` expired tombstones, none of them owned by `skipUserIds`. */
+  findRows: (take: number, skipUserIds: string[]) => Promise<TombstoneRow[]>;
+  /** Remove exactly those ids inside `tx`; returns the rows removed. */
+  deleteIds: (tx: Prisma.TransactionClient, ids: string[]) => Promise<number>;
+  batchSize?: number;
+  maxBatches?: number;
+}
+
+/**
+ * Walk the expired tombstones in bounded batches, deleting each batch one
+ * account at a time under the account's restore lock.
+ *
+ * A restore clears every row of the account, tombstones included, in one
+ * long transaction. The purge used to delete a batch spanning accounts in one
+ * statement, so on an account under restore it waited on the restore's row
+ * locks, for as long as the restore ran and well past `statement_timeout`,
+ * or, holding some of the account's rows while the restore held others,
+ * deadlocked with it. So each account's share of a batch is deleted in a
+ * transaction that first takes the restore lock in shared mode, the way the
+ * consolidation passes do (`restore-lock.ts`). An account under restore is
+ * refused at once, dropped from the rest of the run, and purged the next
+ * night, when its tombstones are whatever the restore left.
+ *
+ * Still one delete statement per account per batch, never one per row.
+ */
+async function purgeTombstonesByAccount({
+  prisma,
+  findRows,
+  deleteIds,
+  batchSize = PURGE_BATCH_SIZE,
+  maxBatches = PURGE_MAX_BATCHES,
+}: TombstonePurgeOptions): Promise<TombstonePurgeOutcome> {
+  let deleted = 0;
+  const deferred = new Set<string>();
+  const outcome = (drained: boolean): TombstonePurgeOutcome => ({
+    deleted,
+    drained,
+    deferredAccounts: deferred.size,
+  });
+
+  for (let batch = 0; batch < maxBatches; batch++) {
+    const rows = await findRows(batchSize, [...deferred]);
+    if (rows.length === 0) return outcome(true);
+
+    const byAccount = new Map<string, string[]>();
+    for (const row of rows) {
+      const ids = byAccount.get(row.userId);
+      if (ids) ids.push(row.id);
+      else byAccount.set(row.userId, [row.id]);
+    }
+    for (const [userId, ids] of byAccount) {
+      try {
+        deleted += await prisma.$transaction(
+          async (tx) => {
+            await holdAccountAgainstRestore(tx, userId);
+            return deleteIds(tx, ids);
+          },
+          // A full batch of one account is one statement over thousands of
+          // rows; the interactive default of five seconds is too tight for
+          // it on a slow disk. The connection's statement_timeout bounds it.
+          { timeout: 60_000 },
+        );
+      } catch (err) {
+        if (!(err instanceof AccountRestoreInProgressError)) throw err;
+        deferred.add(userId);
+      }
+    }
+
+    // A short batch means the predicate is exhausted, the deferred accounts
+    // aside; asking again would cost a round-trip to learn nothing.
+    if (rows.length < batchSize) return outcome(true);
+  }
+
+  return outcome(false);
+}
+
+/** The retention predicate, minus the accounts a run has deferred. */
+function expiredTombstones(cutoff: Date, skipUserIds: string[]) {
+  return {
+    deletedAt: { not: null, lt: cutoff },
+    ...(skipUserIds.length > 0 ? { userId: { notIn: skipUserIds } } : {}),
+  };
+}
+
 /**
  * Hard-delete soft-deleted measurement rows whose `deletedAt` is older than
  * the retention horizon.
@@ -44,20 +149,18 @@ function tombstoneCutoff(now: Date): Date {
 export async function cleanupExpiredMeasurementTombstones(
   prisma: PrismaClient,
   now: Date = new Date(),
-): Promise<PurgeOutcome> {
+): Promise<TombstonePurgeOutcome> {
   const cutoff = tombstoneCutoff(now);
-  return purgeInBatches({
-    findIds: async (take) =>
-      (
-        await prisma.measurement.findMany({
-          where: { deletedAt: { not: null, lt: cutoff } },
-          select: { id: true },
-          take,
-        })
-      ).map((row) => row.id),
-    deleteIds: async (ids) =>
-      (await prisma.measurement.deleteMany({ where: { id: { in: ids } } }))
-        .count,
+  return purgeTombstonesByAccount({
+    prisma,
+    findRows: (take, skipUserIds) =>
+      prisma.measurement.findMany({
+        where: expiredTombstones(cutoff, skipUserIds),
+        select: { id: true, userId: true },
+        take,
+      }),
+    deleteIds: async (tx, ids) =>
+      (await tx.measurement.deleteMany({ where: { id: { in: ids } } })).count,
   });
 }
 
@@ -65,19 +168,18 @@ export async function cleanupExpiredMeasurementTombstones(
 export async function cleanupExpiredMoodTombstones(
   prisma: PrismaClient,
   now: Date = new Date(),
-): Promise<PurgeOutcome> {
+): Promise<TombstonePurgeOutcome> {
   const cutoff = tombstoneCutoff(now);
-  return purgeInBatches({
-    findIds: async (take) =>
-      (
-        await prisma.moodEntry.findMany({
-          where: { deletedAt: { not: null, lt: cutoff } },
-          select: { id: true },
-          take,
-        })
-      ).map((row) => row.id),
-    deleteIds: async (ids) =>
-      (await prisma.moodEntry.deleteMany({ where: { id: { in: ids } } })).count,
+  return purgeTombstonesByAccount({
+    prisma,
+    findRows: (take, skipUserIds) =>
+      prisma.moodEntry.findMany({
+        where: expiredTombstones(cutoff, skipUserIds),
+        select: { id: true, userId: true },
+        take,
+      }),
+    deleteIds: async (tx, ids) =>
+      (await tx.moodEntry.deleteMany({ where: { id: { in: ids } } })).count,
   });
 }
 
@@ -88,20 +190,19 @@ export async function cleanupExpiredMoodTombstones(
 export async function cleanupExpiredIntakeTombstones(
   prisma: PrismaClient,
   now: Date = new Date(),
-): Promise<PurgeOutcome> {
+): Promise<TombstonePurgeOutcome> {
   const cutoff = tombstoneCutoff(now);
-  return purgeInBatches({
-    findIds: async (take) =>
+  return purgeTombstonesByAccount({
+    prisma,
+    findRows: (take, skipUserIds) =>
+      prisma.medicationIntakeEvent.findMany({
+        where: expiredTombstones(cutoff, skipUserIds),
+        select: { id: true, userId: true },
+        take,
+      }),
+    deleteIds: async (tx, ids) =>
       (
-        await prisma.medicationIntakeEvent.findMany({
-          where: { deletedAt: { not: null, lt: cutoff } },
-          select: { id: true },
-          take,
-        })
-      ).map((row) => row.id),
-    deleteIds: async (ids) =>
-      (
-        await prisma.medicationIntakeEvent.deleteMany({
+        await tx.medicationIntakeEvent.deleteMany({
           where: { id: { in: ids } },
         })
       ).count,

src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts

@@ -62,15 +62,16 @@ import {
 } from "../cleanup-handlers";
 
 const DRAINED = { deleted: 1, drained: true };
+const TOMBSTONES_DRAINED = { ...DRAINED, deferredAccounts: 0 };
 
 beforeEach(() => {
   vi.clearAllMocks();
   for (const key of Object.keys(h.meta)) delete h.meta[key];
   h.warnings.length = 0;
   h.cleanupOldAuditLogs.mockResolvedValue(DRAINED);
-  h.cleanupExpiredMeasurementTombstones.mockResolvedValue(DRAINED);
-  h.cleanupExpiredMoodTombstones.mockResolvedValue(DRAINED);
-  h.cleanupExpiredIntakeTombstones.mockResolvedValue(DRAINED);
+  h.cleanupExpiredMeasurementTombstones.mockResolvedValue(TOMBSTONES_DRAINED);
+  h.cleanupExpiredMoodTombstones.mockResolvedValue(TOMBSTONES_DRAINED);
+  h.cleanupExpiredIntakeTombstones.mockResolvedValue(TOMBSTONES_DRAINED);
 });
 
 describe("audit-log cleanup handler", () => {
@@ -115,6 +116,7 @@ describe("tombstone cleanup handler", () => {
     h.cleanupExpiredMoodTombstones.mockResolvedValue({
       deleted: 200_000,
       drained: false,
+      deferredAccounts: 0,
     });
     await handleMeasurementTombstoneCleanup([]);
     expect(h.meta.tombstone_cleanup_drained).toBe(false);
@@ -122,6 +124,16 @@ describe("tombstone cleanup handler", () => {
     expect(h.warnings.join(" ")).toContain("backlog remains");
   });
 
+  it("reports the accounts it left because a restore was running", async () => {
+    h.cleanupExpiredMeasurementTombstones.mockResolvedValue({
+      ...TOMBSTONES_DRAINED,
+      deferredAccounts: 2,
+    });
+    const outcome = await handleMeasurementTombstoneCleanup([]);
+    expect(h.meta.tombstone_cleanup_deferred_accounts).toBe(2);
+    expect(outcome).toMatchObject({ did: { deferred_accounts: 2 } });
+  });
+
   it("stays quiet when all three legs drained", async () => {
     await handleMeasurementTombstoneCleanup([]);
     expect(h.meta.tombstone_cleanup_drained).toBe(true);

src/lib/jobs/reminder/cleanup-handlers.ts

@@ -254,6 +254,13 @@ export async function handleMeasurementTombstoneCleanup(
         // outside, which is how a purge that never finished stayed invisible.
         const drained = measurements.drained && mood.drained && intakes.drained;
         evt.addMeta("tombstone_cleanup_drained", drained);
+        // Accounts under restore are left alone and purged the next night.
+        const deferredAccounts = Math.max(
+          measurements.deferredAccounts,
+          mood.deferredAccounts,
+          intakes.deferredAccounts,
+        );
+        evt.addMeta("tombstone_cleanup_deferred_accounts", deferredAccounts);
         if (!drained) {
           evt.addWarning(
             "tombstone-cleanup stopped at the batch cap; a backlog remains for the next run",
@@ -264,6 +271,7 @@ export async function handleMeasurementTombstoneCleanup(
           mood_pruned: mood.deleted,
           intakes_pruned: intakes.deleted,
           drained,
+          deferred_accounts: deferredAccounts,
         });
       } catch (err) {
         // Rethrow. A retention purge that fails silently every night looks

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

tests/integration/sync-changes.test.ts

@@ -399,7 +399,7 @@ describe("GET /api/sync/changes (real Postgres)", () => {
     // pruned; the recent reachable tombstone survives. `drained` says the
     // batched walk reached the end of the backlog rather than stopping at
     // its per-run cap.
-    expect(pruned).toEqual({ deleted: 1, drained: true });
+    expect(pruned).toEqual({ deleted: 1, drained: true, deferredAccounts: 0 });
     const survivors = await prisma.measurement.findMany({
       where: { userId: TEST_USER_ID },
       select: { externalId: true },

tests/integration/tombstone-cleanup-restore-lock.test.ts

@@ -0,0 +1,152 @@
+/**
+ * #1031 — the nightly tombstone purge against a running restore.
+ *
+ * A restore clears every row of the account, tombstones included, inside one
+ * long transaction that holds the account's restore lock exclusively. The
+ * purge used to delete expired tombstones across accounts in one statement
+ * per batch, without the lock: on an account under restore it waited on the
+ * restore's row locks for as long as the restore ran, or deadlocked with it.
+ *
+ * Here a transaction stands in for the restore: it takes the lock, deletes the
+ * account's rows, and holds for three seconds before rolling back. The purge
+ * runs while it holds. What must hold: the purge does not wait, leaves the
+ * account under restore alone and says so, and still purges every other
+ * account.
+ *
+ * Mutation check: drop the `holdAccountAgainstRestore` call from
+ * `purgeTombstonesByAccount` and the purge waits out the hold, then deletes
+ * the restoring account's tombstones once the stand-in rolls back.
+ */
+import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const RESTORING = "tombstone-restoring";
+const OTHER = "tombstone-other";
+const HOLD_MS = 3_000;
+const LONG_AGO = new Date("2020-01-01T00:00:00.000Z");
+
+async function seed() {
+  const prisma = getPrismaClient();
+  for (const id of [RESTORING, OTHER]) {
+    await prisma.user.create({
+      data: { id, username: id, role: "USER", timezone: "UTC" },
+    });
+    for (let i = 0; i < 3; i++) {
+      await prisma.measurement.create({
+        data: {
+          id: `${id}-tomb-${i}`,
+          userId: id,
+          type: "WEIGHT",
+          value: 70 + i,
+          unit: "kg",
+          source: "MANUAL",
+          measuredAt: new Date(`2020-01-0${i + 1}T08:00:00.000Z`),
+          deletedAt: LONG_AGO,
+        },
+      });
+    }
+    await prisma.moodEntry.create({
+      data: {
+        id: `${id}-mood-tomb`,
+        userId: id,
+        date: "2020-01-01",
+        mood: "OKAY",
+        score: 3,
+        moodLoggedAt: new Date("2020-01-01T08:00:00.000Z"),
+        deletedAt: LONG_AGO,
+      },
+    });
+  }
+}
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+});
+
+afterAll(async () => {
+  await truncateAllTables(getPrismaClient());
+});
+
+describe("tombstone purge against a running restore (#1031)", () => {
+  it("leaves the account under restore for the next run without waiting on it", async () => {
+    const prisma = getPrismaClient();
+    await seed();
+    const { takeRestoreLock } = await import("@/lib/export/restore-lock");
+    const {
+      cleanupExpiredMeasurementTombstones,
+      cleanupExpiredMoodTombstones,
+    } = await import("@/lib/jobs/measurement-tombstone-cleanup");
+
+    let holding!: () => void;
+    const isHolding = new Promise<void>((resolve) => (holding = resolve));
+    const restore = prisma
+      .$transaction(
+        async (tx) => {
+          await takeRestoreLock(tx, RESTORING);
+          await tx.measurement.deleteMany({ where: { userId: RESTORING } });
+          await tx.moodEntry.deleteMany({ where: { userId: RESTORING } });
+          holding();
+          await new Promise((resolve) => setTimeout(resolve, HOLD_MS));
+          throw new Error("rolled back");
+        },
+        { timeout: 30_000 },
+      )
+      .catch(() => undefined);
+    await isHolding;
+
+    const started = Date.now();
+    const measurements = await cleanupExpiredMeasurementTombstones(prisma);
+    const moods = await cleanupExpiredMoodTombstones(prisma);
+    const elapsed = Date.now() - started;
+    await restore;
+
+    expect(elapsed).toBeLessThan(HOLD_MS / 2);
+    expect(measurements).toEqual({
+      deleted: 3,
+      drained: true,
+      deferredAccounts: 1,
+    });
+    expect(moods).toEqual({ deleted: 1, drained: true, deferredAccounts: 1 });
+    // The stand-in rolled back, so the restoring account still has its
+    // tombstones; the next run purges them.
+    expect(
+      await prisma.measurement.count({ where: { userId: RESTORING } }),
+    ).toBe(3);
+    expect(await prisma.measurement.count({ where: { userId: OTHER } })).toBe(
+      0,
+    );
+    expect(await prisma.moodEntry.count({ where: { userId: OTHER } })).toBe(0);
+
+    const next = await cleanupExpiredMeasurementTombstones(prisma);
+    expect(next).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 });
+  }, 60_000);
+
+  it("a restore that starts during the purge waits only for the account's batch", async () => {
+    const prisma = getPrismaClient();
+    await seed();
+    const { takeRestoreLock } = await import("@/lib/export/restore-lock");
+    const { cleanupExpiredMeasurementTombstones } =
+      await import("@/lib/jobs/measurement-tombstone-cleanup");
+
+    const purge = cleanupExpiredMeasurementTombstones(prisma);
+    const restore = prisma.$transaction(async (tx) => {
+      await takeRestoreLock(tx, RESTORING);
+      return tx.measurement.deleteMany({ where: { userId: RESTORING } });
+    });
+    const [purged] = await Promise.all([purge, restore]);
+
+    // Whichever went first, both finished and nothing is left.
+    expect(purged.deferredAccounts).toBeLessThanOrEqual(1);
+    expect(await prisma.measurement.count({ where: { userId: OTHER } })).toBe(
+      0,
+    );
+    expect(
+      await prisma.measurement.count({ where: { userId: RESTORING } }),
+    ).toBe(0);
+  }, 60_000);
+});