diff --git a/docs/integrations/google-health.md b/docs/integrations/google-health.md index 7fc61e208..b9bf5e095 100644 --- a/docs/integrations/google-health.md +++ b/docs/integrations/google-health.md @@ -122,22 +122,87 @@ The Google Cloud client id/secret are per-user (Settings), not env vars. ## Backfill, the hourly sync and Google's rate limit -Only one sync runs per account at a time. While the history backfill that -follows a connect is running, the hourly sync leaves that account alone, and -**Sync now** reports that a sync is already in progress and shows its progress -instead of starting a second one. +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 +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 +type read to the end is not read again, and a data type read page by page +(heart rate, weight, body fat, blood glucose, core temperature) keeps the +time of the oldest reading it has written, so the next slice asks Google only +for what is older. A restart or an update in the middle loses at most the page +that was being read. + +The hourly sync waits for the slice that is running (up to eight minutes) +instead of skipping the account, so recent data keeps arriving while an old +history is still coming in. Google limits how many requests one account may make per minute. When a sync -reaches that limit, Google answers `429`, and HealthLog waits and repeats the -request; the connection is never marked as needing a reconnect because of it. -Access tokens last an hour, and a long backfill renews its token as it goes. -A backfill that stops part-way (a restart, the job's time budget) picks up -with the data types it has not finished yet on its next attempt. +reaches that limit Google answers `429`, and HealthLog waits and repeats the +request. If Google asks for a longer wait than a minute, the slice ends there +and the next one starts five minutes later. A server error (`500`, `502`, +`503`, `504`), a request that times out and a dropped connection are repeated +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. + +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. +Everything already written stays. If an older version left your Google Health card asking for a reconnect after a backfill, the first sync after updating checks the stored grant and resumes on its own when the grant is still valid. +### Reading the backfill in the logs + +Every slice writes one log line when it ends, with +`"background":{"task_name":"job.integration_backfill"}`. The parts worth +reading: + +- `meta.kind` is `google-health-backfill`, `meta.slice` the slice number + (0 for the first), `meta.job_id` the pg-boss job. +- `external_calls` lists the requests the slice made to Google, one entry per + request: `method` names the data type and page (`fetchHeartRate(page=212)`), + `status` the HTTP status, `duration_ms` how long it took. A request that was + repeated carries a `note` (`rate_limited_retrying_in_ms=4000`, + `server_error_retrying_in_ms=2000`, `no_response_retrying_in_ms=2000`, + `token_rejected_retrying_with_fresh_token`); one that failed for good + carries an `error`. Successful requests are listed up to 60 per line and + everything up to 200; `external_calls_omitted` counts the rest. +- `meta["googleHealth.backfill.slice"]` sums the slice up: + - `pages`, `elapsedMs` and `imported`: what the slice read, how long it + took, how many readings it wrote. + - `stopReason`: why it ended early. `time_budget` and `page_budget` are the + normal end of a slice, `rate_limited` means Google would not take more + requests for now, `job_budget` that the job itself ran out of time. + `null` means the slice read everything it started. + - `collectionsDone` of `collectionsTotal`: data types read to the end so + far. + - `collections`: one entry per data type the slice touched, with its + `status` (`done`, `stopped` short on a budget or a rate limit, + `deferred` to the next slice, or `failed`), its `pages`, and for the + page-by-page types + `edgeFrom` and `edgeTo`, the oldest reading written before and after the + slice. + - `edges`: where every unfinished page-by-page data type stands now. +- `action.name` is `google_health.backfill.continued` for a slice that made + progress and queued the next one, `google_health.backfill.complete` for the + one that finished the history, and `google_health.backfill.stuck` for one + that moved nothing. A stuck slice also fails the job with a message that + names each unfinished data type and its status, for example + `no progress in slice 4, stopped on rate_limited; fetchHeartRate:stopped`. + +While a slice runs, the usual `job.progress` line for the +`integration-backfill-admission` queue reports +`google_health_backfill_slice`, `google_health_backfill_pages` and +`google_health_backfill_collections_done` every two minutes. + ## The 7-day re-consent caveat While the OAuth consent screen stays in **Testing** publishing mode, Google diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 9834872e9..95b316c3a 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -4410,8 +4410,10 @@ model GoogleHealthConnection { /// pre-fix externalId (a re-scored night minted parallel copies). Null = /// repair pending (or the connection predates the fix). sleepRepairedAt DateTime? @map("sleep_repaired_at") - /// The collections the in-flight history backfill has already walked to - /// the end (`{ done: string[] }`), so a retry resumes with the rest + /// 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. backfillProgress Json? @map("backfill_progress") diff --git a/src/lib/google-health/__tests__/backfill-resume-progress.test.ts b/src/lib/google-health/__tests__/backfill-resume-progress.test.ts index aa80d605d..69e9715e9 100644 --- a/src/lib/google-health/__tests__/backfill-resume-progress.test.ts +++ b/src/lib/google-health/__tests__/backfill-resume-progress.test.ts @@ -20,9 +20,17 @@ vi.mock("@/lib/logging/context", () => ({ })); import { + readGoogleHealthBackfillCursors, + recordGoogleHealthBackfillPage, runGoogleHealthCollection, runWithGoogleHealthBackfillResume, + type GoogleHealthBackfillCursor, } from "../sync-core"; +import { + googleHealthSliceBudget, + runWithGoogleHealthClientOutcome, + runWithGoogleHealthStop, +} from "../client"; beforeEach(() => { vi.clearAllMocks(); @@ -41,7 +49,9 @@ describe("runGoogleHealthCollection under a backfill", () => { expect([...done]).toEqual(["fetchSteps", "sleep"]); expect(prismaMock.googleHealthConnection.update).toHaveBeenCalledWith({ where: { userId: "u1" }, - data: { backfillProgress: { done: ["fetchSteps", "sleep"] } }, + data: { + backfillProgress: { done: ["fetchSteps", "sleep"], cursors: {} }, + }, }); }); @@ -63,3 +73,183 @@ describe("runGoogleHealthCollection under a backfill", () => { ); }); }); + +/** + * #1220 — a collection read page by page keeps its place inside the + * collection, not only once it is finished. A heart-rate history of years + * that one attempt could not finish used to start over from its newest page + * on every attempt. + */ +describe("the resume edge of a page-by-page collection", () => { + const at = (iso: string) => new Date(iso); + + it("moves the edge to the oldest sample of each page and persists it", async () => { + prismaMock.googleHealthConnection.update.mockResolvedValue({}); + const cursors = new Map(); + + const { advanced } = await runWithGoogleHealthBackfillResume( + "u1", + new Set(["sleep"]), + async () => { + await recordGoogleHealthBackfillPage("fetchHeartRate", { + first: at("2026-10-01T12:00:00Z"), + last: at("2026-10-01T00:00:00Z"), + }); + await recordGoogleHealthBackfillPage("fetchHeartRate", { + first: at("2026-09-30T23:59:00Z"), + last: at("2026-09-30T08:00:00Z"), + }); + }, + cursors, + ); + + expect(cursors.get("fetchHeartRate")).toEqual({ + at: "2026-09-30T08:00:00.000Z", + order: "desc", + pages: 2, + }); + expect(advanced.get("fetchHeartRate")).toBeNull(); + const last = prismaMock.googleHealthConnection.update.mock.calls.at( + -1, + )![0] as { data: { backfillProgress: unknown } }; + expect(last.data.backfillProgress).toEqual({ + done: ["sleep"], + cursors: { + fetchHeartRate: { + at: "2026-09-30T08:00:00.000Z", + order: "desc", + pages: 2, + }, + }, + }); + // And it reads back the way it was written. + expect( + readGoogleHealthBackfillCursors(last.data.backfillProgress).get( + "fetchHeartRate", + ), + ).toEqual(cursors.get("fetchHeartRate")); + }); + + it("follows a walk that runs oldest first", async () => { + prismaMock.googleHealthConnection.update.mockResolvedValue({}); + const cursors = new Map(); + await runWithGoogleHealthBackfillResume( + "u1", + new Set(), + () => + recordGoogleHealthBackfillPage("fetchWeight", { + first: at("2024-01-01T00:00:00Z"), + last: at("2024-03-01T00:00:00Z"), + }), + cursors, + ); + expect(cursors.get("fetchWeight")).toMatchObject({ + at: "2024-03-01T00:00:00.000Z", + order: "asc", + }); + }); + + it("does not count a page that moved nothing as progress", async () => { + prismaMock.googleHealthConnection.update.mockResolvedValue({}); + const cursors = new Map([ + [ + "fetchHeartRate", + { at: "2026-01-01T00:00:00.000Z", order: "desc", pages: 7 }, + ], + ]); + const { advanced } = await runWithGoogleHealthBackfillResume( + "u1", + new Set(), + () => + // The edge sample itself, read again: no further than before. + recordGoogleHealthBackfillPage("fetchHeartRate", { + first: at("2026-01-01T00:00:00Z"), + last: at("2026-01-01T00:00:00Z"), + }), + cursors, + ); + expect(advanced.size).toBe(0); + expect(cursors.get("fetchHeartRate")).toMatchObject({ + at: "2026-01-01T00:00:00.000Z", + pages: 8, + }); + }); + + it("counts no progress when the edge cannot be written", async () => { + prismaMock.googleHealthConnection.update.mockRejectedValue( + new Error("connection reset"), + ); + const cursors = new Map(); + const { advanced } = await runWithGoogleHealthBackfillResume( + "u1", + new Set(), + () => + recordGoogleHealthBackfillPage("fetchHeartRate", { + first: at("2026-10-01T12:00:00Z"), + last: at("2026-10-01T00:00:00Z"), + }), + cursors, + ); + expect(advanced.size).toBe(0); + expect(cursors.size).toBe(0); + expect(warnings.some((w) => w.includes("progress write failed"))).toBe( + true, + ); + }); + + it("clears a finished collection's edge with the same write that marks it done", async () => { + prismaMock.googleHealthConnection.update.mockResolvedValue({}); + const cursors = new Map([ + [ + "fetchHeartRate", + { at: "2026-01-01T00:00:00.000Z", order: "desc", pages: 7 }, + ], + ]); + const done = new Set(); + await runWithGoogleHealthBackfillResume( + "u1", + done, + () => runGoogleHealthCollection("fetchHeartRate", async () => 1), + cursors, + ); + expect([...done]).toEqual(["fetchHeartRate"]); + expect(cursors.size).toBe(0); + expect(prismaMock.googleHealthConnection.update).toHaveBeenCalledWith({ + where: { userId: "u1" }, + data: { backfillProgress: { done: ["fetchHeartRate"], cursors: {} } }, + }); + }); + + it("leaves a collection for the next slice once the slice has no budget left", async () => { + prismaMock.googleHealthConnection.update.mockResolvedValue({}); + const walk = vi.fn(async () => 5); + const budget = googleHealthSliceBudget(0, 100); + const { result } = await runWithGoogleHealthClientOutcome(() => + runWithGoogleHealthBackfillResume("u1", new Set(), () => + runWithGoogleHealthStop( + () => false, + () => runGoogleHealthCollection("fetchHeartRate", walk), + budget, + ), + ), + ); + expect(walk).not.toHaveBeenCalled(); + expect(result.collections).toEqual([ + { key: "fetchHeartRate", status: "deferred", pages: 0, durationMs: 0 }, + ]); + expect(budget.stopReason).toBe("time_budget"); + }); + + it("marks the cycle incomplete when it leaves a collection for later", async () => { + const { outcome } = await runWithGoogleHealthClientOutcome(() => + runWithGoogleHealthBackfillResume("u1", new Set(), () => + runWithGoogleHealthStop( + () => false, + () => runGoogleHealthCollection("fetchWeight", async () => 0), + googleHealthSliceBudget(0, 100), + ), + ), + ); + expect(outcome.truncated).toBe(true); + }); +}); diff --git a/src/lib/google-health/__tests__/rollup-request.test.ts b/src/lib/google-health/__tests__/rollup-request.test.ts index 54e23c31b..bed00a773 100644 --- a/src/lib/google-health/__tests__/rollup-request.test.ts +++ b/src/lib/google-health/__tests__/rollup-request.test.ts @@ -208,7 +208,8 @@ describe("fetchDailyRollUp — chunk walk + conservative fallback", () => { }); it("propagates a non-400 first-request failure without a fallback walk", async () => { - safeFetchMock.mockResolvedValueOnce(jsonResponse(500, {})); + // A 403 is neither the fallback's 400 nor a server error worth repeating. + safeFetchMock.mockResolvedValueOnce(jsonResponse(403, {})); await expect( fetchDailyRollUp(GOOGLE_HEALTH_DATA_TYPES.steps, "token", "fetchSteps", { start: new Date("2026-06-01T00:00:00Z"), diff --git a/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts index b4bf66c8e..becd43db9 100644 --- a/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts +++ b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts @@ -32,6 +32,10 @@ vi.mock("../sync-core", () => ({ (_userId: string, info: { accessToken: string }) => async () => info.accessToken, runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), + // Outside a backfill: no resume edge, nothing to record. + googleHealthBackfillCursor: () => null, + googleHealthHardFailureCount: () => 0, + recordGoogleHealthBackfillPage: vi.fn(async () => {}), })); vi.mock("@/lib/db", () => ({ diff --git a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts index 0a4c021e0..1f0ce79d2 100644 --- a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts +++ b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts @@ -58,6 +58,10 @@ vi.mock("../sync-core", () => ({ (_userId: string, info: { accessToken: string }) => async () => info.accessToken, runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), + // Outside a backfill: no resume edge, nothing to record. + googleHealthBackfillCursor: () => null, + googleHealthHardFailureCount: () => 0, + recordGoogleHealthBackfillPage: vi.fn(async () => {}), upsertGoogleHealthMeasurements: vi.fn(async () => ({ imported: 0, touched: [], diff --git a/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts b/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts new file mode 100644 index 000000000..ecaba31c6 --- /dev/null +++ b/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts @@ -0,0 +1,192 @@ +/** + * #1220 — the list walk under a backfill slice. + * + * - A server error or a request that got no answer is repeated a few times + * instead of ending a walk of thousands of pages. + * - A resumable walk asks only for the samples beyond the edge an earlier + * attempt reached. + * - Inside a slice, a resumable walk stops on the slice's budget and on a + * rate limit it cannot wait out, and reports itself truncated rather than + * failing; a walk that is not resumable runs on under the job's limit. + */ +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; + +const { safeFetchMock } = vi.hoisted(() => ({ safeFetchMock: vi.fn() })); +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, safeFetch: safeFetchMock }; +}); +vi.mock("@/lib/logging/context", () => ({ + annotate: () => {}, + getEvent: () => null, +})); + +import { SafeFetchError } from "@/lib/safe-fetch"; +import { + GOOGLE_HEALTH_DATA_TYPES, + GOOGLE_HEALTH_TRANSIENT_RETRY, + forEachDataPointPage, + googleHealthSliceBudget, + runWithGoogleHealthClientOutcome, + runWithGoogleHealthStop, +} from "../client"; +import { GoogleHealthApiError } from "../response-classifier"; + +function page(n: number, next: boolean, status = 200, headers = {}): Response { + return new Response( + JSON.stringify( + status === 200 + ? { + dataPoints: [{ n }], + ...(next ? { nextPageToken: String(n + 1) } : {}), + } + : { error: { code: status } }, + ), + { status, headers: { "content-type": "application/json", ...headers } }, + ); +} + +const baseDelay = GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs; +beforeEach(() => { + safeFetchMock.mockReset(); + GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = 0; +}); +afterAll(() => { + GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = baseDelay; +}); + +const walk = (query: Parameters[3] = {}) => + runWithGoogleHealthClientOutcome(() => + forEachDataPointPage( + GOOGLE_HEALTH_DATA_TYPES.heartRate, + "token", + "fetchHeartRate", + query, + () => {}, + ), + ); + +describe("a server error or a lost request mid-walk", () => { + it("is repeated and the walk goes on", async () => { + safeFetchMock + .mockResolvedValueOnce(page(0, true)) + .mockResolvedValueOnce(page(0, false, 503)) + .mockRejectedValueOnce(new SafeFetchError("aborted", "timeout")) + .mockResolvedValueOnce(page(1, false)); + + const { outcome } = await walk(); + + expect(safeFetchMock).toHaveBeenCalledTimes(4); + expect(outcome).toMatchObject({ pages: 2, truncated: false }); + }); + + it("fails the walk once the retries are spent", async () => { + safeFetchMock.mockResolvedValue(page(0, false, 500)); + await expect(walk()).rejects.toBeInstanceOf(GoogleHealthApiError); + expect(safeFetchMock).toHaveBeenCalledTimes( + GOOGLE_HEALTH_TRANSIENT_RETRY.maxRetries + 1, + ); + }); + + it("does not repeat an error that is not about the server", async () => { + safeFetchMock.mockRejectedValue(new TypeError("bad request init")); + await expect(walk()).rejects.toBeInstanceOf(TypeError); + expect(safeFetchMock).toHaveBeenCalledTimes(1); + }); +}); + +describe("a resumable walk", () => { + it("asks only for the samples at or beyond the recorded edge", async () => { + safeFetchMock.mockResolvedValue(page(0, false)); + await walk({ + resumable: true, + edge: { at: new Date("2025-03-01T08:00:00Z"), order: "desc" }, + }); + const url = new URL(safeFetchMock.mock.calls[0]![0] as string); + expect(url.searchParams.get("filter")).toBe( + 'heart_rate.sample_time.physical_time <= "2025-03-01T08:00:00.000Z"', + ); + }); + + it("stops on the slice's page budget and reports itself truncated", async () => { + safeFetchMock.mockImplementation(async (url: string) => + page(Number(new URL(url).searchParams.get("pageToken") ?? 0), true), + ); + const budget = googleHealthSliceBudget(60_000, 3); + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk({ resumable: true }), + budget, + ); + expect(safeFetchMock).toHaveBeenCalledTimes(3); + expect(outcome.truncated).toBe(true); + expect(budget.stopReason).toBe("page_budget"); + }); + + it("starts the slice's clock at the first walk, not when the slice was set up", async () => { + safeFetchMock.mockResolvedValue(page(0, false)); + let clock = 0; + const budget = googleHealthSliceBudget(1_000, 500, () => clock); + // The slice waited ten minutes for the account's lock before walking. + clock = 600_000; + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk({ resumable: true }), + budget, + ); + expect(outcome).toMatchObject({ pages: 1, truncated: false }); + expect(budget.deadline).toBe(601_000); + }); + + it("ends the slice on a rate limit it cannot wait out, without failing", async () => { + safeFetchMock + .mockResolvedValueOnce(page(0, true)) + .mockResolvedValueOnce(page(1, true, 429, { "retry-after": "3600" })); + const budget = googleHealthSliceBudget(60_000, 500); + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk({ resumable: true }), + budget, + ); + expect(outcome).toMatchObject({ pages: 1, truncated: true }); + expect(budget.stopReason).toBe("rate_limited"); + }); + + it("still fails on a rate limit outside a slice", async () => { + safeFetchMock.mockResolvedValue( + page(0, true, 429, { "retry-after": "3600" }), + ); + await expect(walk({ resumable: true })).rejects.toBeInstanceOf( + GoogleHealthApiError, + ); + }); +}); + +describe("a walk that is not resumable", () => { + it("runs past the slice's budget to its end", async () => { + safeFetchMock.mockImplementation(async (url: string) => { + const n = Number(new URL(url).searchParams.get("pageToken") ?? 0); + return page(n, n < 5); + }); + const budget = googleHealthSliceBudget(0, 1); + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk(), + budget, + ); + expect(safeFetchMock).toHaveBeenCalledTimes(6); + expect(outcome.truncated).toBe(false); + }); + + it("still stops on the job's own budget", async () => { + safeFetchMock.mockImplementation(async () => page(0, true)); + let asked = 0; + const { outcome } = await runWithGoogleHealthStop( + () => ++asked > 2, + () => walk(), + googleHealthSliceBudget(60_000, 500), + ); + expect(safeFetchMock).toHaveBeenCalledTimes(2); + expect(outcome.truncated).toBe(true); + }); +}); diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index 3bdab3aec..1bcdee440 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -36,7 +36,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { createHash, randomBytes } from "node:crypto"; import { annotate, getEvent } from "@/lib/logging/context"; -import { safeFetch } from "@/lib/safe-fetch"; +import { SafeFetchError, safeFetch } from "@/lib/safe-fetch"; import { wallClockInTz } from "@/lib/tz/wall-clock"; import { GoogleHealthApiError, @@ -134,18 +134,140 @@ export async function runWithGoogleHealthClientOutcome( * reaching the page ceiling: the resource is incomplete, the cycle is not * stamped, and the reason shown is that only part of the history was * processed. Outside such a scope the walks never stop early. + * + * A backfill slice adds a second, tighter limit (`GoogleHealthSliceBudget`): + * minutes and pages rather than the job's hours. Only a walk that keeps its + * place across attempts (`resumable`) honours it, because only such a walk + * loses nothing by stopping; a walk that would start over from its first page + * runs on to the end under the job's limit alone. */ -const stopStorage = new AsyncLocalStorage<() => boolean>(); +interface GoogleHealthWalkLimits { + shouldStop: () => boolean; + slice?: GoogleHealthSliceBudget; +} -function stopRequested(): boolean { - return stopStorage.getStore()?.() === true; +/** Why a backfill slice stopped walking. */ +export type GoogleHealthSliceStopReason = + "time_budget" | "page_budget" | "rate_limited" | "job_budget"; + +/** + * One backfill slice's share of the work: a span of wall-clock time and a + * page count, both well inside the job's own expiry. The clock starts the + * first time a walk asks, which is after the slice has the account's lock: a + * slice that waited for an hourly sync to finish does not start out with its + * time already spent. `pages` counts every page the slice's walks have read; + * `stopReason` is set the first time a limit is reached and stays set, so + * every walk after it stops too. + */ +export interface GoogleHealthSliceBudget { + ms: number; + /** When the slice's time is up; null until the first walk asks. */ + deadline: number | null; + maxPages: number; + pages: number; + stopReason: GoogleHealthSliceStopReason | null; + now: () => number; +} + +/** A fresh slice budget of `ms` milliseconds and `maxPages` pages. */ +export function googleHealthSliceBudget( + ms: number, + maxPages: number, + now: () => number = Date.now, +): GoogleHealthSliceBudget { + return { + ms, + deadline: null, + maxPages, + pages: 0, + stopReason: null, + now, + }; +} + +const stopStorage = new AsyncLocalStorage(); + +/** Whether the slice in scope has used its budget (and why, once it has). */ +function sliceExhausted(slice: GoogleHealthSliceBudget): boolean { + if (slice.stopReason) return true; + slice.deadline ??= slice.now() + slice.ms; + if (slice.now() >= slice.deadline) slice.stopReason = "time_budget"; + else if (slice.pages >= slice.maxPages) slice.stopReason = "page_budget"; + return slice.stopReason !== null; +} + +/** + * Whether a walk should stop before its next request. Every walk stops on the + * job's limit; a resumable one also stops on the slice's. + */ +function stopRequested(resumable = false): boolean { + const limits = stopStorage.getStore(); + if (!limits) return false; + if (limits.shouldStop()) { + if (limits.slice && !limits.slice.stopReason) { + limits.slice.stopReason = "job_budget"; + } + return true; + } + return ( + resumable && limits.slice !== undefined && sliceExhausted(limits.slice) + ); +} + +/** + * Whether the backfill slice in scope has used its budget. A collection that + * has not started yet waits for the next slice rather than starting now. + * False outside a slice. + */ +export function googleHealthSliceExhausted(): boolean { + const limits = stopStorage.getStore(); + if (!limits) return false; + if (limits.shouldStop()) { + if (limits.slice && !limits.slice.stopReason) { + limits.slice.stopReason = "job_budget"; + } + return true; + } + return limits.slice !== undefined && sliceExhausted(limits.slice); +} + +/** Pages the slice in scope has read so far (0 outside a slice). */ +export function googleHealthSlicePagesRead(): number { + return stopStorage.getStore()?.slice?.pages ?? 0; +} + +/** Count one page read against the slice in scope, if any. */ +function noteSlicePage(): void { + const slice = stopStorage.getStore()?.slice; + if (slice) slice.pages += 1; +} + +/** End the slice in scope for `reason` (no-op outside a slice). */ +function endSlice(reason: GoogleHealthSliceStopReason): void { + const slice = stopStorage.getStore()?.slice; + if (slice && !slice.stopReason) slice.stopReason = reason; +} + +/** Whether a resumable walk can stop and keep its place here. */ +function inSlice(): boolean { + return stopStorage.getStore()?.slice !== undefined; } export async function runWithGoogleHealthStop( shouldStop: () => boolean, fn: () => Promise, + slice?: GoogleHealthSliceBudget, ): Promise { - return stopStorage.run(shouldStop, fn); + return stopStorage.run({ shouldStop, slice }, fn); +} + +/** + * Mark the resource in scope truncated without a walk: a collection a slice + * left for the next one. The cycle then reports itself incomplete, so the + * backfill does not mistake a skipped collection for an empty one. + */ +export function noteGoogleHealthDeferred(): void { + notePagination(0, 0, true); } export const GOOGLE_HEALTH_API_BASE = "https://health.googleapis.com/v4"; @@ -527,6 +649,7 @@ export function resolveGoogleHealthUserId( // camelCase nested objects (`sampleTime.physicalTime`, // `interval.startTime`, `civilStartTime.date`). // + // proto3 int64 fields arrive as JSON **strings** (`"12345"`) — every numeric // extractor coerces numeric strings before the finite check. @@ -553,6 +676,9 @@ export const GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE = 25; */ export const GOOGLE_HEALTH_DENSE_MAX_PAGES = 10_000; +/** Page ceiling for every other list walk. */ +export const GOOGLE_HEALTH_DEFAULT_MAX_PAGES = 1000; + /** `dataPoints.list` envelope: `{ dataPoints, nextPageToken }`. */ interface GoogleHealthDataPointPage { dataPoints?: GoogleHealthDataPoint[]; @@ -578,6 +704,20 @@ interface DataPointQuery { * surfaces it so a live account reports which grammar Google accepted. */ onDateFilterStyle?: (style: GoogleHealthDateFilterStyle) => void; + /** + * The walk keeps its place across attempts (a backfill's resume cursor), so + * it may stop on the backfill slice's budget and on a rate limit it cannot + * wait out, and report itself truncated instead of failing: the next slice + * picks up from the edge it recorded. + */ + resumable?: boolean; + /** + * Where an earlier attempt of this walk got to: the oldest sample it wrote + * when the walk ran newest first (`desc`), the newest when it ran oldest + * first (`asc`). The walk asks only for the samples at or beyond it; the + * one sample on the edge itself is read again and overwrites in place. + */ + edge?: { at: Date; order: "desc" | "asc" }; } const pad2 = (n: number): string => String(n).padStart(2, "0"); @@ -797,6 +937,21 @@ export function googleHealthRetryDelayMs( return undefined; } +/** + * How a data request rides out a server that did not answer properly: a + * 500, 502, 503 or 504, a timeout, or a broken connection. A short history + * reads a handful of pages and rarely meets one; a full-history heart-rate + * walk reads thousands, and a single one used to end the walk and throw away + * everything it had read. Three tries more, 2, 4 and 8 s apart (or the + * `Retry-After` a 503 names, up to a minute), before the request fails. + */ +export const GOOGLE_HEALTH_TRANSIENT_RETRY = { + maxRetries: 3, + baseDelayMs: 2_000, +}; + +const GOOGLE_HEALTH_TRANSIENT_STATUSES = new Set([500, 502, 503, 504]); + function waitMs(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } @@ -811,10 +966,12 @@ interface GoogleHealthDataResponse { /** * Send one data request and answer with the response it finally got. * - * Two answers are handled here rather than handed back: + * Three answers are handled here rather than handed back: * - 429: the request waits (see `GOOGLE_HEALTH_RATE_LIMIT`) and is sent * again, up to `maxRetries` times, unless the caller's stop condition * says to end the walk. + * - a server error or no answer at all (see + * `GOOGLE_HEALTH_TRANSIENT_RETRY`): the same, with its own count. * - 401 with a token source: the token expired or was revoked between * the source's last check and this request. The source is asked once for * a fresh token and the request is repeated. A second 401, or a 401 on a @@ -830,6 +987,7 @@ async function googleHealthDataRequest( isExpectedFallback: (status: number) => boolean = () => false, ): Promise> { let rateLimited = 0; + let transient = 0; let refreshed = false; let forceRefresh = false; for (;;) { @@ -837,7 +995,33 @@ async function googleHealthDataRequest( typeof token === "string" ? token : await token({ forceRefresh }); forceRefresh = false; const started = performance.now(); - const res = await send(accessToken); + let res: Response; + try { + res = await send(accessToken); + } catch (err) { + // No answer at all: the request timed out or the connection broke. + // On a walk of a thousand pages one of those is a matter of time, and + // it cost the whole walk. + const retry = + err instanceof SafeFetchError && + (err.kind === "timeout" || err.kind === "network") && + transient < GOOGLE_HEALTH_TRANSIENT_RETRY.maxRetries && + !stopRequested(); + const delayMs = + GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs * 2 ** transient; + getEvent()?.addExternalCall({ + service: "google-health", + method: label, + duration_ms: Math.round(performance.now() - started), + ...(retry + ? { note: `no_response_retrying_in_ms=${delayMs}` } + : { error: err instanceof Error ? err.message : String(err) }), + }); + if (!retry) throw err; + transient += 1; + await waitMs(delayMs); + continue; + } const json = (await res.json().catch(() => null)) as T | null; const verdict = classifyGoogleHealthResponse(res.status); const apiErrorDetail = @@ -859,6 +1043,17 @@ async function googleHealthDataRequest( !stopRequested(); const willRefresh = res.status === 401 && typeof token !== "string" && !refreshed; + const transientDelayMs = GOOGLE_HEALTH_TRANSIENT_STATUSES.has(res.status) + ? Math.min( + googleHealthRetryDelayMs(res, json) ?? + GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs * 2 ** transient, + GOOGLE_HEALTH_RATE_LIMIT.maxDelayMs, + ) + : 0; + const willRetryTransient = + GOOGLE_HEALTH_TRANSIENT_STATUSES.has(res.status) && + transient < GOOGLE_HEALTH_TRANSIENT_RETRY.maxRetries && + !stopRequested(); getEvent()?.addExternalCall({ service: "google-health", @@ -871,14 +1066,16 @@ async function googleHealthDataRequest( ? { note: `rate_limited_retrying_in_ms=${backoffMs}` } : willRefresh ? { note: "token_rejected_retrying_with_fresh_token" } - : { - error: - verdict.classification === "success" - ? undefined - : apiErrorDetail - ? `${verdict.reason} ${apiErrorDetail}` - : verdict.reason, - }), + : willRetryTransient + ? { note: `server_error_retrying_in_ms=${transientDelayMs}` } + : { + error: + verdict.classification === "success" + ? undefined + : apiErrorDetail + ? `${verdict.reason} ${apiErrorDetail}` + : verdict.reason, + }), }); if (willWait) { @@ -886,6 +1083,11 @@ async function googleHealthDataRequest( await waitMs(backoffMs); continue; } + if (willRetryTransient) { + transient += 1; + await waitMs(transientDelayMs); + continue; + } if (willRefresh) { refreshed = true; forceRefresh = true; @@ -947,8 +1149,9 @@ export async function forEachDataPointPage( query: DataPointQuery, onPage: (points: GoogleHealthDataPoint[]) => void | Promise, ): Promise { - const maxPages = query.maxPages ?? 1000; + const maxPages = query.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES; const pageSize = query.pageSize ?? GOOGLE_HEALTH_PAGE_SIZE; + const resumable = query.resumable === true; let requestCount = 0; @@ -961,11 +1164,12 @@ export async function forEachDataPointPage( let stopped = false; do { - if (stopRequested()) { + if (stopRequested(resumable)) { stopped = true; break; } const params = new URLSearchParams({ pageSize: String(pageSize) }); + const filters: string[] = []; if (query.start) { const { field, bound } = incrementalFilter( dataType, @@ -973,8 +1177,20 @@ export async function forEachDataPointPage( query.tz, dateStyle, ); - params.set("filter", `${field} >= "${bound}"`); + filters.push(`${field} >= "${bound}"`); } + if (query.edge) { + const { field, bound } = incrementalFilter( + dataType, + query.edge.at, + query.tz, + dateStyle, + ); + filters.push( + `${field} ${query.edge.order === "desc" ? "<=" : ">="} "${bound}"`, + ); + } + if (filters.length > 0) params.set("filter", filters.join(" AND ")); if (pageToken) params.set("pageToken", pageToken); requestCount += 1; @@ -985,7 +1201,7 @@ export async function forEachDataPointPage( // an error next to the 200 that followed it. It is recorded as a note // now; any other 400 still carries its error. const firstRequest = requestCount === 1; - const { json, verdict, apiErrorDetail } = + const { status, json, verdict, apiErrorDetail } = await googleHealthDataRequest( accessToken, `${verb}(page=${pageCount})`, @@ -1003,6 +1219,15 @@ export async function forEachDataPointPage( firstRequest && status === 400, ); + // A rate limit the request could not wait out ends a resumable walk + // inside a backfill slice where it is: the pages before it are written + // and their edge recorded, so this is a pause, not a failure. The slice + // ends with it, since every other walk would meet the same quota. + if (status === 429 && resumable && inSlice()) { + endSlice("rate_limited"); + stopped = true; + break; + } if (verdict.classification !== "success") { throw new GoogleHealthApiError({ verb, @@ -1017,6 +1242,7 @@ export async function forEachDataPointPage( pageToken = json?.nextPageToken ?? null; pageCount += 1; fetched += page.length; + noteSlicePage(); await onPage(page); } while (pageToken && pageCount < maxPages); @@ -1245,7 +1471,7 @@ export async function fetchDailyRollUp( let totalPages = 0; let truncated = false; for (const chunk of chunkCivilRange(startCivil, endCivil, maxDays)) { - if (stopRequested()) { + if (stopRequested(false)) { truncated = true; break; } @@ -1295,6 +1521,7 @@ export async function fetchDailyRollUp( 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); diff --git a/src/lib/google-health/mappers.ts b/src/lib/google-health/mappers.ts index d73666dc6..194a782ff 100644 --- a/src/lib/google-health/mappers.ts +++ b/src/lib/google-health/mappers.ts @@ -387,6 +387,21 @@ function resolveMeasuredAt( return fallback; } +/** + * The sample instant of a spot-reading `DataPoint` (`sample` data types), or + * null for any other shape or a point without a parseable time. A backfill + * reads it to record how far a walk has got. + */ +export function googleHealthSampleInstant( + point: GoogleHealthDataPoint, + dataType: GoogleHealthDataType, + tz?: string, +): Date | null { + if (dataType.timeField !== "sample") return null; + const t = readPath(point, `${dataType.key}.sampleTime.physicalTime`); + return typeof t === "string" ? parseLocalInstant(t, tz) : null; +} + /** * Stable anchor for a `DataPoint`'s externalId. A spot reading anchors on its * sample time; a daily summary on its civil date. Combined with a type-specific diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index dc215266d..80004abf5 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -37,6 +37,9 @@ import { import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { googleHealthOutcomeTruncated, + googleHealthSliceExhausted, + googleHealthSlicePagesRead, + noteGoogleHealthDeferred, noteGoogleHealthMapped, noteGoogleHealthOutcomeFailure, noteGoogleHealthWritten, @@ -402,33 +405,75 @@ export async function runWithGoogleHealthSyncCycle( } /** - * Collections a full-history backfill already walked to the end, carried from - * one attempt of the backfill to the next. + * How far a full-history backfill has got, carried from one attempt of the + * backfill to the next. * - * A backfill that ended early (its job budget ran out, a collection failed) - * used to start the next attempt from the first page of the first collection, - * so the dense heart-rate walk, thousands of pages, was paid for again on - * every retry, and an account whose walk never fit into one attempt never - * finished at all. Each collection that completes cleanly is now written to - * `backfillProgress`, and the next attempt skips it and goes on with the - * collections it did not reach. A collection is the unit because Google - * offers no cursor that survives between attempts; its page tokens are not - * documented to outlive the walk that received them. + * Two grains. A collection walked to the end is noted in `done`, and the next + * attempt skips it. A collection read page by page (the spot readings: heart + * rate, weight, glucose, ...) also notes, after every page it writes, the + * edge it has reached: the oldest sample written so far when the walk runs + * newest first, as Google's list does. The next attempt asks Google only for + * the samples at or beyond that edge, so a collection that does not fit into + * one attempt is worked through over several instead of being read from its + * first page every time. + * + * The edge is a sample time, not Google's page token: page tokens are not + * documented to outlive the walk that received them, a sample time is a + * filter Google documents. The one sample on the edge itself is read twice + * and overwrites its own row. + * + * Before the edge existed only `done` was kept, and a heart-rate history of + * years that one attempt could not finish (its budget, a rate limit, one + * failed page among thousands) started again from the newest page on every + * attempt and never reached the end (#1220). */ +export interface GoogleHealthBackfillCursor { + /** The edge: the sample instant (ISO 8601) the walk has written up to. */ + 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: number; +} + +/** What one collection came to in one backfill attempt. */ +export interface GoogleHealthBackfillCollectionReport { + key: string; + /** + * `done`: walked to the end. `stopped`: ended short on a budget, a page + * ceiling or a rate limit, with nothing failed. `deferred`: not started, + * left for the next slice. `failed`: a request or a write failed, or the + * collection answered 403. + */ + status: "done" | "stopped" | "deferred" | "failed"; + pages: number; + durationMs: number; + /** The edge before and after this attempt, for a resumable collection. */ + edgeFrom?: string | null; + edgeTo?: string | null; +} + interface BackfillResumeTracker { userId: string; done: Set; + cursors: Map; /** Collections this attempt skipped because an earlier one finished them. */ skipped: string[]; + /** Collections whose edge this attempt moved, with the edge they started from. */ + advanced: Map; + collections: GoogleHealthBackfillCollectionReport[]; } const backfillResumeStorage = new AsyncLocalStorage(); +function progressRecord(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; +} + /** Read the `backfillProgress` column into the set of finished collections. */ export function readGoogleHealthBackfillDone(value: unknown): Set { - const done = - value && typeof value === "object" && !Array.isArray(value) - ? (value as { done?: unknown }).done - : undefined; + const done = progressRecord(value).done; return new Set( Array.isArray(done) ? done.filter( @@ -438,20 +483,188 @@ export function readGoogleHealthBackfillDone(value: unknown): Set { ); } +/** Read the `backfillProgress` column into the per-collection edges. */ +export function readGoogleHealthBackfillCursors( + value: unknown, +): Map { + const out = new Map(); + const cursors = progressRecord(progressRecord(value).cursors); + for (const [key, raw] of Object.entries(cursors)) { + if (key.length > 64) continue; + const c = progressRecord(raw); + const at = typeof c.at === "string" ? c.at : null; + if (!at || Number.isNaN(Date.parse(at))) continue; + out.set(key, { + at: new Date(at).toISOString(), + order: c.order === "asc" ? "asc" : "desc", + pages: + typeof c.pages === "number" && Number.isFinite(c.pages) && c.pages > 0 + ? Math.trunc(c.pages) + : 0, + }); + } + return out; +} + +/** + * Write the tracker's progress with one change applied, and apply it to the + * tracker only once the write has landed: progress that is not on disk is + * progress the next attempt cannot see, so it must not count in this one. + */ +async function persistBackfillProgress( + tracker: BackfillResumeTracker, + next: { + done: Set; + cursors: Map; + }, +): Promise { + await prisma.googleHealthConnection.update({ + where: { userId: tracker.userId }, + data: { + backfillProgress: { + done: [...next.done], + cursors: Object.fromEntries( + [...next.cursors].map(([key, c]) => [ + key, + { at: c.at, order: c.order, pages: c.pages }, + ]), + ), + }, + }, + }); + // `next` may be the tracker's own collections: copy before clearing. + const done = [...next.done]; + const cursors = [...next.cursors]; + tracker.done.clear(); + for (const key of done) tracker.done.add(key); + tracker.cursors.clear(); + for (const [key, c] of cursors) tracker.cursors.set(key, c); +} + +/** What one backfill attempt did, as `runWithGoogleHealthBackfillResume` reports it. */ +export interface GoogleHealthBackfillAttempt { + result: T; + skipped: string[]; + /** Collections whose edge moved, each with the edge it started from. */ + advanced: Map; + collections: GoogleHealthBackfillCollectionReport[]; +} + /** * Run a full-history backfill attempt that resumes: collections in `done` are - * skipped, and each collection finished cleanly during this attempt is added - * to it and persisted at once, so an attempt that dies part-way keeps what it - * finished. + * skipped, a collection with an edge in `cursors` starts from it, and each + * collection finished or page written during this attempt is persisted at + * once, so an attempt that dies part-way keeps what it did. */ export async function runWithGoogleHealthBackfillResume( userId: string, done: Set, fn: () => Promise, -): Promise<{ result: T; skipped: string[] }> { - const tracker: BackfillResumeTracker = { userId, done, skipped: [] }; + cursors: Map = new Map(), +): Promise> { + const tracker: BackfillResumeTracker = { + userId, + done, + cursors, + skipped: [], + advanced: new Map(), + collections: [], + }; const result = await backfillResumeStorage.run(tracker, fn); - return { result, skipped: tracker.skipped }; + return { + result, + skipped: tracker.skipped, + advanced: tracker.advanced, + collections: tracker.collections, + }; +} + +/** + * Where a resumable collection's walk starts in this backfill attempt: the + * edge an earlier attempt recorded (if any) and the pages it has already + * read. Null outside a backfill, where every walk reads its whole window. + */ +export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): { + edge?: { at: Date; order: "desc" | "asc" }; + pagesSoFar: number; +} | null { + const resume = backfillResumeStorage.getStore(); + if (!resume) return null; + const cursor = resume.cursors.get(key); + return cursor + ? { + edge: { at: new Date(cursor.at), order: cursor.order }, + pagesSoFar: cursor.pages, + } + : { pagesSoFar: 0 }; +} + +/** + * Record that a resumable collection wrote one more page, holding samples + * from `first` to `last` in the order Google returned them. Moves the + * collection's edge to the far end of the page and persists it. Outside a + * backfill this does nothing. + * + * The order is read off the first page whose two ends differ and kept for + * the rest of the backfill; Google returns its lists newest first, which is + * also what a page that cannot tell is taken to mean. + */ +export async function recordGoogleHealthBackfillPage( + key: GoogleHealthCollectionKey, + page: { first: Date | null; last: Date | null }, +): Promise { + const resume = backfillResumeStorage.getStore(); + if (!resume) return; + const current = resume.cursors.get(key); + const order: "desc" | "asc" = + current?.order ?? + (page.first && page.last && page.first.getTime() < page.last.getTime() + ? "asc" + : "desc"); + const ends = [page.first, page.last].filter((d): d is Date => d !== null); + const far = + ends.length === 0 + ? null + : new Date( + order === "desc" + ? Math.min(...ends.map((d) => d.getTime())) + : Math.max(...ends.map((d) => d.getTime())), + ); + const currentAt = current ? Date.parse(current.at) : null; + const moved = + far !== null && + (currentAt === null || + (order === "desc" + ? far.getTime() < currentAt + : far.getTime() > currentAt)); + if (!current && !moved) return; + const nextCursor: GoogleHealthBackfillCursor = { + at: moved ? far.toISOString() : current!.at, + order, + pages: (current?.pages ?? 0) + 1, + }; + const cursors = new Map(resume.cursors); + cursors.set(key, nextCursor); + try { + await persistBackfillProgress(resume, { done: resume.done, cursors }); + if (moved && !resume.advanced.has(key)) { + resume.advanced.set(key, current?.at ?? null); + } + } catch (err) { + // The page's rows are written; only the note of how far the walk got is + // missing, so the next attempt reads this page again. Not a failure. + getEvent()?.addWarning( + `google-health: backfill progress write failed for ${key}: ${err}`, + ); + annotate({ + meta: { "googleHealth.backfill.progressWriteFailed": key }, + }); + } +} + +/** Hard failures the cycle in scope has recorded so far. */ +export function googleHealthHardFailureCount(): number { + return hardFailStorage.getStore()?.failures.length ?? 0; } /** @@ -459,10 +672,12 @@ export async function runWithGoogleHealthBackfillResume( * one in scope. Outside a backfill this is just `fn()`. * * Inside one, a collection already walked to the end by an earlier attempt is - * skipped. Otherwise it runs, and it counts as finished only when nothing - * went wrong while it ran: no hard failure on the cycle's ledger, no 403 - * soft-skip, and no walk that ended short (its page ceiling, or the job's - * budget asking it to stop). Anything less is walked again next attempt. + * skipped, and so is one the slice has no budget left to start: it is left + * for the next slice, and the cycle reports itself incomplete. Otherwise it + * runs, and it counts as finished only when nothing went wrong while it ran: + * no hard failure on the cycle's ledger, no 403 soft-skip, and no walk that + * ended short (its page ceiling, or a budget asking it to stop). Anything + * less is walked again next attempt, from its edge when it has one. */ export async function runGoogleHealthCollection( key: GoogleHealthCollectionKey, @@ -474,15 +689,29 @@ export async function runGoogleHealthCollection( resume.skipped.push(key); return 0; } + if (googleHealthSliceExhausted()) { + noteGoogleHealthDeferred(); + resume.collections.push({ + key, + status: "deferred", + pages: 0, + durationMs: 0, + }); + return 0; + } + const startedAt = performance.now(); + const pagesBefore = googleHealthSlicePagesRead(); + const edgeFrom = resume.cursors.get(key)?.at ?? null; const hardBefore = hardFailStorage.getStore()?.failures.length ?? 0; const softBefore = softSkipStorage.getStore()?.count ?? 0; const truncatedBefore = googleHealthOutcomeTruncated(); const imported = await fn(); - const clean = - (hardFailStorage.getStore()?.failures.length ?? 0) === hardBefore && - (softSkipStorage.getStore()?.count ?? 0) === softBefore && - !truncatedBefore && - !googleHealthOutcomeTruncated(); + const failedHard = + (hardFailStorage.getStore()?.failures.length ?? 0) !== hardBefore || + (softSkipStorage.getStore()?.count ?? 0) !== softBefore; + const truncated = truncatedBefore || googleHealthOutcomeTruncated(); + const clean = !failedHard && !truncated; + let finished = false; if (clean) { // The collection counts as finished only once the note says so. The job // reads the size of `done` as this attempt's progress and sends another @@ -490,11 +719,13 @@ export async function runGoogleHealthCollection( // progress nobody can see next attempt, and a chain of slices could keep // "progressing" over the same collection forever. try { - await prisma.googleHealthConnection.update({ - where: { userId: resume.userId }, - data: { backfillProgress: { done: [...resume.done, key] } }, + const cursors = new Map(resume.cursors); + cursors.delete(key); + await persistBackfillProgress(resume, { + done: new Set([...resume.done, key]), + cursors, }); - resume.done.add(key); + finished = true; } catch (err) { // Not finished, then: the collection is walked again next attempt. It // must not fail a collection whose rows are written. @@ -506,6 +737,14 @@ export async function runGoogleHealthCollection( }); } } + const edgeTo = finished ? null : (resume.cursors.get(key)?.at ?? null); + resume.collections.push({ + key, + status: finished ? "done" : failedHard ? "failed" : "stopped", + pages: googleHealthSlicePagesRead() - pagesBefore, + durationMs: Math.round(performance.now() - startedAt), + ...(edgeFrom !== null || edgeTo !== null ? { edgeFrom, edgeTo } : {}), + }); return imported; } diff --git a/src/lib/google-health/sync-lock.ts b/src/lib/google-health/sync-lock.ts index db8ca51c8..ddb16523d 100644 --- a/src/lib/google-health/sync-lock.ts +++ b/src/lib/google-health/sync-lock.ts @@ -16,7 +16,9 @@ * the hourly poll skips the account for this tick, the manual trigger answers * that a sync is already running, and the backfill and the repair wait a * bounded time for a short run to finish before handing their job back to - * pg-boss to retry. + * pg-boss to retry. The hourly poll waits too, for about one backfill slice: + * a backfill lets go of the account between its slices, so the hourly sync + * gets its turn instead of being skipped for as long as the import runs. */ import { withJobLock, type GuardedRun } from "@/lib/jobs/job-lock"; @@ -32,6 +34,16 @@ export function googleHealthSyncLockKey(userId: string): string { */ export const GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS = 10 * 60 * 1000; +/** + * How long the hourly poll waits for an account another run holds. A backfill + * slice walks for five minutes and then lets go + * (`GOOGLE_HEALTH_BACKFILL_SLICE_MS`); the page in flight, a rate-limit wait + * and the slice's own bookkeeping can add a minute or two, and the next slice + * starts a minute after this one ends. Eight minutes covers that, and leaves + * the hourly job well inside its fifteen-minute expiry. + */ +export const GOOGLE_HEALTH_POLL_LOCK_WAIT_MS = 8 * 60 * 1000; + /** * Run `run` while holding the account's Google Health lock. Resolves * `{ ran: false }` without calling `run` when another run holds it and does diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index a86dfb59e..2bcad8002 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -27,12 +27,14 @@ */ import { GOOGLE_HEALTH_DATA_TYPES, + GOOGLE_HEALTH_DEFAULT_MAX_PAGES, GOOGLE_HEALTH_DENSE_MAX_PAGES, type GoogleHealthDataType, type GoogleHealthMappedMeasurement, type GoogleHealthTokenSource, fetchDataPoints, forEachDataPointPage, + googleHealthSampleInstant, mapBloodGlucose, mapBodyFat, mapCoreBodyTemperature, @@ -44,11 +46,15 @@ import { mapRestingHeartRate, mapWeight, mapWristTemperature, + noteGoogleHealthDeferred, } from "./client"; import { getValidToken, + googleHealthBackfillCursor, + googleHealthHardFailureCount, googleHealthTokenSource, handleCollectionFetchError, + recordGoogleHealthBackfillPage, runGoogleHealthCollection, upsertGoogleHealthMeasurements, type GoogleHealthMeasurementUpsert, @@ -108,13 +114,6 @@ const METRIC_RESOURCES: MetricResource[] = [ map: mapRespiratoryRate, verb: "fetchRespiratoryRate", }, - { - dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate, - map: mapHeartRate, - verb: "fetchHeartRate", - maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES, - intraday: true, - }, { dataType: GOOGLE_HEALTH_DATA_TYPES.bloodGlucose, map: mapBloodGlucose, @@ -133,8 +132,34 @@ const METRIC_RESOURCES: MetricResource[] = [ map: mapWristTemperature, verb: "fetchSleepTemperature", }, + // Last: the dense stream. A backfill finishes every light collection in + // its first slice and then works through this one slice by slice; ahead + // of them it held them back until the whole stream was in. + { + dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate, + map: mapHeartRate, + verb: "fetchHeartRate", + maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES, + intraday: true, + }, ]; +/** + * Whether a metric is walked resumably by a backfill: written page by page, + * and a spot reading whose sample time can mark how far the walk got. + */ +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 = {}, @@ -245,6 +270,27 @@ async function syncMetricCollection( opts: GoogleHealthResourceSyncOptions, ): Promise { 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. + const cursor = isResumable(resource) + ? googleHealthBackfillCursor(resource.verb) + : null; + const maxPages = + (resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES) - + (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; + } + // 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. + let edgeHeld = false; try { await forEachDataPointPage( resource.dataType, @@ -254,7 +300,8 @@ async function syncMetricCollection( start: resource.intraday ? (opts.intradayStart ?? opts.start) : opts.start, - maxPages: resource.maxPages, + maxPages, + ...(cursor ? { resumable: true, edge: cursor.edge } : {}), }, async (points) => { const readings: GoogleHealthMeasurementUpsert[] = []; @@ -269,11 +316,23 @@ async function syncMetricCollection( }); } } + const failuresBefore = googleHealthHardFailureCount(); imported += ( await upsertGoogleHealthMeasurements(userId, readings, { deferRollup: opts.deferRollup, }) ).imported; + if (!cursor || points.length === 0) 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, + ), + }); }, ); } 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 4e5e5a269..796893945 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -31,7 +31,13 @@ const { bossSend: vi.fn(), // Collections the mocked walk "finishes" in this attempt, added to the // resume set the way `runGoogleHealthCollection` adds them. - resumeState: { finishes: [] as string[], doneBefore: [] as string[] }, + resumeState: { + finishes: [] as string[], + doneBefore: [] as string[], + /** Collections whose resume edge the mocked walk moves. */ + advances: [] as string[], + collections: [] as Array<{ key: string; status: string }>, + }, })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); @@ -54,6 +60,7 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { vi.mock("@/lib/google-health/sync-core", () => ({ GOOGLE_HEALTH_INTEGRATION_KEY: "google-health", readGoogleHealthBackfillDone: () => new Set(resumeState.doneBefore), + readGoogleHealthBackfillCursors: () => new Map(), runWithGoogleHealthBackfillResume: async ( _userId: string, done: Set, @@ -61,7 +68,12 @@ vi.mock("@/lib/google-health/sync-core", () => ({ ) => { const result = await fn(); for (const key of resumeState.finishes) done.add(key); - return { result, skipped: [] }; + return { + result, + skipped: [], + advanced: new Map(resumeState.advances.map((key) => [key, null])), + collections: resumeState.collections, + }; }, })); vi.mock("@/lib/google-health/sync", () => ({ @@ -71,8 +83,11 @@ vi.mock("@/lib/google-health/sync", () => ({ import { GOOGLE_HEALTH_BACKFILL_MAX_SLICE, + 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, forEachDataPointPage, @@ -87,6 +102,8 @@ beforeEach(() => { bossSend.mockResolvedValue("job-next"); resumeState.finishes = []; resumeState.doneBefore = []; + resumeState.advances = []; + resumeState.collections = []; }); describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { @@ -261,8 +278,71 @@ describe("runGoogleHealthBackfillForUser — slices", () => { expect(bossSend).not.toHaveBeenCalled(); }); - it("caps the chain at the number of collections plus two", () => { - expect(GOOGLE_HEALTH_BACKFILL_MAX_SLICE).toBe(17 + 2); + it("derives the chain's ceiling from the work a backfill can do", () => { + // Every collection once, plus every page the resumable collections may + // ever read (heart rate's 10 000 and four others' 1 000), plus slack. + expect(GOOGLE_HEALTH_BACKFILL_MAX_SLICE).toBe( + 17 + GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING + 2, + ); + expect(GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING).toBe(10_000 + 4 * 1000); + }); + + it("continues a slice that only moved a collection's edge", async () => { + resumeState.advances = ["fetchHeartRate"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 900, failed: true }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 7), + ).resolves.toEqual({ imported: 900, continued: true }); + expect(bossSend).toHaveBeenCalledTimes(1); + expect(bossSend.mock.calls[0]![1]).toMatchObject({ + data: { userId: "u1", slice: 8 }, + }); + expect(bossSend.mock.calls[0]![2]).toMatchObject({ + startAfter: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + }); + }); + + it("waits longer before the next slice when this one ended on a rate limit", async () => { + resumeState.advances = ["fetchHeartRate"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true }); + // The slice ended the way a resumable walk ends it when Google keeps + // answering 429 past every wait. + const client = await import("@/lib/google-health/client"); + const spy = vi + .spyOn(client, "googleHealthSliceBudget") + .mockImplementation(() => ({ + ms: 300_000, + deadline: null, + maxPages: 500, + pages: 3, + stopReason: "rate_limited", + now: Date.now, + })); + try { + await runGoogleHealthBackfillForUser("u1"); + } finally { + spy.mockRestore(); + } + expect(bossSend.mock.calls[0]![2]).toMatchObject({ + startAfter: GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS, + }); + }); + + it("names where it is stuck when a slice moved nothing", async () => { + resumeState.collections = [ + { key: "fetchWeight", status: "done" }, + { key: "fetchHeartRate", status: "failed" }, + { key: "fetchBloodGlucose", status: "deferred" }, + ]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: true }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 3), + ).rejects.toThrow( + /marker not stamped \(no progress in slice 3; fetchHeartRate:failed,fetchBloodGlucose:deferred\)/, + ); + expect(bossSend).not.toHaveBeenCalled(); }); it("stamps the marker on the slice that finishes the history", async () => { diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index 4f3a00294..05a3a27bd 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -21,6 +21,7 @@ import { integrationBackfillSourceOptions, } from "@/lib/jobs/integration-backfill-admission"; import { + readGoogleHealthBackfillCursors, readGoogleHealthBackfillDone, runWithGoogleHealthBackfillResume, } from "@/lib/google-health/sync-core"; @@ -29,7 +30,12 @@ import { syncUserGoogleHealth, } from "@/lib/google-health/sync"; import { GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock"; -import { runWithGoogleHealthStop } from "@/lib/google-health/client"; +import { + 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"; export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill"; @@ -42,15 +48,52 @@ export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill"; export const GOOGLE_HEALTH_BACKFILL_CONCURRENCY = 1; /** - * The highest slice a backfill chain may reach. Every continued slice has - * finished at least one more collection than the one before it, so a healthy - * chain is at most one slice per collection long; the two extra slices are - * slack for the first one and the one that stamps the marker. A slice past - * this is a chain that counts progress it does not make, and it fails rather - * than queueing another. + * One slice of a backfill: how long it walks and how many pages it reads + * before it lets go of the account and hands the rest to the next slice. + * + * Far inside the admission job's expiry (sixteen hours) on purpose. A slice + * holds the account's sync lock while it walks, and the hourly sync of that + * account waits for the lock (`GOOGLE_HEALTH_POLL_LOCK_WAIT_MS`), so a slice + * has to be short enough for the hourly sync to get its turn: before slices, + * one attempt held the lock for as long as the import ran and every hourly + * sync of the account in that time was skipped. The page budget keeps a + * slice on a fast link from spending a whole quota window in one go. + */ +export const GOOGLE_HEALTH_BACKFILL_SLICE_MS = 5 * 60 * 1000; +export const GOOGLE_HEALTH_BACKFILL_SLICE_PAGES = 500; + +/** + * The pause before the next slice when this one ended on a rate limit it + * could not wait out: long enough for Google's per-minute window to refill + * several times over, so the next slice does not open on the same 429. + */ +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. + * + * A slice is continued only when it made progress that is on disk: it + * finished a collection, or it moved a resumable collection's edge after + * writing at least one page. Collections finish once each, and a resumable + * collection's pages are capped across all attempts by its page ceiling + * (`GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING`, the sum of them). So no chain of + * progressing slices is longer than the collections plus those pages; the two + * extra are slack for the first slice and the one that stamps the marker. In + * practice a chain is a handful of slices per year of dense history. A slice + * past this is counting progress it does not make, and fails rather than + * queueing another. */ export const GOOGLE_HEALTH_BACKFILL_MAX_SLICE = - GOOGLE_HEALTH_COLLECTION_KEYS.length + 2; + GOOGLE_HEALTH_COLLECTION_KEYS.length + + GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING + + 2; + +/** Test seam: a slice's budget, when not the production one. */ +export interface GoogleHealthBackfillSliceOptions { + sliceMs?: number; + slicePages?: number; +} export interface GoogleHealthBackfillPayload { userId: string; @@ -58,42 +101,40 @@ export interface GoogleHealthBackfillPayload { } /** - * Per-user backfill handler. Runs a full-history sync for one account and - * stamps `backfillCompletedAt` ONLY on a clean run, so the discovery query - * drops it. Idempotent: the per-resource upserts are key-stable, so a re-run - * (e.g. a reboot mid-walk) overwrites rather than duplicating. Mirrors - * `runFitbitBackfillForUser`. + * Per-user backfill handler. Runs one slice of a full-history sync for one + * account and stamps `backfillCompletedAt` ONLY on a clean run, so the + * discovery query drops it. Idempotent: the per-resource upserts are + * key-stable, so a re-run (e.g. a reboot mid-walk) overwrites rather than + * duplicating. Mirrors `runFitbitBackfillForUser`. * * Verdict-gated: a partial hard failure (one collection 400ing, a write - * failing) THROWS so pg-boss retries the job — stamping the marker over a - * partial walk would freeze the gap in forever. A connection parked at - * `error_reauth` is NOT clean either, but throwing against a dead grant would - * just burn the retry budget on the same no-op: return WITHOUT stamping — the - * boot discovery re-enqueues the account on the next boot, and the stamp lands - * once the user reconnects and a clean walk completes. + * failing) is not complete, and stamping the marker over a partial walk would + * freeze the gap in forever. A connection parked at `error_reauth` is NOT + * clean either, but throwing against a dead grant would just burn the retry + * budget on the same no-op: return WITHOUT stamping — the boot discovery + * re-enqueues the account on the next boot, and the stamp lands once the user + * reconnects and a clean walk completes. * - * `shouldStop` is the admission job's budget (`jobBudget`). A per-minute - * heart-rate history runs to thousands of pages, and the walk asks it before - * every page: once it says stop, the walk ends where it is, the cycle reports - * itself truncated, and this throws like any other incomplete run, without - * the stamp. The rows already written stay. + * Resumable at two grains (`GoogleHealthBackfillCursor`): every collection the + * walk finishes cleanly is noted in `backfillProgress.done` and skipped from + * then on, and every spot-reading collection notes the edge it has written up + * to after each page, so the next attempt starts there instead of at the + * newest page. The note is cleared once the stamp lands, and on every + * reconnect. * - * Resumable: every collection the walk finishes cleanly is noted in - * `backfillProgress`, and the next attempt (the pg-boss retry, or the next - * boot's discovery) skips those and walks only what is left - * (`runWithGoogleHealthBackfillResume`). The note is cleared once the stamp - * lands, and on every reconnect. + * Runs in slices. A slice walks until its own budget is spent + * (`GOOGLE_HEALTH_BACKFILL_SLICE_MS` / `_PAGES`), Google throttles it past + * what a wait covers, or the job's budget (`shouldStop`) runs out, and then + * lets go of the account. A slice that made progress on disk (a collection + * finished, an edge moved) is not a failure: it sends the next slice + * (`slice + 1`) and returns `continued`, without the error line or the + * queue's failure count. Only a slice that moved nothing throws, with what + * each collection came to in the message, because that one is stuck: the + * pg-boss retry and, after it, the next boot take it from there. The chain is + * capped at `GOOGLE_HEALTH_BACKFILL_MAX_SLICE`. * - * Runs in slices: an attempt that ends before the whole history is in but - * finished at least one more collection than it started with made progress. - * It is not a failure: it sends the next slice (`slice + 1`) and returns - * `continued`, without the error line or the queue's failure count. Only an - * attempt that finished nothing new throws, because that one is stuck: the - * pg-boss retry and, after it, the next boot take it from there. Progress is - * counted only once it is persisted (`runGoogleHealthCollection` adds a - * collection to `done` after its note is written), and the chain is capped at - * `GOOGLE_HEALTH_BACKFILL_MAX_SLICE`: a slice past it throws instead of - * running or sending another. + * What the slice did is annotated on the run's event as + * `googleHealth.backfill.slice` (see `docs/integrations/google-health.md`). * * Never beside another run of the same account: the sync takes the account's * lock, waiting a bounded time for an hourly or manual run to finish. If the @@ -104,6 +145,7 @@ export async function runGoogleHealthBackfillForUser( userId: string, shouldStop: () => boolean = () => false, slice: number = 0, + sliceOptions: GoogleHealthBackfillSliceOptions = {}, ): Promise<{ imported: number; continued?: boolean }> { if (slice > GOOGLE_HEALTH_BACKFILL_MAX_SLICE) { annotate({ @@ -142,22 +184,56 @@ export async function runGoogleHealthBackfillForUser( return { imported: 0 }; } const done = readGoogleHealthBackfillDone(connection?.backfillProgress); + const cursors = readGoogleHealthBackfillCursors(connection?.backfillProgress); const doneBefore = done.size; + const budget = googleHealthSliceBudget( + sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS, + sliceOptions.slicePages ?? GOOGLE_HEALTH_BACKFILL_SLICE_PAGES, + ); + const startedAt = Date.now(); const { result: { imported, failed, busy }, skipped, - } = await runWithGoogleHealthBackfillResume(userId, done, () => - runWithGoogleHealthStop(shouldStop, () => - syncUserGoogleHealth(userId, { - fullSync: true, - waitForLockMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS, - }), - ), + advanced, + collections, + } = await runWithGoogleHealthBackfillResume( + userId, + done, + () => + runWithGoogleHealthStop( + shouldStop, + () => + syncUserGoogleHealth(userId, { + fullSync: true, + waitForLockMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS, + }), + budget, + ), + cursors, ); - if (skipped.length > 0) { - annotate({ meta: { "googleHealth.backfill.resumedPast": skipped } }); - } + + const finished = done.size - doneBefore; + const sliceReport = { + slice, + elapsedMs: Date.now() - startedAt, + pages: budget.pages, + stopReason: budget.stopReason, + imported, + collectionsDone: done.size, + 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])), + }; + 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, + }); if (busy) { throw new Error( @@ -166,10 +242,10 @@ export async function runGoogleHealthBackfillForUser( } if (failed) { - // `done` is the resume set this attempt added to as it finished - // collections; a larger set than it started with is progress. - const finished = done.size - doneBefore; - if (finished > 0 && slice + 1 <= GOOGLE_HEALTH_BACKFILL_MAX_SLICE) { + // Progress is what this slice put on disk: collections finished, and + // edges moved. Either is enough to go on; neither means stuck. + const progressed = finished > 0 || advanced.size > 0; + if (progressed && slice + 1 <= GOOGLE_HEALTH_BACKFILL_MAX_SLICE) { const boss = getGlobalBoss(); const sent = boss ? await enqueueIntegrationBackfillAdmission( @@ -182,7 +258,9 @@ export async function runGoogleHealthBackfillForUser( slice: slice + 1, }, }, - INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + budget.stopReason === "rate_limited" + ? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS + : INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, ) : null; if (sent) { @@ -192,8 +270,10 @@ export async function runGoogleHealthBackfillForUser( details: { imported, slice, + stopReason: budget.stopReason, collectionsFinished: finished, collectionsDone: done.size, + edgesMoved: [...advanced.keys()], }, }, }); @@ -202,12 +282,23 @@ export async function runGoogleHealthBackfillForUser( // No follow-up could be queued; fall through to the retry path so the // work is not dropped. } - // No collection finished in this attempt: the backfill is stuck, not - // slow. Surface through the pg-boss retry path (retryLimit 3, backoff). - // If the failure parked the connection at error_reauth meanwhile, the - // next attempt takes the reauth return above instead of failing again. + // Nothing moved in this slice: the backfill is stuck, not slow. Surface + // through the pg-boss retry path (retryLimit 3, backoff) with what each + // collection came to, so the log line says where it is stuck. If the + // failure parked the connection at error_reauth meanwhile, the next + // attempt takes the reauth return above instead of failing again. + annotate({ + action: { + name: "google_health.backfill.stuck", + details: { slice, stopReason: budget.stopReason }, + }, + }); + const where = collections + .filter((c) => c.status !== "done") + .map((c) => `${c.key}:${c.status}`) + .join(","); throw new Error( - `google-health backfill incomplete for user ${userId} — marker not stamped`, + `google-health backfill incomplete for user ${userId} — marker not stamped (no progress in slice ${slice}${budget.stopReason ? `, stopped on ${budget.stopReason}` : ""}${where ? `; ${where}` : ""})`, ); } @@ -219,7 +310,7 @@ export async function runGoogleHealthBackfillForUser( annotate({ action: { name: "google_health.backfill.complete", - details: { imported }, + details: { imported, slice }, }, }); return { imported }; diff --git a/src/lib/jobs/reminder/google-health-sync.ts b/src/lib/jobs/reminder/google-health-sync.ts index f58329bc0..7ac2e7d0f 100644 --- a/src/lib/jobs/reminder/google-health-sync.ts +++ b/src/lib/jobs/reminder/google-health-sync.ts @@ -15,6 +15,7 @@ import { isGoogleHealthParked, syncUserGoogleHealth, } from "@/lib/google-health/sync"; +import { GOOGLE_HEALTH_POLL_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup"; import { @@ -69,9 +70,15 @@ export async function handleGoogleHealthSync( } try { - const result = await syncUserGoogleHealth(userId); + // A backfill holds the account for one slice at a time, so + // the tick waits for the slice to end rather than skipping the + // account for as long as the whole import runs. + const result = await syncUserGoogleHealth(userId, { + waitForLockMs: GOOGLE_HEALTH_POLL_LOCK_WAIT_MS, + }); // Another run of this account (its backfill, a manual sync) - // holds the account; this tick leaves it to that run. + // still held the account after the wait; this tick leaves it + // to that run. if (result.busy) return { status: "skipped", imported: 0 }; const parked = result.failed && diff --git a/tests/integration/google-health-backfill-rate-limit.test.ts b/tests/integration/google-health-backfill-rate-limit.test.ts index 2cf7bd9a1..3ddb049fa 100644 --- a/tests/integration/google-health-backfill-rate-limit.test.ts +++ b/tests/integration/google-health-backfill-rate-limit.test.ts @@ -57,6 +57,8 @@ const fake = vi.hoisted(() => ({ throttledPages: new Set(), /** Revoke the current access token when this heart-rate page is asked. */ revokeAtPage: -1, + /** Answer every heart-rate request with a 500. */ + failHeartRate: false, /** Hold the request for this heart-rate page until `release` is called. */ holdAtPage: -1, release: null as null | (() => void), @@ -138,13 +140,24 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { { "retry-after": "0" }, ); } + if (fake.failHeartRate) { + return json({ error: { code: 500, status: "INTERNAL" } }, 500); + } if (page === fake.holdAtPage) { fake.holdAtPage = -1; await fake.held; } - const end = Math.min(fake.heartRatePoints, (page + 1) * 10); + // Newest first, one point a minute; a resumed walk's `<=` edge + // filter starts the list at the edge. + const edge = /physical_time <= "([^"]+)"/.exec( + u.searchParams.get("filter") ?? "", + )?.[1]; + const first = edge + ? Math.ceil((fake.now - Date.parse(edge)) / 60_000) + : 0; + const end = Math.min(fake.heartRatePoints, first + (page + 1) * 10); const dataPoints = []; - for (let g = page * 10; g < end; g++) { + for (let g = first + page * 10; g < end; g++) { dataPoints.push({ name: `users/me/dataTypes/heart-rate/dataPoints/${1_000_000 + g}`, heartRate: { @@ -221,6 +234,7 @@ beforeEach(async () => { fake.throttledPages = new Set(); fake.revokeAtPage = -1; fake.holdAtPage = -1; + fake.failHeartRate = false; fake.release = null; fake.held = null; fake.requestsByPath = new Map(); @@ -398,8 +412,19 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () ); expect(done).not.toContain("fetchHeartRate"); + // And the heart-rate stream keeps the edge it reached: the oldest of + // the thirty samples it wrote. + expect( + (afterFirst?.backfillProgress as { cursors: Record }) + .cursors.fetchHeartRate, + ).toMatchObject({ + at: new Date(fake.now - 29 * 60_000).toISOString(), + pages: 3, + }); + // Second attempt, the queued slice: the finished collections are not - // read again. + // read again, and the stream starts at its edge, not at its newest page: + // the 31 samples from the edge on are four pages, not six. fake.requestsByPath = new Map(); await runGoogleHealthBackfillForUser(USER, undefined, 1); @@ -408,7 +433,7 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () expect( fake.requestsByPath.get("steps/dataPoints:dailyRollUp"), ).toBeUndefined(); - expect(fake.requestsByPath.get("heart-rate/dataPoints")).toBe(6); + expect(fake.requestsByPath.get("heart-rate/dataPoints")).toBe(4); expect(await pulseRows()).toBe(60); const afterSecond = await getPrismaClient().googleHealthConnection.findUnique({ @@ -418,7 +443,7 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () expect(afterSecond?.backfillProgress).toBeNull(); }); - it("keeps failing a backfill whose slice finished nothing new", async () => { + it("continues a slice that only moved the stream's edge, and fails one that moved nothing", async () => { const { runGoogleHealthBackfillForUser } = await import("@/lib/jobs/google-health-backfill"); @@ -428,13 +453,29 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () await runGoogleHealthBackfillForUser(USER, stopAtThirdPage); expect(sent.jobs).toHaveLength(1); - // The next slice cannot get through the heart-rate stream either: no - // collection finishes, so this one is stuck and says so. + // The next slice finishes no collection either, but it moves the edge + // thirty samples further back: progress, so it continues. fake.requestsByPath = new Map(); await expect( runGoogleHealthBackfillForUser(USER, stopAtThirdPage, 1), - ).rejects.toThrow(/incomplete/); - expect(sent.jobs).toHaveLength(1); + ).resolves.toMatchObject({ continued: true }); + expect(sent.jobs).toHaveLength(2); + + // A slice whose every heart-rate request fails moves nothing: it is + // stuck, says where, and queues nothing. + const { GOOGLE_HEALTH_TRANSIENT_RETRY } = + await import("@/lib/google-health/client"); + const baseDelay = GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs; + GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = 0; + fake.failHeartRate = true; + try { + await expect( + runGoogleHealthBackfillForUser(USER, undefined, 2), + ).rejects.toThrow(/no progress in slice 2; fetchHeartRate:failed/); + } finally { + GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = baseDelay; + } + expect(sent.jobs).toHaveLength(2); const row = await getPrismaClient().googleHealthConnection.findUnique({ where: { userId: USER }, }); @@ -459,27 +500,32 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () } expect(fake.holdAtPage).toBe(-1); - const requestsBefore = [...fake.requestsByPath.values()].reduce( - (a, b) => a + b, - 0, - ); + const total = () => + [...fake.requestsByPath.values()].reduce((a, b) => a + b, 0); + const requestsBefore = total(); const manual = await syncUserGoogleHealth(USER); expect(manual.busy).toBe(true); - const hourly = await handleGoogleHealthSync([] as never); - expect(hourly).toMatchObject({ - ok: true, - did: { users_skipped: 1, users_parked: 0, users_failed: 0 }, + // The hourly poll waits for the account instead of skipping it: it does + // nothing while the backfill holds the lock ... + let hourlySettled = false; + const hourly = handleGoogleHealthSync([] as never).finally(() => { + hourlySettled = true; }); - const requestsAfter = [...fake.requestsByPath.values()].reduce( - (a, b) => a + b, - 0, - ); - expect(requestsAfter).toBe(requestsBefore); + await new Promise((r) => setTimeout(r, 2_500)); + expect(hourlySettled).toBe(false); + expect(total()).toBe(requestsBefore); + // ... and runs once the backfill lets go. fake.release?.(); await backfill; + const requestsAfterBackfill = total(); + expect(await hourly).toMatchObject({ + ok: true, + did: { users_skipped: 0, users_parked: 0, users_failed: 0 }, + }); + expect(total()).toBeGreaterThan(requestsAfterBackfill); expect(await pulseRows()).toBe(60); expect(await ledgerState()).toBe("connected"); - }); + }, 60_000); });