From 5eaa813b34ba3de02921a6f42489aa69405872c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 06:27:58 +0200 Subject: [PATCH 1/4] fix(google-health): round imported values through importValue The mapper kept its own parseFloat(n.toFixed(2)), which rounds a half at the third decimal down whenever the double sits just below it: a weight of 72 345 g stored as 72.34 kg, a distance of 1 000 005 mm as 1000 m. The shared importValue rounds on the decimal the value prints as. The rounding guard drops its last pending entry and also catches a round2 declared as a const. --- src/__tests__/provider-rounding-guard.test.ts | 28 +++----------- .../google-health/__tests__/mappers.test.ts | 37 +++++++++++++++++++ src/lib/google-health/mappers.ts | 19 +++++----- 3 files changed, 51 insertions(+), 33 deletions(-) diff --git a/src/__tests__/provider-rounding-guard.test.ts b/src/__tests__/provider-rounding-guard.test.ts index 79ff4ba24..10fe0b655 100644 --- a/src/__tests__/provider-rounding-guard.test.ts +++ b/src/__tests__/provider-rounding-guard.test.ts @@ -26,17 +26,7 @@ const PROVIDER_DIRS = [ "whoop", "withings", ]; -/** - * Files that still round on their own, with the reason. Google Health's - * mapper is the one left: it moves to `importValue` with the next change to - * that directory. - */ -const PENDING: Readonly> = { - "google-health/mappers.ts": - "moves to importValue with the next Google Health change", -}; - -const ROUNDING = /\.toFixed\s*\(|\bfunction\s+round2\b/; +const ROUNDING = /\.toFixed\s*\(|\b(?:function|const|let)\s+round2\b/; function strip(text: string): string { return text @@ -56,25 +46,17 @@ describe("provider clients round through importValue", () => { expect(files.length).toBeGreaterThanOrEqual(60); }); - it("finds no rounding of their own outside the pending list", () => { - const offenders = files.filter( - (f) => - !(f in PENDING) && - ROUNDING.test(strip(readFileSync(join(LIB, f), "utf8"))), + it("finds no rounding of their own", () => { + const offenders = files.filter((f) => + ROUNDING.test(strip(readFileSync(join(LIB, f), "utf8"))), ); expect(offenders).toEqual([]); }); - it("keeps the pending list honest", () => { - const stale = Object.keys(PENDING).filter( - (f) => !ROUNDING.test(strip(readFileSync(join(LIB, f), "utf8"))), - ); - expect(stale).toEqual([]); - }); - it("catches a planted copy", () => { expect(ROUNDING.test("const v = parseFloat(n.toFixed(2));")).toBe(true); expect(ROUNDING.test("function round2(n: number) {}")).toBe(true); + expect(ROUNDING.test("const round2 = (n: number) => n;")).toBe(true); expect(ROUNDING.test("value: importValue(n)")).toBe(false); }); }); diff --git a/src/lib/google-health/__tests__/mappers.test.ts b/src/lib/google-health/__tests__/mappers.test.ts index c20559389..16d119f23 100644 --- a/src/lib/google-health/__tests__/mappers.test.ts +++ b/src/lib/google-health/__tests__/mappers.test.ts @@ -73,6 +73,43 @@ describe("mapWeight — spot sample", () => { }); }); +describe("imported values round half away from zero (importValue)", () => { + // Each of these sits on a half at the third decimal whose double lies just + // below it, where `toFixed(2)` rounded down. + it("rounds a gram-precise weight up on the half", () => { + const [r] = mapWeight({ + weight: { + weightGrams: 72345, + sampleTime: { physicalTime: "2026-06-01T08:00:00.000Z" }, + }, + }); + expect(r!.value).toBe(72.35); + }); + + it("rounds a daily distance total up on the half", () => { + const [r] = mapDistance({ + civilStartTime: { date: { year: 2026, month: 6, day: 1 } }, + distance: { millimetersSum: "1000005" }, + }); + expect(r!.value).toBe(1000.01); + }); + + it("rounds a session distance up on the half", () => { + const w = mapWorkout({ + name: "users/me/dataTypes/exercise/dataPoints/half", + exercise: { + exerciseType: "RUNNING", + interval: { + startTime: "2026-06-03T07:00:00.000Z", + endTime: "2026-06-03T07:45:00.000Z", + }, + metricsSummary: { distanceMillimeters: "8000025" }, + }, + }); + expect(w!.totalDistanceM).toBe(8000.03); + }); +}); + describe("mapBodyFat — camelCase union key", () => { it("reads the percentage under the camelCase bodyFat key", () => { const rows = mapBodyFat({ diff --git a/src/lib/google-health/mappers.ts b/src/lib/google-health/mappers.ts index 194a782ff..0e6bfe02a 100644 --- a/src/lib/google-health/mappers.ts +++ b/src/lib/google-health/mappers.ts @@ -14,6 +14,7 @@ * this file performs I/O — everything is synchronous and deterministic given a * payload (+ optional user timezone). */ +import { importValue } from "@/lib/integrations/import-value"; import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; import { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; @@ -217,10 +218,6 @@ export interface GoogleHealthRollupPoint { /** Millimetres → centimetres (Google `height.heightMillimeters` → `User.heightCm`). */ const MM_TO_CM = 0.1; -function round2(n: number): number { - return parseFloat(n.toFixed(2)); -} - /** * Coerce a raw payload value to a finite number, or null. proto3 int64 fields * arrive as JSON strings (`"12345"`); `Number()` handles both int and decimal @@ -445,7 +442,7 @@ function mapSimple( return [ { type: spec.type, - value: round2(value), + value: importValue(value), unit: spec.unit, measuredAt: resolveMeasuredAt(point, dataType, new Date(), tz), fieldTag: `${externalAnchor(point, dataType)}:${spec.fieldTag}`, @@ -678,7 +675,7 @@ export function mapHeight( if (mm === null) return null; const t = readPath(point, `${dt.key}.sampleTime.physicalTime`); const sampledAt = typeof t === "string" ? parseLocalInstant(t) : null; - return { cm: round2(mm * MM_TO_CM), sampledAt }; + return { cm: importValue(mm * MM_TO_CM), sampledAt }; } // ─── Activity mappers: daily roll-up totals ──────────────────── @@ -732,7 +729,7 @@ function mapDailyRollup( return [ { type: spec.type, - value: round2(value), + value: importValue(value), unit: spec.unit, measuredAt: canonicalDailyTimestamp(dayKey, tz), // `stats::` — the sync layer reads `cumulativeDaily` @@ -833,7 +830,7 @@ export function mapVo2Max( return [ { type: "VO2_MAX", - value: round2(value), + value: importValue(value), unit: "mL/(kg·min)", measuredAt, fieldTag: `vo2_max:${externalAnchor(point, dt)}`, @@ -1035,7 +1032,7 @@ export function mapSleepSessionDetailed( if (!windowEnd || end > windowEnd) windowEnd = end; rows.push({ type: "SLEEP_DURATION", - value: round2(mins), + value: importValue(mins), unit: "minutes", measuredAt: end, fieldTag: `${anchor}:sleep:${start.toISOString()}`, @@ -1287,7 +1284,9 @@ export function mapWorkout( durationSec, totalEnergyKcal: energyKcal !== null ? Math.round(energyKcal) : null, totalDistanceM: - distanceMm !== null && distanceMm >= 0 ? round2(distanceMm / 1000) : null, + distanceMm !== null && distanceMm >= 0 + ? importValue(distanceMm / 1000) + : null, avgHeartRate: avgHr !== null && avgHr > 0 ? Math.round(avgHr) : null, // metricsSummary carries no maximum/minimum heart rate. maxHeartRate: null, From 93b3f36d96ced1f68bab537bb4880c47e161e957 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 06:49:44 +0200 Subject: [PATCH 2/4] fix(strava): walk activity pages through walkPages and gate the backfill on it The activity loop stopped at its page ceiling and returned what it had read, and the backfill stamped the account complete over it: a history longer than forty pages was cut off for good. The walk now runs through walkPages, which reports a walk cut off at its ceiling, and the backfill reads backfillVerdict like the WHOOP and Fitbit backfills: a truncated walk leaves the marker unset without a retry, and a connection waiting for re-authorisation returns without walking. --- src/lib/jobs/strava-backfill.ts | 45 ++++- .../__tests__/sync-cursor-durability.test.ts | 1 + .../strava/__tests__/sync-page-walk.test.ts | 191 ++++++++++++++++++ src/lib/strava/sync.ts | 98 +++++---- 4 files changed, 289 insertions(+), 46 deletions(-) create mode 100644 src/lib/strava/__tests__/sync-page-walk.test.ts diff --git a/src/lib/jobs/strava-backfill.ts b/src/lib/jobs/strava-backfill.ts index 9301bb11e..539396dad 100644 --- a/src/lib/jobs/strava-backfill.ts +++ b/src/lib/jobs/strava-backfill.ts @@ -18,6 +18,11 @@ import { prisma } from "@/lib/db"; import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; +import { isReauthRequired } from "@/lib/integrations/status"; +import { + backfillVerdict, + withPageWalkLedger, +} from "@/lib/integrations/page-walk"; import { syncUserStrava } from "@/lib/strava/sync"; export const STRAVA_BACKFILL_QUEUE = "strava-backfill"; @@ -36,14 +41,46 @@ export interface StravaBackfillPayload { /** * Per-user backfill handler. Runs a full-history sync for one account and - * stamps `stravaBackfillCompletedAt` so the discovery query drops it. - * Idempotent: the `(userId, source, externalId)` workout upsert is key-stable, - * so a re-run (e.g. a reboot mid-walk) overwrites rather than duplicating. + * stamps `stravaBackfillCompletedAt` only when the walk read the whole + * history, so the discovery query drops it. Idempotent: the + * `(userId, source, externalId)` workout upsert is key-stable, so a re-run + * (e.g. a reboot mid-walk) overwrites rather than duplicating. + * + * Verdict-gated (`backfillVerdict`, `integrations/page-walk.ts`), like the + * WHOOP and Fitbit backfills. A failed sync throws out of `syncUserStrava` + * (failure recorded first) and pg-boss retries. An activity walk that reached + * its page ceiling with history left is not complete: the marker stays unset + * without a retry, and boot discovery offers the account again. A connection + * parked for re-authorisation returns without stamping and without a retry, + * since every attempt would meet the same dead grant. */ export async function runStravaBackfillForUser( userId: string, ): Promise<{ imported: number }> { - const imported = await syncUserStrava(userId, { fullSync: true }); + if (await isReauthRequired(userId, "strava")) { + annotate({ + action: { + name: "strava.backfill.skipped_reauth", + details: { imported: 0 }, + }, + }); + return { imported: 0 }; + } + + const { result: imported, truncated } = await withPageWalkLedger(() => + syncUserStrava(userId, { fullSync: true }), + ); + // A failure has already thrown; what is left to decide is whether the walk + // reached the beginning of the history. + if (backfillVerdict({ failed: false, truncated }) === "incomplete") { + annotate({ + action: { + name: "strava.backfill.truncated", + details: { imported, collections: truncated.join(",") }, + }, + }); + return { imported }; + } await prisma.user.update({ where: { id: userId }, diff --git a/src/lib/strava/__tests__/sync-cursor-durability.test.ts b/src/lib/strava/__tests__/sync-cursor-durability.test.ts index dc71edf55..636e529ba 100644 --- a/src/lib/strava/__tests__/sync-cursor-durability.test.ts +++ b/src/lib/strava/__tests__/sync-cursor-durability.test.ts @@ -38,6 +38,7 @@ vi.mock("@/lib/logging/context", () => ({ vi.mock("@/lib/integrations/status", () => ({ recordSyncFailure: mocks.recordSyncFailure, recordSyncSuccess: mocks.recordSyncSuccess, + isReauthRequired: vi.fn(async () => false), toFailureKind: vi.fn(() => "transient"), markSyncFailureRecorded: (err: T) => err, })); diff --git a/src/lib/strava/__tests__/sync-page-walk.test.ts b/src/lib/strava/__tests__/sync-page-walk.test.ts new file mode 100644 index 000000000..3f2f09f0e --- /dev/null +++ b/src/lib/strava/__tests__/sync-page-walk.test.ts @@ -0,0 +1,191 @@ +/** + * The Strava activity walk runs through `walkPages`, and the backfill reads + * its verdict (`integrations/page-walk.ts`): a walk that reaches its page + * ceiling with a full page last is truncated, and the backfill leaves its + * marker unset instead of stamping a history it did not finish. Before, the + * loop stopped at forty pages and the backfill stamped the account complete. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + fetchActivities: vi.fn(), + annotate: vi.fn(), + updateUser: vi.fn(), + isReauthRequired: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + findUnique: vi.fn(async () => ({ stravaLastActivityAt: null })), + updateMany: vi.fn(async () => ({ count: 1 })), + update: mocks.updateUser, + }, + workout: { + createManyAndReturn: vi.fn(async ({ data }) => [ + { id: `w-${data.externalId}`, startedAt: data.startedAt }, + ]), + update: vi.fn(), + }, + }, +})); +vi.mock("@/lib/arrivals/workout-emit", () => ({ + emitInsertedWorkoutArrival: vi.fn(async () => {}), +})); +vi.mock("@/lib/cache/invalidate", () => ({ + invalidateUserMeasurements: vi.fn(), +})); +vi.mock("@/lib/logging/context", () => ({ + annotate: mocks.annotate, + getEvent: () => ({ addWarning: vi.fn() }), +})); +vi.mock("@/lib/integrations/status", () => ({ + recordSyncFailure: vi.fn(), + recordSyncSuccess: vi.fn(), + isReauthRequired: mocks.isReauthRequired, + toFailureKind: vi.fn(() => "transient"), + markSyncFailureRecorded: (err: T) => err, +})); +vi.mock("../credentials", () => ({ + getStravaConnection: vi.fn(async () => ({ + accessToken: "access-token", + refreshToken: "refresh-token", + refreshTokenCiphertext: Buffer.from("ciphertext"), + })), + getStravaClientCredentials: vi.fn(), + storeStravaTokens: vi.fn(), +})); +vi.mock("../client", async (importOriginal) => ({ + ...(await importOriginal()), + fetchActivities: mocks.fetchActivities, + // Summaries carry HR; the detail fetch only fills calories. + fetchActivityById: vi.fn(async () => ({})), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill"; +import { syncUserStrava } from "../sync"; + +const PER_PAGE = 100; +const BACKFILL_MAX_PAGES = 40; + +/** A page of `n` activities; ids continue across pages. */ +function page(pageNo: number, n: number) { + return Array.from({ length: n }, (_, i) => { + const id = (pageNo - 1) * PER_PAGE + i + 1; + return { + id, + sport_type: "Run", + start_date: new Date(Date.UTC(2020, 0, 1) + id * 3_600_000).toISOString(), + moving_time: 1800, + elapsed_time: 1900, + average_heartrate: 140, + }; + }); +} + +/** A history of `total` activities, served page by page. */ +function history(total: number) { + mocks.fetchActivities.mockImplementation( + async (_t: string, q: { page: number; perPage: number }) => { + const remaining = total - (q.page - 1) * q.perPage; + return page(q.page, Math.max(0, Math.min(q.perPage, remaining))); + }, + ); +} + +function truncatedAnnotations() { + return mocks.annotate.mock.calls.filter( + ([e]) => e?.action?.name === "integration.page_walk.truncated", + ); +} + +beforeEach(() => { + vi.clearAllMocks(); + mocks.isReauthRequired.mockResolvedValue(false); + mocks.updateUser.mockResolvedValue({ id: "user-1" }); +}); + +describe("Strava activity walk", () => { + it("reads pages until a short one and stops there", async () => { + history(2 * PER_PAGE + 7); + + await expect(syncUserStrava("user-1", { fullSync: true })).resolves.toBe( + 207, + ); + expect(mocks.fetchActivities.mock.calls.map(([, q]) => q.page)).toEqual([ + 1, 2, 3, + ]); + expect(truncatedAnnotations()).toEqual([]); + }); + + it("ends on an empty page without calling it truncated", async () => { + history(PER_PAGE); + + await expect(syncUserStrava("user-1", { fullSync: true })).resolves.toBe( + PER_PAGE, + ); + expect(mocks.fetchActivities).toHaveBeenCalledTimes(2); + expect(truncatedAnnotations()).toEqual([]); + }); + + it("reports a walk cut off at its ceiling", async () => { + history((BACKFILL_MAX_PAGES + 5) * PER_PAGE); + + await syncUserStrava("user-1", { fullSync: true }); + + expect(mocks.fetchActivities).toHaveBeenCalledTimes(BACKFILL_MAX_PAGES); + expect(truncatedAnnotations()).toEqual([ + [ + expect.objectContaining({ + meta: { + walk: "strava.activities", + pages: BACKFILL_MAX_PAGES, + reason: "ceiling", + }, + }), + ], + ]); + }); +}); + +describe("Strava backfill verdict", () => { + it("stamps the marker when the walk reached the start of the history", async () => { + history(3 * PER_PAGE - 1); + + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: 3 * PER_PAGE - 1, + }); + expect(mocks.updateUser).toHaveBeenCalledWith({ + where: { id: "user-1" }, + data: { stravaBackfillCompletedAt: expect.any(Date) }, + }); + }); + + it("leaves the marker unset, without a retry, when the walk was cut off", async () => { + history((BACKFILL_MAX_PAGES + 1) * PER_PAGE); + + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: BACKFILL_MAX_PAGES * PER_PAGE, + }); + expect(mocks.updateUser).not.toHaveBeenCalled(); + expect(mocks.annotate).toHaveBeenCalledWith( + expect.objectContaining({ + action: expect.objectContaining({ name: "strava.backfill.truncated" }), + }), + ); + }); + + it("returns without walking or stamping while the connection waits for re-authorisation", async () => { + mocks.isReauthRequired.mockResolvedValue(true); + history(PER_PAGE); + + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: 0, + }); + expect(mocks.fetchActivities).not.toHaveBeenCalled(); + expect(mocks.updateUser).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/strava/sync.ts b/src/lib/strava/sync.ts index 38e5600ba..89070f8fd 100644 --- a/src/lib/strava/sync.ts +++ b/src/lib/strava/sync.ts @@ -19,7 +19,8 @@ * `reauth_required` on the `strava` ledger. * * Rate limits: 200 req / 15 min + 2000 / day per app. The walk is page-bounded - * and the per-activity detail fetch (for `calories`, absent from the summary) + * (`walkPages`, which reports a walk cut off at its ceiling instead of + * passing it off as the whole history) and the per-activity detail fetch (for `calories`, absent from the summary) * is budget-capped; a 429 mid-walk records a transient failure and the next * cron tick resumes from the cursor. */ @@ -34,6 +35,7 @@ import { toFailureKind, type FailureKind, } from "@/lib/integrations/status"; +import { walkPages } from "@/lib/integrations/page-walk"; import { fetchActivities, fetchActivityById, @@ -163,58 +165,70 @@ export async function syncUserStrava( ? BACKFILL_DETAIL_BUDGET : INCREMENTAL_DETAIL_BUDGET; - const rows: StravaWorkoutRow[] = []; + let rows: StravaWorkoutRow[]; let newestStart: Date | null = null; try { - for (let page = 1; page <= maxPages; page++) { - const activities = await authed((t) => - fetchActivities(t, { after: afterEpoch, page, perPage: PER_PAGE }), - ); - if (activities.length === 0) break; - - for (const a of activities) { - // Fetch detail only to fill `calories` (absent from the summary), and - // only while the budget allows. When the summary already lacks HR we - // also let the detail top it up. A detail failure is non-fatal — map - // from the summary — but a 429 aborts the whole walk (transient). - let detail: StravaDetailedActivity | null = null; - if (detailBudget > 0) { - try { - detail = await authed((t) => fetchActivityById(t, a.id)); - detailBudget -= 1; - } catch (err) { - if (err instanceof StravaApiError && err.httpStatus === 429) { - throw err; + // Strava pages by number; a short page is the last one. A walk that + // reaches its ceiling with a full page last reports itself truncated, and + // the backfill leaves its marker unset (`integrations/page-walk.ts`). + const walk = await walkPages( + "strava.activities", + maxPages, + async (cursor) => { + const page = cursor ?? 1; + const activities = await authed((t) => + fetchActivities(t, { after: afterEpoch, page, perPage: PER_PAGE }), + ); + const mapped: StravaWorkoutRow[] = []; + for (const a of activities) { + // Fetch detail only to fill `calories` (absent from the summary), + // and only while the budget allows. When the summary already lacks + // HR we also let the detail top it up. A detail failure is + // non-fatal — map from the summary — but a 429 aborts the whole + // walk (transient). + let detail: StravaDetailedActivity | null = null; + if (detailBudget > 0) { + try { + detail = await authed((t) => fetchActivityById(t, a.id)); + detailBudget -= 1; + } catch (err) { + if (err instanceof StravaApiError && err.httpStatus === 429) { + throw err; + } + getEvent()?.addWarning( + `strava: activity detail fetch failed for ${a.id}: ${err}`, + ); + detail = null; } + } else if (!summaryHasHeartRate(a)) { getEvent()?.addWarning( - `strava: activity detail fetch failed for ${a.id}: ${err}`, + `strava: detail budget exhausted; mapping activity ${a.id} from summary only`, ); - detail = null; } - } else if (!summaryHasHeartRate(a)) { - getEvent()?.addWarning( - `strava: detail budget exhausted; mapping activity ${a.id} from summary only`, - ); - } - const row = mapActivity(a, detail); - if (row) { - rows.push(row); - if (a.start_date) { - const d = new Date(a.start_date); - if ( - !Number.isNaN(d.getTime()) && - (!newestStart || d > newestStart) - ) { - newestStart = d; + const row = mapActivity(a, detail); + if (row) { + mapped.push(row); + if (a.start_date) { + const d = new Date(a.start_date); + if ( + !Number.isNaN(d.getTime()) && + (!newestStart || d > newestStart) + ) { + newestStart = d; + } } } } - } - - if (activities.length < PER_PAGE) break; - } + return { + items: mapped, + next: activities.length < PER_PAGE ? null : page + 1, + }; + }, + 1, + ); + rows = walk.items; } catch (err) { await recordStravaFailure(userId, err); throw markSyncFailureRecorded(err); From 5c1b9889ec74e7cdc2d36288502a56c5c3008105 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 06:49:45 +0200 Subject: [PATCH 3/4] fix(google-health): page dailyRollUp chunks through walkPages Each rollup chunk kept its own token loop with a hundred-page bound. It now walks through walkPages: a token handed out twice ends the chunk at once instead of re-reading the same page up to the bound, and a chunk cut off is reported on the wide event as well as through the client outcome. --- .../__tests__/rollup-request.test.ts | 77 +++++++++++++ src/lib/google-health/client.ts | 107 ++++++++++-------- 2 files changed, 137 insertions(+), 47 deletions(-) diff --git a/src/lib/google-health/__tests__/rollup-request.test.ts b/src/lib/google-health/__tests__/rollup-request.test.ts index bed00a773..770010be4 100644 --- a/src/lib/google-health/__tests__/rollup-request.test.ts +++ b/src/lib/google-health/__tests__/rollup-request.test.ts @@ -26,6 +26,7 @@ import { buildDailyRollUpBody, extractGoogleApiErrorDetail, fetchDailyRollUp, + runWithGoogleHealthClientOutcome, type GoogleHealthRollupShape, } from "../client"; import { GoogleHealthApiError } from "../response-classifier"; @@ -219,6 +220,82 @@ describe("fetchDailyRollUp — chunk walk + conservative fallback", () => { }); }); +describe("fetchDailyRollUp — pages within a chunk (walkPages)", () => { + const oneChunk = { start: new Date("2026-06-20T00:00:00Z") }; + + it("follows nextPageToken to the end of a chunk", async () => { + safeFetchMock + .mockResolvedValueOnce( + jsonResponse(200, { + rollupDataPoints: [{ steps: { countSum: "1" } }], + nextPageToken: "p2", + }), + ) + .mockResolvedValueOnce( + jsonResponse(200, { + rollupDataPoints: [{ steps: { countSum: "2" } }], + nextPageToken: "", + }), + ); + const { result, outcome } = await runWithGoogleHealthClientOutcome(() => + fetchDailyRollUp( + GOOGLE_HEALTH_DATA_TYPES.steps, + "token", + "fetchSteps", + oneChunk, + ), + ); + expect(result).toHaveLength(2); + expect(safeFetchMock).toHaveBeenCalledTimes(2); + expect( + JSON.parse((safeFetchMock.mock.calls[1]![1] as { body: string }).body) + .pageToken, + ).toBe("p2"); + expect(outcome).toMatchObject({ pages: 2, truncated: false }); + }); + + it("stops a chunk whose page token comes back again and reports it truncated", async () => { + safeFetchMock.mockResolvedValue( + jsonResponse(200, { + rollupDataPoints: [{ steps: { countSum: "1" } }], + nextPageToken: "again", + }), + ); + const { result, outcome } = await runWithGoogleHealthClientOutcome(() => + fetchDailyRollUp( + GOOGLE_HEALTH_DATA_TYPES.steps, + "token", + "fetchSteps", + oneChunk, + ), + ); + // The first page hands out "again", the second hands it out again. + expect(safeFetchMock).toHaveBeenCalledTimes(2); + expect(result).toHaveLength(2); + expect(outcome).toMatchObject({ + truncated: true, + reasonCode: "existing_page_limit", + }); + }); + + it("stops a chunk at its page ceiling and reports it truncated", async () => { + let n = 0; + safeFetchMock.mockImplementation(async () => + jsonResponse(200, { rollupDataPoints: [], nextPageToken: `p${++n}` }), + ); + const { outcome } = await runWithGoogleHealthClientOutcome(() => + fetchDailyRollUp( + GOOGLE_HEALTH_DATA_TYPES.steps, + "token", + "fetchSteps", + oneChunk, + ), + ); + expect(safeFetchMock).toHaveBeenCalledTimes(100); + expect(outcome).toMatchObject({ pages: 100, truncated: true }); + }); +}); + describe("extractGoogleApiErrorDetail — AIP-193 field violations", () => { it("keeps the plain STATUS: message shape without details", () => { expect( diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index fcf70ecd5..d12006fc1 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -55,6 +55,7 @@ import { googleHealthWalkProgress, noteGoogleHealthWalkPage, } from "./backfill-bounds"; +import { walkPages } from "@/lib/integrations/page-walk"; export * from "./mappers"; @@ -1605,6 +1606,13 @@ export const GOOGLE_HEALTH_ROLLUP_FALLBACK_RANGE_DAYS = 14; */ export const GOOGLE_HEALTH_ROLLUP_BACKFILL_DAYS = 5 * 365; +/** + * Page ceiling for one dailyRollUp chunk. A chunk of at most 90 daily windows + * fits on one page of the documented default size, so a chunk that pages at + * all is unusual and one that reaches this is a runaway token. + */ +const GOOGLE_HEALTH_ROLLUP_MAX_PAGES = 100; + /** A civil calendar date (1-based month), the dailyRollUp range unit. */ export interface GoogleHealthCivilDate { year: number; @@ -1778,56 +1786,61 @@ export async function fetchDailyRollUp( truncated = true; break; } - let pageToken: string | null | undefined; - let pageCount = 0; - do { - const body = buildDailyRollUpBody(chunk, pageToken ?? undefined); + const index = chunkIndex; + const chunkWalk = await walkPages( + `google_health.rollup.${dataType.path}`, + GOOGLE_HEALTH_ROLLUP_MAX_PAGES, + async (pageToken, pageCount) => { + const body = buildDailyRollUpBody(chunk, pageToken); - requestCount += 1; - const { json, verdict, apiErrorDetail } = - await googleHealthDataRequest( - accessToken, - `${verb}(chunk=${chunkIndex},page=${pageCount})`, - (token) => - safeFetch( - `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`, - { - method: "POST", - headers: { - Authorization: `Bearer ${token}`, - "Content-Type": "application/json", + requestCount += 1; + const { json, verdict, apiErrorDetail } = + await googleHealthDataRequest( + accessToken, + `${verb}(chunk=${index},page=${pageCount})`, + (token) => + safeFetch( + `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`, + { + method: "POST", + headers: { + Authorization: `Bearer ${token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(body), }, - body: JSON.stringify(body), - }, - ), - ); - if (verdict.classification !== "success") { - throw new GoogleHealthApiError({ - verb, - classification: verdict.classification, - httpStatus: verdict.httpStatus, - reason: verdict.reason, - upstreamError: apiErrorDetail, - }); - } + ), + ); + if (verdict.classification !== "success") { + throw new GoogleHealthApiError({ + verb, + classification: verdict.classification, + httpStatus: verdict.httpStatus, + reason: verdict.reason, + upstreamError: apiErrorDetail, + }); + } - // Surface the raw envelope key names of the very first page (names - // only, never values) — see `RollupQuery.onEnvelopeKeys`. - if ( - chunkIndex === 0 && - pageCount === 0 && - json && - typeof json === "object" - ) { - query.onEnvelopeKeys?.(Object.keys(json)); - } - for (const p of json?.rollupDataPoints ?? []) points.push(p); - pageToken = json?.nextPageToken ?? null; - pageCount += 1; - noteSlicePage(); - } while (pageToken && pageCount < 100); - totalPages += pageCount; - truncated ||= Boolean(pageToken); + // Surface the raw envelope key names of the very first page (names + // only, never values) — see `RollupQuery.onEnvelopeKeys`. + if ( + index === 0 && + pageCount === 0 && + json && + typeof json === "object" + ) { + query.onEnvelopeKeys?.(Object.keys(json)); + } + noteSlicePage(); + return { + items: json?.rollupDataPoints ?? [], + next: json?.nextPageToken || null, + }; + }, + ); + for (const p of chunkWalk.items) points.push(p); + totalPages += chunkWalk.pages; + truncated ||= chunkWalk.truncated; chunkIndex += 1; } notePagination(totalPages, points.length, truncated); From 2e92069c16732d0ef3eb3270a01834a0ccb47d02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 06:49:47 +0200 Subject: [PATCH 4/4] test(guards): name the list walk's exception instead of pending entries The page-walk guard flagged the backfill progress codec's page > 0 check, a validity test rather than a ceiling, and missed the list walk's own loop (pageCount - skipped < maxPages). The matcher now reads a counter offset and ignores a literal bound of 0 or 1. Strava and the rollup chunks walk through walkPages, so their pending entries go. The dataPoints.list walk stays outside as a named exception with a match allowance: it is bounded by progress rather than by a cursor ceiling, streams pages, stops on a slice budget and keeps its place, and its backfill decides with a sliced verdict. --- src/__tests__/page-walk-guard.test.ts | 69 +++++++++++++++++++-------- src/lib/integrations/page-walk.ts | 13 +++-- 2 files changed, 57 insertions(+), 25 deletions(-) diff --git a/src/__tests__/page-walk-guard.test.ts b/src/__tests__/page-walk-guard.test.ts index 3d7f7900f..bd843d704 100644 --- a/src/__tests__/page-walk-guard.test.ts +++ b/src/__tests__/page-walk-guard.test.ts @@ -26,20 +26,37 @@ const PROVIDER_DIRS = [ "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. + * Page loops that stay outside `walkPages` on purpose, with how many matches + * each file may hold and why. A file with more matches than its allowance has + * grown a loop of its own and fails like any other. + * + * Google Health's list walk (`forEachDataPointPage`) is bounded by progress, + * not by a cursor ceiling: it stops after `GOOGLE_HEALTH_NO_PROGRESS_PAGES` + * pages in a row that reach no further back (the same page token handed out + * again counts toward that, rather than ending the walk at once as + * `walkPages` does), streams each page to the writer instead of collecting + * the collection (a dense heart-rate history does not fit in the heap), stops + * mid-walk on the backfill slice's budget and on a rate limit it cannot wait + * out, passes over pages beyond the resume edge without charging them to its + * ceiling, and keeps its place in the list for the next slice. Its truncation + * reaches the backfill through the client outcome, and the backfill decides + * from it with its own sliced verdict (continue, wait for a retry, stuck, + * complete; `jobs/google-health-backfill.ts`), which `backfillVerdict`'s + * three outcomes cannot carry. */ -const PENDING: Readonly> = { - "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", +const EXCEPTIONS: Readonly> = { + "google-health/client.ts": { + matches: 1, + why: "progress-bounded streaming list walk with slice budget and resume (forEachDataPointPage)", + }, }; -/** A page counter compared against a bound, in any of the shapes used. */ +/** + * A page counter compared against a bound, in any of the shapes used. A + * literal bound of 0 or 1 is a validity check (`page > 0`), not a ceiling. + */ const PAGE_LOOP = - /\b(?:pages?|pageCount|pagesRead|page)\s*(?:<|<=|>|>=)\s*(?:[A-Z_][A-Z0-9_]*|maxPages|\d+)\b/; + /\b(?:pages?|pageCount|pagesRead|page)(?:\s*-\s*[A-Za-z_]\w*)?\s*(?:<|<=|>|>=)\s*(?:[A-Z_][A-Z0-9_]*|maxPages|[2-9]|[1-9]\d+)\b/g; function strip(text: string): string { return text @@ -54,19 +71,26 @@ describe("provider clients walk pages through walkPages", () => { !f.includes("__tests__") && !f.endsWith(".test.ts"), ); - const offending = (f: string) => - PAGE_LOOP.test(strip(readFileSync(join(LIB, f), "utf8"))); + const matches = (f: string) => + strip(readFileSync(join(LIB, f), "utf8")).match(PAGE_LOOP)?.length ?? 0; 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("finds no page loop of their own beyond the exceptions", () => { + const over = files + .map((f) => ({ f, n: matches(f), allowed: EXCEPTIONS[f]?.matches ?? 0 })) + .filter(({ n, allowed }) => n > allowed) + .map(({ f, n }) => `${f}: ${n}`); + expect(over).toEqual([]); }); - it("keeps the pending list honest", () => { - expect(Object.keys(PENDING).filter((f) => !offending(f))).toEqual([]); + it("keeps the exceptions honest", () => { + const stale = Object.entries(EXCEPTIONS) + .filter(([f, { matches: allowed }]) => matches(f) !== allowed) + .map(([f]) => f); + expect(stale).toEqual([]); }); it("catches the shapes the loops used", () => { @@ -74,11 +98,16 @@ describe("provider clients walk pages through walkPages", () => { "while (pages < WORKOUT_MAX_PAGES) {", "if (pageCount > 1000) break;", "} while (nextToken && pageCount < maxPages);", + "} while (pageToken && pageCount - skipped < maxPages);", + "for (let page = 1; page <= maxPages; page++) {", ]) { - expect(PAGE_LOOP.test(planted), planted).toBe(true); + expect(planted.match(PAGE_LOOP), planted).not.toBeNull(); + } + for (const fine of [ + "const walk = await walkPages(label, 12, f);", + "r.page > 0 ? r.page : undefined", + ]) { + expect(fine.match(PAGE_LOOP), fine).toBeNull(); } - expect(PAGE_LOOP.test("const walk = await walkPages(label, 12, f);")).toBe( - false, - ); }); }); diff --git a/src/lib/integrations/page-walk.ts b/src/lib/integrations/page-walk.ts index df126c8a5..08af1771d 100644 --- a/src/lib/integrations/page-walk.ts +++ b/src/lib/integrations/page-walk.ts @@ -19,11 +19,14 @@ * - 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. + * Google Health's dailyRollUp chunks walk through here. Its `dataPoints.list` + * walk (`forEachDataPointPage`) stays outside on purpose: it is bounded by + * progress in time rather than by a cursor ceiling (`google-health/ + * backfill-bounds.ts`), streams pages instead of collecting them, stops on a + * backfill slice's budget and keeps its place for the next slice, and its + * backfill decides with a sliced verdict these three outcomes cannot carry. + * `src/__tests__/page-walk-guard.test.ts` keeps hand-rolled page loops out of + * the provider directories and names that one exception with its reasons. */ import { AsyncLocalStorage } from "node:async_hooks";