fix(google-health): keep the skip's place across backfill slices

A walk that reads without the edge bound passed over everything earlier
slices wrote on every slice, and once it crossed the edge the skipped
pages were counted against the walk's page ceiling. The skip now stays
counted for the whole walk, runs on the slice's clock and its own
three-minute limit, and saves the page token it stopped at (on a limit,
a rate limit or the slice budget) so the next slice re-reads that one
page and goes on. A saved place counts as progress only when it moves
forward; a token Google refuses, or one that no longer reaches the edge,
sends the walk back to the newest page as before.

docs/integrations/google-health.md

@@ -143,12 +143,17 @@ Google answers that first resumed request with `400`, the slice reads the data
 type again without the bound, passes over the readings newer than its note
 without writing them, and carries on writing from there. The refusal is noted
 per data type, so later slices read without the bound straight away. The pages
-passed over are not counted against the slice's five minutes or 500 pages:
-otherwise a history deeper than one slice would spend every slice passing over
-what earlier slices wrote and never write anything new. Only the job's own
-time limit stops a slice while it is passing over, so on such an account each
-slice takes longer the further back the backfill has got. A `400` that names
-something else (the page token, the page size) is still a failure.
+passed over are not counted against the slice's 500 pages, but they do run on
+its five minutes, and passing over stops after three minutes at most, so a
+slice never holds the account longer than a slice. Wherever the walk stops, on
+a limit or on a rate limit, it keeps Google's page token for the page it read
+last, and the next slice reads that one page again and goes on from there
+instead of passing over everything earlier slices wrote: the pages read stay
+in proportion to the history. If Google no longer takes the token, or the page
+it returns no longer reaches the noted reading, the slice starts again from the
+newest page and passes over as before. A `400` that names something else (the
+page size, or the page token of a walk that did not start from a kept one) is
+still a failure.
 
 The hourly sync waits for the slice that is running (up to eight minutes)
 instead of skipping the account, so recent data keeps arriving while an old

src/lib/google-health/__tests__/edge-filter-fallback.test.ts

@@ -33,6 +33,7 @@ vi.mock("@/lib/logging/context", () => ({
 
 import {
   GOOGLE_HEALTH_DATA_TYPES,
+  GOOGLE_HEALTH_SKIP_MS,
   type GoogleHealthDataPoint,
   forEachDataPointPage,
   googleHealthSliceBudget,
@@ -47,7 +48,9 @@ import {
   readGoogleHealthBackfillCursors,
   readGoogleHealthBackfillEdgeFilterUnsupported,
   recordGoogleHealthBackfillPage,
+  recordGoogleHealthBackfillResumePoint,
   runWithGoogleHealthBackfillResume,
+  type GoogleHealthBackfillCursor,
 } from "../sync-core";
 
 const MINUTE = 60_000;
@@ -284,31 +287,181 @@ describe("passing over the samples beyond the edge inside a slice", () => {
     expect(outcome.truncated).toBe(true);
   });
 
-  it("does not stop on the slice's clock while it is still passing over", async () => {
+  it("stops on the slice's clock while it is still passing over, and keeps its place", async () => {
     let clock = 0;
     safeFetchMock.mockImplementation(async (url: string) => {
       clock += 10_000;
       return respond(url);
     });
-    // Each request takes ten seconds and the slice has fifteen, counted from
-    // when the collection started; passing over the first three pages takes
-    // forty.
+    // Each request takes ten seconds and the slice has fifteen; passing over
+    // the first three pages takes thirty. The slice does not run on past its
+    // clock for the skip any more: it stops and saves where it got to.
     const budget = googleHealthSliceBudget(15_000, 500, () => clock);
-    budget.deadline = 15_000;
-    const { pages } = await runWithGoogleHealthStop(
+    const onResumePoint = vi.fn();
+    const { pages, outcome } = await runWithGoogleHealthStop(
       () => false,
       () =>
         walk({
           resumable: true,
-          edge: { at: at(15), order: "desc", filter: false },
+          edge: { at: at(15), order: "desc", filter: false, onResumePoint },
         }),
       budget,
     );
-    // The crossing page and the two after it, inside the fifteen seconds the
-    // slice still had once the skip was given back.
-    expect(pages).toHaveLength(3);
-    expect(budget.deadline).toBe(55_000);
+    expect(pages).toHaveLength(0);
+    expect(budget.deadline).toBe(15_000);
     expect(budget.stopReason).toBe("time_budget");
+    expect(outcome.truncated).toBe(true);
+    expect(onResumePoint).toHaveBeenCalledWith({ token: "1", page: 1 });
+  });
+
+  it("stops passing over on its own limit inside a long slice, and keeps its place", async () => {
+    let clock = 0;
+    safeFetchMock.mockImplementation(async (url: string) => {
+      clock += 60_000;
+      return respond(url);
+    });
+    // The edge is six pages deep; each request takes a minute, and the skip
+    // may run for three of the slice's sixty.
+    const budget = googleHealthSliceBudget(3_600_000, 500, () => clock);
+    const onResumePoint = vi.fn();
+    const { pages, outcome } = await runWithGoogleHealthStop(
+      () => false,
+      () =>
+        walk({
+          resumable: true,
+          edge: { at: at(30), order: "desc", filter: false, onResumePoint },
+        }),
+      budget,
+    );
+    expect(GOOGLE_HEALTH_SKIP_MS).toBe(3 * 60_000);
+    expect(pages).toHaveLength(0);
+    expect(safeFetchMock).toHaveBeenCalledTimes(3);
+    expect(outcome.truncated).toBe(true);
+    // The slice itself is not over: other collections may still run.
+    expect(budget.stopReason).toBeNull();
+    expect(onResumePoint).toHaveBeenCalledWith({ token: "2", page: 2 });
+  });
+
+  it("goes on from the saved place, rereading only its page", async () => {
+    const filtered = await walk({
+      resumable: true,
+      edge: { at: at(30), order: "desc" },
+    });
+    server.requests = [];
+    safeFetchMock.mockClear();
+    const tokens: Array<string | null> = [];
+    safeFetchMock.mockImplementation(async (url: string) => {
+      tokens.push(new URL(url).searchParams.get("pageToken"));
+      return respond(url);
+    });
+    const { rows, outcome } = await walk({
+      resumable: true,
+      edge: {
+        at: at(30),
+        order: "desc",
+        filter: false,
+        resumeFrom: { token: "2", page: 2 },
+      },
+    });
+    expect(rows).toEqual(filtered.rows);
+    expect(tokens).toEqual(["2", "3", "4", "5", "6", "7"]);
+    expect(outcome.truncated).toBe(false);
+  });
+
+  it("saves its place on a rate limit it cannot wait out", async () => {
+    let n = 0;
+    safeFetchMock.mockImplementation(async (url: string) =>
+      ++n === 3
+        ? new Response(
+            JSON.stringify({
+              error: { code: 429, status: "RESOURCE_EXHAUSTED" },
+            }),
+            {
+              status: 429,
+              headers: {
+                "content-type": "application/json",
+                "retry-after": "3600",
+              },
+            },
+          )
+        : respond(url),
+    );
+    const budget = googleHealthSliceBudget(3_600_000, 500);
+    const onResumePoint = vi.fn();
+    const { pages, outcome } = await runWithGoogleHealthStop(
+      () => false,
+      () =>
+        walk({
+          resumable: true,
+          edge: { at: at(30), order: "desc", filter: false, onResumePoint },
+        }),
+      budget,
+    );
+    expect(pages).toHaveLength(0);
+    expect(budget.stopReason).toBe("rate_limited");
+    expect(outcome.truncated).toBe(true);
+    expect(onResumePoint).toHaveBeenCalledWith({ token: "1", page: 1 });
+  });
+
+  it("reads from the newest page when Google refuses the saved token", async () => {
+    let n = 0;
+    safeFetchMock.mockImplementation(async (url: string) =>
+      ++n === 1
+        ? json(
+            {
+              error: {
+                code: 400,
+                status: "INVALID_ARGUMENT",
+                message: "Invalid page token.",
+              },
+            },
+            400,
+          )
+        : respond(url),
+    );
+    const { rows, outcome } = await walk({
+      resumable: true,
+      edge: {
+        at: at(30),
+        order: "desc",
+        filter: false,
+        resumeFrom: { token: "2", page: 2 },
+      },
+    });
+    expect(rows).toHaveLength(TOTAL - 30);
+    expect(outcome.truncated).toBe(false);
+  });
+
+  it("reads from the newest page when the saved place no longer reaches the edge", async () => {
+    // Page 7 holds only samples older than the edge: the list moved under
+    // the token, and going on from it could step over samples nobody wrote.
+    const { rows } = await walk({
+      resumable: true,
+      edge: {
+        at: at(30),
+        order: "desc",
+        filter: false,
+        resumeFrom: { token: "7", page: 7 },
+      },
+    });
+    expect(rows.map((p) => p.heartRate)).toEqual(
+      Array.from({ length: TOTAL - 30 }, (_, i) => ({
+        beatsPerMinute: String(60 + 30 + i),
+        sampleTime: { physicalTime: at(30 + i).toISOString() },
+      })),
+    );
+  });
+
+  it("does not charge the pages it passed over against the walk's own ceiling", async () => {
+    // Three pages passed over, then a ceiling of two written pages: both are
+    // written, not one.
+    const { pages, outcome } = await walk({
+      resumable: true,
+      maxPages: 2,
+      edge: { at: at(15), order: "desc", filter: false },
+    });
+    expect(pages).toHaveLength(2);
+    expect(outcome.truncated).toBe(true);
   });
 
   it("still stops on the job's own budget", async () => {
@@ -410,6 +563,75 @@ describe("the note that a collection refuses the edge filter", () => {
     expect(lastProgress()).not.toHaveProperty("edgeFilterUnsupported");
   });
 
+  it("keeps a saved place in the list, counts it as progress only when it moves forward, and hands it to the next slice", async () => {
+    server.refuseUpperBound = true;
+    const cursor = {
+      at: at(30).toISOString(),
+      order: "desc" as const,
+      pages: 4,
+    };
+    const run = (
+      cursors: Map<string, GoogleHealthBackfillCursor>,
+      point: { token: string; page: number },
+    ) =>
+      runWithGoogleHealthBackfillResume(
+        "u1",
+        new Set(),
+        () => recordGoogleHealthBackfillResumePoint("fetchHeartRate", point),
+        cursors,
+        new Set(["fetchHeartRate"]),
+      );
+
+    const first = await run(new Map([["fetchHeartRate", cursor]]), {
+      token: "4",
+      page: 4,
+    });
+    expect(first.advanced.has("fetchHeartRate")).toBe(true);
+    const saved = readGoogleHealthBackfillCursors(lastProgress());
+    expect(saved.get("fetchHeartRate")).toEqual({
+      ...cursor,
+      resume: { token: "4", page: 4, reached: 4 },
+    });
+
+    // The same place again, or one short of it, is not progress.
+    const again = await run(new Map(saved), { token: "4", page: 4 });
+    expect(again.advanced.size).toBe(0);
+    const back = await run(new Map(saved), { token: "2", page: 2 });
+    expect(back.advanced.size).toBe(0);
+    expect(
+      readGoogleHealthBackfillCursors(lastProgress()).get("fetchHeartRate")
+        ?.resume,
+    ).toEqual({ token: "2", page: 2, reached: 4 });
+
+    // A page written keeps the place; the next slice is handed it.
+    await runWithGoogleHealthBackfillResume(
+      "u1",
+      new Set(),
+      () =>
+        recordGoogleHealthBackfillPage("fetchHeartRate", {
+          first: at(30),
+          last: at(34),
+        }),
+      new Map(saved),
+      new Set(["fetchHeartRate"]),
+    );
+    const next = readGoogleHealthBackfillCursors(lastProgress());
+    expect(next.get("fetchHeartRate")?.resume).toEqual(
+      saved.get("fetchHeartRate")?.resume,
+    );
+    await runWithGoogleHealthBackfillResume(
+      "u1",
+      new Set(),
+      async () => {
+        expect(
+          googleHealthBackfillCursor("fetchHeartRate")?.edge,
+        ).toMatchObject({ filter: false, resumeFrom: { token: "4", page: 4 } });
+      },
+      next,
+      new Set(["fetchHeartRate"]),
+    );
+  });
+
   it("reads only well-formed keys", () => {
     expect([
       ...readGoogleHealthBackfillEdgeFilterUnsupported({

src/lib/google-health/client.ts

@@ -169,7 +169,11 @@ export interface GoogleHealthSliceBudget {
   /**
    * Of `pages`, those a walk only passed over on its way to its edge, because
    * Google refused the edge filter (see `forEachDataPointPage`). They are not
-   * charged against `maxPages`, and the time they took moves `deadline` on.
+   * charged against `maxPages`. Their time is: a skip runs on the slice's
+   * clock like any other read, and on its own shorter limit
+   * (`GOOGLE_HEALTH_SKIP_MS`), and keeps its place for the next slice when it
+   * stops. Only a skip that had to start over from the newest page, because
+   * its saved place was refused, moves `deadline` on by the time it took.
    */
   skippedPages: number;
   stopReason: GoogleHealthSliceStopReason | null;
@@ -253,14 +257,21 @@ function noteSlicePage(): void {
 }
 
 /**
- * Give back to the slice in scope the pages and time a walk spent passing
- * over samples beyond its edge (no-op outside a slice).
+ * Note one page the walk only passed over on its way to its edge: read, but
+ * not charged against the slice's pages (no-op outside a slice).
  */
-function refundSliceSkip(pages: number, ms: number): void {
+function noteSliceSkip(): void {
   const slice = stopStorage.getStore()?.slice;
-  if (!slice) return;
-  slice.skippedPages += pages;
-  if (slice.deadline !== null) slice.deadline += ms;
+  if (slice) slice.skippedPages += 1;
+}
+
+/**
+ * Give back to the slice in scope the time a skip took that had to start over
+ * from the newest page (no-op outside a slice).
+ */
+function extendSliceDeadline(ms: number): void {
+  const slice = stopStorage.getStore()?.slice;
+  if (slice && slice.deadline !== null) slice.deadline += ms;
 }
 
 /** The slice clock in scope, or the wall clock outside a slice. */
@@ -747,8 +758,13 @@ interface DataPointQuery {
    * beyond the edge here instead (see `forEachDataPointPage`); the walk falls
    * back to that on its own when Google refuses the bound, and calls
    * `onFilterUnsupported` once when it does. `tz` reads an offset-less sample
-   * time the way the edge was recorded. `maxSkipPages` bounds how many pages
-   * the walk may pass over before it reaches the edge.
+   * time the way the edge was recorded. `maxSkipPages` bounds how far into
+   * the unbounded list the walk may get before it reaches the edge.
+   *
+   * A walk without the bound keeps its place in that list across slices:
+   * `resumeFrom` is where an earlier slice's walk stopped, and
+   * `onResumePoint` is told where this one stops (see
+   * `GoogleHealthResumePoint`).
    */
   edge?: {
     at: Date;
@@ -757,9 +773,38 @@ interface DataPointQuery {
     tz?: string;
     onFilterUnsupported?: () => void | Promise<void>;
     maxSkipPages?: number;
+    resumeFrom?: GoogleHealthResumePoint;
+    onResumePoint?: (point: GoogleHealthResumePoint) => void | Promise<void>;
   };
 }
 
+/**
+ * Where a walk without the edge bound stopped in the unbounded list: the page
+ * token that fetched the last page it read, and that page's position (pages
+ * from the newest one, which is position 0). The next slice starts by reading
+ * that page again rather than the newest one, so it passes over only what is
+ * left between it and the edge instead of everything earlier slices wrote.
+ *
+ * Re-reading the last page, not the one after it, is what lets the walk check
+ * the token still points where it did: the page must hold a sample at or
+ * beyond the edge. A list that lost samples in between could otherwise move
+ * the token past samples nobody wrote. A page that fails the check, or a
+ * token Google refuses, sends the walk back to the newest page, the way it
+ * read before it kept its place.
+ */
+export interface GoogleHealthResumePoint {
+  token: string;
+  page: number;
+}
+
+/**
+ * The longest a walk passes over samples beyond its edge in one slice before
+ * it stops and keeps its place for the next. Shorter than the slice, so a
+ * slice that has to skip still writes, and never longer than it: a skip runs
+ * on the slice's clock and the account's lock is held no longer than a slice.
+ */
+export const GOOGLE_HEALTH_SKIP_MS = 3 * 60 * 1000;
+
 const pad2 = (n: number): string => String(n).padStart(2, "0");
 
 /**
@@ -1202,158 +1247,202 @@ export async function forEachDataPointPage(
   const walk = async (
     dateStyle: GoogleHealthDateFilterStyle,
     edgeBy: "filter" | "skip",
+    resumeFrom: GoogleHealthResumePoint | undefined,
+    restarted: boolean,
   ): Promise<void> => {
     let fetched = 0;
-    let pageToken: string | null | undefined;
+    let pageToken: string | null | undefined = resumeFrom?.token;
+    // The token that fetched the page last read, and that page's position.
+    let lastToken: string | undefined;
+    let lastPosition = -1;
     let pageCount = 0;
     let stopped = false;
     // Reading without the edge bound: the pages before the edge are passed
     // over, not written, until the first sample at or beyond it.
     const skipTo = edge && edgeBy === "skip" ? edge : undefined;
     let crossed = skipTo === undefined;
+    // Pages passed over in this walk. Never reset: the walk's page ceiling
+    // counts the pages it wrote, `pageCount - skipped`, before the crossing
+    // and after it alike.
     let skipped = 0;
+    const startPosition = resumeFrom?.page ?? 0;
     const skipStartedAt = skipTo ? sliceNow() : 0;
     const maxSkip = skipTo?.maxSkipPages ?? GOOGLE_HEALTH_DENSE_MAX_PAGES;
+    const sampleAt = (point: GoogleHealthDataPoint): number | null =>
+      googleHealthSampleInstant(point, dataType, skipTo?.tz)?.getTime() ?? null;
     const beyondEdge = (point: GoogleHealthDataPoint): boolean | null => {
-      const t = googleHealthSampleInstant(point, dataType, skipTo?.tz);
+      const t = sampleAt(point);
       if (t === null || !skipTo) return null;
       return skipTo.order === "desc"
-        ? t.getTime() > skipTo.at.getTime()
-        : t.getTime() < skipTo.at.getTime();
+        ? t > skipTo.at.getTime()
+        : t < skipTo.at.getTime();
     };
-    const refundSkip = () => {
-      if (skipped > 0) refundSliceSkip(skipped, sliceNow() - skipStartedAt);
-      skipped = 0;
+    const reachesEdge = (point: GoogleHealthDataPoint): boolean => {
+      const t = sampleAt(point);
+      if (t === null || !skipTo) return false;
+      return skipTo.order === "desc"
+        ? t >= skipTo.at.getTime()
+        : t <= skipTo.at.getTime();
     };
 
-    try {
-      do {
-        // A walk still passing over the samples beyond its edge does not stop
-        // on the slice's budget, only on the job's: the skip writes nothing,
-        // so a slice that stopped inside it would have nothing to show, and
-        // the next slice would start the same skip over (see
-        // `GoogleHealthSliceBudget.skippedPages`).
-        if (crossed ? stopRequested(resumable) : stopRequested(false)) {
-          stopped = true;
-          break;
-        }
-        if (!crossed && skipped >= maxSkip) {
-          getEvent()?.addWarning(
-            `google-health ${verb}: passed over ${skipped} pages without reaching the resume edge`,
-          );
-          stopped = true;
-          break;
-        }
-        const params = new URLSearchParams({ pageSize: String(pageSize) });
-        const filters: string[] = [];
-        if (query.start) {
-          const { field, bound } = incrementalFilter(
-            dataType,
-            query.start,
-            query.tz,
-            dateStyle,
-          );
-          filters.push(`${field} >= "${bound}"`);
-        }
-        const edgeFiltered = edge !== undefined && edgeBy === "filter";
-        if (edgeFiltered) {
-          const { field, bound } = incrementalFilter(
-            dataType,
-            edge.at,
-            query.tz,
-            dateStyle,
-          );
-          filters.push(
-            `${field} ${edge.order === "desc" ? "<=" : ">="} "${bound}"`,
-          );
-        }
-        if (filters.length > 0) params.set("filter", filters.join(" AND "));
-        if (pageToken) params.set("pageToken", pageToken);
+    do {
+      // A skip that had to start over from the newest page runs as the walk
+      // did before it kept its place: on the job's budget alone, its time
+      // given back to the slice once it crosses. Every other read stops on
+      // the slice's budget, and a skip also on its own shorter limit; it
+      // keeps its place below, so the next slice goes on from there.
+      const untimed = !crossed && restarted;
+      if (untimed ? stopRequested(false) : stopRequested(resumable)) {
+        stopped = true;
+        break;
+      }
+      if (!crossed && startPosition + pageCount >= maxSkip) {
+        getEvent()?.addWarning(
+          `google-health ${verb}: passed over ${startPosition + pageCount} pages without reaching the resume edge`,
+        );
+        stopped = true;
+        break;
+      }
+      if (
+        !crossed &&
+        !untimed &&
+        resumable &&
+        inSlice() &&
+        sliceNow() - skipStartedAt >= GOOGLE_HEALTH_SKIP_MS
+      ) {
+        stopped = true;
+        break;
+      }
+      const params = new URLSearchParams({ pageSize: String(pageSize) });
+      const filters: string[] = [];
+      if (query.start) {
+        const { field, bound } = incrementalFilter(
+          dataType,
+          query.start,
+          query.tz,
+          dateStyle,
+        );
+        filters.push(`${field} >= "${bound}"`);
+      }
+      const edgeFiltered = edge !== undefined && edgeBy === "filter";
+      if (edgeFiltered) {
+        const { field, bound } = incrementalFilter(
+          dataType,
+          edge.at,
+          query.tz,
+          dateStyle,
+        );
+        filters.push(
+          `${field} ${edge.order === "desc" ? "<=" : ">="} "${bound}"`,
+        );
+      }
+      if (filters.length > 0) params.set("filter", filters.join(" AND "));
+      if (pageToken) params.set("pageToken", pageToken);
 
-        requestCount += 1;
-        // v1.42 (#1023) — the first filtered daily-summary read answering 400
-        // is the expected trigger for the snake_case retry below, not a
-        // failure: some accounts reject the documented camelCase prefix on
-        // some types, and the very next request answers 200. It was logged as
-        // an error next to the 200 that followed it. It is recorded as a note
-        // now; any other 400 still carries its error. The same holds for the
-        // first resumed request refusing the edge bound.
-        const firstRequest = requestCount === 1;
-        const firstOfWalk = pageCount === 0;
-        const edgeRefused = (status: number, detail: string | undefined) =>
-          edgeFiltered &&
-          firstOfWalk &&
-          status === 400 &&
-          refusesEdgeFilter(detail);
-        const { status, json, verdict, apiErrorDetail } =
-          await googleHealthDataRequest<GoogleHealthDataPointPage>(
-            accessToken,
-            `${verb}(page=${pageCount})`,
-            (token) =>
-              safeFetch(
-                `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`,
-                {
-                  method: "GET",
-                  headers: { Authorization: `Bearer ${token}` },
-                },
-              ),
-            (status, detail) =>
-              canFallBack &&
-              dateStyle === "camel" &&
-              firstRequest &&
-              status === 400
-                ? "date_filter_rejected_retrying_snake_case"
-                : edgeRefused(status, detail)
-                  ? "edge_filter_rejected_reading_without_it"
+      requestCount += 1;
+      // v1.42 (#1023) — the first filtered daily-summary read answering 400
+      // is the expected trigger for the snake_case retry below, not a
+      // failure: some accounts reject the documented camelCase prefix on
+      // some types, and the very next request answers 200. It was logged as
+      // an error next to the 200 that followed it. It is recorded as a note
+      // now; any other 400 still carries its error. The same holds for the
+      // first resumed request refusing the edge bound, and for a saved
+      // place in the unbounded list Google no longer takes.
+      const firstRequest = requestCount === 1;
+      const firstOfWalk = pageCount === 0;
+      const edgeRefused = (status: number, detail: string | undefined) =>
+        edgeFiltered &&
+        firstOfWalk &&
+        status === 400 &&
+        refusesEdgeFilter(detail);
+      const resumeRefused = (status: number) =>
+        resumeFrom !== undefined && firstOfWalk && status === 400;
+      const requestToken = pageToken ?? undefined;
+      const { status, json, verdict, apiErrorDetail } =
+        await googleHealthDataRequest<GoogleHealthDataPointPage>(
+          accessToken,
+          `${verb}(page=${pageCount})`,
+          (token) =>
+            safeFetch(
+              `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`,
+              {
+                method: "GET",
+                headers: { Authorization: `Bearer ${token}` },
+              },
+            ),
+          (status, detail) =>
+            canFallBack &&
+            dateStyle === "camel" &&
+            firstRequest &&
+            status === 400
+              ? "date_filter_rejected_retrying_snake_case"
+              : edgeRefused(status, detail)
+                ? "edge_filter_rejected_reading_without_it"
+                : resumeRefused(status)
+                  ? "resume_token_rejected_reading_from_newest"
                   : undefined,
-          );
-        if (edgeRefused(status, apiErrorDetail)) {
-          throw new EdgeFilterRefused();
-        }
-        // A rate limit the request could not wait out ends a resumable walk
-        // inside a backfill slice where it is: the pages before it are written
-        // and their edge recorded, so this is a pause, not a failure. The slice
-        // ends with it, since every other walk would meet the same quota.
-        if (status === 429 && resumable && inSlice()) {
-          endSlice("rate_limited");
-          stopped = true;
-          break;
-        }
-        if (verdict.classification !== "success") {
-          throw new GoogleHealthApiError({
-            verb,
-            classification: verdict.classification,
-            httpStatus: verdict.httpStatus,
-            reason: verdict.reason,
-            upstreamError: apiErrorDetail,
-          });
-        }
+        );
+      if (edgeRefused(status, apiErrorDetail)) {
+        throw new EdgeFilterRefused();
+      }
+      if (resumeRefused(status)) throw new ResumePointLost();
+      // A rate limit the request could not wait out ends a resumable walk
+      // inside a backfill slice where it is: the pages before it are written
+      // and their edge recorded, so this is a pause, not a failure. The slice
+      // ends with it, since every other walk would meet the same quota.
+      if (status === 429 && resumable && inSlice()) {
+        endSlice("rate_limited");
+        stopped = true;
+        break;
+      }
+      if (verdict.classification !== "success") {
+        throw new GoogleHealthApiError({
+          verb,
+          classification: verdict.classification,
+          httpStatus: verdict.httpStatus,
+          reason: verdict.reason,
+          upstreamError: apiErrorDetail,
+        });
+      }
 
-        let page = json?.dataPoints ?? [];
-        pageToken = json?.nextPageToken ?? null;
-        pageCount += 1;
-        fetched += page.length;
-        noteSlicePage();
-        if (!crossed) {
-          // Placed samples beyond the edge were written by an earlier slice;
-          // one without a sample time cannot be placed and is passed over
-          // with them until the walk is past the edge.
-          const kept = page.filter((p) => beyondEdge(p) === false);
-          if (kept.length === 0) {
-            skipped += 1;
-            continue;
-          }
-          crossed = true;
-          refundSkip();
-          page = page.filter((p) => beyondEdge(p) !== true);
+      let page = json?.dataPoints ?? [];
+      // A saved place has to still hold a sample at or beyond the edge;
+      // past it, samples nobody wrote could lie behind the token.
+      if (resumeFrom && firstOfWalk && !page.some(reachesEdge)) {
+        throw new ResumePointLost();
+      }
+      if (requestToken !== undefined) {
+        lastToken = requestToken;
+        lastPosition = startPosition + pageCount;
+      }
+      pageToken = json?.nextPageToken ?? null;
+      pageCount += 1;
+      fetched += page.length;
+      noteSlicePage();
+      if (!crossed) {
+        // Placed samples beyond the edge were written by an earlier slice;
+        // one without a sample time cannot be placed and is passed over
+        // with them until the walk is past the edge.
+        const kept = page.filter((p) => beyondEdge(p) === false);
+        if (kept.length === 0) {
+          skipped += 1;
+          noteSliceSkip();
+          continue;
         }
-        await onPage(page);
-      } while (pageToken && pageCount - skipped < maxPages);
-    } finally {
-      refundSkip();
+        crossed = true;
+        if (restarted) extendSliceDeadline(sliceNow() - skipStartedAt);
+        page = page.filter((p) => beyondEdge(p) !== true);
+      }
+      await onPage(page);
+    } while (pageToken && pageCount - skipped < maxPages);
+
+    const truncated = stopped || Boolean(pageToken);
+    // Keep the place for the next slice. Only a walk without the bound has
+    // one, and only one that ended short needs it.
+    if (skipTo && truncated && lastToken !== undefined) {
+      await skipTo.onResumePoint?.({ token: lastToken, page: lastPosition });
     }
-
-    notePagination(pageCount, fetched, stopped || Boolean(pageToken));
+    notePagination(pageCount, fetched, truncated);
   };
 
   // The fallback only exists for filtered daily-summary reads — every other
@@ -1363,15 +1452,28 @@ export async function forEachDataPointPage(
 
   let style: GoogleHealthDateFilterStyle = "camel";
   let edgeBy: "filter" | "skip" = edge?.filter === false ? "skip" : "filter";
-  const run = async (dateStyle: GoogleHealthDateFilterStyle) => {
+  // From the saved place in the unbounded list when there is one; once from
+  // the newest page when that place is lost.
+  const walkSkipping = async (dateStyle: GoogleHealthDateFilterStyle) => {
+    const resumeFrom = edge?.resumeFrom;
+    if (!resumeFrom) return walk(dateStyle, "skip", undefined, false);
     try {
-      await walk(dateStyle, edgeBy);
+      await walk(dateStyle, "skip", resumeFrom, false);
+    } catch (err) {
+      if (!(err instanceof ResumePointLost)) throw err;
+      await walk(dateStyle, "skip", undefined, true);
+    }
+  };
+  const run = async (dateStyle: GoogleHealthDateFilterStyle) => {
+    if (edgeBy === "skip") return walkSkipping(dateStyle);
+    try {
+      await walk(dateStyle, edgeBy, undefined, false);
     } catch (err) {
       if (!(err instanceof EdgeFilterRefused)) throw err;
       // Once: the walk without the bound sends no edge filter to refuse.
       edgeBy = "skip";
       await edge?.onFilterUnsupported?.();
-      await walk(dateStyle, edgeBy);
+      await walk(dateStyle, edgeBy, undefined, false);
     }
   };
   try {
@@ -1403,6 +1505,18 @@ class EdgeFilterRefused extends Error {
   }
 }
 
+/**
+ * Thrown inside a walk that started from a saved place in the unbounded list
+ * (`GoogleHealthResumePoint`) when Google refused the token or the page it
+ * returned no longer reaches the edge; the walk starts over from the newest
+ * page.
+ */
+class ResumePointLost extends Error {
+  constructor() {
+    super("google-health: saved place in the list lost");
+  }
+}
+
 /**
  * Whether a 400 on the first resumed request is Google refusing the edge
  * bound, rather than something else about the request. The request differs

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

@@ -44,6 +44,7 @@ import {
   noteGoogleHealthOutcomeFailure,
   noteGoogleHealthWritten,
   refreshAccessToken,
+  type GoogleHealthResumePoint,
   type GoogleHealthTokenSource,
 } from "./client";
 import { getUserGoogleHealthCredentials } from "./credentials";
@@ -441,6 +442,14 @@ export interface GoogleHealthBackfillCursor {
   order: "desc" | "asc";
   /** Pages this collection has read across every attempt so far. */
   pages: number;
+  /**
+   * Only for a collection read without the edge bound: where its last walk
+   * stopped in the unbounded list (see `GoogleHealthResumePoint`), and the
+   * furthest position any walk of it has saved. A saved place counts as the
+   * slice's progress only when it is further than every one before it, so a
+   * chain of slices that keep passing over the same pages ends.
+   */
+  resume?: { token: string; page: number; reached: number };
 }
 
 /** What one collection came to in one backfill attempt. */
@@ -503,13 +512,25 @@ export function readGoogleHealthBackfillCursors(
     const c = progressRecord(raw);
     const at = typeof c.at === "string" ? c.at : null;
     if (!at || Number.isNaN(Date.parse(at))) continue;
+    const r = progressRecord(c.resume);
+    const count = (v: unknown): number =>
+      typeof v === "number" && Number.isFinite(v) && v > 0 ? Math.trunc(v) : 0;
+    const resume =
+      typeof r.token === "string" &&
+      r.token.length > 0 &&
+      r.token.length <= 4096 &&
+      count(r.page) > 0
+        ? {
+            token: r.token,
+            page: count(r.page),
+            reached: Math.max(count(r.reached), count(r.page)),
+          }
+        : undefined;
     out.set(key, {
       at: new Date(at).toISOString(),
       order: c.order === "asc" ? "asc" : "desc",
-      pages:
-        typeof c.pages === "number" && Number.isFinite(c.pages) && c.pages > 0
-          ? Math.trunc(c.pages)
-          : 0,
+      pages: count(c.pages),
+      ...(resume ? { resume } : {}),
     });
   }
   return out;
@@ -555,7 +576,12 @@ async function persistBackfillProgress(
         cursors: Object.fromEntries(
           [...next.cursors].map(([key, c]) => [
             key,
-            { at: c.at, order: c.order, pages: c.pages },
+            {
+              at: c.at,
+              order: c.order,
+              pages: c.pages,
+              ...(c.resume ? { resume: c.resume } : {}),
+            },
           ]),
         ),
         // Only once there is something to say, so a backfill that never met
@@ -626,24 +652,80 @@ export async function runWithGoogleHealthBackfillResume<T>(
  * where every walk reads its whole window.
  */
 export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): {
-  edge?: { at: Date; order: "desc" | "asc"; filter: boolean };
+  edge?: {
+    at: Date;
+    order: "desc" | "asc";
+    filter: boolean;
+    resumeFrom?: GoogleHealthResumePoint;
+  };
   pagesSoFar: number;
 } | null {
   const resume = backfillResumeStorage.getStore();
   if (!resume) return null;
   const cursor = resume.cursors.get(key);
+  const filter = !resume.edgeFilterUnsupported.has(key);
   return cursor
     ? {
         edge: {
           at: new Date(cursor.at),
           order: cursor.order,
-          filter: !resume.edgeFilterUnsupported.has(key),
+          filter,
+          ...(!filter && cursor.resume
+            ? {
+                resumeFrom: {
+                  token: cursor.resume.token,
+                  page: cursor.resume.page,
+                },
+              }
+            : {}),
         },
         pagesSoFar: cursor.pages,
       }
     : { pagesSoFar: 0 };
 }
 
+/**
+ * Record where a collection's walk without the edge bound stopped in the
+ * unbounded list, so the next slice goes on from there. A place further than
+ * every one saved before counts as this slice's progress (the backfill sends
+ * another slice on it), even when no page was written: a skip deeper than one
+ * slice is worked through over several. Outside a backfill, or for a
+ * collection without an edge, this does nothing.
+ */
+export async function recordGoogleHealthBackfillResumePoint(
+  key: GoogleHealthCollectionKey,
+  point: GoogleHealthResumePoint,
+): Promise<void> {
+  const resume = backfillResumeStorage.getStore();
+  if (!resume) return;
+  const current = resume.cursors.get(key);
+  if (!current) return;
+  const reached = current.resume?.reached ?? 0;
+  const further = point.page > reached;
+  const cursors = new Map(resume.cursors);
+  cursors.set(key, {
+    ...current,
+    resume: {
+      token: point.token,
+      page: point.page,
+      reached: Math.max(reached, point.page),
+    },
+  });
+  try {
+    await persistBackfillProgress(resume, { done: resume.done, cursors });
+    if (further && !resume.advanced.has(key)) {
+      resume.advanced.set(key, current.at);
+    }
+  } catch (err) {
+    getEvent()?.addWarning(
+      `google-health: backfill progress write failed for ${key}: ${err}`,
+    );
+    annotate({
+      meta: { "googleHealth.backfill.progressWriteFailed": key },
+    });
+  }
+}
+
 /**
  * Note that Google refused the edge filter on a resumed request of this
  * collection, so every later slice reads without it from the start instead
@@ -718,6 +800,8 @@ export async function recordGoogleHealthBackfillPage(
     at: moved ? far.toISOString() : current!.at,
     order,
     pages: (current?.pages ?? 0) + 1,
+    // The saved place stays until the walk records where it stopped.
+    ...(current?.resume ? { resume: current.resume } : {}),
   };
   const cursors = new Map(resume.cursors);
   cursors.set(key, nextCursor);

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

@@ -56,6 +56,7 @@ import {
   handleCollectionFetchError,
   noteGoogleHealthEdgeFilterUnsupported,
   recordGoogleHealthBackfillPage,
+  recordGoogleHealthBackfillResumePoint,
   runGoogleHealthCollection,
   upsertGoogleHealthMeasurements,
   type GoogleHealthMeasurementUpsert,
@@ -312,6 +313,15 @@ async function syncMetricCollection(
                 maxSkipPages: ceiling,
                 onFilterUnsupported: () =>
                   noteGoogleHealthEdgeFilterUnsupported(resource.verb),
+                // A place past a page whose write failed would leave that
+                // page's rows behind it, like a moved edge would.
+                onResumePoint: (point) =>
+                  edgeHeld
+                    ? undefined
+                    : recordGoogleHealthBackfillResumePoint(
+                        resource.verb,
+                        point,
+                      ),
               },
             }
           : {}),

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

@@ -281,9 +281,10 @@ describe("runGoogleHealthBackfillForUser — slices", () => {
 
   it("derives the chain's ceiling from the work a backfill can do", () => {
     // Every collection once, plus every page the resumable collections may
-    // ever read (heart rate's 10 000 and four others' 1 000), plus slack.
+    // ever write (heart rate's 10 000 and four others' 1 000) and as many
+    // again for a saved place in the list moving forward, plus slack.
     expect(GOOGLE_HEALTH_BACKFILL_MAX_SLICE).toBe(
-      17 + GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING + 2,
+      17 + 2 * GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING + 2,
     );
     expect(GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING).toBe(10_000 + 4 * 1000);
   });

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

@@ -75,19 +75,23 @@ export const GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS = 5 * 60;
  * backfill can do rather than guessed.
  *
  * A slice is continued only when it made progress that is on disk: it
- * finished a collection, or it moved a resumable collection's edge after
- * writing at least one page. Collections finish once each, and a resumable
- * collection's pages are capped across all attempts by its page ceiling
- * (`GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING`, the sum of them). So no chain of
- * progressing slices is longer than the collections plus those pages; the two
- * extra are slack for the first slice and the one that stamps the marker. In
+ * finished a collection, it moved a resumable collection's edge after
+ * writing at least one page, or a walk without the edge bound saved a place
+ * in the list further than any before it. Collections finish once each, a
+ * resumable collection's pages are capped across all attempts by its page
+ * ceiling (`GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING`, the sum of them), and a
+ * saved place only ever moves forward, through no more pages than the ones
+ * written plus the ones passed over, which the same ceiling bounds. So no
+ * chain of progressing slices is longer than the collections plus twice
+ * those pages; the two extra are slack for the first slice and the one that
+ * stamps the marker. In
  * practice a chain is a handful of slices per year of dense history. A slice
  * past this is counting progress it does not make, and fails rather than
  * queueing another.
  */
 export const GOOGLE_HEALTH_BACKFILL_MAX_SLICE =
   GOOGLE_HEALTH_COLLECTION_KEYS.length +
-  GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING +
+  2 * GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING +
   2;
 
 /** Test seam: a slice's budget, when not the production one. */

tests/integration/google-health-backfill-multi-year.test.ts

@@ -24,10 +24,12 @@
  *   - every slice writes a log event naming its requests and where it got to;
  *   - the heap does not grow with the history (run it under the production
  *     ceiling: `NODE_OPTIONS=--max-old-space-size=524`);
- *   - and, in a second run over two years, a Google that refuses the resume
+ *   - and, in a second run over four years, a Google that refuses the resume
  *     edge's upper bound (`<=`): the backfill falls back to reading without
- *     it, notes the refusal so it is met only once, and still imports every
- *     reading exactly once.
+ *     it, notes the refusal so it is met only once, keeps its place in the
+ *     unbounded list from slice to slice so the pages it reads stay linear in
+ *     the history (before, every slice passed over everything the slices
+ *     before it wrote), and still imports every reading exactly once.
  *
  * Google is replaced at the `safeFetch` seam by a generator that builds each
  * page on request from the filter and page token it is asked with, so the
@@ -433,9 +435,11 @@ describe("Google Health backfill over three years of heart rate (#1220)", () =>
   }, 3_600_000);
 
   it("completes when Google refuses the resume edge's upper bound, meeting the refusal once", async () => {
-    // Two years: the first slice reads without an edge, the second resumes
-    // with the bound and is refused, the third reads without it from the start.
-    const points = 2 * 365 * 24 * 60;
+    // Four years: the first slice reads without an edge, the second resumes
+    // with the bound and is refused, every later one reads without it. Long
+    // enough that passing over everything already written on every slice
+    // would read the history several times over.
+    const points = 4 * 365 * 24 * 60;
     await seed(points);
     fake.refuseUpperBound = true;
 
@@ -445,6 +449,8 @@ describe("Google Health backfill over three years of heart rate (#1220)", () =>
       await import("@/lib/jobs/reminder/google-health-sync");
     const { GOOGLE_HEALTH_TRANSIENT_RETRY } =
       await import("@/lib/google-health/client");
+    const { GOOGLE_HEALTH_BACKFILL_SLICE_PAGES } =
+      await import("@/lib/jobs/google-health-backfill");
     GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = 0;
 
     const prisma = getPrismaClient();
@@ -452,6 +458,7 @@ describe("Google Health backfill over three years of heart rate (#1220)", () =>
       kind: "google-health-backfill",
       data: { userId: USER, enqueuedAt: new Date().toISOString() },
     };
+    let hourlyRuns = 0;
     const slices: Array<{
       imported: number;
       continued: number;
@@ -495,6 +502,7 @@ describe("Google Health backfill over three years of heart rate (#1220)", () =>
           ok: true,
           did: { users_skipped: 0, users_failed: 0, users_parked: 0 },
         });
+        hourlyRuns += 1;
       }
     }
 
@@ -506,6 +514,22 @@ describe("Google Health backfill over three years of heart rate (#1220)", () =>
     expect(connection.backfillProgress).toBeNull();
     expect(payload).toBeNull();
     expect(slices.slice(0, -1).every((s) => s.continued === 1)).toBe(true);
+    // The hourly sync ran between every two slices.
+    expect(slices.length).toBeGreaterThan(4);
+    expect(hourlyRuns).toBe(slices.length - 1);
+
+    // The pages read stay linear in the history: every page once, the one
+    // pass over what the first slice wrote when the refusal was met, and a
+    // page or two per slice where it picks up again. Passing over everything
+    // already written on every slice read more than three times as many.
+    const historyPages = Math.ceil(points / 1000);
+    report(
+      `refusing run: backfill requests=${fake.backfillRequests} ` +
+        `history pages=${historyPages} slices=${slices.length}`,
+    );
+    expect(fake.backfillRequests).toBeLessThanOrEqual(
+      historyPages + GOOGLE_HEALTH_BACKFILL_SLICE_PAGES + 2 * slices.length,
+    );
 
     // The bound was asked for once, refused once, and never again: the
     // refusal was noted on the slice that met it and honoured after.
@@ -514,10 +538,11 @@ describe("Google Health backfill over three years of heart rate (#1220)", () =>
     expect(slices.filter((s) => s.refusalNoted !== undefined)).toEqual([
       expect.objectContaining({ refusalNoted: ["fetchHeartRate"] }),
     ]);
-    // More than one slice had to pass over what the ones before it wrote,
-    // and each still wrote its share.
+    // Only the slice that met the refusal passed over what the slice before
+    // it wrote, and it still wrote its share; every later one went on from
+    // where the last stopped.
     const skipping = slices.filter((s) => s.skippedPages > 0);
-    expect(skipping.length).toBeGreaterThan(1);
+    expect(skipping).toHaveLength(1);
     expect(skipping.every((s) => s.imported > 0)).toBe(true);
 
     // Every reading exactly once, none of them rewritten.