diff --git a/.env.production.example b/.env.production.example index 84d0fe10e..9159f8a88 100644 --- a/.env.production.example +++ b/.env.production.example @@ -49,6 +49,11 @@ API_TOKEN_HMAC_KEY="" # "/api/google-health/callback"; it must stay same-origin # with NEXT_PUBLIC_APP_URL or the handshake is refused. # GOOGLE_HEALTH_REDIRECT_URI="" +# +# How many years back the history backfill reads after a connect: a whole +# number from 1 to 100, 20 when unset. A data type whose history goes further +# back ends at that point, and the integration card names the date. +# GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS="20" # ----------------------------------------------------------------------------- diff --git a/docker-compose.yml b/docker-compose.yml index e03a72d4e..cfd6d6b9e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -226,6 +226,10 @@ x-healthlog-env: &healthlog_env # the authorization code somewhere else. FITBIT_REDIRECT_URI: "${FITBIT_REDIRECT_URI:-}" GOOGLE_HEALTH_REDIRECT_URI: "${GOOGLE_HEALTH_REDIRECT_URI:-}" + # How many years back the Google Health history backfill reads (a whole + # number from 1 to 100; 20 when unset). A data type whose history goes + # further ends there, and the integration card says so. + GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS: "${GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS:-}" # Nightscout origins on your own network that the per-user connection # validator should accept, comma-separated (e.g. "http://nightscout.lan"). # Without it a LAN/private Nightscout URL is refused as an SSRF target. diff --git a/docs/integrations/google-health.md b/docs/integrations/google-health.md index 600a52fc7..c4ddac290 100644 --- a/docs/integrations/google-health.md +++ b/docs/integrations/google-health.md @@ -126,8 +126,11 @@ Only one sync runs per account at a time. **Sync now** reports that a sync is already in progress while the history backfill that follows a connect is running, and shows its progress instead of starting a second one. -The backfill reads the account's whole history, and for a watch that records -heart rate every minute that is a thousand-odd pages per year. It runs in +The backfill reads the account's whole history. Heart rate is by far the +largest part: a page holds up to 1,000 readings, so a watch that records once +a minute fills some 525 pages a year, and one that records every few seconds +(as many Pixel and Fitbit watches do during the day) several thousand; one +account measured about 4,100 pages a year. It runs in slices: a slice reads for up to five minutes or 500 pages, writes what it read, notes how far it got and lets go of the account. The next slice starts a minute later where the last one stopped. Two kinds of note are kept: a data @@ -171,6 +174,35 @@ three times before they count as a failure. None of these mark the connection as needing a reconnect. Access tokens last an hour, and a long backfill renews its token as it goes. +### How far back the backfill reads + +The backfill is not bounded by a number of pages, because how many pages a +year of history fills depends on the device. Two bounds apply instead: + +- **The horizon.** It reads back at most 20 years. Set + `GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS` to a whole number from 1 to 100 to + change that (it is on the compose `environment:` list; any other value + falls back to 20). A data type whose history goes back further is imported + up to the horizon and ends there. +- **Progress.** A walk has to reach older readings as it goes. If Google + hands out 20 pages in a row that reach no further back than the ones before + them (the same page again, or a page token that goes round in a circle), + the walk stops, and that data type's history ends at the oldest reading + written so far. Retrying would only meet the same wall, so it is not + retried; a reconnect starts the backfill over. + +Either way the data type counts as finished, the backfill completes once +every other type is in, and the integration card stops saying that older +history is loading. Instead it says, per data type, how far back the import +reaches and why it ends there. + +Before 1.43.6 every page-by-page data type had a ceiling of pages counted +across all slices (10,000 for heart rate, 1,000 for the others). A dense +heart-rate history reaches 10,000 pages two to three years back, the count +never reset, and from then on every slice stopped at the same place without +the backfill ever completing. A backfill left in that state continues from +where it stopped with the first slice after the update. + A slice that moved nothing forward (no data type finished, no reading older than before) is a real failure: the job fails with the reason, pg-boss retries it a few times, and after that the next restart offers the backfill again. @@ -215,6 +247,13 @@ reading: `edgeFrom` and `edgeTo`, the oldest reading written before and after the slice. - `edges`: where every unfinished page-by-page data type stands now. + - `ended`: data types whose history ended short of its beginning, each + with its measurement `type`, the instant it ends at (`at`) and the + `reason` (`horizon` or `no_progress`). +- `meta["googleHealth.backfill.ended"]` appears on the slice where a data + type ended short, with the same entries. A walk stopped for lack of + progress also leaves a warning: + `fetchHeartRate: 20 pages in a row reached no further back than `. - `meta["googleHealth.backfill.edgeFilterUnsupported"]` appears on the slice where Google refused the upper-bound filter, listing the data types noted as refusing it. diff --git a/prisma/schema.prisma b/prisma/schema.prisma index b74c17864..6abeda554 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -4463,11 +4463,14 @@ model GoogleHealthConnection { /// repair pending (or the connection predates the fix). sleepRepairedAt DateTime? @map("sleep_repaired_at") /// How far the in-flight history backfill has got - /// (`{ done: string[], cursors: { [collection]: { at, order, pages } } }`): - /// the collections it walked to the end, and for each collection read page - /// by page the edge it has written up to, so the next slice resumes there - /// instead of starting over. Null once the backfill completes, and reset - /// on every reconnect. + /// (`{ done: string[], cursors: { [collection]: { at, order, pages } }, + /// ended?: { [collection]: { type, at, reason } } }`): the collections it + /// walked to the end, for each collection read page by page the edge it has + /// written up to, so the next slice resumes there instead of starting over, + /// and the collections it ended short of their history's start (the + /// horizon, or a walk that stopped getting anywhere). Once the backfill + /// completes only `ended` is kept, or null when nothing ended short; the + /// status routes publish it as `backfillEnded`. Reset on every reconnect. backfillProgress Json? @map("backfill_progress") /// One bounded, owner-scoped current sync run. This is replaced at the start /// of every run and never retains provider payloads or historical runs. diff --git a/src/lib/google-health/__tests__/backfill-bounds.test.ts b/src/lib/google-health/__tests__/backfill-bounds.test.ts new file mode 100644 index 000000000..2e3cbb6b7 --- /dev/null +++ b/src/lib/google-health/__tests__/backfill-bounds.test.ts @@ -0,0 +1,334 @@ +/** + * #1220 — a Google Health backfill is bounded by whether it gets anywhere and + * by a horizon in years, not by a count of the pages it has read. + * + * The count (ten thousand pages per data type, across every slice) ran out + * about two and a half years into a heart-rate history recorded every few + * seconds, never reset, and left the backfill stopping at the same place on + * every slice without ever being marked complete. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { prismaMock, safeFetchMock } = vi.hoisted(() => ({ + prismaMock: { googleHealthConnection: { update: vi.fn() } }, + safeFetchMock: vi.fn(), +})); +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/safe-fetch", async (importOriginal) => ({ + ...(await importOriginal()), + safeFetch: safeFetchMock, +})); +vi.mock("@/lib/logging/context", () => ({ + annotate: () => {}, + getEvent: () => null, +})); + +import { + GOOGLE_HEALTH_NO_PROGRESS_PAGES, + googleHealthBackfillHorizon, + googleHealthBackfillHorizonYears, + googleHealthWalkProgress, + noteGoogleHealthWalkPage, +} from "../backfill-bounds"; +import { + googleHealthBackfillEndedList, + googleHealthBackfillProgressAfterCompletion, + readGoogleHealthBackfillEnded, + recordGoogleHealthBackfillEnd, + runGoogleHealthCollection, + runWithGoogleHealthBackfillResume, +} from "../sync-core"; +import { + forEachDataPointPage, + GOOGLE_HEALTH_DATA_TYPES, + noteGoogleHealthDeferred, + runWithGoogleHealthClientOutcome, +} from "../client"; + +const MINUTE = 60_000; +const T0 = Date.parse("2026-10-17T12:00:00Z"); + +/** A newest-first page of `size` readings, `offset` minutes back from T0. */ +function descPage(offset: number, size = 1000): number[] { + return Array.from({ length: size }, (_, i) => T0 - (offset + i) * MINUTE); +} + +describe("the progress guard", () => { + it("never stops a walk that reaches further back on every page", () => { + // Fifty thousand pages: five times the old page count, still moving. + const progress = googleHealthWalkProgress(); + for (let page = 0; page < 50_000; page++) { + expect(noteGoogleHealthWalkPage(progress, descPage(page * 1000, 2))).toBe( + false, + ); + } + expect(progress.stalled).toBe(0); + expect(progress.order).toBe("desc"); + }); + + it("stops a walk that keeps getting the same page", () => { + const progress = googleHealthWalkProgress(); + const page = descPage(0, 50); + expect(noteGoogleHealthWalkPage(progress, page)).toBe(false); + for (let i = 1; i < GOOGLE_HEALTH_NO_PROGRESS_PAGES; i++) { + expect(noteGoogleHealthWalkPage(progress, page)).toBe(false); + } + expect(noteGoogleHealthWalkPage(progress, page)).toBe(true); + }); + + it("starts over counting once a page reaches further again", () => { + const progress = googleHealthWalkProgress(); + noteGoogleHealthWalkPage(progress, descPage(0, 10)); + for (let i = 0; i < GOOGLE_HEALTH_NO_PROGRESS_PAGES - 1; i++) { + noteGoogleHealthWalkPage(progress, descPage(0, 10)); + } + expect(noteGoogleHealthWalkPage(progress, descPage(10, 10))).toBe(false); + expect(progress.stalled).toBe(0); + }); + + it("counts a cycle of pages it has already seen as no progress", () => { + const progress = googleHealthWalkProgress(); + const pages = [descPage(0, 5), descPage(5, 5), descPage(10, 5)]; + for (const page of pages) noteGoogleHealthWalkPage(progress, page); + let stopped = false; + for (let i = 0; i < GOOGLE_HEALTH_NO_PROGRESS_PAGES && !stopped; i++) { + stopped = noteGoogleHealthWalkPage(progress, pages[i % 3]!); + } + expect(stopped).toBe(true); + }); + + it("measures a walk resumed at an edge from the edge", () => { + // Pages newer than the edge are what the walk already wrote. + const edge = new Date(T0 - 5000 * MINUTE); + const progress = googleHealthWalkProgress({ at: edge, order: "desc" }); + for (let i = 0; i < GOOGLE_HEALTH_NO_PROGRESS_PAGES - 1; i++) { + expect(noteGoogleHealthWalkPage(progress, descPage(i * 10, 10))).toBe( + false, + ); + } + expect(noteGoogleHealthWalkPage(progress, descPage(500, 10))).toBe(true); + }); + + it("counts a page without a readable time as no progress", () => { + const progress = googleHealthWalkProgress(); + noteGoogleHealthWalkPage(progress, descPage(0, 3)); + for (let i = 0; i < GOOGLE_HEALTH_NO_PROGRESS_PAGES - 1; i++) { + noteGoogleHealthWalkPage(progress, [null, null]); + } + expect(noteGoogleHealthWalkPage(progress, [])).toBe(true); + }); + + it("follows an oldest-first walk forward", () => { + const progress = googleHealthWalkProgress(); + for (let page = 0; page < 100; page++) { + const asc = Array.from( + { length: 10 }, + (_, i) => T0 + (page * 10 + i) * MINUTE, + ); + expect(noteGoogleHealthWalkPage(progress, asc)).toBe(false); + } + expect(progress.order).toBe("asc"); + }); +}); + +describe("the walk under the progress guard", () => { + const respond = (body: unknown) => + new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); + const reading = (t: number) => ({ + heartRate: { + beatsPerMinute: "60", + sampleTime: { physicalTime: new Date(t).toISOString() }, + }, + }); + const walk = async (onPage: (n: number) => void = () => {}) => { + const onNoProgress = vi.fn(); + const { outcome } = await runWithGoogleHealthClientOutcome(() => + forEachDataPointPage( + GOOGLE_HEALTH_DATA_TYPES.heartRate, + "token", + "fetchHeartRate", + { onNoProgress, maxPages: 10_000 }, + (points) => onPage(points.length), + ), + ); + return { outcome, onNoProgress }; + }; + + beforeEach(() => { + safeFetchMock.mockReset(); + }); + + it("is not stopped by a long history that keeps moving", async () => { + const pages = 3 * GOOGLE_HEALTH_NO_PROGRESS_PAGES; + safeFetchMock.mockImplementation(async (url: string) => { + const n = Number(new URL(url).searchParams.get("pageToken") ?? 0); + return respond({ + dataPoints: [reading(T0 - n * MINUTE)], + ...(n + 1 < pages ? { nextPageToken: String(n + 1) } : {}), + }); + }); + const { outcome, onNoProgress } = await walk(); + expect(safeFetchMock).toHaveBeenCalledTimes(pages); + expect(outcome.truncated).toBe(false); + expect(onNoProgress).not.toHaveBeenCalled(); + }); + + it("stops a walk that Google keeps answering with the same page", async () => { + safeFetchMock.mockImplementation(async () => + respond({ dataPoints: [reading(T0)], nextPageToken: "again" }), + ); + const written: number[] = []; + const { outcome, onNoProgress } = await walk((n) => written.push(n)); + expect(safeFetchMock).toHaveBeenCalledTimes( + 1 + GOOGLE_HEALTH_NO_PROGRESS_PAGES, + ); + // The page that tipped it over is not handed on. + expect(written).toHaveLength(GOOGLE_HEALTH_NO_PROGRESS_PAGES); + expect(outcome.truncated).toBe(true); + expect(onNoProgress).toHaveBeenCalledTimes(1); + }); + + it("ends as complete, not truncated, when the caller says it has read enough", async () => { + safeFetchMock.mockImplementation(async (url: string) => { + const n = Number(new URL(url).searchParams.get("pageToken") ?? 0); + return respond({ + dataPoints: [reading(T0 - n * MINUTE)], + nextPageToken: String(n + 1), + }); + }); + let pages = 0; + const { outcome } = await runWithGoogleHealthClientOutcome(() => + forEachDataPointPage( + GOOGLE_HEALTH_DATA_TYPES.heartRate, + "token", + "fetchHeartRate", + {}, + () => (++pages === 3 ? "end" : undefined), + ), + ); + expect(pages).toBe(3); + expect(outcome.truncated).toBe(false); + }); +}); + +describe("the backfill horizon", () => { + const saved = process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS; + afterEach(() => { + if (saved === undefined) + delete process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS; + else process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS = saved; + }); + + it("reads back twenty years unless the operator says otherwise", () => { + delete process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS; + expect(googleHealthBackfillHorizonYears()).toBe(20); + expect( + googleHealthBackfillHorizon( + new Date("2026-10-17T12:00:00Z"), + ).toISOString(), + ).toBe("2006-10-17T12:00:00.000Z"); + }); + + it("takes a whole number of years from 1 to 100", () => { + process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS = "5"; + expect(googleHealthBackfillHorizonYears()).toBe(5); + expect( + googleHealthBackfillHorizon( + new Date("2026-10-17T12:00:00Z"), + ).toISOString(), + ).toBe("2021-10-17T12:00:00.000Z"); + for (const bad of ["0", "101", "2.5", "-3", "ten", " "]) { + process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS = bad; + expect(googleHealthBackfillHorizonYears()).toBe(20); + } + }); +}); + +describe("a collection ended short", () => { + beforeEach(() => { + vi.clearAllMocks(); + prismaMock.googleHealthConnection.update.mockResolvedValue({}); + }); + + const END = { + type: "PULSE" as const, + at: "2024-05-11T00:00:00.000Z", + reason: "no_progress" as const, + }; + + it("counts as finished even though its walk ended truncated, and is noted with where and why", async () => { + const done = new Set(); + const { ended, collections } = await runWithGoogleHealthClientOutcome(() => + runWithGoogleHealthBackfillResume("u1", done, () => + runGoogleHealthCollection("fetchHeartRate", async () => { + // The walk stopped for no progress: truncated, and the end noted. + noteGoogleHealthDeferred(); + await recordGoogleHealthBackfillEnd("fetchHeartRate", END); + return 0; + }), + ), + ).then((r) => r.result); + expect(collections).toEqual([ + expect.objectContaining({ key: "fetchHeartRate", status: "done" }), + ]); + expect([...ended]).toEqual([["fetchHeartRate", END]]); + const written = prismaMock.googleHealthConnection.update.mock.calls.at( + -1, + )![0].data.backfillProgress as unknown; + expect( + readGoogleHealthBackfillEnded(written).get("fetchHeartRate"), + ).toEqual(END); + }); + + it("is skipped by the next slice instead of walked into the same wall", async () => { + const walk = vi.fn(async () => 1); + const { skipped } = await runWithGoogleHealthBackfillResume( + "u1", + new Set(), + () => runGoogleHealthCollection("fetchHeartRate", walk), + new Map(), + new Set(), + new Map([["fetchHeartRate", END]]), + ); + expect(walk).not.toHaveBeenCalled(); + expect(skipped).toEqual(["fetchHeartRate"]); + }); + + it("outlives the completion stamp, and nothing else does", () => { + expect(googleHealthBackfillProgressAfterCompletion(new Map())).toBeNull(); + expect( + googleHealthBackfillProgressAfterCompletion( + new Map([["fetchHeartRate", END]]), + ), + ).toEqual({ ended: { fetchHeartRate: END } }); + }); + + it("is published per type, and a malformed note is dropped", () => { + expect( + googleHealthBackfillEndedList({ + ended: { + fetchHeartRate: END, + fetchWeight: { + type: "WEIGHT", + at: "2006-10-17T12:00:00.000Z", + reason: "horizon", + }, + fetchBodyFat: { type: "BODY_FAT", at: "nope", reason: "horizon" }, + fetchSpo2: { type: "NOT_A_TYPE", at: END.at, reason: "horizon" }, + fetchHrv: { type: "PULSE", at: END.at, reason: "because" }, + }, + }), + ).toEqual([ + { type: "PULSE", endsAt: END.at, reason: "no_progress" }, + { + type: "WEIGHT", + endsAt: "2006-10-17T12:00:00.000Z", + reason: "horizon", + }, + ]); + expect(googleHealthBackfillEndedList(null)).toEqual([]); + }); +}); diff --git a/src/lib/google-health/__tests__/edge-filter-fallback.test.ts b/src/lib/google-health/__tests__/edge-filter-fallback.test.ts index af018c13e..69a8cfc7e 100644 --- a/src/lib/google-health/__tests__/edge-filter-fallback.test.ts +++ b/src/lib/google-health/__tests__/edge-filter-fallback.test.ts @@ -42,6 +42,7 @@ import { runWithGoogleHealthStop, } from "../client"; import { GoogleHealthApiError } from "../response-classifier"; +import { GOOGLE_HEALTH_NO_PROGRESS_PAGES } from "../backfill-bounds"; import { googleHealthBackfillCursor, noteGoogleHealthEdgeFilterUnsupported, @@ -479,14 +480,42 @@ describe("passing over the samples beyond the edge inside a slice", () => { expect(outcome.truncated).toBe(true); }); - it("gives up past its own ceiling instead of reading without end", async () => { + it("stops passing over once the pages it reads stop reaching further back", async () => { + // A list that hands out its newest page whatever token it is asked with: + // passing over it would never reach the edge. There is no page count to + // run into any more; the walk stops because it gets nowhere. + safeFetchMock.mockImplementation(async (url: string) => { + const u = new URL(url); + u.searchParams.delete("pageToken"); + const res = respond(u.toString()); + const body = (await res.json()) as object; + return json({ ...body, nextPageToken: "1" }); + }); + const onNoProgress = vi.fn(); const { pages, outcome } = await walk({ resumable: true, - edge: { at: at(30), order: "desc", filter: false, maxSkipPages: 2 }, + edge: { at: at(30), order: "desc", filter: false }, + onNoProgress, }); expect(pages).toHaveLength(0); - expect(safeFetchMock).toHaveBeenCalledTimes(2); + // The first page reaches somewhere; the twenty after it repeat it. + expect(safeFetchMock).toHaveBeenCalledTimes( + 1 + GOOGLE_HEALTH_NO_PROGRESS_PAGES, + ); expect(outcome.truncated).toBe(true); + expect(onNoProgress).toHaveBeenCalledTimes(1); + }); + + it("passes over as many pages as it takes while they move toward the edge", async () => { + // One reading a page, so reaching the edge takes more pages than the + // no-progress limit and more than any page count a walk once had to fit. + const { rows, outcome } = await walk({ + resumable: true, + pageSize: 1, + edge: { at: at(30), order: "desc", filter: false }, + }); + expect(rows).toHaveLength(TOTAL - 30); + expect(outcome.truncated).toBe(false); }); }); diff --git a/src/lib/google-health/backfill-bounds.ts b/src/lib/google-health/backfill-bounds.ts new file mode 100644 index 000000000..695be1e95 --- /dev/null +++ b/src/lib/google-health/backfill-bounds.ts @@ -0,0 +1,143 @@ +/** + * How far a Google Health list walk is allowed to go, expressed in what the + * walk actually reaches rather than in how many pages it has read (#1220). + * + * Two bounds: + * + * - **Progress.** A walk that keeps reading pages but no longer reaches any + * further back in time (Google handing out the same page again, or a + * page token that cycles) is a runaway. It is stopped after + * `GOOGLE_HEALTH_NO_PROGRESS_PAGES` such pages in a row. A walk that + * moves to older readings on every page is never stopped by this, + * however long the history is. + * - **Horizon.** A history backfill reads back at most + * `GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS` years (twenty unless the + * operator sets the variable), so even a walk that does progress has an + * end that does not depend on how densely the device records. + * + * Before, a backfill counted every page a data type had read across all of + * its slices and stopped at ten thousand. A watch that records heart rate + * every few seconds fills some four thousand pages a year, so the count ran + * out about two and a half years back, and since it never reset, every later + * slice stopped at the same place and the backfill was never marked complete. + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { envValue } from "@/lib/env"; + +/** + * Pages in a row that may fail to reach past the furthest reading the walk + * has seen before the walk is taken for a runaway and stopped. A real + * history moves on every page (a page is up to a thousand readings), so this + * only ever counts pages that repeat what came before. + */ +export const GOOGLE_HEALTH_NO_PROGRESS_PAGES = 20; + +/** The backfill horizon when `GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS` is unset. */ +const DEFAULT_HORIZON_YEARS = 20; +const MAX_HORIZON_YEARS = 100; + +/** + * How many years back a history backfill reads: the operator's + * `GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS` when it is a whole number from 1 to + * 100, else twenty. + */ +export function googleHealthBackfillHorizonYears(): number { + const raw = envValue("GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS"); + if (raw === undefined || !/^\d+$/.test(raw)) return DEFAULT_HORIZON_YEARS; + const years = Number.parseInt(raw, 10); + return years >= 1 && years <= MAX_HORIZON_YEARS + ? years + : DEFAULT_HORIZON_YEARS; +} + +/** The oldest instant a backfill started at `now` reads back to. */ +export function googleHealthBackfillHorizon(now: Date = new Date()): Date { + const horizon = new Date(now); + horizon.setUTCFullYear( + horizon.getUTCFullYear() - googleHealthBackfillHorizonYears(), + ); + return horizon; +} + +/** + * Where one walk has got to in time. `order` is the order Google returns the + * readings in (newest first unless a page shows otherwise); `furthest` is the + * oldest reading seen so far on a newest-first walk, the newest on an + * oldest-first one; `stalled` counts the pages in a row that did not reach + * past it. + */ +export interface GoogleHealthWalkProgress { + order: "desc" | "asc" | null; + furthest: number | null; + stalled: number; +} + +/** + * A fresh progress note. A walk that resumes from an edge with the edge as a + * filter starts from it, so a page that only repeats readings at or beyond + * the edge does not count as getting anywhere. + */ +export function googleHealthWalkProgress( + from?: { at: Date; order: "desc" | "asc" } | null, +): GoogleHealthWalkProgress { + return from + ? { order: from.order, furthest: from.at.getTime(), stalled: 0 } + : { order: null, furthest: null, stalled: 0 }; +} + +/** + * Note one page, given the sample instants of its readings in the order + * Google returned them (null for a reading without a readable time). Returns + * true once the walk has gone `GOOGLE_HEALTH_NO_PROGRESS_PAGES` pages in a + * row without reaching further. A page with no readable instant at all + * reaches nowhere and counts toward the limit. + */ +export function noteGoogleHealthWalkPage( + progress: GoogleHealthWalkProgress, + instants: ReadonlyArray, +): boolean { + let first: number | null = null; + let last: number | null = null; + let min = Infinity; + let max = -Infinity; + for (const t of instants) { + if (t === null) continue; + first ??= t; + last = t; + if (t < min) min = t; + if (t > max) max = t; + } + if (progress.order === null && first !== null && last !== null) { + if (first !== last) progress.order = first < last ? "asc" : "desc"; + } + const order = progress.order ?? "desc"; + const far = first === null ? null : order === "desc" ? min : max; + const moved = + far !== null && + (progress.furthest === null || + (order === "desc" ? far < progress.furthest : far > progress.furthest)); + if (moved) { + progress.furthest = far; + progress.stalled = 0; + } else { + progress.stalled += 1; + } + return progress.stalled >= GOOGLE_HEALTH_NO_PROGRESS_PAGES; +} + +/** + * Why a backfill stopped reading a data type before the end of its history: + * it reached the horizon, or Google stopped handing out older readings. + */ +export type GoogleHealthBackfillEndReason = "horizon" | "no_progress"; + +/** + * A data type whose backfill ended short of the history's beginning, and + * where: the oldest instant the import reaches back to. + */ +export interface GoogleHealthBackfillEnd { + /** The measurement type whose history this is. */ + type: MeasurementType; + at: string; + reason: GoogleHealthBackfillEndReason; +} diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index 2cbf02b73..fcf70ecd5 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -50,6 +50,11 @@ import { type GoogleHealthRollupPoint, } from "./mappers"; import { envValue } from "@/lib/env"; +import { + GOOGLE_HEALTH_NO_PROGRESS_PAGES, + googleHealthWalkProgress, + noteGoogleHealthWalkPage, +} from "./backfill-bounds"; export * from "./mappers"; @@ -700,16 +705,15 @@ export const GOOGLE_HEALTH_PAGE_SIZE = 1000; export const GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE = 25; /** - * Page ceiling for the intraday heart-rate walk. + * Page ceiling for one intraday heart-rate walk. * - * The default of 1000 pages is a million points, which a watch that records - * a reading a minute passes in under two years. A walk that hits the ceiling - * reports `truncated`, the backfill treats that as incomplete and throws, and - * pg-boss retries it — so an account with a longer history re-ran the whole - * walk on every retry and every boot and never completed. Ten thousand pages - * is nineteen years at that rate: still a bound on a runaway cursor, no longer - * a bound a real history reaches. The walk is paged (`forEachDataPointPage`), - * so the ceiling costs time, not memory. + * Per walk, never across walks: a backfill reads at most a slice's pages per + * walk and keeps its place between slices, and an hourly sync reads a few + * pages. What stops a walk that keeps going without getting anywhere is the + * progress guard (`GOOGLE_HEALTH_NO_PROGRESS_PAGES`), and what bounds a + * backfill that does get somewhere is its horizon in years (see + * `backfill-bounds.ts`). The walk is paged (`forEachDataPointPage`), so the + * ceiling costs time, not memory. */ export const GOOGLE_HEALTH_DENSE_MAX_PAGES = 10_000; @@ -758,8 +762,9 @@ 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 far into - * the unbounded list the walk may get before it reaches the edge. + * time the way the edge was recorded. Passing over has no page bound of + * its own: the pages passed over move toward the edge, and a walk whose + * pages stop moving is stopped by the progress guard like any other. * * A walk without the bound keeps its place in that list across slices: * `resumeFrom` is where an earlier slice's walk stopped, and @@ -772,10 +777,16 @@ interface DataPointQuery { filter?: boolean; tz?: string; onFilterUnsupported?: () => void | Promise; - maxSkipPages?: number; resumeFrom?: GoogleHealthResumePoint; onResumePoint?: (point: GoogleHealthResumePoint) => void | Promise; }; + /** + * Told once when the walk is stopped for going + * `GOOGLE_HEALTH_NO_PROGRESS_PAGES` pages in a row without reaching any + * further back (see `backfill-bounds.ts`). The walk then reports itself + * truncated, as it does on its page ceiling. + */ + onNoProgress?: () => void | Promise; } /** @@ -1229,13 +1240,24 @@ export async function fetchDataPoints( * applies back-pressure to the fetch. The date-filter fallback still works: * it only fires when the first request is rejected, which is before any page * reaches `onPage`. + * + * `onPage` answers `"end"` when the caller has read as far as it wants (a + * backfill that reached its horizon): the walk ends there as complete, not + * truncated. + * + * A walk over spot readings (`timeField: "sample"`) also watches whether it + * is getting anywhere: `GOOGLE_HEALTH_NO_PROGRESS_PAGES` pages in a row that + * reach no further back than the pages before them end it, truncated, and + * tell `onNoProgress` (see `backfill-bounds.ts`). */ export async function forEachDataPointPage( dataType: GoogleHealthDataType, accessToken: GoogleHealthAccessToken, verb: string, query: DataPointQuery, - onPage: (points: GoogleHealthDataPoint[]) => void | Promise, + onPage: ( + points: GoogleHealthDataPoint[], + ) => void | "end" | Promise, ): Promise { const maxPages = query.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES; const pageSize = query.pageSize ?? GOOGLE_HEALTH_PAGE_SIZE; @@ -1267,7 +1289,15 @@ export async function forEachDataPointPage( let skipped = 0; const startPosition = resumeFrom?.page ?? 0; const skipStartedAt = skipTo ? sliceNow() : 0; - const maxSkip = skipTo?.maxSkipPages ?? GOOGLE_HEALTH_DENSE_MAX_PAGES; + // Whether the walk is still reaching further back. A walk filtered to + // the edge starts from it; one passing over the samples beyond the edge + // starts from nothing, since those pages move toward the edge too. + const progress = + dataType.timeField === "sample" + ? googleHealthWalkProgress(edge && edgeBy === "filter" ? edge : null) + : null; + const progressTz = edge?.tz ?? query.tz; + let noProgress = false; const sampleAt = (point: GoogleHealthDataPoint): number | null => googleHealthSampleInstant(point, dataType, skipTo?.tz)?.getTime() ?? null; const beyondEdge = (point: GoogleHealthDataPoint): boolean | null => { @@ -1296,13 +1326,6 @@ export async function forEachDataPointPage( 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 && @@ -1419,6 +1442,24 @@ export async function forEachDataPointPage( pageCount += 1; fetched += page.length; noteSlicePage(); + if ( + progress && + noteGoogleHealthWalkPage( + progress, + page.map( + (p) => + googleHealthSampleInstant(p, dataType, progressTz)?.getTime() ?? + null, + ), + ) + ) { + getEvent()?.addWarning( + `google-health ${verb}: ${GOOGLE_HEALTH_NO_PROGRESS_PAGES} pages in a row reached no further back than ${progress.furthest === null ? "the first" : new Date(progress.furthest).toISOString()}; walk stopped`, + ); + noProgress = true; + stopped = true; + break; + } 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 @@ -1433,7 +1474,10 @@ export async function forEachDataPointPage( if (restarted) extendSliceDeadline(sliceNow() - skipStartedAt); page = page.filter((p) => beyondEdge(p) !== true); } - await onPage(page); + if ((await onPage(page)) === "end") { + pageToken = null; + break; + } } while (pageToken && pageCount - skipped < maxPages); const truncated = stopped || Boolean(pageToken); @@ -1443,6 +1487,7 @@ export async function forEachDataPointPage( await skipTo.onResumePoint?.({ token: lastToken, page: lastPosition }); } notePagination(pageCount, fetched, truncated); + if (noProgress) await query.onNoProgress?.(); }; // The fallback only exists for filtered daily-summary reads — every other diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index 4940e43ea..19dc1d523 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -18,9 +18,11 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; +import { MeasurementType as MeasurementTypes } from "@/generated/prisma/enums"; import { encrypt, decrypt } from "@/lib/crypto"; import { annotate, getEvent } from "@/lib/logging/context"; import type { GoogleHealthCollectionKey } from "./collection-keys"; +import type { GoogleHealthBackfillEnd } from "./backfill-bounds"; import { userDayKey } from "@/lib/tz/format"; import { isP2002 } from "@/lib/prisma-errors"; import { @@ -440,7 +442,14 @@ export interface GoogleHealthBackfillCursor { at: string; /** The order the walk returned samples in: newest first, or oldest first. */ order: "desc" | "asc"; - /** Pages this collection has read across every attempt so far. */ + /** + * Pages this collection has written across every attempt so far. Kept for + * the record only: it bounds nothing. It once capped the walk at ten + * thousand pages, which a dense heart-rate history passes in under three + * years, and since it never reset the backfill stopped there for good + * (#1220). The walk is bounded by its progress and the backfill horizon + * instead (`backfill-bounds.ts`). + */ pages: number; /** * Only for a collection read without the edge bound: where its last walk @@ -480,6 +489,8 @@ interface BackfillResumeTracker { collections: GoogleHealthBackfillCollectionReport[]; /** Collections whose resumed request Google refused the edge filter on. */ edgeFilterUnsupported: Set; + /** Collections whose walk ended short of the history's start, and where. */ + ended: Map; } const backfillResumeStorage = new AsyncLocalStorage(); @@ -536,6 +547,79 @@ export function readGoogleHealthBackfillCursors( return out; } +/** + * Read the `backfillProgress` column into the collections whose backfill + * ended short of the start of their history (the horizon, or a walk that + * stopped getting anywhere), each with where it ended and why. Survives the + * completion stamp, so the integration card can say where history ends. + */ +export function readGoogleHealthBackfillEnded( + value: unknown, +): Map { + const out = new Map(); + const ended = progressRecord(progressRecord(value).ended); + for (const [key, raw] of Object.entries(ended)) { + if (key.length > 64) continue; + const e = progressRecord(raw); + const at = typeof e.at === "string" ? e.at : null; + if (!at || Number.isNaN(Date.parse(at))) continue; + if (e.reason !== "horizon" && e.reason !== "no_progress") continue; + if (typeof e.type !== "string" || !(e.type in MeasurementTypes)) { + continue; + } + out.set(key, { + type: e.type as MeasurementType, + at: new Date(at).toISOString(), + reason: e.reason, + }); + } + return out; +} + +/** + * Where each data type's imported history ends short of its beginning, and + * why, as the status routes publish it (`backfillEnded`), by type. + */ +export interface GoogleHealthBackfillEndedEntry { + type: MeasurementType; + /** The oldest instant the imported history reaches back to. */ + endsAt: string; + reason: GoogleHealthBackfillEnd["reason"]; +} + +/** The `backfillProgress` column's ended collections, as published. */ +export function googleHealthBackfillEndedList( + value: unknown, +): GoogleHealthBackfillEndedEntry[] { + return [...readGoogleHealthBackfillEnded(value).values()] + .map((e) => ({ type: e.type, endsAt: e.at, reason: e.reason })) + .sort((a, b) => a.type.localeCompare(b.type)); +} + +/** + * The `backfillProgress` value to keep once the backfill is stamped complete: + * only where collections ended short, or nothing at all. + */ +export function googleHealthBackfillProgressAfterCompletion( + ended: Map, +): { + ended: Record; +} | null { + return ended.size > 0 ? { ended: endedJson(ended) } : null; +} + +/** The ended collections as the column stores them. */ +function endedJson( + ended: Map, +): Record { + return Object.fromEntries( + [...ended].map(([key, e]) => [ + key, + { type: e.type, at: e.at, reason: e.reason }, + ]), + ); +} + /** * Read the `backfillProgress` column into the collections Google refused the * edge filter on. @@ -564,10 +648,12 @@ async function persistBackfillProgress( done: Set; cursors: Map; edgeFilterUnsupported?: Set; + ended?: Map; }, ): Promise { const unsupported = next.edgeFilterUnsupported ?? tracker.edgeFilterUnsupported; + const ended = next.ended ?? tracker.ended; await prisma.googleHealthConnection.update({ where: { userId: tracker.userId }, data: { @@ -589,12 +675,16 @@ async function persistBackfillProgress( ...(unsupported.size > 0 ? { edgeFilterUnsupported: [...unsupported] } : {}), + ...(ended.size > 0 ? { ended: endedJson(ended) } : {}), }, }, }); if (unsupported !== tracker.edgeFilterUnsupported) { for (const key of unsupported) tracker.edgeFilterUnsupported.add(key); } + if (ended !== tracker.ended) { + for (const [key, e] of ended) tracker.ended.set(key, e); + } // `next` may be the tracker's own collections: copy before clearing. const done = [...next.done]; const cursors = [...next.cursors]; @@ -611,6 +701,8 @@ export interface GoogleHealthBackfillAttempt { /** Collections whose edge moved, each with the edge it started from. */ advanced: Map; collections: GoogleHealthBackfillCollectionReport[]; + /** Every collection ended short so far, this attempt's and earlier ones. */ + ended: Map; } /** @@ -625,6 +717,7 @@ export async function runWithGoogleHealthBackfillResume( fn: () => Promise, cursors: Map = new Map(), edgeFilterUnsupported: Set = new Set(), + ended: Map = new Map(), ): Promise> { const tracker: BackfillResumeTracker = { userId, @@ -634,6 +727,7 @@ export async function runWithGoogleHealthBackfillResume( advanced: new Map(), collections: [], edgeFilterUnsupported, + ended, }; const result = await backfillResumeStorage.run(tracker, fn); return { @@ -641,15 +735,16 @@ export async function runWithGoogleHealthBackfillResume( skipped: tracker.skipped, advanced: tracker.advanced, collections: tracker.collections, + ended: tracker.ended, }; } /** * Where a resumable collection's walk starts in this backfill attempt: the - * edge an earlier attempt recorded (if any), whether Google may be asked for - * it with a filter (false once Google refused that filter for the - * collection), and the pages it has already read. Null outside a backfill, - * where every walk reads its whole window. + * edge an earlier attempt recorded (if any), and whether Google may be asked + * for it with a filter (false once Google refused that filter for the + * collection). Null outside a backfill, where every walk reads its whole + * window. */ export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): { edge?: { @@ -658,7 +753,6 @@ export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): { filter: boolean; resumeFrom?: GoogleHealthResumePoint; }; - pagesSoFar: number; } | null { const resume = backfillResumeStorage.getStore(); if (!resume) return null; @@ -679,9 +773,44 @@ export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): { } : {}), }, - pagesSoFar: cursor.pages, } - : { pagesSoFar: 0 }; + : {}; +} + +/** + * Record that a collection's backfill ended short of the start of its + * history, at `at`, because of `reason` (see `backfill-bounds.ts`). The + * collection then counts as finished for this backfill + * (`runGoogleHealthCollection`), and the note outlives the completion stamp + * so the card can say where the history ends. Outside a backfill this does + * nothing. + */ +export async function recordGoogleHealthBackfillEnd( + key: GoogleHealthCollectionKey, + end: GoogleHealthBackfillEnd, +): Promise { + const resume = backfillResumeStorage.getStore(); + if (!resume) return; + const ended = new Map(resume.ended); + ended.set(key, end); + annotate({ + meta: { "googleHealth.backfill.ended": Object.fromEntries(ended) }, + }); + try { + await persistBackfillProgress(resume, { + done: resume.done, + cursors: resume.cursors, + ended, + }); + } catch (err) { + // Not ended, then: the next slice walks it again and meets the same end. + getEvent()?.addWarning( + `google-health: backfill progress write failed for ${key}: ${err}`, + ); + annotate({ + meta: { "googleHealth.backfill.progressWriteFailed": key }, + }); + } } /** @@ -845,7 +974,7 @@ export async function runGoogleHealthCollection( ): Promise { const resume = backfillResumeStorage.getStore(); if (!resume) return fn(); - if (resume.done.has(key)) { + if (resume.done.has(key) || resume.ended.has(key)) { resume.skipped.push(key); return 0; } @@ -870,7 +999,11 @@ export async function runGoogleHealthCollection( (hardFailStorage.getStore()?.failures.length ?? 0) !== hardBefore || (softSkipStorage.getStore()?.count ?? 0) !== softBefore; const truncated = truncatedBefore || googleHealthOutcomeTruncated(); - const clean = !failedHard && !truncated; + // A walk that ended short on purpose (the horizon, or no progress) has + // read all it is going to: finished, though the walk may report itself + // truncated. + const endedShort = resume.ended.has(key); + const clean = !failedHard && (endedShort || !truncated); let finished = false; if (clean) { // The collection counts as finished only once the note says so. The job diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index 743cccf6c..2d832f6c3 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -29,6 +29,7 @@ import { GOOGLE_HEALTH_DATA_TYPES, GOOGLE_HEALTH_DEFAULT_MAX_PAGES, GOOGLE_HEALTH_DENSE_MAX_PAGES, + type GoogleHealthDataPoint, type GoogleHealthDataType, type GoogleHealthMappedMeasurement, type GoogleHealthTokenSource, @@ -46,8 +47,8 @@ import { mapRestingHeartRate, mapWeight, mapWristTemperature, - noteGoogleHealthDeferred, } from "./client"; +import { googleHealthBackfillHorizon } from "./backfill-bounds"; import { getValidToken, googleHealthBackfillCursor, @@ -55,6 +56,7 @@ import { googleHealthTokenSource, handleCollectionFetchError, noteGoogleHealthEdgeFilterUnsupported, + recordGoogleHealthBackfillEnd, recordGoogleHealthBackfillPage, recordGoogleHealthBackfillResumePoint, runGoogleHealthCollection, @@ -65,6 +67,7 @@ import { import { getEvent } from "@/lib/logging/context"; import type { GoogleHealthCollectionKey } from "./collection-keys"; import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; import { resolveUserTimezone } from "@/lib/tz/resolver"; /** One mappable metric: its data-type encoding + the per-point mapper + a verb. */ @@ -75,6 +78,12 @@ interface MetricResource { tz: string, ) => GoogleHealthMappedMeasurement[]; verb: GoogleHealthCollectionKey; + /** + * The measurement type the collection's readings land as (heart rate's + * main one for the intraday stream), which names its history when a + * backfill ends it short. + */ + type: MeasurementType; /** Page ceiling for the walk; the client default when omitted. */ maxPages?: number; /** @@ -90,41 +99,49 @@ const METRIC_RESOURCES: MetricResource[] = [ dataType: GOOGLE_HEALTH_DATA_TYPES.weight, map: mapWeight, verb: "fetchWeight", + type: "WEIGHT", }, { dataType: GOOGLE_HEALTH_DATA_TYPES.bodyFat, map: mapBodyFat, verb: "fetchBodyFat", + type: "BODY_FAT", }, { dataType: GOOGLE_HEALTH_DATA_TYPES.oxygenSaturation, map: mapOxygenSaturation, verb: "fetchSpo2", + type: "OXYGEN_SATURATION", }, { dataType: GOOGLE_HEALTH_DATA_TYPES.heartRateVariability, map: mapHeartRateVariability, verb: "fetchHrv", + type: "HEART_RATE_VARIABILITY", }, { dataType: GOOGLE_HEALTH_DATA_TYPES.restingHeartRate, map: mapRestingHeartRate, verb: "fetchRhr", + type: "RESTING_HEART_RATE", }, { dataType: GOOGLE_HEALTH_DATA_TYPES.respiratoryRate, map: mapRespiratoryRate, verb: "fetchRespiratoryRate", + type: "RESPIRATORY_RATE", }, { dataType: GOOGLE_HEALTH_DATA_TYPES.bloodGlucose, map: mapBloodGlucose, verb: "fetchBloodGlucose", + type: "BLOOD_GLUCOSE", }, { dataType: GOOGLE_HEALTH_DATA_TYPES.coreBodyTemperature, map: mapCoreBodyTemperature, verb: "fetchCoreBodyTemperature", + type: "BODY_TEMPERATURE", }, // Nightly sleep skin temperature — the documented `nightlyTemperatureCelsius` // is an absolute reading (not the once-assumed signed deviation), so it lands @@ -133,6 +150,7 @@ const METRIC_RESOURCES: MetricResource[] = [ dataType: GOOGLE_HEALTH_DATA_TYPES.sleepTemperature, map: mapWristTemperature, verb: "fetchSleepTemperature", + type: "WRIST_TEMPERATURE", }, // Last: the dense stream. A backfill finishes every light collection in // its first slice and then works through this one slice by slice; ahead @@ -141,6 +159,7 @@ const METRIC_RESOURCES: MetricResource[] = [ dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate, map: mapHeartRate, verb: "fetchHeartRate", + type: "PULSE", maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES, intraday: true, }, @@ -154,14 +173,6 @@ function isResumable(resource: MetricResource): boolean { return resource.dataType.timeField === "sample"; } -/** - * The page ceilings of every resumable collection, summed: the most pages a - * backfill can ever spend on them across all of its attempts. - */ -export const GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING = METRIC_RESOURCES.filter( - isResumable, -).reduce((sum, r) => sum + (r.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES), 0); - export async function syncUserMetrics( userId: string, opts: GoogleHealthResourceSyncOptions = {}, @@ -274,20 +285,30 @@ async function syncMetricCollection( let imported = 0; // Inside a backfill a spot-reading collection keeps its place: it starts // from the edge an earlier attempt reached and records its new edge after - // every page it writes (see `GoogleHealthBackfillCursor`). Its page ceiling - // then counts across attempts, so a cursor that never ends is still bounded. + // every page it writes (see `GoogleHealthBackfillCursor`). Its walk is + // bounded per attempt by the slice, across attempts by the edge having to + // move back on every page and by the horizon: readings older than it are + // not read, and the collection ends there (see `backfill-bounds.ts`). const cursor = isResumable(resource) ? googleHealthBackfillCursor(resource.verb) : null; - const ceiling = resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES; - const maxPages = ceiling - (cursor?.pagesSoFar ?? 0); - if (cursor && maxPages <= 0) { - getEvent()?.addWarning( - `google-health ${resource.verb}: backfill reached its page ceiling of ${resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES} pages for ${userId}`, - ); - noteGoogleHealthDeferred(); - return 0; - } + const maxPages = resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES; + const horizon = cursor ? googleHealthBackfillHorizon() : null; + // The order the walk returns readings in: the one its edge has recorded, + // else the page's own (as `recordGoogleHealthBackfillPage` reads it), + // newest first when a page cannot tell. + const walkOrder = (page: GoogleHealthDataPoint[]): "desc" | "asc" => { + const known = googleHealthBackfillCursor(resource.verb)?.edge?.order; + if (known) return known; + const instants = page + .map((p) => + googleHealthSampleInstant(p, resource.dataType, tz)?.getTime(), + ) + .filter((t): t is number => t !== undefined); + return instants.length > 1 && instants[0]! < instants.at(-1)! + ? "asc" + : "desc"; + }; // Once one page's write fails, the edge stays where it is for the rest of // the walk: moving it past the failed page would leave that page's rows // behind it for good. @@ -305,12 +326,22 @@ async function syncMetricCollection( ...(cursor ? { resumable: true, + // Google keeps handing out pages that reach no further back: + // the history ends where the walk got to, and the backfill + // goes on without it instead of meeting the same wall again. + onNoProgress: async () => { + // A page whose write failed fails the collection instead. + if (edgeHeld) return; + const at = googleHealthBackfillCursor(resource.verb)?.edge?.at; + await recordGoogleHealthBackfillEnd(resource.verb, { + type: resource.type, + at: (at ?? new Date()).toISOString(), + reason: "no_progress", + }); + }, edge: cursor.edge && { ...cursor.edge, tz, - // Without the bound the walk passes over what earlier slices - // wrote, and they wrote no more than the collection's ceiling. - maxSkipPages: ceiling, onFilterUnsupported: () => noteGoogleHealthEdgeFilterUnsupported(resource.verb), // A place past a page whose write failed would leave that @@ -326,7 +357,21 @@ async function syncMetricCollection( } : {}), }, - async (points) => { + async (page) => { + // Past the horizon nothing is written; on a newest-first walk + // everything after it is older still, so the walk ends there. An + // oldest-first walk starts beyond the horizon and works toward it, + // so it is read as before: dropping its first pages would write + // nothing and move no edge, and the slice would fail as stuck. + let points = page; + let pastHorizon = false; + if (horizon && walkOrder(page) === "desc") { + points = page.filter((p) => { + const t = googleHealthSampleInstant(p, resource.dataType, tz); + return t === null || t.getTime() >= horizon.getTime(); + }); + pastHorizon = points.length < page.length; + } const readings: GoogleHealthMeasurementUpsert[] = []; for (const point of points) { for (const m of resource.map(point, tz)) { @@ -345,17 +390,30 @@ async function syncMetricCollection( deferRollup: opts.deferRollup, }) ).imported; - if (!cursor || points.length === 0) return; + if (!cursor) return; if (googleHealthHardFailureCount() !== failuresBefore) edgeHeld = true; - if (edgeHeld) return; - await recordGoogleHealthBackfillPage(resource.verb, { - first: googleHealthSampleInstant(points[0]!, resource.dataType, tz), - last: googleHealthSampleInstant( - points[points.length - 1]!, - resource.dataType, - tz, - ), - }); + if (points.length > 0 && !edgeHeld) { + await recordGoogleHealthBackfillPage(resource.verb, { + first: googleHealthSampleInstant(points[0]!, resource.dataType, tz), + last: googleHealthSampleInstant( + points[points.length - 1]!, + resource.dataType, + tz, + ), + }); + } + if (pastHorizon) { + // With a page's write failed the collection is not finished + // either way; it is walked again next slice. + if (!edgeHeld) { + await recordGoogleHealthBackfillEnd(resource.verb, { + type: resource.type, + at: horizon!.toISOString(), + reason: "horizon", + }); + } + return "end"; + } }, ); } catch (err) { diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index b9fb2c788..390c0b603 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -37,6 +37,8 @@ const { /** Collections whose resume edge the mocked walk moves. */ advances: [] as string[], collections: [] as Array<{ key: string; status: string }>, + /** Collections the mocked walk ends short (horizon / no progress). */ + ends: [] as string[], }, })); @@ -62,18 +64,32 @@ vi.mock("@/lib/google-health/sync-core", () => ({ readGoogleHealthBackfillDone: () => new Set(resumeState.doneBefore), readGoogleHealthBackfillCursors: () => new Map(), readGoogleHealthBackfillEdgeFilterUnsupported: () => new Set(), + readGoogleHealthBackfillEnded: () => new Map(), + googleHealthBackfillProgressAfterCompletion: (ended: Map) => + ended.size > 0 ? { ended: Object.fromEntries(ended) } : null, runWithGoogleHealthBackfillResume: async ( _userId: string, done: Set, fn: () => Promise, + _cursors: unknown, + _unsupported: unknown, + ended: Map, ) => { const result = await fn(); for (const key of resumeState.finishes) done.add(key); + for (const key of resumeState.ends) { + ended.set(key, { + type: "PULSE", + at: "2024-05-11T00:00:00.000Z", + reason: "no_progress", + }); + } return { result, skipped: [], advanced: new Map(resumeState.advances.map((key) => [key, null])), collections: resumeState.collections, + ended, }; }, })); @@ -87,7 +103,6 @@ import { GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS, runGoogleHealthBackfillForUser, } from "../google-health-backfill"; -import { GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING } from "@/lib/google-health/sync-metrics"; import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission"; import { GOOGLE_HEALTH_DATA_TYPES, @@ -105,6 +120,7 @@ beforeEach(() => { resumeState.doneBefore = []; resumeState.advances = []; resumeState.collections = []; + resumeState.ends = []; }); describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { @@ -279,14 +295,40 @@ describe("runGoogleHealthBackfillForUser — slices", () => { expect(bossSend).not.toHaveBeenCalled(); }); - it("derives the chain's ceiling from the work a backfill can do", () => { - // Every collection once, plus every page the resumable collections may - // 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 + 2 * GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING + 2, - ); - expect(GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING).toBe(10_000 + 4 * 1000); + it("bounds the chain in time, not in pages", () => { + // Every collection once, sixty days of five-minute slices back to back, + // and two of slack. No page count: a dense history must not run into it + // (#1220). + expect(GOOGLE_HEALTH_BACKFILL_MAX_SLICE).toBe(17 + (60 * 24 * 60) / 5 + 2); + }); + + it("continues a slice that only ended a collection short", async () => { + resumeState.ends = ["fetchHeartRate"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: true }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 4), + ).resolves.toEqual({ imported: 0, continued: true }); + expect(bossSend).toHaveBeenCalledTimes(1); + }); + + it("keeps where a collection ended when it stamps the marker", async () => { + resumeState.ends = ["fetchHeartRate"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: false }); + + await runGoogleHealthBackfillForUser("u1", undefined, 2); + const data = prismaMock.googleHealthConnection.update.mock.calls[0]![0] + .data as Record; + expect(data.backfillCompletedAt).toBeInstanceOf(Date); + expect(data.backfillProgress).toEqual({ + ended: { + fetchHeartRate: { + type: "PULSE", + at: "2024-05-11T00:00:00.000Z", + reason: "no_progress", + }, + }, + }); }); it("continues a slice that only moved a collection's edge", async () => { diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index 4b07bda36..4c143ff16 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -21,9 +21,11 @@ import { integrationBackfillSourceOptions, } from "@/lib/jobs/integration-backfill-admission"; import { + googleHealthBackfillProgressAfterCompletion, readGoogleHealthBackfillCursors, readGoogleHealthBackfillDone, readGoogleHealthBackfillEdgeFilterUnsupported, + readGoogleHealthBackfillEnded, runWithGoogleHealthBackfillResume, } from "@/lib/google-health/sync-core"; import { @@ -35,7 +37,6 @@ import { googleHealthSliceBudget, runWithGoogleHealthStop, } from "@/lib/google-health/client"; -import { GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING } from "@/lib/google-health/sync-metrics"; import { reportJobProgress } from "@/lib/jobs/job-observer"; import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys"; @@ -71,27 +72,34 @@ export const GOOGLE_HEALTH_BACKFILL_SLICE_PAGES = 500; export const GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS = 5 * 60; /** - * The highest slice a backfill chain may reach, derived from the work a - * backfill can do rather than guessed. + * How long a chain of slices may keep going, run back to back, before it is + * taken for counting progress it does not make: sixty days. + */ +const GOOGLE_HEALTH_BACKFILL_MAX_CHAIN_MS = 60 * 24 * 60 * 60 * 1000; + +/** + * The highest slice a backfill chain may reach: every collection once, the + * slices sixty days of back-to-back slicing hold, and two of slack for the + * first slice and the one that stamps the marker. * * A slice is continued only when it made progress that is on disk: it - * 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. + * finished a collection (or ended one short, at the horizon or on a walk + * that stopped getting anywhere), it moved a resumable collection's edge + * back after writing at least one page, or a walk without the edge bound + * saved a place in the list further than any before it. Each of those moves + * toward an end the backfill has (the edge cannot pass the horizon, and a + * walk that repeats itself is stopped, see `backfill-bounds.ts`), so this + * ceiling bounds nothing a working backfill reaches. It used to be derived + * from a page count that a dense heart-rate history passes in under three + * years (#1220); in time it stays meaningful however densely a device + * records. In practice a chain is a handful of slices per year of dense + * history. A slice past this fails rather than queueing another. */ export const GOOGLE_HEALTH_BACKFILL_MAX_SLICE = GOOGLE_HEALTH_COLLECTION_KEYS.length + - 2 * GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING + + Math.ceil( + GOOGLE_HEALTH_BACKFILL_MAX_CHAIN_MS / GOOGLE_HEALTH_BACKFILL_SLICE_MS, + ) + 2; /** Test seam: a slice's budget, when not the production one. */ @@ -193,7 +201,14 @@ export async function runGoogleHealthBackfillForUser( const edgeFilterUnsupported = readGoogleHealthBackfillEdgeFilterUnsupported( connection?.backfillProgress, ); - const doneBefore = done.size; + const endedBefore = readGoogleHealthBackfillEnded( + connection?.backfillProgress, + ); + // A collection that ended short (`endedBefore`) has read all it is going + // to, so it counts as finished alongside the ones read to the end. + const finishedCount = (): number => + new Set([...done, ...endedBefore.keys()]).size; + const doneBefore = finishedCount(); const budget = googleHealthSliceBudget( sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS, @@ -205,6 +220,7 @@ export async function runGoogleHealthBackfillForUser( skipped, advanced, collections, + ended, } = await runWithGoogleHealthBackfillResume( userId, done, @@ -220,9 +236,12 @@ export async function runGoogleHealthBackfillForUser( ), cursors, edgeFilterUnsupported, + endedBefore, ); - const finished = done.size - doneBefore; + // `ended` is `endedBefore`, grown by this slice. + const doneAfter = finishedCount(); + const finished = doneAfter - doneBefore; const sliceReport = { slice, elapsedMs: Date.now() - startedAt, @@ -232,19 +251,21 @@ export async function runGoogleHealthBackfillForUser( skippedPages: budget.skippedPages, stopReason: budget.stopReason, imported, - collectionsDone: done.size, + collectionsDone: doneAfter, collectionsTotal: GOOGLE_HEALTH_COLLECTION_KEYS.length, collectionsFinished: finished, resumedPast: skipped.length, collections, // Where each unfinished resumable collection has got to, after this slice. edges: Object.fromEntries([...cursors].map(([key, c]) => [key, c.at])), + // Collections ended short of their history's start, and where. + ...(ended.size > 0 ? { ended: Object.fromEntries(ended) } : {}), }; annotate({ meta: { "googleHealth.backfill.slice": sliceReport } }); reportJobProgress({ google_health_backfill_slice: slice, google_health_backfill_pages: budget.pages, - google_health_backfill_collections_done: done.size, + google_health_backfill_collections_done: doneAfter, }); if (busy) { @@ -284,7 +305,7 @@ export async function runGoogleHealthBackfillForUser( slice, stopReason: budget.stopReason, collectionsFinished: finished, - collectionsDone: done.size, + collectionsDone: doneAfter, edgesMoved: [...advanced.keys()], }, }, @@ -314,9 +335,16 @@ export async function runGoogleHealthBackfillForUser( ); } + // The note of how far it got goes with the stamp, except where a + // collection ended short: that stays, so the card can say where its + // history ends and why. + const kept = googleHealthBackfillProgressAfterCompletion(ended); await prisma.googleHealthConnection.update({ where: { userId }, - data: { backfillCompletedAt: new Date(), backfillProgress: Prisma.DbNull }, + data: { + backfillCompletedAt: new Date(), + backfillProgress: kept ?? Prisma.DbNull, + }, }); annotate({ diff --git a/tests/integration/google-health-backfill-progress-bound.test.ts b/tests/integration/google-health-backfill-progress-bound.test.ts new file mode 100644 index 000000000..d9116e5d8 --- /dev/null +++ b/tests/integration/google-health-backfill-progress-bound.test.ts @@ -0,0 +1,380 @@ +/** + * #1220 — a Google Health backfill bounded by progress and a horizon in + * years instead of a count of pages, against Postgres. + * + * What went wrong. Every page-by-page data type counted the pages it had + * written across all of its slices and stopped at a ceiling (10,000 for heart + * rate). A watch that records heart rate every few seconds fills some 4,000 + * pages a year, so the count ran out about two and a half years back. It + * never reset: from then on every slice found heart rate at its ceiling, + * wrote nothing, failed as stuck, and the backfill was never marked complete, + * so the integration card said older history was loading for good. + * + * What this pins, each through the real backfill job, sync, walk and + * progress note: + * - a heart-rate walk that crosses the old ceiling, slice by slice, + * completes and imports every reading once; + * - an account left at the old ceiling resumes with the next slice, from + * where it stopped, and completes; + * - a Google that keeps answering with the same page is stopped after a + * bounded number of requests, and the backfill still completes, with the + * data type noted as ending there because of it; + * - a history older than the horizon is imported up to the horizon, and the + * data type is noted as ending at it; + * - a history handed out oldest first is read whole, not ended at the + * horizon on its first page. + * + * Google is replaced at the `safeFetch` seam by a generator that builds each + * heart-rate page from the filter and page token it is asked with. Run under + * the production heap ceiling: `NODE_OPTIONS=--max-old-space-size=524`. + */ +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER = "gh-progress-bound-owner"; +const HOUR = 3_600_000; +const DAY = 24 * HOUR; +const NOW = Date.parse("2026-10-17T12:00:00.000Z"); + +const fake = vi.hoisted(() => ({ + now: 0, + /** Readings in the history, newest first, `spacing` ms apart. */ + total: 0, + spacing: 0, + /** Readings per page Google hands out. */ + perPage: 1000, + /** Answer every heart-rate request with the newest page and a next token. */ + loop: false, + /** Hand the history out oldest first instead. */ + asc: false, + heartRateRequests: 0, +})); + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + const respond = (body: unknown) => + new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); + return { + ...actual, + safeFetch: async (url: string) => { + const u = new URL(url); + if (!u.pathname.endsWith("/dataTypes/heart-rate/dataPoints")) { + return respond({}); + } + fake.heartRateRequests += 1; + const filter = u.searchParams.get("filter") ?? ""; + const upTo = /physical_time <= "([^"]+)"/.exec(filter)?.[1]; + const from = /physical_time >= "([^"]+)"/.exec(filter)?.[1]; + // Reading `g` is `g * spacing` old and sits at position `g` of the + // list, or `total - 1 - g` oldest first. The first position asked for + // is the first reading at or beyond the edge. + const first = upTo + ? Math.max(0, Math.ceil((fake.now - Date.parse(upTo)) / fake.spacing)) + : from + ? Math.max( + 0, + fake.total - + 1 - + Math.floor((fake.now - Date.parse(from)) / fake.spacing), + ) + : 0; + const page = fake.loop ? 0 : Number(u.searchParams.get("pageToken") ?? 0); + const start = first + page * fake.perPage; + const end = Math.min(fake.total, start + fake.perPage); + const dataPoints = []; + for (let position = start; position < end; position++) { + const g = fake.asc ? fake.total - 1 - position : position; + dataPoints.push({ + name: `users/me/dataTypes/heart-rate/dataPoints/${8_000_000_000 + g}`, + dataSource: { platform: "FITBIT" }, + heartRate: { + beatsPerMinute: String(50 + (g % 70)), + sampleTime: { + physicalTime: new Date(fake.now - g * fake.spacing).toISOString(), + }, + }, + }); + } + return respond({ + dataPoints, + ...(fake.loop || end < fake.total + ? { nextPageToken: String(page + 1) } + : {}), + }); + }, + }; +}); + +/** The next slices the backfill queues, recorded. */ +const sent = vi.hoisted(() => ({ slices: [] as number[] })); +vi.mock("@/lib/jobs/boss-instance", async (importOriginal) => ({ + ...(await importOriginal()), + getGlobalBoss: () => ({ + send: async (_queue: string, data: { data?: { slice?: number } }) => { + sent.slices.push(data.data?.slice ?? -1); + return `job-${sent.slices.length}`; + }, + }), +})); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const OTHER_COLLECTIONS = [ + "fetchSteps", + "fetchDistance", + "fetchActiveEnergy", + "fetchFloors", + "fetchVo2Max", + "fetchWeight", + "fetchBodyFat", + "fetchSpo2", + "fetchHrv", + "fetchRhr", + "fetchRespiratoryRate", + "fetchBloodGlucose", + "fetchCoreBodyTemperature", + "fetchSleepTemperature", + "fetchExercise", + "sleep", +]; + +/** + * A fresh account. Every collection but heart rate is noted as done unless + * `backfillProgress` says otherwise: the walks this file is about are heart + * rate's, and the others would only add requests. + */ +async function seed(opts: { + total: number; + spacing: number; + perPage: number; + backfillProgress?: object; +}): Promise { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + const { encrypt } = await import("@/lib/crypto"); + await prisma.user.create({ + data: { id: USER, username: USER, timezone: "Europe/Berlin" }, + }); + await prisma.googleHealthConnection.create({ + data: { + userId: USER, + googleUserId: `google-${USER}`, + accessToken: encrypt("access"), + refreshToken: encrypt("refresh"), + tokenExpiresAt: new Date(Date.now() + 7 * DAY), + backfillProgress: opts.backfillProgress ?? { done: OTHER_COLLECTIONS }, + }, + }); + fake.now = NOW; + fake.total = opts.total; + fake.spacing = opts.spacing; + fake.perPage = opts.perPage; + fake.loop = false; + fake.asc = false; + fake.heartRateRequests = 0; + sent.slices = []; +} + +/** Pages a slice reads here: a fifth of production, so slices stay cheap. */ +const SLICE_PAGES = 100; + +/** + * Run slices the way the admission drain does, each one sending the next, + * until one ends the chain. Returns how many ran. + */ +async function runChain(maxSlices: number): Promise { + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + let slice = 0; + let ran = 0; + for (; ran < maxSlices; ran++) { + const before = sent.slices.length; + const outcome = await runGoogleHealthBackfillForUser( + USER, + () => false, + slice, + { slicePages: SLICE_PAGES }, + ); + if (!outcome.continued) return ran + 1; + expect(sent.slices.length).toBe(before + 1); + slice = sent.slices.at(-1)!; + } + throw new Error(`backfill did not finish within ${maxSlices} slices`); +} + +async function connection() { + return getPrismaClient().googleHealthConnection.findUniqueOrThrow({ + where: { userId: USER }, + select: { backfillCompletedAt: true, backfillProgress: true }, + }); +} + +async function pulseRows(): Promise { + return getPrismaClient().measurement.count({ + where: { userId: USER, type: "PULSE", source: "GOOGLE_HEALTH" }, + }); +} + +const savedHorizon = process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS; +beforeEach(() => { + delete process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS; +}); +afterAll(async () => { + if (savedHorizon !== undefined) { + process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS = savedHorizon; + } + await truncateAllTables(getPrismaClient()); +}); + +describe("Google Health backfill bounded by progress, not pages (#1220)", () => { + it("walks across the old page ceiling, slice after slice, to the end", async () => { + // Three years of heart rate, one reading a page, so a slice's pages are + // cheap. The stored count starts just short of the old ceiling of + // ten thousand, standing in for the dense pages a real watch fills + // before it: the old walk stopped a hundred pages in and never + // completed. Since the count no longer bounds anything, the walk + // crosses it over several slices and reaches the start of the history. + const pages = 350; + await seed({ + total: pages, + spacing: Math.floor((3 * 365 * DAY) / pages), + perPage: 1, + backfillProgress: { + done: OTHER_COLLECTIONS, + cursors: { + fetchHeartRate: { + at: new Date(NOW).toISOString(), + order: "desc", + pages: 9_900, + }, + }, + }, + }); + + const slices = await runChain(10); + + const c = await connection(); + expect(c.backfillCompletedAt).toBeInstanceOf(Date); + // Nothing ended short: the note goes with the stamp. + expect(c.backfillProgress).toBeNull(); + expect(await pulseRows()).toBe(pages); + // Back to the start of the three years, well past where the old count + // ran out. + const oldest = await getPrismaClient().measurement.findFirstOrThrow({ + where: { userId: USER, type: "PULSE" }, + orderBy: { measuredAt: "asc" }, + select: { measuredAt: true }, + }); + expect(NOW - oldest.measuredAt.getTime()).toBeGreaterThan( + (3 * 365 - 7) * DAY, + ); + // More than one slice, each one before the last sending the next. + expect(slices).toBeGreaterThanOrEqual(3); + expect(sent.slices).toEqual( + Array.from({ length: slices - 1 }, (_, i) => i + 1), + ); + }, 900_000); + + it("resumes an account left at the old ceiling and completes it", async () => { + // The state the report showed: every other collection done, heart rate + // at its ten thousand pages with its edge some way back. + const edgeIndex = 20_000; + const total = edgeIndex + 10 * 1000; + const edge = new Date(NOW - edgeIndex * 60_000).toISOString(); + await seed({ + total, + spacing: 60_000, + perPage: 1000, + backfillProgress: { + done: OTHER_COLLECTIONS, + cursors: { + fetchHeartRate: { at: edge, order: "desc", pages: 10_000 }, + }, + }, + }); + + const slices = await runChain(10); + + const c = await connection(); + expect(c.backfillCompletedAt).toBeInstanceOf(Date); + expect(c.backfillProgress).toBeNull(); + // Everything from the edge back was written, none of what was newer. + expect(await pulseRows()).toBe(total - edgeIndex); + expect(slices).toBeGreaterThanOrEqual(1); + // Read from the edge on, not from the newest page. + expect(fake.heartRateRequests).toBeLessThanOrEqual(10 + 2 * slices); + }, 600_000); + + it("stops a walk Google keeps answering with the same page, and completes with the reason noted", async () => { + await seed({ total: 5_000, spacing: 60_000, perPage: 1000 }); + fake.loop = true; + + await runChain(5); + + const { GOOGLE_HEALTH_NO_PROGRESS_PAGES } = + await import("@/lib/google-health/backfill-bounds"); + const { googleHealthBackfillEndedList } = + await import("@/lib/google-health/sync-core"); + // The first page and the twenty that repeated it, then never again. + expect(fake.heartRateRequests).toBe(1 + GOOGLE_HEALTH_NO_PROGRESS_PAGES); + const c = await connection(); + expect(c.backfillCompletedAt).toBeInstanceOf(Date); + expect(googleHealthBackfillEndedList(c.backfillProgress)).toEqual([ + { + type: "PULSE", + // The oldest reading the one real page wrote. + endsAt: new Date(NOW - 999 * 60_000).toISOString(), + reason: "no_progress", + }, + ]); + expect(await pulseRows()).toBe(1000); + }, 600_000); + + it("reads back to the horizon and notes where the history ends", async () => { + process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS = "1"; + // Three years of one reading a day. + await seed({ total: 3 * 365, spacing: DAY, perPage: 100 }); + + await runChain(5); + + const { googleHealthBackfillEndedList } = + await import("@/lib/google-health/sync-core"); + const c = await connection(); + expect(c.backfillCompletedAt).toBeInstanceOf(Date); + const ended = googleHealthBackfillEndedList(c.backfillProgress); + expect(ended).toHaveLength(1); + expect(ended[0]).toMatchObject({ type: "PULSE", reason: "horizon" }); + const horizon = Date.parse(ended[0]!.endsAt); + // About a year back from when the slice ran. + expect(Math.abs(Date.now() - 365 * DAY - horizon)).toBeLessThan(2 * DAY); + // Only the readings inside the horizon, and the walk stopped at the + // first page that crossed it instead of reading the rest. + const inside = Array.from({ length: fake.total }, (_, g) => g).filter( + (g) => NOW - g * DAY >= horizon, + ).length; + expect(await pulseRows()).toBe(inside); + expect(fake.heartRateRequests).toBe(Math.ceil((inside + 1) / 100)); + }, 600_000); + + it("reads an oldest-first history whole instead of ending it at the horizon", async () => { + // A list handed out oldest first starts beyond the horizon and works + // toward it. Ending it there would drop the whole history; it is read + // as it always was, and nothing is noted as ending short. + process.env.GOOGLE_HEALTH_BACKFILL_HORIZON_YEARS = "1"; + await seed({ total: 3 * 365, spacing: DAY, perPage: 100 }); + fake.asc = true; + + await runChain(5); + + const c = await connection(); + expect(c.backfillCompletedAt).toBeInstanceOf(Date); + expect(c.backfillProgress).toBeNull(); + expect(await pulseRows()).toBe(fake.total); + }, 600_000); +});