fix(google-health): let a progressing backfill slice end quietly and continue

Since the backfill resumes per collection, a slice that ends before the
whole history is in threw "backfill incomplete" so pg-boss retried it,
which logged job.run.failed at error and grew the queue's failure count
although nothing had failed. A slice that finished at least one more
collection now queues the next slice on the admission queue under its
own singleton key and completes. Only a slice that finished nothing new
still throws, so a backfill that is actually stuck keeps its error. The
import lock keys on the import identity without the slice number, and a
slice that finds the import already stamped walks nothing.

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

@@ -7,24 +7,37 @@
  */
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
-const { prismaMock, syncUserGoogleHealthMock, isParkedMock } = vi.hoisted(
-  () => ({
-    prismaMock: {
-      googleHealthConnection: {
-        findMany: vi.fn(),
-        findUnique: vi.fn(async () => null),
-        update: vi.fn(),
-      },
+const {
+  prismaMock,
+  syncUserGoogleHealthMock,
+  isParkedMock,
+  bossSend,
+  resumeState,
+} = vi.hoisted(() => ({
+  prismaMock: {
+    googleHealthConnection: {
+      findMany: vi.fn(),
+      findUnique: vi.fn(
+        async (): Promise<{
+          backfillProgress: unknown;
+          backfillCompletedAt: Date | null;
+        } | null> => null,
+      ),
+      update: vi.fn(),
     },
-    syncUserGoogleHealthMock: vi.fn(),
-    isParkedMock: vi.fn(async () => false),
-  }),
-);
+  },
+  syncUserGoogleHealthMock: vi.fn(),
+  isParkedMock: vi.fn(async () => false),
+  bossSend: vi.fn(),
+  // Collections the mocked walk "finishes" in this attempt, added to the
+  // resume set the way `runGoogleHealthCollection` adds them.
+  resumeState: { finishes: [] as string[], doneBefore: [] as string[] },
+}));
 
 vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
 
 vi.mock("@/lib/jobs/boss-instance", () => ({
-  getGlobalBoss: () => ({ send: vi.fn() }),
+  getGlobalBoss: () => ({ send: bossSend }),
 }));
 
 vi.mock("@/lib/logging/context", () => ({
@@ -40,12 +53,16 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
 
 vi.mock("@/lib/google-health/sync-core", () => ({
   GOOGLE_HEALTH_INTEGRATION_KEY: "google-health",
-  readGoogleHealthBackfillDone: () => new Set<string>(),
+  readGoogleHealthBackfillDone: () => new Set<string>(resumeState.doneBefore),
   runWithGoogleHealthBackfillResume: async (
     _userId: string,
-    _done: Set<string>,
+    done: Set<string>,
     fn: () => Promise<unknown>,
-  ) => ({ result: await fn(), skipped: [] }),
+  ) => {
+    const result = await fn();
+    for (const key of resumeState.finishes) done.add(key);
+    return { result, skipped: [] };
+  },
 }));
 vi.mock("@/lib/google-health/sync", () => ({
   isGoogleHealthParked: isParkedMock,
@@ -63,6 +80,10 @@ beforeEach(() => {
   vi.clearAllMocks();
   isParkedMock.mockResolvedValue(false);
   prismaMock.googleHealthConnection.update.mockResolvedValue({});
+  prismaMock.googleHealthConnection.findUnique.mockResolvedValue(null);
+  bossSend.mockResolvedValue("job-next");
+  resumeState.finishes = [];
+  resumeState.doneBefore = [];
 });
 
 describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
@@ -150,3 +171,92 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
     expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
   });
 });
+
+/**
+ * A long history walks in slices. A slice that finished more collections than
+ * it started with is progressing, not failing: it queues the next slice and
+ * ends quietly, without the error line or the failure count. Only a slice that
+ * finished nothing new is stuck and keeps throwing.
+ */
+describe("runGoogleHealthBackfillForUser — slices", () => {
+  it("continues a progressing slice instead of failing it", async () => {
+    resumeState.doneBefore = ["steps"];
+    resumeState.finishes = ["sleep", "weight"];
+    syncUserGoogleHealthMock.mockResolvedValue({ imported: 9, failed: true });
+
+    await expect(
+      runGoogleHealthBackfillForUser("u1", undefined, 2),
+    ).resolves.toEqual({
+      imported: 9,
+      continued: true,
+    });
+
+    expect(bossSend).toHaveBeenCalledTimes(1);
+    const [queue, payload, options] = bossSend.mock.calls[0]!;
+    expect(queue).toBe("integration-backfill-admission");
+    expect(payload).toMatchObject({
+      kind: "google-health-backfill",
+      data: { userId: "u1", slice: 3 },
+    });
+    // Its own key, so the exclusive policy does not swallow it while this
+    // slice is still active.
+    expect(options).toMatchObject({
+      singletonKey: "google-health-backfill|u1|slice-3",
+      startAfter: expect.any(Number),
+    });
+    // Not complete: no stamp.
+    expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
+  });
+
+  it("still throws for a slice that finished nothing new", async () => {
+    resumeState.doneBefore = ["steps"];
+    resumeState.finishes = [];
+    syncUserGoogleHealthMock.mockResolvedValue({ imported: 3, failed: true });
+
+    await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow(
+      /incomplete/,
+    );
+    expect(bossSend).not.toHaveBeenCalled();
+  });
+
+  it("throws rather than drop the work when the next slice cannot be queued", async () => {
+    resumeState.finishes = ["sleep"];
+    bossSend.mockResolvedValue(null);
+    syncUserGoogleHealthMock.mockResolvedValue({ imported: 1, failed: true });
+
+    await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow(
+      /incomplete/,
+    );
+  });
+
+  it("stamps the marker on the slice that finishes the history", async () => {
+    resumeState.doneBefore = ["steps", "sleep"];
+    resumeState.finishes = ["weight"];
+    syncUserGoogleHealthMock.mockResolvedValue({ imported: 4, failed: false });
+
+    await expect(
+      runGoogleHealthBackfillForUser("u1", undefined, 5),
+    ).resolves.toEqual({
+      imported: 4,
+    });
+    expect(bossSend).not.toHaveBeenCalled();
+    expect(
+      prismaMock.googleHealthConnection.update.mock.calls[0]![0].data
+        .backfillCompletedAt,
+    ).toBeInstanceOf(Date);
+  });
+
+  it("walks nothing again once an earlier slice completed the import", async () => {
+    prismaMock.googleHealthConnection.findUnique.mockResolvedValue({
+      backfillProgress: null,
+      backfillCompletedAt: new Date(),
+    });
+
+    await expect(
+      runGoogleHealthBackfillForUser("u1", undefined, 4),
+    ).resolves.toEqual({
+      imported: 0,
+    });
+    expect(syncUserGoogleHealthMock).not.toHaveBeenCalled();
+  });
+});

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

@@ -15,7 +15,11 @@ import { Prisma } from "@/generated/prisma/client";
 import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
-import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
+import {
+  enqueueIntegrationBackfillAdmission,
+  INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
+  integrationBackfillSourceOptions,
+} from "@/lib/jobs/integration-backfill-admission";
 import {
   readGoogleHealthBackfillDone,
   runWithGoogleHealthBackfillResume,
@@ -68,6 +72,14 @@ export interface GoogleHealthBackfillPayload {
  * (`runWithGoogleHealthBackfillResume`). The note is cleared once the stamp
  * lands, and on every reconnect.
  *
+ * Runs in slices: an attempt that ends before the whole history is in but
+ * finished at least one more collection than it started with made progress.
+ * It is not a failure: it sends the next slice (`slice + 1`) and returns
+ * `continued`, without the error line or the queue's failure count. Only an
+ * attempt that finished nothing new throws, because that one is stuck: the
+ * pg-boss retry and, after it, the next boot take it from there. Progress is
+ * bounded by the number of collections, so the chain of slices is too.
+ *
  * Never beside another run of the same account: the sync takes the account's
  * lock, waiting a bounded time for an hourly or manual run to finish. If the
  * lock does not come free the attempt throws without doing anything, and the
@@ -76,7 +88,8 @@ export interface GoogleHealthBackfillPayload {
 export async function runGoogleHealthBackfillForUser(
   userId: string,
   shouldStop: () => boolean = () => false,
-): Promise<{ imported: number }> {
+  slice: number = 0,
+): Promise<{ imported: number; continued?: boolean }> {
   if (await isGoogleHealthParked(userId)) {
     annotate({
       action: {
@@ -89,9 +102,21 @@ export async function runGoogleHealthBackfillForUser(
 
   const connection = await prisma.googleHealthConnection.findUnique({
     where: { userId },
-    select: { backfillProgress: true },
+    select: { backfillProgress: true, backfillCompletedAt: true },
   });
+  if (connection?.backfillCompletedAt) {
+    // A slice queued behind the one that finished the import, or a boot
+    // delivery that raced it: the history is in, nothing to walk again.
+    annotate({
+      action: {
+        name: "google_health.backfill.already_complete",
+        details: { slice },
+      },
+    });
+    return { imported: 0 };
+  }
   const done = readGoogleHealthBackfillDone(connection?.backfillProgress);
+  const doneBefore = done.size;
 
   const {
     result: { imported, failed, busy },
@@ -115,9 +140,46 @@ export async function runGoogleHealthBackfillForUser(
   }
 
   if (failed) {
-    // Surface through the pg-boss retry path (retryLimit 3, backoff). If the
-    // failure parked the connection at error_reauth meanwhile, the next
-    // attempt takes the reauth return above instead of failing again.
+    // `done` is the resume set this attempt added to as it finished
+    // collections; a larger set than it started with is progress.
+    const finished = done.size - doneBefore;
+    if (finished > 0) {
+      const boss = getGlobalBoss();
+      const sent = boss
+        ? await enqueueIntegrationBackfillAdmission(
+            boss,
+            {
+              kind: "google-health-backfill",
+              data: {
+                userId,
+                enqueuedAt: new Date().toISOString(),
+                slice: slice + 1,
+              },
+            },
+            INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
+          )
+        : null;
+      if (sent) {
+        annotate({
+          action: {
+            name: "google_health.backfill.continued",
+            details: {
+              imported,
+              slice,
+              collectionsFinished: finished,
+              collectionsDone: done.size,
+            },
+          },
+        });
+        return { imported, continued: true };
+      }
+      // No follow-up could be queued; fall through to the retry path so the
+      // work is not dropped.
+    }
+    // No collection finished in this attempt: the backfill is stuck, not
+    // slow. Surface through the pg-boss retry path (retryLimit 3, backoff).
+    // If the failure parked the connection at error_reauth meanwhile, the
+    // next attempt takes the reauth return above instead of failing again.
     throw new Error(
       `google-health backfill incomplete for user ${userId} — marker not stamped`,
     );

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

@@ -28,6 +28,14 @@ export type IntegrationBackfillData = {
   userId: string;
   enqueuedAt: string;
   provider?: "WHOOP" | "WITHINGS";
+  /**
+   * The continuation number of an import that runs in slices: absent (0) for
+   * the job discovery or a connect sends, 1, 2, ... for each follow-up a
+   * slice that made progress sent for the rest. Part of the job's singleton
+   * key, so the follow-up is not swallowed by the `exclusive` policy while the
+   * slice that sends it is still active; never part of the import identity.
+   */
+  slice?: number;
 };
 
 export interface IntegrationBackfillAdmissionPayload {
@@ -88,7 +96,12 @@ export function integrationBackfillSourceOptions(
   };
 }
 
-export function integrationBackfillAdmissionSingletonKey(
+/**
+ * The import a job belongs to: one kind for one account (and provider). The
+ * import lock keys on it, so no two slices or deliveries of one import ever
+ * run at once.
+ */
+export function integrationBackfillIdentity(
   payload: IntegrationBackfillAdmissionPayload,
 ): string {
   if (payload.kind === "sleep-timeline-backfill") {
@@ -100,17 +113,29 @@ export function integrationBackfillAdmissionSingletonKey(
   return `${payload.kind}|${payload.data.userId}`;
 }
 
+/** The job's singleton key: the import identity, plus its slice number. */
+export function integrationBackfillAdmissionSingletonKey(
+  payload: IntegrationBackfillAdmissionPayload,
+): string {
+  const identity = integrationBackfillIdentity(payload);
+  const slice = payload.data.slice ?? 0;
+  return slice > 0 ? `${identity}|slice-${slice}` : identity;
+}
+
 export async function enqueueIntegrationBackfillAdmission(
   boss: BossSender,
   payload: IntegrationBackfillAdmissionPayload,
+  startAfterSeconds: number = 0,
 ): Promise<string | null> {
-  return boss.send(
-    INTEGRATION_BACKFILL_ADMISSION_QUEUE,
-    payload,
-    integrationBackfillAdmissionSendOptions(payload),
-  );
+  return boss.send(INTEGRATION_BACKFILL_ADMISSION_QUEUE, payload, {
+    ...integrationBackfillAdmissionSendOptions(payload),
+    ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
+  });
 }
 
+/** Breathing room between one slice of an import and the next. */
+export const INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS = 60;
+
 /**
  * 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

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

@@ -25,7 +25,7 @@ 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,
+  integrationBackfillIdentity,
   type IntegrationBackfillAdmissionPayload,
 } from "@/lib/jobs/integration-backfill-admission";
 import { withIntegrationBackfillLock } from "@/lib/jobs/integration-backfill-lock";
@@ -36,6 +36,8 @@ import { withBackgroundEvent } from "@/lib/logging/background";
 
 interface RunCounts {
   imported: number;
+  /** Imports that ended a slice with progress and sent the next one. */
+  continued: number;
   removed: number;
   deleted: number;
   markers: number;
@@ -66,14 +68,16 @@ async function runAdmitted(
       return;
     }
     case "google-health-backfill": {
-      const { imported } = await runGoogleHealthBackfillForUser(
+      const { imported, continued } = await runGoogleHealthBackfillForUser(
         userId,
         shouldStop,
+        data.slice ?? 0,
       );
       totals.imported += imported;
+      if (continued) totals.continued += 1;
       workerLog(
         "info",
-        `[google-health-backfill] user=${userId} imported=${imported}`,
+        `[google-health-backfill] user=${userId} imported=${imported}${continued ? " continued" : ""}`,
       );
       return;
     }
@@ -152,6 +156,7 @@ export async function drainIntegrationBackfillAdmission(
   const shouldStop = jobBudget(jobs);
   const totals: RunCounts = {
     imported: 0,
+    continued: 0,
     removed: 0,
     deleted: 0,
     markers: 0,
@@ -159,7 +164,7 @@ export async function drainIntegrationBackfillAdmission(
   };
   let alreadyRunning = 0;
   for (const job of jobs) {
-    const identity = integrationBackfillAdmissionSingletonKey(job.data);
+    const identity = integrationBackfillIdentity(job.data);
     const run = await withIntegrationBackfillLock(identity, () =>
       runAdmitted(job.data, shouldStop, totals),
     );
@@ -183,6 +188,7 @@ export async function drainIntegrationBackfillAdmission(
     jobs: jobs.length,
     already_running: alreadyRunning,
     imported: totals.imported,
+    continued: totals.continued,
     removed: totals.removed,
     deleted: totals.deleted,
     markers: totals.markers,

tests/integration/google-health-backfill-rate-limit.test.ts

@@ -168,6 +168,20 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
   };
 });
 
+/** The follow-up slices a progressing backfill queues, recorded. */
+const sent = vi.hoisted(() => ({
+  jobs: [] as Array<{ queue: string; data: unknown; options: unknown }>,
+}));
+vi.mock("@/lib/jobs/boss-instance", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/jobs/boss-instance")>()),
+  getGlobalBoss: () => ({
+    send: async (queue: string, data: unknown, options: unknown) => {
+      sent.jobs.push({ queue, data, options });
+      return `job-${sent.jobs.length}`;
+    },
+  }),
+}));
+
 vi.mock("@/lib/db-compat", () => ({
   ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
 }));
@@ -212,6 +226,7 @@ beforeEach(async () => {
   fake.requestsByPath = new Map();
   fake.status429 = 0;
   fake.status401 = 0;
+  sent.jobs = [];
   await seed();
 }, 120_000);
 
@@ -353,9 +368,19 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
         fake.requestsByPath.get("heart-rate/dataPoints") ?? 0;
       return heartRatePagesSeen >= 3;
     };
+    // It finished four collections, so it is progressing, not failing: it
+    // ends quietly and queues the next slice of the import.
     await expect(
       runGoogleHealthBackfillForUser(USER, stopAfterThreeHeartRatePages),
-    ).rejects.toThrow(/incomplete/);
+    ).resolves.toMatchObject({ continued: true });
+    expect(sent.jobs).toHaveLength(1);
+    expect(sent.jobs[0]).toMatchObject({
+      queue: "integration-backfill-admission",
+      data: {
+        kind: "google-health-backfill",
+        data: { userId: USER, slice: 1 },
+      },
+    });
 
     const afterFirst =
       await getPrismaClient().googleHealthConnection.findUnique({
@@ -373,9 +398,10 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
     );
     expect(done).not.toContain("fetchHeartRate");
 
-    // Second attempt: the finished collections are not read again.
+    // Second attempt, the queued slice: the finished collections are not
+    // read again.
     fake.requestsByPath = new Map();
-    await runGoogleHealthBackfillForUser(USER);
+    await runGoogleHealthBackfillForUser(USER, undefined, 1);
 
     expect(fake.requestsByPath.get("exercise/dataPoints")).toBeUndefined();
     expect(fake.requestsByPath.get("sleep/dataPoints")).toBeUndefined();
@@ -392,6 +418,29 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
     expect(afterSecond?.backfillProgress).toBeNull();
   });
 
+  it("keeps failing a backfill whose slice finished nothing new", async () => {
+    const { runGoogleHealthBackfillForUser } =
+      await import("@/lib/jobs/google-health-backfill");
+
+    // A first slice finishes everything but the heart-rate stream.
+    const stopAtThirdPage = () =>
+      (fake.requestsByPath.get("heart-rate/dataPoints") ?? 0) >= 3;
+    await runGoogleHealthBackfillForUser(USER, stopAtThirdPage);
+    expect(sent.jobs).toHaveLength(1);
+
+    // The next slice cannot get through the heart-rate stream either: no
+    // collection finishes, so this one is stuck and says so.
+    fake.requestsByPath = new Map();
+    await expect(
+      runGoogleHealthBackfillForUser(USER, stopAtThirdPage, 1),
+    ).rejects.toThrow(/incomplete/);
+    expect(sent.jobs).toHaveLength(1);
+    const row = await getPrismaClient().googleHealthConnection.findUnique({
+      where: { userId: USER },
+    });
+    expect(row?.backfillCompletedAt).toBeNull();
+  });
+
   it("never runs the hourly poll or a manual sync beside a running backfill", async () => {
     fake.holdAtPage = 2;
     fake.held = new Promise<void>((resolve) => {