fix(google-health): count backfill progress only once it is persisted and cap the slice chain

src/lib/google-health/__tests__/backfill-resume-progress.test.ts

@@ -0,0 +1,65 @@
+/**
+ * A collection counts toward a backfill's progress only once its resume note
+ * is written. The job reads the size of `done` as progress and queues another
+ * slice on it, so a key added before a write that failed would let a chain of
+ * slices "progress" over the same collection without end.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const { prismaMock, warnings } = vi.hoisted(() => ({
+  prismaMock: {
+    googleHealthConnection: { update: vi.fn() },
+  },
+  warnings: [] as string[],
+}));
+
+vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
+vi.mock("@/lib/logging/context", () => ({
+  annotate: () => {},
+  getEvent: () => ({ addWarning: (w: string) => warnings.push(w) }),
+}));
+
+import {
+  runGoogleHealthCollection,
+  runWithGoogleHealthBackfillResume,
+} from "../sync-core";
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  warnings.length = 0;
+});
+
+describe("runGoogleHealthCollection under a backfill", () => {
+  it("adds a clean collection to done once its note is written", async () => {
+    prismaMock.googleHealthConnection.update.mockResolvedValue({});
+    const done = new Set<string>(["fetchSteps"]);
+
+    await runWithGoogleHealthBackfillResume("u1", done, () =>
+      runGoogleHealthCollection("sleep", async () => 3),
+    );
+
+    expect([...done]).toEqual(["fetchSteps", "sleep"]);
+    expect(prismaMock.googleHealthConnection.update).toHaveBeenCalledWith({
+      where: { userId: "u1" },
+      data: { backfillProgress: { done: ["fetchSteps", "sleep"] } },
+    });
+  });
+
+  it("counts no progress when the note cannot be written", async () => {
+    prismaMock.googleHealthConnection.update.mockRejectedValue(
+      new Error("connection reset"),
+    );
+    const done = new Set<string>(["fetchSteps"]);
+
+    const { result } = await runWithGoogleHealthBackfillResume("u1", done, () =>
+      runGoogleHealthCollection("sleep", async () => 3),
+    );
+
+    // The rows are in, the collection did not fail, but it is not finished.
+    expect(result).toBe(3);
+    expect([...done]).toEqual(["fetchSteps"]);
+    expect(warnings.some((w) => w.includes("progress write failed"))).toBe(
+      true,
+    );
+  });
+});

src/lib/google-health/collection-keys.ts

@@ -0,0 +1,29 @@
+/**
+ * Every collection a Google Health backfill walks, by the key its resume note
+ * stores (`runGoogleHealthCollection`). The key type is what that function
+ * accepts, so a collection added without an entry here does not compile, and
+ * the backfill's slice ceiling (`GOOGLE_HEALTH_BACKFILL_MAX_SLICE`) is derived
+ * from this list rather than counted by hand.
+ */
+export const GOOGLE_HEALTH_COLLECTION_KEYS = [
+  "fetchSteps",
+  "fetchDistance",
+  "fetchActiveEnergy",
+  "fetchFloors",
+  "fetchVo2Max",
+  "fetchWeight",
+  "fetchBodyFat",
+  "fetchSpo2",
+  "fetchHrv",
+  "fetchRhr",
+  "fetchRespiratoryRate",
+  "fetchHeartRate",
+  "fetchBloodGlucose",
+  "fetchCoreBodyTemperature",
+  "fetchSleepTemperature",
+  "fetchExercise",
+  "sleep",
+] as const;
+
+export type GoogleHealthCollectionKey =
+  (typeof GOOGLE_HEALTH_COLLECTION_KEYS)[number];

src/lib/google-health/sync-activity.ts

@@ -46,6 +46,7 @@ import {
   type GoogleHealthResourceSyncOptions,
 } from "./sync-core";
 import { annotate } from "@/lib/logging/context";
+import type { GoogleHealthCollectionKey } from "./collection-keys";
 import { resolveUserTimezone } from "@/lib/tz/resolver";
 
 /** One mappable activity metric: its data-type encoding + the mapper + a verb. */
@@ -55,7 +56,7 @@ interface ActivityResource {
     point: Record<string, unknown>,
     tz: string,
   ) => GoogleHealthMappedMeasurement[];
-  verb: string;
+  verb: GoogleHealthCollectionKey;
 }
 
 /** The four cumulative daily totals — read via `POST :dailyRollUp`. */

src/lib/google-health/sync-core.ts

@@ -20,6 +20,7 @@ import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { encrypt, decrypt } from "@/lib/crypto";
 import { annotate, getEvent } from "@/lib/logging/context";
+import type { GoogleHealthCollectionKey } from "./collection-keys";
 import { userDayKey } from "@/lib/tz/format";
 import { isP2002 } from "@/lib/prisma-errors";
 import {
@@ -464,7 +465,7 @@ export async function runWithGoogleHealthBackfillResume<T>(
  * budget asking it to stop). Anything less is walked again next attempt.
  */
 export async function runGoogleHealthCollection(
-  key: string,
+  key: GoogleHealthCollectionKey,
   fn: () => Promise<number>,
 ): Promise<number> {
   const resume = backfillResumeStorage.getStore();
@@ -483,19 +484,27 @@ export async function runGoogleHealthCollection(
     !truncatedBefore &&
     !googleHealthOutcomeTruncated();
   if (clean) {
-    resume.done.add(key);
-    await prisma.googleHealthConnection
-      .update({
+    // The collection counts as finished only once the note says so. The job
+    // reads the size of `done` as this attempt's progress and sends another
+    // slice on it; a key added before a write that then failed would be
+    // progress nobody can see next attempt, and a chain of slices could keep
+    // "progressing" over the same collection forever.
+    try {
+      await prisma.googleHealthConnection.update({
         where: { userId: resume.userId },
-        data: { backfillProgress: { done: [...resume.done] } },
-      })
-      .catch((err) =>
-        // Losing the note costs a re-walk of this collection next attempt,
-        // nothing more; it must not fail a collection that did complete.
-        getEvent()?.addWarning(
-          `google-health: backfill progress write failed: ${err}`,
-        ),
+        data: { backfillProgress: { done: [...resume.done, key] } },
+      });
+      resume.done.add(key);
+    } catch (err) {
+      // Not finished, then: the collection is walked again next attempt. It
+      // must not fail a collection whose rows are written.
+      getEvent()?.addWarning(
+        `google-health: backfill progress write failed for ${key}: ${err}`,
       );
+      annotate({
+        meta: { "googleHealth.backfill.progressWriteFailed": key },
+      });
+    }
   }
   return imported;
 }

src/lib/google-health/sync-metrics.ts

@@ -55,6 +55,7 @@ import {
   type GoogleHealthResourceSyncOptions,
 } from "./sync-core";
 import { getEvent } from "@/lib/logging/context";
+import type { GoogleHealthCollectionKey } from "./collection-keys";
 import { prisma } from "@/lib/db";
 import { resolveUserTimezone } from "@/lib/tz/resolver";
 
@@ -65,7 +66,7 @@ interface MetricResource {
     point: Record<string, unknown>,
     tz: string,
   ) => GoogleHealthMappedMeasurement[];
-  verb: string;
+  verb: GoogleHealthCollectionKey;
   /** Page ceiling for the walk; the client default when omitted. */
   maxPages?: number;
   /**

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

@@ -69,7 +69,10 @@ vi.mock("@/lib/google-health/sync", () => ({
   syncUserGoogleHealth: (...a: unknown[]) => syncUserGoogleHealthMock(...a),
 }));
 
-import { runGoogleHealthBackfillForUser } from "../google-health-backfill";
+import {
+  GOOGLE_HEALTH_BACKFILL_MAX_SLICE,
+  runGoogleHealthBackfillForUser,
+} from "../google-health-backfill";
 import {
   GOOGLE_HEALTH_DATA_TYPES,
   forEachDataPointPage,
@@ -229,6 +232,39 @@ describe("runGoogleHealthBackfillForUser — slices", () => {
     );
   });
 
+  it("fails a slice past the ceiling without walking or queueing", async () => {
+    syncUserGoogleHealthMock.mockResolvedValue({ imported: 1, failed: true });
+    resumeState.finishes = ["sleep"];
+
+    await expect(
+      runGoogleHealthBackfillForUser(
+        "u1",
+        undefined,
+        GOOGLE_HEALTH_BACKFILL_MAX_SLICE + 1,
+      ),
+    ).rejects.toThrow(/past the ceiling/);
+    expect(syncUserGoogleHealthMock).not.toHaveBeenCalled();
+    expect(bossSend).not.toHaveBeenCalled();
+  });
+
+  it("does not queue a slice past the ceiling even when this one progressed", async () => {
+    resumeState.finishes = ["sleep"];
+    syncUserGoogleHealthMock.mockResolvedValue({ imported: 1, failed: true });
+
+    await expect(
+      runGoogleHealthBackfillForUser(
+        "u1",
+        undefined,
+        GOOGLE_HEALTH_BACKFILL_MAX_SLICE,
+      ),
+    ).rejects.toThrow(/incomplete/);
+    expect(bossSend).not.toHaveBeenCalled();
+  });
+
+  it("caps the chain at the number of collections plus two", () => {
+    expect(GOOGLE_HEALTH_BACKFILL_MAX_SLICE).toBe(17 + 2);
+  });
+
   it("stamps the marker on the slice that finishes the history", async () => {
     resumeState.doneBefore = ["steps", "sleep"];
     resumeState.finishes = ["weight"];

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

@@ -30,6 +30,7 @@ import {
 } from "@/lib/google-health/sync";
 import { GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock";
 import { runWithGoogleHealthStop } from "@/lib/google-health/client";
+import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys";
 
 export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill";
 
@@ -40,6 +41,17 @@ export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill";
  */
 export const GOOGLE_HEALTH_BACKFILL_CONCURRENCY = 1;
 
+/**
+ * The highest slice a backfill chain may reach. Every continued slice has
+ * finished at least one more collection than the one before it, so a healthy
+ * chain is at most one slice per collection long; the two extra slices are
+ * slack for the first one and the one that stamps the marker. A slice past
+ * this is a chain that counts progress it does not make, and it fails rather
+ * than queueing another.
+ */
+export const GOOGLE_HEALTH_BACKFILL_MAX_SLICE =
+  GOOGLE_HEALTH_COLLECTION_KEYS.length + 2;
+
 export interface GoogleHealthBackfillPayload {
   userId: string;
   enqueuedAt: string;
@@ -78,7 +90,10 @@ export interface GoogleHealthBackfillPayload {
  * `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.
+ * counted only once it is persisted (`runGoogleHealthCollection` adds a
+ * collection to `done` after its note is written), and the chain is capped at
+ * `GOOGLE_HEALTH_BACKFILL_MAX_SLICE`: a slice past it throws instead of
+ * running or sending another.
  *
  * 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
@@ -90,6 +105,17 @@ export async function runGoogleHealthBackfillForUser(
   shouldStop: () => boolean = () => false,
   slice: number = 0,
 ): Promise<{ imported: number; continued?: boolean }> {
+  if (slice > GOOGLE_HEALTH_BACKFILL_MAX_SLICE) {
+    annotate({
+      action: {
+        name: "google_health.backfill.slice_ceiling",
+        details: { slice, maxSlice: GOOGLE_HEALTH_BACKFILL_MAX_SLICE },
+      },
+    });
+    throw new Error(
+      `google-health backfill for user ${userId} reached slice ${slice}, past the ceiling of ${GOOGLE_HEALTH_BACKFILL_MAX_SLICE}`,
+    );
+  }
   if (await isGoogleHealthParked(userId)) {
     annotate({
       action: {
@@ -143,7 +169,7 @@ export async function runGoogleHealthBackfillForUser(
     // `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) {
+    if (finished > 0 && slice + 1 <= GOOGLE_HEALTH_BACKFILL_MAX_SLICE) {
       const boss = getGlobalBoss();
       const sent = boss
         ? await enqueueIntegrationBackfillAdmission(