fix(integrations): report a page walk that stops short instead of ending quietly

The WHOOP, Oura and Withings list walks and the Fitbit exercise walk each
kept a page loop that, at its ceiling or on a cursor that did not advance,
returned what it had read and said nothing, and the backfill around it
stamped the history complete. `walkPages` keeps a ceiling but reports a
walk that stops with a page pending, or on a cursor it already followed,
as truncated. A backfill reads that through one verdict: a failure
retries, a truncated collection leaves the marker unset without a retry,
a clean run stamps it. The Fitbit exercise walk may read a hundred pages
during a backfill instead of twelve. A guard keeps page-counting loops out
of the provider directories, with Google Health and Strava listed until
their next change.

src/__tests__/page-walk-guard.test.ts

@@ -0,0 +1,84 @@
+/**
+ * Structural guard: a provider client walks a paged collection through
+ * `walkPages` (`src/lib/integrations/page-walk.ts`), never a loop of its own.
+ *
+ * Each hand-rolled loop ended at its page ceiling by returning what it had,
+ * silently, and the backfill around it stamped the history complete. A
+ * `while (pages < MAX)`, a `pageCount > N` break or a `do … while (next &&
+ * pageCount < max)` in a provider directory is that class coming back.
+ */
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import { describe, expect, it } from "vitest";
+
+import { walkSourceFiles } from "./helpers/source-files";
+
+const LIB = join(process.cwd(), "src", "lib");
+const PROVIDER_DIRS = [
+  "dawarich",
+  "fitbit",
+  "google-health",
+  "nightscout",
+  "oura",
+  "polar",
+  "strava",
+  "whoop",
+  "withings",
+];
+/**
+ * Loops still counting pages on their own, with the reason. Each is bounded
+ * and reports nothing at its bound; each moves to `walkPages` with the next
+ * change to its file.
+ */
+const PENDING: Readonly<Record<string, string>> = {
+  "google-health/client.ts":
+    "progress-bounded walk (backfill-bounds.ts); moves with the next Google Health change",
+  "strava/sync.ts":
+    "activity pages per sync; moves with the next Strava change",
+};
+
+/** A page counter compared against a bound, in any of the shapes used. */
+const PAGE_LOOP =
+  /\b(?:pages?|pageCount|pagesRead|page)\s*(?:<|<=|>|>=)\s*(?:[A-Z_][A-Z0-9_]*|maxPages|\d+)\b/;
+
+function strip(text: string): string {
+  return text
+    .replace(/\/\*[\s\S]*?\*\//g, "")
+    .replace(/(^|[^:])\/\/.*$/gm, "$1");
+}
+
+describe("provider clients walk pages through walkPages", () => {
+  const files = walkSourceFiles(LIB, { floor: 1000 }).filter(
+    (f) =>
+      PROVIDER_DIRS.some((d) => f.startsWith(`${d}/`)) &&
+      !f.includes("__tests__") &&
+      !f.endsWith(".test.ts"),
+  );
+  const offending = (f: string) =>
+    PAGE_LOOP.test(strip(readFileSync(join(LIB, f), "utf8")));
+
+  it("walks the provider directories", () => {
+    expect(files.length).toBeGreaterThanOrEqual(60);
+  });
+
+  it("finds no page loop of their own outside the pending list", () => {
+    expect(files.filter((f) => !(f in PENDING) && offending(f))).toEqual([]);
+  });
+
+  it("keeps the pending list honest", () => {
+    expect(Object.keys(PENDING).filter((f) => !offending(f))).toEqual([]);
+  });
+
+  it("catches the shapes the loops used", () => {
+    for (const planted of [
+      "while (pages < WORKOUT_MAX_PAGES) {",
+      "if (pageCount > 1000) break;",
+      "} while (nextToken && pageCount < maxPages);",
+    ]) {
+      expect(PAGE_LOOP.test(planted), planted).toBe(true);
+    }
+    expect(PAGE_LOOP.test("const walk = await walkPages(label, 12, f);")).toBe(
+      false,
+    );
+  });
+});

src/lib/fitbit/sync-workout.ts

@@ -13,8 +13,10 @@
  * collapse for a server-owned source pair.
  *
  * The page-walk is bounded (the 150 req/h budget is tight): a small page cap
- * covers the incremental window comfortably; a deep backfill is bounded by the
- * backfill horizon `afterDate`. A per-endpoint 403 soft-skips the resource.
+ * covers the incremental window comfortably, and a backfill, bounded by the
+ * horizon `afterDate`, may read up to ten thousand sessions. A walk that
+ * reaches its cap is reported truncated (`walkPages`), so the backfill is not
+ * stamped complete over it. A per-endpoint 403 soft-skips the resource.
  */
 import { fetchActivityList, mapWorkout, readActivityList } from "./client";
 import type { FitbitMappedWorkout } from "./client";
@@ -24,11 +26,14 @@ import { prisma, toJson } from "@/lib/db";
 import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit";
 import { annotate, getEvent } from "@/lib/logging/context";
 import { resolveUserTimezone } from "@/lib/tz/resolver";
+import { walkPages } from "@/lib/integrations/page-walk";
 
 /** Activities per page on the `activities/list` walk. */
 const WORKOUT_PAGE_SIZE = 100;
-/** Hard ceiling on pages walked per cycle (rate-budget guard). */
+/** Ceiling on pages walked per incremental cycle (rate-budget guard). */
 const WORKOUT_MAX_PAGES = 12;
+/** Ceiling on pages walked by a backfill. */
+const WORKOUT_BACKFILL_MAX_PAGES = 100;
 
 function metadataRecord(value: unknown): Record<string, unknown> {
   return typeof value === "object" && value !== null && !Array.isArray(value)
@@ -36,6 +41,32 @@ function metadataRecord(value: unknown): Record<string, unknown> {
     : {};
 }
 
+function walkActivityList(
+  accessToken: string,
+  afterDate: Date,
+  maxPages: number,
+) {
+  return walkPages<number, ReturnType<typeof readActivityList>[number]>(
+    "fitbit.exercise",
+    maxPages,
+    async (cursor) => {
+      const offset = cursor ?? 0;
+      const page = readActivityList(
+        await fetchActivityList(accessToken, afterDate, {
+          limit: WORKOUT_PAGE_SIZE,
+          offset,
+        }),
+      );
+      return {
+        items: page,
+        next:
+          page.length < WORKOUT_PAGE_SIZE ? null : offset + WORKOUT_PAGE_SIZE,
+      };
+    },
+    0,
+  );
+}
+
 export async function syncUserWorkout(
   userId: string,
   opts: FitbitResourceSyncOptions = {},
@@ -50,28 +81,19 @@ export async function syncUserWorkout(
   const afterDate =
     opts.start ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
 
-  const entries: ReturnType<typeof readActivityList> = [];
-  let offset = 0;
-  let pages = 0;
-  while (pages < WORKOUT_MAX_PAGES) {
-    let body: unknown;
-    try {
-      body = await fetchActivityList(tokenInfo.accessToken, afterDate, {
-        limit: WORKOUT_PAGE_SIZE,
-        offset,
-      });
-    } catch (err) {
-      return handleCollectionFetchError("fetchExercise", userId, err);
-    }
-    const page = readActivityList(body);
-    entries.push(...page);
-    pages += 1;
-    if (page.length < WORKOUT_PAGE_SIZE) break;
-    offset += WORKOUT_PAGE_SIZE;
+  let walk: Awaited<ReturnType<typeof walkActivityList>>;
+  try {
+    walk = await walkActivityList(
+      tokenInfo.accessToken,
+      afterDate,
+      opts.fullSync ? WORKOUT_BACKFILL_MAX_PAGES : WORKOUT_MAX_PAGES,
+    );
+  } catch (err) {
+    return handleCollectionFetchError("fetchExercise", userId, err);
   }
 
   let imported = 0;
-  for (const entry of entries) {
+  for (const entry of walk.items) {
     const w: FitbitMappedWorkout | null = mapWorkout(entry, tz);
     if (!w) continue; // no usable time span — not a workout
 

src/lib/integrations/__tests__/page-walk.test.ts

@@ -0,0 +1,66 @@
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
+
+import { backfillVerdict, walkPages, withPageWalkLedger } from "../page-walk";
+
+describe("walkPages", () => {
+  it("reads until a page has no next", async () => {
+    const walk = await walkPages<number, string>(
+      "test.done",
+      10,
+      async (cursor) => ({
+        items: [`p${cursor ?? 0}`],
+        next: (cursor ?? 0) < 2 ? (cursor ?? 0) + 1 : null,
+      }),
+    );
+    expect(walk).toEqual({
+      items: ["p0", "p1", "p2"],
+      pages: 3,
+      truncated: false,
+    });
+  });
+
+  it("reports a walk that reaches its ceiling with a page pending", async () => {
+    const walk = await walkPages<number, number>(
+      "test.ceiling",
+      4,
+      async (cursor) => ({ items: [cursor ?? 0], next: (cursor ?? 0) + 1 }),
+    );
+    expect(walk.truncated).toBe(true);
+    expect(walk.pages).toBe(4);
+    expect(walk.items).toEqual([0, 1, 2, 3]);
+  });
+
+  it("stops on a cursor it already followed, the first one included", async () => {
+    const fetchPage = vi.fn(async () => ({ items: [1], next: 0 }));
+    const walk = await walkPages<number, number>(
+      "test.repeat",
+      50,
+      fetchPage,
+      0,
+    );
+    expect(walk.truncated).toBe(true);
+    expect(fetchPage).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe("withPageWalkLedger and backfillVerdict", () => {
+  it("collects the truncated walks of a run", async () => {
+    const { result, truncated } = await withPageWalkLedger(async () => {
+      await walkPages("b.walk", 1, async () => ({ items: [], next: "x" }));
+      await walkPages("a.walk", 5, async () => ({ items: [], next: null }));
+      return "done";
+    });
+    expect(result).toBe("done");
+    expect(truncated).toEqual(["b.walk"]);
+  });
+
+  it("retries a failure, leaves a truncated history incomplete, stamps a clean one", () => {
+    expect(backfillVerdict({ failed: true, truncated: ["a"] })).toBe("retry");
+    expect(backfillVerdict({ failed: false, truncated: ["a"] })).toBe(
+      "incomplete",
+    );
+    expect(backfillVerdict({ failed: false, truncated: [] })).toBe("complete");
+  });
+});

src/lib/integrations/page-walk.ts

@@ -0,0 +1,112 @@
+/**
+ * The one way a provider client walks a paged collection, and the contract a
+ * history backfill reads off it.
+ *
+ * Every client used to keep its own `while (pages < MAX)` loop, and each one
+ * ended the same way at its ceiling: it returned what it had read, said
+ * nothing, and the backfill around it stamped the connection complete. A
+ * history longer than the ceiling, or a cursor that never advanced, was cut
+ * off for good without a trace. `walkPages` keeps the ceiling, because a
+ * runaway cursor must stop somewhere, but a walk that stops with a page still
+ * pending, or on a cursor it has already followed, reports itself
+ * `truncated`: on the run's wide event, and in the ledger a backfill opens
+ * with `withPageWalkLedger`.
+ *
+ * The backfill contract (`backfillVerdict`), the same for every provider:
+ *   - a failed collection is transient: the job throws and pg-boss retries;
+ *   - a truncated collection is not complete: the marker is not stamped, and
+ *     boot discovery offers the account again, without a retry storm;
+ *   - a connection parked for re-authorisation is permanent until the person
+ *     reconnects: the runner returns without stamping and without a retry.
+ *
+ * Google Health keeps its own progress-based walk (`google-health/
+ * backfill-bounds.ts`), whose `no_progress` end is this module's runaway case
+ * measured in time rather than in cursors; it moves onto this contract with
+ * its next change. `src/__tests__/page-walk-guard.test.ts` keeps hand-rolled
+ * page loops out of the provider directories.
+ */
+import { AsyncLocalStorage } from "node:async_hooks";
+
+import { annotate } from "@/lib/logging/context";
+
+/** One page as the client read it. A null `next` ends the walk. */
+export interface WalkPage<C, T> {
+  items: readonly T[];
+  next: C | null | undefined;
+}
+
+/** What the walk came to. */
+export interface PageWalk<T> {
+  items: T[];
+  pages: number;
+  /** Stopped at the ceiling or on a repeated cursor with history left. */
+  truncated: boolean;
+}
+
+const ledger = new AsyncLocalStorage<Set<string>>();
+
+/**
+ * Walk `fetchPage` from `first` (or no cursor) until a page has no `next`,
+ * at most `maxPages` pages. `label` names the collection (`whoop.recovery`)
+ * on the wide event and in the ledger.
+ */
+export async function walkPages<C, T>(
+  label: string,
+  maxPages: number,
+  fetchPage: (cursor: C | undefined, page: number) => Promise<WalkPage<C, T>>,
+  first?: C,
+): Promise<PageWalk<T>> {
+  const items: T[] = [];
+  const seen = new Set<C>(first === undefined ? [] : [first]);
+  let cursor = first;
+  for (let page = 0; page < maxPages; page += 1) {
+    const result = await fetchPage(cursor, page);
+    items.push(...result.items);
+    const next = result.next;
+    if (next === null || next === undefined) {
+      return { items, pages: page + 1, truncated: false };
+    }
+    if (seen.has(next)) return truncated(label, items, page + 1, "repeat");
+    seen.add(next);
+    cursor = next;
+  }
+  return truncated(label, items, maxPages, "ceiling");
+}
+
+function truncated<T>(
+  label: string,
+  items: T[],
+  pages: number,
+  reason: "ceiling" | "repeat",
+): PageWalk<T> {
+  annotate({
+    action: { name: "integration.page_walk.truncated" },
+    meta: { walk: label, pages, reason },
+  });
+  ledger.getStore()?.add(label);
+  return { items, pages, truncated: true };
+}
+
+/**
+ * Run `run` with a ledger of the walks inside it that ended truncated. A
+ * backfill wraps its full sync in this and reads the ledger before it
+ * stamps its marker.
+ */
+export async function withPageWalkLedger<R>(
+  run: () => Promise<R>,
+): Promise<{ result: R; truncated: string[] }> {
+  const walks = new Set<string>();
+  const result = await ledger.run(walks, run);
+  return { result, truncated: [...walks].sort() };
+}
+
+/** What a backfill does with its full sync's outcome. */
+export type BackfillVerdict = "complete" | "retry" | "incomplete";
+
+export function backfillVerdict(outcome: {
+  failed: boolean;
+  truncated: readonly string[];
+}): BackfillVerdict {
+  if (outcome.failed) return "retry";
+  return outcome.truncated.length > 0 ? "incomplete" : "complete";
+}

src/lib/jobs/__tests__/fitbit-backfill.test.ts

@@ -125,3 +125,24 @@ describe("runFitbitBackfillForUser — verdict-gated marker", () => {
     expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
   });
 });
+
+describe("runFitbitBackfillForUser — a walk that reached its ceiling", () => {
+  it("leaves the marker unset and does not throw", async () => {
+    const { walkPages } = await import("@/lib/integrations/page-walk");
+    syncUserFitbit.mockImplementation(async () => {
+      // A history longer than the walk may read: every page has a next.
+      await walkPages<number, number>(
+        "fitbit.exercise",
+        3,
+        async (cursor) => ({ items: [1], next: (cursor ?? 0) + 1 }),
+        0,
+      );
+      return { imported: 3, failed: false };
+    });
+
+    await expect(runFitbitBackfillForUser("u1")).resolves.toEqual({
+      imported: 3,
+    });
+    expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
+  });
+});

src/lib/jobs/fitbit-backfill.ts

@@ -15,6 +15,10 @@ import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
 import { isReauthRequired } from "@/lib/integrations/status";
+import {
+  backfillVerdict,
+  withPageWalkLedger,
+} from "@/lib/integrations/page-walk";
 import { syncUserFitbit } from "@/lib/fitbit/sync";
 
 export const FITBIT_BACKFILL_QUEUE = "fitbit-backfill";
@@ -39,14 +43,17 @@ export interface FitbitBackfillPayload {
  * `runWhoopBackfillForUser` (whose identical unconditional-stamp bug is fixed
  * separately — this is the template for that port).
  *
- * Verdict-gated: a partial hard failure (one collection 400ing, a write
- * failing, a dead token) THROWS so pg-boss retries the job — stamping the
- * marker over a partial walk would freeze the history gap in forever and the
- * configured `retryLimit` would be dead code. A connection parked at
- * `error_reauth` is NOT clean either, but throwing against a dead grant would
- * 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.
+ * Verdict-gated (`backfillVerdict`, `integrations/page-walk.ts`): a partial
+ * hard failure (one collection 400ing, a write failing, a dead token) THROWS
+ * so pg-boss retries the job — stamping the marker over a partial walk would
+ * freeze the history gap in forever and the configured `retryLimit` would be
+ * dead code. A collection whose walk reached its page ceiling leaves the
+ * marker unset WITHOUT a retry: nothing failed, and boot discovery offers the
+ * account again. A connection parked at `error_reauth` is NOT clean either,
+ * but throwing against a dead grant would 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.
  */
 export async function runFitbitBackfillForUser(
   userId: string,
@@ -61,9 +68,28 @@ export async function runFitbitBackfillForUser(
     return { imported: 0 };
   }
 
-  const { imported, failed } = await syncUserFitbit(userId, { fullSync: true });
+  const {
+    result: { imported, failed },
+    truncated,
+  } = await withPageWalkLedger(() =>
+    syncUserFitbit(userId, { fullSync: true }),
+  );
+  const verdict = backfillVerdict({ failed, truncated });
 
-  if (failed) {
+  if (verdict === "incomplete") {
+    // A collection longer than its walk may read in one run: the history is
+    // not complete, but nothing failed, so no retry. Boot discovery offers
+    // the account again while the marker stays unset.
+    annotate({
+      action: {
+        name: "fitbit.backfill.truncated",
+        details: { imported, collections: truncated.join(",") },
+      },
+    });
+    return { imported };
+  }
+
+  if (verdict === "retry") {
     // Surface through the pg-boss retry path (retryLimit 3, backoff, set by
     // `integrationBackfillSourceOptions`). If the failure parked the connection
     // at error_reauth meanwhile, the next attempt takes the reauth return above

src/lib/jobs/whoop-backfill.ts

@@ -21,6 +21,10 @@ import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
 import { isReauthRequired } from "@/lib/integrations/status";
+import {
+  backfillVerdict,
+  withPageWalkLedger,
+} from "@/lib/integrations/page-walk";
 import { syncUserWhoop } from "@/lib/whoop/sync";
 
 export const WHOOP_BACKFILL_QUEUE = "whoop-backfill";
@@ -44,16 +48,19 @@ export interface WhoopBackfillPayload {
  * reboot mid-walk) overwrites rather than duplicating. Mirrors the landed
  * `runFitbitBackfillForUser`.
  *
- * Verdict-gated: a partial hard failure (a dead token, a write failing, a
- * fetch hard-fail) THROWS so pg-boss retries the job — stamping the marker over
- * a partial multi-year walk would freeze the history gap forever and the
- * configured `retryLimit` would be dead code. A connection parked at
- * `error_reauth` is NOT clean either, but throwing against a dead grant would
- * just burn the retry budget on the same no-op: return WITHOUT stamping — boot
- * discovery re-offers the account after the user reconnects and a clean walk
- * completes. `syncUserWhoop` has its own reauth short-circuit (returns
- * `failed: true`), but the guard must still live here so a parked account
- * RETURNS instead of throwing through three retries.
+ * Verdict-gated (`backfillVerdict`, `integrations/page-walk.ts`): a partial
+ * hard failure (a dead token, a write failing, a fetch hard-fail) THROWS so
+ * pg-boss retries the job — stamping the marker over a partial multi-year
+ * walk would freeze the history gap forever and the configured `retryLimit`
+ * would be dead code. A collection whose walk reached its page ceiling leaves
+ * the marker unset WITHOUT a retry: nothing failed, and boot discovery offers
+ * the account again. A connection parked at `error_reauth` is NOT clean
+ * either, but throwing against a dead grant would just burn the retry budget
+ * on the same no-op: return WITHOUT stamping — boot discovery re-offers the
+ * account after the user reconnects and a clean walk completes.
+ * `syncUserWhoop` has its own reauth short-circuit (returns `failed: true`),
+ * but the guard must still live here so a parked account RETURNS instead of
+ * throwing through three retries.
  */
 export async function runWhoopBackfillForUser(
   userId: string,
@@ -68,9 +75,26 @@ export async function runWhoopBackfillForUser(
     return { imported: 0 };
   }
 
-  const { imported, failed } = await syncUserWhoop(userId, { fullSync: true });
+  const {
+    result: { imported, failed },
+    truncated,
+  } = await withPageWalkLedger(() => syncUserWhoop(userId, { fullSync: true }));
+  const verdict = backfillVerdict({ failed, truncated });
 
-  if (failed) {
+  if (verdict === "incomplete") {
+    // A collection longer than its walk may read in one run: the history is
+    // not complete, but nothing failed, so no retry. Boot discovery offers
+    // the account again while the marker stays unset.
+    annotate({
+      action: {
+        name: "whoop.backfill.truncated",
+        details: { imported, collections: truncated.join(",") },
+      },
+    });
+    return { imported };
+  }
+
+  if (verdict === "retry") {
     // Surface through the pg-boss retry path (retryLimit + backoff set by
     // `integrationBackfillSourceOptions`). If the failure parked the connection
     // at error_reauth meanwhile, the next attempt takes the reauth return above.

src/lib/oura/client.ts

@@ -22,6 +22,7 @@ import { safeFetch } from "@/lib/safe-fetch";
 import { OuraApiError, classifyOuraResponse } from "./response-classifier";
 import { envOr, envValue } from "@/lib/env";
 import { importValue } from "@/lib/integrations/import-value";
+import { walkPages } from "@/lib/integrations/page-walk";
 
 const OURA_API_BASE = "https://api.ouraring.com";
 const OURA_OAUTH_AUTH_URL = "https://cloud.ouraring.com/oauth/authorize";
@@ -476,51 +477,48 @@ async function fetchCollection<T>(
   verb: string,
   query: DateRangeQuery,
 ): Promise<T[]> {
-  const records: T[] = [];
-  let nextToken: string | null | undefined;
-  let pageCount = 0;
-  const maxPages = query.maxPages ?? 100;
-
-  do {
-    const params = new URLSearchParams({
-      start_date: query.startDate,
-      end_date: query.endDate,
-    });
-    if (nextToken) params.set("next_token", nextToken);
-
-    const pageStart = performance.now();
-    const res = await safeFetch(`${OURA_API_BASE}${path}?${params}`, {
-      method: "GET",
-      headers: {
-        Authorization: `Bearer ${accessToken}`,
-        Accept: "application/json",
-      },
-    });
-    const json = (await res
-      .json()
-      .catch(() => null)) as OuraCollection<T> | null;
-    const verdict = classifyOuraResponse(res.status);
-    getEvent()?.addExternalCall({
-      service: "oura",
-      method: `${verb}(page=${pageCount})`,
-      duration_ms: Math.round(performance.now() - pageStart),
-      status: res.status,
-      error: verdict.classification === "success" ? undefined : verdict.reason,
-    });
-    if (verdict.classification !== "success") {
-      throw new OuraApiError({
-        verb,
-        classification: verdict.classification,
-        httpStatus: verdict.httpStatus,
-        reason: verdict.reason,
+  const walk = await walkPages<string, T>(
+    `oura.${verb}`,
+    query.maxPages ?? 100,
+    async (nextToken, page) => {
+      const params = new URLSearchParams({
+        start_date: query.startDate,
+        end_date: query.endDate,
       });
-    }
-    for (const r of json?.data ?? []) records.push(r);
-    nextToken = json?.next_token ?? null;
-    pageCount += 1;
-  } while (nextToken && pageCount < maxPages);
+      if (nextToken) params.set("next_token", nextToken);
 
-  return records;
+      const pageStart = performance.now();
+      const res = await safeFetch(`${OURA_API_BASE}${path}?${params}`, {
+        method: "GET",
+        headers: {
+          Authorization: `Bearer ${accessToken}`,
+          Accept: "application/json",
+        },
+      });
+      const json = (await res
+        .json()
+        .catch(() => null)) as OuraCollection<T> | null;
+      const verdict = classifyOuraResponse(res.status);
+      getEvent()?.addExternalCall({
+        service: "oura",
+        method: `${verb}(page=${page})`,
+        duration_ms: Math.round(performance.now() - pageStart),
+        status: res.status,
+        error:
+          verdict.classification === "success" ? undefined : verdict.reason,
+      });
+      if (verdict.classification !== "success") {
+        throw new OuraApiError({
+          verb,
+          classification: verdict.classification,
+          httpStatus: verdict.httpStatus,
+          reason: verdict.reason,
+        });
+      }
+      return { items: json?.data ?? [], next: json?.next_token ?? null };
+    },
+  );
+  return walk.items;
 }
 
 export function fetchSleep(

src/lib/whoop/client.ts

@@ -21,6 +21,7 @@ import { reconstructContiguousSleepTimeline } from "@/lib/sleep/reconstruct-time
 import { WhoopApiError, classifyWhoopResponse } from "./response-classifier";
 import { envOr, envValue } from "@/lib/env";
 import { importValue } from "@/lib/integrations/import-value";
+import { walkPages } from "@/lib/integrations/page-walk";
 
 const WHOOP_API_BASE = "https://api.prod.whoop.com/developer";
 const WHOOP_OAUTH_AUTH_URL = "https://api.prod.whoop.com/oauth/oauth2/auth";
@@ -269,7 +270,10 @@ export interface WhoopProfile {
 interface CollectionQuery {
   start?: Date;
   end?: Date;
-  /** Hard ceiling on pages walked (defence against a runaway cursor). */
+  /**
+   * Ceiling on pages walked (defence against a runaway cursor). A walk that
+   * reaches it is reported truncated (`walkPages`).
+   */
   maxPages?: number;
 }
 
@@ -279,49 +283,45 @@ async function fetchCollection<T>(
   verb: string,
   query: CollectionQuery = {},
 ): Promise<T[]> {
-  const records: T[] = [];
-  let nextToken: string | null | undefined;
-  let pageCount = 0;
-  const maxPages = query.maxPages ?? 1000;
+  const walk = await walkPages<string, T>(
+    `whoop.${verb}`,
+    query.maxPages ?? 1000,
+    async (nextToken, page) => {
+      const params = new URLSearchParams({ limit: String(WHOOP_PAGE_LIMIT) });
+      if (query.start) params.set("start", query.start.toISOString());
+      if (query.end) params.set("end", query.end.toISOString());
+      if (nextToken) params.set("nextToken", nextToken);
 
-  do {
-    const params = new URLSearchParams({ limit: String(WHOOP_PAGE_LIMIT) });
-    if (query.start) params.set("start", query.start.toISOString());
-    if (query.end) params.set("end", query.end.toISOString());
-    if (nextToken) params.set("nextToken", nextToken);
-
-    const pageStart = performance.now();
-    const res = await safeFetch(`${WHOOP_API_BASE}${path}?${params}`, {
-      method: "GET",
-      headers: { Authorization: `Bearer ${accessToken}` },
-    });
-
-    const json = (await res
-      .json()
-      .catch(() => null)) as WhoopCollection<T> | null;
-    const verdict = classifyWhoopResponse(res.status);
-    getEvent()?.addExternalCall({
-      service: "whoop",
-      method: `${verb}(page=${pageCount})`,
-      duration_ms: Math.round(performance.now() - pageStart),
-      status: res.status,
-      error: verdict.classification === "success" ? undefined : verdict.reason,
-    });
-    if (verdict.classification !== "success") {
-      throw new WhoopApiError({
-        verb,
-        classification: verdict.classification,
-        httpStatus: verdict.httpStatus,
-        reason: verdict.reason,
+      const pageStart = performance.now();
+      const res = await safeFetch(`${WHOOP_API_BASE}${path}?${params}`, {
+        method: "GET",
+        headers: { Authorization: `Bearer ${accessToken}` },
       });
-    }
 
-    for (const r of json?.records ?? []) records.push(r);
-    nextToken = json?.next_token ?? null;
-    pageCount += 1;
-  } while (nextToken && pageCount < maxPages);
-
-  return records;
+      const json = (await res
+        .json()
+        .catch(() => null)) as WhoopCollection<T> | null;
+      const verdict = classifyWhoopResponse(res.status);
+      getEvent()?.addExternalCall({
+        service: "whoop",
+        method: `${verb}(page=${page})`,
+        duration_ms: Math.round(performance.now() - pageStart),
+        status: res.status,
+        error:
+          verdict.classification === "success" ? undefined : verdict.reason,
+      });
+      if (verdict.classification !== "success") {
+        throw new WhoopApiError({
+          verb,
+          classification: verdict.classification,
+          httpStatus: verdict.httpStatus,
+          reason: verdict.reason,
+        });
+      }
+      return { items: json?.records ?? [], next: json?.next_token ?? null };
+    },
+  );
+  return walk.items;
 }
 
 async function fetchSingle<T>(

src/lib/withings/client.ts

@@ -10,6 +10,7 @@ import {
 } from "./response-classifier";
 import { envOr, envValue } from "@/lib/env";
 import { importValue } from "@/lib/integrations/import-value";
+import { walkPages } from "@/lib/integrations/page-walk";
 
 const WITHINGS_OAUTH_URL =
   "https://account.withings.com/oauth2_user/authorize2";
@@ -249,6 +250,24 @@ export interface WithingsMeasureGroup {
 export const WITHINGS_MEASTYPES: readonly number[] =
   Object.keys(MEASURE_TYPE_MAP).map(Number);
 
+/** Ceiling on the pages one Withings list walk reads (`walkPages`). */
+export const WITHINGS_MAX_PAGES = 1000;
+
+/**
+ * The offset of the next page of a Withings list body, or null when it has
+ * no more. A body that says `more` without an offset past the current one
+ * hands back the current offset, which `walkPages` reports as a repeat
+ * instead of stopping quietly.
+ */
+export function withingsNextOffset(
+  body: { more?: unknown; offset?: unknown },
+  offset: number,
+): number | null {
+  if (body.more !== true && body.more !== 1) return null;
+  const next = Number(body.offset);
+  return Number.isFinite(next) && next > offset ? next : offset;
+}
+
 /**
  * Fetch measurements from Withings API.
  *
@@ -273,77 +292,69 @@ export async function fetchMeasurements(
     baseParams.enddate = String(Math.floor(endDate.getTime() / 1000));
   }
 
-  const results: WithingsMeasure[] = [];
-  let offset = 0;
-  let pageCount = 0;
-
-  while (true) {
-    const params = new URLSearchParams({
-      ...baseParams,
-      offset: String(offset),
-    });
-
-    const pageStart = performance.now();
-    const res = await safeFetch(WITHINGS_MEASURE_URL, {
-      method: "POST",
-      headers: {
-        "Content-Type": "application/x-www-form-urlencoded",
-        Authorization: `Bearer ${accessToken}`,
-      },
-      body: params.toString(),
-    });
-
-    const json = await res.json();
-    const verdict = classifyWithingsResponse(res.status, json);
-    getEvent()?.addExternalCall({
-      service: "withings",
-      method: `fetchMeasurements(page=${pageCount})`,
-      duration_ms: Math.round(performance.now() - pageStart),
-      status: res.status,
-      error: verdict.classification === "success" ? undefined : verdict.reason,
-    });
-    if (verdict.classification !== "success") {
-      throw new WithingsApiError({
-        verb: "measure",
-        classification: verdict.classification,
-        withingsStatus: verdict.withingsStatus,
-        reason: verdict.reason,
-        upstreamError: typeof json?.error === "string" ? json.error : undefined,
+  const walk = await walkPages<number, WithingsMeasure>(
+    "withings.measure",
+    WITHINGS_MAX_PAGES,
+    async (cursor, page) => {
+      const offset = cursor ?? 0;
+      const params = new URLSearchParams({
+        ...baseParams,
+        offset: String(offset),
       });
-    }
 
-    const body = json.body ?? {};
-    const groups: WithingsMeasureGroup[] = body.measuregrps ?? [];
+      const pageStart = performance.now();
+      const res = await safeFetch(WITHINGS_MEASURE_URL, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/x-www-form-urlencoded",
+          Authorization: `Bearer ${accessToken}`,
+        },
+        body: params.toString(),
+      });
 
-    for (const group of groups) {
-      const measuredAt = new Date(group.date * 1000);
-
-      for (const m of group.measures) {
-        const mapping = MEASURE_TYPE_MAP[m.type];
-        if (!mapping) continue;
-
-        // Withings stores value * 10^unit (unit is usually negative)
-        const value = m.value * Math.pow(10, m.unit);
-
-        results.push({
-          type: mapping.type,
-          value: importValue(value),
-          measuredAt,
+      const json = await res.json();
+      const verdict = classifyWithingsResponse(res.status, json);
+      getEvent()?.addExternalCall({
+        service: "withings",
+        method: `fetchMeasurements(page=${page})`,
+        duration_ms: Math.round(performance.now() - pageStart),
+        status: res.status,
+        error:
+          verdict.classification === "success" ? undefined : verdict.reason,
+      });
+      if (verdict.classification !== "success") {
+        throw new WithingsApiError({
+          verb: "measure",
+          classification: verdict.classification,
+          withingsStatus: verdict.withingsStatus,
+          reason: verdict.reason,
+          upstreamError:
+            typeof json?.error === "string" ? json.error : undefined,
         });
       }
-    }
 
-    const hasMore = body.more === true || body.more === 1;
-    if (!hasMore) break;
-
-    const nextOffset = Number(body.offset);
-    if (!Number.isFinite(nextOffset) || nextOffset <= offset) break;
-    offset = nextOffset;
-    pageCount += 1;
-    if (pageCount > 1000) break;
-  }
-
-  return results;
+      const body = json.body ?? {};
+      const groups: WithingsMeasureGroup[] = body.measuregrps ?? [];
+      const items: WithingsMeasure[] = [];
+      for (const group of groups) {
+        const measuredAt = new Date(group.date * 1000);
+        for (const m of group.measures) {
+          const mapping = MEASURE_TYPE_MAP[m.type];
+          if (!mapping) continue;
+          // Withings stores value * 10^unit (unit is usually negative)
+          const value = m.value * Math.pow(10, m.unit);
+          items.push({
+            type: mapping.type,
+            value: importValue(value),
+            measuredAt,
+          });
+        }
+      }
+      return { items, next: withingsNextOffset(body, offset) };
+    },
+    0,
+  );
+  return walk.items;
 }
 
 /**

src/lib/withings/sync-activity.ts

@@ -49,7 +49,11 @@ import {
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 
-import { hasActivityScope } from "./client";
+import {
+  hasActivityScope,
+  WITHINGS_MAX_PAGES,
+  withingsNextOffset,
+} from "./client";
 import {
   WithingsApiError,
   classifyWithingsResponse,
@@ -66,6 +70,7 @@ import {
   recordSyncFailure,
   recordSyncSuccess,
 } from "@/lib/integrations/status";
+import { walkPages } from "@/lib/integrations/page-walk";
 
 const WITHINGS_MEASURE_URL = "https://wbsapi.withings.net/v2/measure";
 
@@ -145,57 +150,51 @@ export async function fetchWithingsActivity(
     data_fields: "steps,distance,calories",
   };
 
-  const results: WithingsActivityEntry[] = [];
-  let offset = 0;
-  let pageCount = 0;
-  while (true) {
-    const params = new URLSearchParams({
-      ...baseParams,
-      offset: String(offset),
-    });
-    const pageStart = performance.now();
-    const res = await safeFetch(WITHINGS_MEASURE_URL, {
-      method: "POST",
-      headers: {
-        "Content-Type": "application/x-www-form-urlencoded",
-        Authorization: `Bearer ${accessToken}`,
-      },
-      body: params.toString(),
-    });
-    const json = await res.json();
-    const verdict = classifyWithingsResponse(res.status, json);
-    getEvent()?.addExternalCall({
-      service: "withings",
-      method: `fetchWithingsActivity(page=${pageCount})`,
-      duration_ms: Math.round(performance.now() - pageStart),
-      status: res.status,
-      error: verdict.classification === "success" ? undefined : verdict.reason,
-    });
-    if (verdict.classification !== "success") {
-      throw new WithingsApiError({
-        verb: "activity",
-        classification: verdict.classification,
-        withingsStatus: verdict.withingsStatus,
-        reason: verdict.reason,
-        upstreamError: typeof json?.error === "string" ? json.error : undefined,
+  const walk = await walkPages<number, WithingsActivityEntry>(
+    "withings.activity",
+    WITHINGS_MAX_PAGES,
+    async (cursor, page) => {
+      const offset = cursor ?? 0;
+      const params = new URLSearchParams({
+        ...baseParams,
+        offset: String(offset),
       });
-    }
-
-    const body = json.body ?? {};
-    const activities: WithingsActivityEntry[] = body.activities ?? [];
-    for (const entry of activities) {
-      results.push(entry);
-    }
-
-    const hasMore = body.more === true || body.more === 1;
-    if (!hasMore) break;
-    const nextOffset = Number(body.offset);
-    if (!Number.isFinite(nextOffset) || nextOffset <= offset) break;
-    offset = nextOffset;
-    pageCount += 1;
-    if (pageCount > 1000) break;
-  }
-  return results;
+      const pageStart = performance.now();
+      const res = await safeFetch(WITHINGS_MEASURE_URL, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/x-www-form-urlencoded",
+          Authorization: `Bearer ${accessToken}`,
+        },
+        body: params.toString(),
+      });
+      const json = await res.json();
+      const verdict = classifyWithingsResponse(res.status, json);
+      getEvent()?.addExternalCall({
+        service: "withings",
+        method: `fetchWithingsActivity(page=${page})`,
+        duration_ms: Math.round(performance.now() - pageStart),
+        status: res.status,
+        error:
+          verdict.classification === "success" ? undefined : verdict.reason,
+      });
+      if (verdict.classification !== "success") {
+        throw new WithingsApiError({
+          verb: "activity",
+          classification: verdict.classification,
+          withingsStatus: verdict.withingsStatus,
+          reason: verdict.reason,
+          upstreamError:
+            typeof json?.error === "string" ? json.error : undefined,
+        });
+      }
+      const body = json.body ?? {};
+      const activities: WithingsActivityEntry[] = body.activities ?? [];
+      return { items: activities, next: withingsNextOffset(body, offset) };
+    },
+    0,
+  );
+  return walk.items;
 }
 
 /**

src/lib/withings/sync-ecg.ts

@@ -80,6 +80,8 @@ import {
   recordSyncFailure,
   recordSyncSuccess,
 } from "@/lib/integrations/status";
+import { walkPages } from "@/lib/integrations/page-walk";
+import { WITHINGS_MAX_PAGES, withingsNextOffset } from "./client";
 
 const WITHINGS_HEART_URL = "https://wbsapi.withings.net/v2/heart";
 
@@ -139,60 +141,55 @@ export async function fetchWithingsHeartList(
   startdate: number,
   enddate: number,
 ): Promise<WithingsHeartEntry[]> {
-  const results: WithingsHeartEntry[] = [];
-  let offset = 0;
-  let pageCount = 0;
-
-  while (true) {
-    const params = new URLSearchParams({
-      action: "list",
-      startdate: String(startdate),
-      enddate: String(enddate),
-      offset: String(offset),
-    });
-
-    const pageStart = performance.now();
-    const res = await safeFetch(WITHINGS_HEART_URL, {
-      method: "POST",
-      headers: {
-        "Content-Type": "application/x-www-form-urlencoded",
-        Authorization: `Bearer ${accessToken}`,
-      },
-      body: params.toString(),
-    });
-    const json = await res.json();
-    const verdict = classifyWithingsResponse(res.status, json);
-    getEvent()?.addExternalCall({
-      service: "withings",
-      method: `fetchWithingsHeartList(page=${pageCount})`,
-      duration_ms: Math.round(performance.now() - pageStart),
-      status: res.status,
-      error: verdict.classification === "success" ? undefined : verdict.reason,
-    });
-    if (verdict.classification !== "success") {
-      throw new WithingsApiError({
-        verb: "heart",
-        classification: verdict.classification,
-        withingsStatus: verdict.withingsStatus,
-        reason: verdict.reason,
-        upstreamError: typeof json?.error === "string" ? json.error : undefined,
+  const walk = await walkPages<number, WithingsHeartEntry>(
+    "withings.heart",
+    WITHINGS_MAX_PAGES,
+    async (cursor, page) => {
+      const offset = cursor ?? 0;
+      const params = new URLSearchParams({
+        action: "list",
+        startdate: String(startdate),
+        enddate: String(enddate),
+        offset: String(offset),
       });
-    }
 
-    const body = json.body ?? {};
-    const series: WithingsHeartEntry[] = body.series ?? [];
-    results.push(...series);
+      const pageStart = performance.now();
+      const res = await safeFetch(WITHINGS_HEART_URL, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/x-www-form-urlencoded",
+          Authorization: `Bearer ${accessToken}`,
+        },
+        body: params.toString(),
+      });
+      const json = await res.json();
+      const verdict = classifyWithingsResponse(res.status, json);
+      getEvent()?.addExternalCall({
+        service: "withings",
+        method: `fetchWithingsHeartList(page=${page})`,
+        duration_ms: Math.round(performance.now() - pageStart),
+        status: res.status,
+        error:
+          verdict.classification === "success" ? undefined : verdict.reason,
+      });
+      if (verdict.classification !== "success") {
+        throw new WithingsApiError({
+          verb: "heart",
+          classification: verdict.classification,
+          withingsStatus: verdict.withingsStatus,
+          reason: verdict.reason,
+          upstreamError:
+            typeof json?.error === "string" ? json.error : undefined,
+        });
+      }
 
-    const hasMore = body.more === true || body.more === 1;
-    if (!hasMore) break;
-    const nextOffset = Number(body.offset);
-    if (!Number.isFinite(nextOffset) || nextOffset <= offset) break;
-    offset = nextOffset;
-    pageCount += 1;
-    if (pageCount > 1000) break;
-  }
-
-  return results;
+      const body = json.body ?? {};
+      const series: WithingsHeartEntry[] = body.series ?? [];
+      return { items: series, next: withingsNextOffset(body, offset) };
+    },
+    0,
+  );
+  return walk.items;
 }
 
 /**