From e4c78e1d0c972c7d8891ef0e87a96426ed8c58d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 19 Oct 2026 20:10:12 +0200 Subject: [PATCH 1/2] fix(google-health): give a stalled walk a second attempt before it ends A walk that stopped for lack of progress ended its data type for good, so a passing fault in Google's paging closed that history until a reconnect. A walk that had written nothing yet was also noted as ending today. The first end without progress now parks the data type with its edge: it is not finished, not published, and is walked again a day later. Once nothing else is left, the next slice is queued for when it is due, and the marker is never stamped over a parked type. A second end without progress is final; a second attempt that moves on is an ordinary walk again. A walk that stalls before writing anything records no end and fails the slice as stuck. Refs #1220 --- docs/integrations/google-health.md | 30 ++- .../__tests__/backfill-bounds.test.ts | 197 +++++++++++++++--- src/lib/google-health/backfill-bounds.ts | 46 ++++ src/lib/google-health/sync-core.ts | 129 ++++++++++-- src/lib/google-health/sync-metrics.ts | 11 +- .../__tests__/google-health-backfill.test.ts | 81 ++++++- src/lib/jobs/google-health-backfill.ts | 106 +++++++--- ...gle-health-backfill-progress-bound.test.ts | 105 ++++++++-- 8 files changed, 601 insertions(+), 104 deletions(-) diff --git a/docs/integrations/google-health.md b/docs/integrations/google-health.md index c4ddac290..50c8890d5 100644 --- a/docs/integrations/google-health.md +++ b/docs/integrations/google-health.md @@ -187,14 +187,23 @@ year of history fills depends on the device. Two bounds apply instead: - **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. + the walk stops where it got to. A passing fault in Google's paging looks + the same as the end of what it holds, so the first time this happens the + data type only waits: it keeps its place, the rest of the backfill goes + on, and a day later it is walked again from there. If that second attempt + also stops without getting anywhere, the data type's history ends at the + oldest reading written so far. If it moves on, it is an ordinary walk + again. A walk that stops before it has written a single reading notes no + end at all (there is no date it could be said to end at); the slice fails + as stuck and is retried like any other. 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. +Either way the data type counts as finished once its end is final, 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. While a type waits +for its second attempt the backfill is not complete, and the card still +says older history is loading. 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 @@ -249,7 +258,12 @@ reading: - `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`). + `reason` (`horizon` or `no_progress`). A `no_progress` entry also + carries `attempts` and `endedAt`; with one attempt it is waiting for + its second, a day after `endedAt`, and is not yet published. +- `action: google_health.backfill.retry_scheduled` marks a slice that found + nothing left but a data type waiting for its second attempt and queued the + next slice for when it is due (`delaySeconds`). - `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: diff --git a/src/lib/google-health/__tests__/backfill-bounds.test.ts b/src/lib/google-health/__tests__/backfill-bounds.test.ts index 2e3cbb6b7..9860b0022 100644 --- a/src/lib/google-health/__tests__/backfill-bounds.test.ts +++ b/src/lib/google-health/__tests__/backfill-bounds.test.ts @@ -29,10 +29,12 @@ import { googleHealthBackfillHorizonYears, googleHealthWalkProgress, noteGoogleHealthWalkPage, + type GoogleHealthBackfillEnd, } from "../backfill-bounds"; import { googleHealthBackfillEndedList, googleHealthBackfillProgressAfterCompletion, + googleHealthBackfillRetryPending, readGoogleHealthBackfillEnded, recordGoogleHealthBackfillEnd, runGoogleHealthCollection, @@ -253,69 +255,202 @@ describe("a collection ended short", () => { prismaMock.googleHealthConnection.update.mockResolvedValue({}); }); + const HOUR = 3_600_000; const END = { type: "PULSE" as const, at: "2024-05-11T00:00:00.000Z", reason: "no_progress" as const, }; + const HORIZON_END = { + type: "WEIGHT" as const, + at: "2006-10-17T12:00:00.000Z", + reason: "horizon" as const, + }; + /** An end without progress `attempts` deep, the last `hoursAgo` back. */ + const parked = (attempts: number, hoursAgo: number) => ({ + ...END, + attempts, + endedAt: new Date(Date.now() - hoursAgo * HOUR).toISOString(), + }); - 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; - }), + /** One slice walking heart rate with `walk`, from `ended`. */ + async function slice( + walk: () => Promise, + ended = new Map(), + done = new Set(), + ) { + return runWithGoogleHealthClientOutcome(() => + runWithGoogleHealthBackfillResume( + "u1", + done, + () => runGoogleHealthCollection("fetchHeartRate", walk), + new Map(), + new Set(), + ended, ), ).then((r) => r.result); + } + + /** A walk stopped for no progress: truncated, and the end noted. */ + const stalls = vi.fn(async () => { + noteGoogleHealthDeferred(); + await recordGoogleHealthBackfillEnd("fetchHeartRate", END); + return 0; + }); + + it("parks a first end without progress: not finished, edge kept, not published", async () => { + const done = new Set(); + const { ended, collections } = await slice(stalls, new Map(), done); expect(collections).toEqual([ - expect.objectContaining({ key: "fetchHeartRate", status: "done" }), + expect.objectContaining({ key: "fetchHeartRate", status: "stopped" }), ]); - expect([...ended]).toEqual([["fetchHeartRate", END]]); + expect(done.has("fetchHeartRate")).toBe(false); + const note = ended.get("fetchHeartRate")!; + expect(note).toMatchObject({ ...END, attempts: 1 }); + expect(Date.now() - Date.parse(note.endedAt!)).toBeLessThan(60_000); const written = prismaMock.googleHealthConnection.update.mock.calls.at( -1, )![0].data.backfillProgress as unknown; expect( readGoogleHealthBackfillEnded(written).get("fetchHeartRate"), - ).toEqual(END); + ).toEqual(note); + // Waiting for another try is not where the history ends. + expect(googleHealthBackfillEndedList(written)).toEqual([]); + expect(googleHealthBackfillProgressAfterCompletion(ended)).toBeNull(); }); - it("is skipped by the next slice instead of walked into the same wall", async () => { + it("is not walked again before its day is up", 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]]), + const { skipped } = await slice( + walk, + new Map([ + ["fetchHeartRate", parked(1, 23)], + ]), ); expect(walk).not.toHaveBeenCalled(); expect(skipped).toEqual(["fetchHeartRate"]); }); - it("outlives the completion stamp, and nothing else does", () => { + it("is walked again once its day is up, and a second end without progress is final", async () => { + const done = new Set(); + const { ended, collections } = await slice( + stalls, + new Map([ + ["fetchHeartRate", parked(1, 25)], + ]), + done, + ); + expect(stalls).toHaveBeenCalledTimes(1); + expect(collections).toEqual([ + expect.objectContaining({ key: "fetchHeartRate", status: "done" }), + ]); + expect(done.has("fetchHeartRate")).toBe(true); + expect(ended.get("fetchHeartRate")).toMatchObject({ ...END, attempts: 2 }); + expect( + googleHealthBackfillProgressAfterCompletion(ended)!.ended.fetchHeartRate, + ).toMatchObject({ ...END, attempts: 2 }); + }); + + it("goes back to an ordinary walk when the second attempt gets somewhere", async () => { + const { ended, collections } = await slice( + async () => { + // Moved on, then ran out of slice: no end this time. + noteGoogleHealthDeferred(); + return 5; + }, + new Map([ + ["fetchHeartRate", parked(1, 25)], + ]), + ); + expect(collections).toEqual([ + expect.objectContaining({ key: "fetchHeartRate", status: "stopped" }), + ]); + expect(ended.has("fetchHeartRate")).toBe(false); + }); + + it("is skipped for good once final, and so is the horizon", async () => { + const walk = vi.fn(async () => 1); + const { skipped } = await slice( + walk, + new Map([ + ["fetchHeartRate", parked(2, 100)], + ]), + ); + expect(walk).not.toHaveBeenCalled(); + expect(skipped).toEqual(["fetchHeartRate"]); + + const atHorizon = await slice( + walk, + new Map([ + ["fetchHeartRate", { ...HORIZON_END, type: "PULSE" as const }], + ]), + ); + expect(walk).not.toHaveBeenCalled(); + expect(atHorizon.skipped).toEqual(["fetchHeartRate"]); + }); + + it("finishes the collection at once at the horizon", async () => { + const done = new Set(); + const { collections } = await slice( + async () => { + await recordGoogleHealthBackfillEnd("fetchHeartRate", { + ...HORIZON_END, + type: "PULSE", + }); + return 3; + }, + new Map(), + done, + ); + expect(collections).toEqual([ + expect.objectContaining({ key: "fetchHeartRate", status: "done" }), + ]); + expect(done.has("fetchHeartRate")).toBe(true); + }); + + it("outlives the completion stamp when final, and nothing else does", () => { expect(googleHealthBackfillProgressAfterCompletion(new Map())).toBeNull(); expect( googleHealthBackfillProgressAfterCompletion( - new Map([["fetchHeartRate", END]]), + new Map([ + ["fetchWeight", HORIZON_END], + ["fetchHeartRate", parked(1, 1)], + ]), ), - ).toEqual({ ended: { fetchHeartRate: END } }); + ).toEqual({ ended: { fetchWeight: HORIZON_END } }); }); - it("is published per type, and a malformed note is dropped", () => { + it("names the earliest retry among the parked ones", () => { + const early = parked(1, 20); + expect( + googleHealthBackfillRetryPending( + new Map([ + ["fetchWeight", HORIZON_END], + ["fetchHeartRate", early], + ["fetchHrv", parked(1, 2)], + ["fetchSpo2", parked(2, 30)], + ]), + ), + ).toEqual({ + keys: ["fetchHeartRate", "fetchHrv"], + dueAt: Date.parse(early.endedAt) + 24 * HOUR, + }); + expect( + googleHealthBackfillRetryPending( + new Map([ + ["fetchWeight", HORIZON_END], + ]), + ), + ).toBeNull(); + }); + + it("is published per type when final, and a malformed note is dropped", () => { expect( googleHealthBackfillEndedList({ ended: { - fetchHeartRate: END, - fetchWeight: { - type: "WEIGHT", - at: "2006-10-17T12:00:00.000Z", - reason: "horizon", - }, + fetchHeartRate: parked(2, 30), + fetchRhr: parked(1, 1), + fetchWeight: HORIZON_END, 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" }, diff --git a/src/lib/google-health/backfill-bounds.ts b/src/lib/google-health/backfill-bounds.ts index 695be1e95..2b5fe8ab0 100644 --- a/src/lib/google-health/backfill-bounds.ts +++ b/src/lib/google-health/backfill-bounds.ts @@ -140,4 +140,50 @@ export interface GoogleHealthBackfillEnd { type: MeasurementType; at: string; reason: GoogleHealthBackfillEndReason; + /** + * `no_progress` only: how many separate attempts have ended this way, and + * when the last one did (see `googleHealthBackfillEndIsFinal`). + */ + attempts?: number; + endedAt?: string; +} + +/** + * Separate attempts that must end without progress before a data type's + * history counts as ending there. One is not enough: Google handing out the + * same page for a while is as likely a passing fault in its paging as the end + * of what it holds, and a reconnect would be the only way back. + */ +export const GOOGLE_HEALTH_NO_PROGRESS_ATTEMPTS = 2; + +/** How long after an attempt ended without progress the next one starts. */ +export const GOOGLE_HEALTH_NO_PROGRESS_RETRY_MS = 24 * 60 * 60 * 1000; + +/** + * Whether an end is where the history stays: always at the horizon, and + * after `GOOGLE_HEALTH_NO_PROGRESS_ATTEMPTS` attempts without progress. + * Until then the data type waits for its next attempt, unfinished, and is not + * published as ending anywhere. + */ +export function googleHealthBackfillEndIsFinal( + end: GoogleHealthBackfillEnd, +): boolean { + return ( + end.reason === "horizon" || + (end.attempts ?? 1) >= GOOGLE_HEALTH_NO_PROGRESS_ATTEMPTS + ); +} + +/** + * When an end that is not final may be walked again (epoch ms), or null for + * a final one. An end without a readable time is due at once. + */ +export function googleHealthBackfillEndRetryAt( + end: GoogleHealthBackfillEnd, +): number | null { + if (googleHealthBackfillEndIsFinal(end)) return null; + const endedAt = end.endedAt ? Date.parse(end.endedAt) : Number.NaN; + return Number.isNaN(endedAt) + ? 0 + : endedAt + GOOGLE_HEALTH_NO_PROGRESS_RETRY_MS; } diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index 19dc1d523..e33735991 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -22,7 +22,11 @@ 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 { + googleHealthBackfillEndIsFinal, + googleHealthBackfillEndRetryAt, + type GoogleHealthBackfillEnd, +} from "./backfill-bounds"; import { userDayKey } from "@/lib/tz/format"; import { isP2002 } from "@/lib/prisma-errors"; import { @@ -491,6 +495,11 @@ interface BackfillResumeTracker { edgeFilterUnsupported: Set; /** Collections whose walk ended short of the history's start, and where. */ ended: Map; + /** + * Collections walked again this attempt after an earlier one ended without + * progress, with how many attempts had ended that way. + */ + retrying: Map; } const backfillResumeStorage = new AsyncLocalStorage(); @@ -567,10 +576,22 @@ export function readGoogleHealthBackfillEnded( if (typeof e.type !== "string" || !(e.type in MeasurementTypes)) { continue; } + const attempts = + typeof e.attempts === "number" && + Number.isInteger(e.attempts) && + e.attempts >= 1 + ? e.attempts + : undefined; + const endedAt = + typeof e.endedAt === "string" && !Number.isNaN(Date.parse(e.endedAt)) + ? new Date(e.endedAt).toISOString() + : undefined; out.set(key, { type: e.type as MeasurementType, at: new Date(at).toISOString(), reason: e.reason, + ...(attempts !== undefined ? { attempts } : {}), + ...(endedAt !== undefined ? { endedAt } : {}), }); } return out; @@ -587,11 +608,16 @@ export interface GoogleHealthBackfillEndedEntry { reason: GoogleHealthBackfillEnd["reason"]; } -/** The `backfillProgress` column's ended collections, as published. */ +/** + * The `backfillProgress` column's ended collections, as published: only the + * final ends. One still waiting for another attempt is not where its history + * ends yet. + */ export function googleHealthBackfillEndedList( value: unknown, ): GoogleHealthBackfillEndedEntry[] { return [...readGoogleHealthBackfillEnded(value).values()] + .filter(googleHealthBackfillEndIsFinal) .map((e) => ({ type: e.type, endsAt: e.at, reason: e.reason })) .sort((a, b) => a.type.localeCompare(b.type)); } @@ -603,23 +629,60 @@ export function googleHealthBackfillEndedList( export function googleHealthBackfillProgressAfterCompletion( ended: Map, ): { - ended: Record; + ended: Record; } | null { - return ended.size > 0 ? { ended: endedJson(ended) } : null; + const final = new Map( + [...ended].filter(([, e]) => googleHealthBackfillEndIsFinal(e)), + ); + return final.size > 0 ? { ended: endedJson(final) } : null; } +/** One ended collection as the column stores it. */ +type GoogleHealthBackfillEndJson = { + type: string; + at: string; + reason: string; + attempts?: number; + endedAt?: string; +}; + /** The ended collections as the column stores them. */ function endedJson( ended: Map, -): Record { +): Record { return Object.fromEntries( [...ended].map(([key, e]) => [ key, - { type: e.type, at: e.at, reason: e.reason }, + { + type: e.type, + at: e.at, + reason: e.reason, + ...(e.attempts !== undefined ? { attempts: e.attempts } : {}), + ...(e.endedAt !== undefined ? { endedAt: e.endedAt } : {}), + }, ]), ); } +/** + * The collections waiting for another attempt after one ended without + * progress, and the earliest instant (epoch ms) one of them is due, or null + * when none waits. + */ +export function googleHealthBackfillRetryPending( + ended: Map, +): { keys: string[]; dueAt: number } | null { + const keys: string[] = []; + let dueAt = Infinity; + for (const [key, e] of ended) { + const at = googleHealthBackfillEndRetryAt(e); + if (at === null) continue; + keys.push(key); + dueAt = Math.min(dueAt, at); + } + return keys.length > 0 ? { keys, dueAt } : null; +} + /** * Read the `backfillProgress` column into the collections Google refused the * edge filter on. @@ -728,6 +791,7 @@ export async function runWithGoogleHealthBackfillResume( collections: [], edgeFilterUnsupported, ended, + retrying: new Map(), }; const result = await backfillResumeStorage.run(tracker, fn); return { @@ -779,20 +843,32 @@ export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): { /** * 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. + * history, at `at`, because of `reason` (see `backfill-bounds.ts`). A final + * end (`googleHealthBackfillEndIsFinal`) counts the collection as finished + * for this backfill (`runGoogleHealthCollection`) and outlives the + * completion stamp, so the card can say where the history ends. A first end + * without progress parks the collection with its edge until its next + * attempt is due. Outside a backfill this does nothing. */ export async function recordGoogleHealthBackfillEnd( key: GoogleHealthCollectionKey, - end: GoogleHealthBackfillEnd, + end: Pick, ): Promise { const resume = backfillResumeStorage.getStore(); if (!resume) return; const ended = new Map(resume.ended); - ended.set(key, end); + ended.set( + key, + end.reason === "no_progress" + ? { + ...end, + // Counted across attempts: the one walked again this attempt + // carries the count it was parked with. + attempts: (resume.retrying.get(key) ?? 0) + 1, + endedAt: new Date().toISOString(), + } + : end, + ); annotate({ meta: { "googleHealth.backfill.ended": Object.fromEntries(ended) }, }); @@ -974,7 +1050,13 @@ export async function runGoogleHealthCollection( ): Promise { const resume = backfillResumeStorage.getStore(); if (!resume) return fn(); - if (resume.done.has(key) || resume.ended.has(key)) { + const parked = resume.ended.get(key); + if ( + resume.done.has(key) || + (parked && + (googleHealthBackfillEndIsFinal(parked) || + googleHealthBackfillEndRetryAt(parked)! > Date.now())) + ) { resume.skipped.push(key); return 0; } @@ -988,6 +1070,13 @@ export async function runGoogleHealthCollection( }); return 0; } + if (parked) { + // Its earlier attempt ended without progress and its wait is over: walk + // it again from its edge. The note goes with the next write; the count + // stays with the attempt, so a second end without progress is final. + resume.retrying.set(key, parked.attempts ?? 1); + resume.ended.delete(key); + } const startedAt = performance.now(); const pagesBefore = googleHealthSlicePagesRead(); const edgeFrom = resume.cursors.get(key)?.at ?? null; @@ -999,11 +1088,13 @@ export async function runGoogleHealthCollection( (hardFailStorage.getStore()?.failures.length ?? 0) !== hardBefore || (softSkipStorage.getStore()?.count ?? 0) !== softBefore; const truncated = truncatedBefore || googleHealthOutcomeTruncated(); - // 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); + // A walk that ended short for good (the horizon, or no progress on its + // last attempt) has read all it is going to: finished, though the walk may + // report itself truncated. One that ended without progress for the first + // time is not: it keeps its edge and waits for its next attempt. + const end = resume.ended.get(key); + const endedFinal = end !== undefined && googleHealthBackfillEndIsFinal(end); + const clean = !failedHard && (endedFinal || (!truncated && !end)); 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 2d832f6c3..95ed4a564 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -327,15 +327,20 @@ async function syncMetricCollection( ? { 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. + // the walk is noted as ending where it got to. The first time + // it waits a day for another attempt; a second such end is + // where the history stays (`backfill-bounds.ts`). onNoProgress: async () => { // A page whose write failed fails the collection instead. if (edgeHeld) return; + // Nothing written yet: there is no place the history could + // be said to end at, and "now" would be false. The walk + // stays unfinished and fails the slice like any stuck one. const at = googleHealthBackfillCursor(resource.verb)?.edge?.at; + if (!at) return; await recordGoogleHealthBackfillEnd(resource.verb, { type: resource.type, - at: (at ?? new Date()).toISOString(), + at: at.toISOString(), reason: "no_progress", }); }, diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index 390c0b603..d7c802d29 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -37,8 +37,12 @@ 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). */ + /** Collections the mocked walk ends short for good. */ ends: [] as string[], + /** Collections the mocked walk parks after a first end without progress. */ + parks: [] as string[], + /** When the parked ones are due again (epoch ms). */ + retryDueAt: 0, }, })); @@ -67,6 +71,14 @@ vi.mock("@/lib/google-health/sync-core", () => ({ readGoogleHealthBackfillEnded: () => new Map(), googleHealthBackfillProgressAfterCompletion: (ended: Map) => ended.size > 0 ? { ended: Object.fromEntries(ended) } : null, + googleHealthBackfillRetryPending: ( + ended: Map, + ) => { + const keys = [...ended] + .filter(([, e]) => e.reason === "no_progress" && (e.attempts ?? 1) < 2) + .map(([key]) => key); + return keys.length > 0 ? { keys, dueAt: resumeState.retryDueAt } : null; + }, runWithGoogleHealthBackfillResume: async ( _userId: string, done: Set, @@ -78,10 +90,21 @@ vi.mock("@/lib/google-health/sync-core", () => ({ const result = await fn(); for (const key of resumeState.finishes) done.add(key); for (const key of resumeState.ends) { + // Final: finished the way `runGoogleHealthCollection` finishes it. ended.set(key, { type: "PULSE", at: "2024-05-11T00:00:00.000Z", reason: "no_progress", + attempts: 2, + }); + done.add(key); + } + for (const key of resumeState.parks) { + ended.set(key, { + type: "PULSE", + at: "2024-05-11T00:00:00.000Z", + reason: "no_progress", + attempts: 1, }); } return { @@ -104,6 +127,7 @@ import { runGoogleHealthBackfillForUser, } from "../google-health-backfill"; import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission"; +import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys"; import { GOOGLE_HEALTH_DATA_TYPES, forEachDataPointPage, @@ -121,6 +145,8 @@ beforeEach(() => { resumeState.advances = []; resumeState.collections = []; resumeState.ends = []; + resumeState.parks = []; + resumeState.retryDueAt = 0; }); describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { @@ -326,11 +352,64 @@ describe("runGoogleHealthBackfillForUser — slices", () => { type: "PULSE", at: "2024-05-11T00:00:00.000Z", reason: "no_progress", + attempts: 2, }, }, }); }); + it("continues a slice that only parked a collection for another try", async () => { + resumeState.parks = ["fetchHeartRate"]; + resumeState.retryDueAt = Date.now() + 24 * 3_600_000; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: true }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 4), + ).resolves.toEqual({ imported: 0, continued: true }); + // Other collections are still to walk: the next slice comes as usual. + expect(bossSend.mock.calls[0]![2]).toMatchObject({ + startAfter: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + }); + expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); + }); + + it("waits for the retry instead of stamping when only a parked collection is left", async () => { + resumeState.doneBefore = GOOGLE_HEALTH_COLLECTION_KEYS.filter( + (k) => k !== "fetchHeartRate", + ); + resumeState.parks = ["fetchHeartRate"]; + resumeState.retryDueAt = Date.now() + 10 * 3_600_000; + // The parked collection was skipped as not yet due: a clean run. + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: false }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 7), + ).resolves.toEqual({ imported: 0, continued: true }); + expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); + expect(bossSend).toHaveBeenCalledTimes(1); + const startAfter = (bossSend.mock.calls[0]![2] as { startAfter: number }) + .startAfter; + expect(Math.abs(startAfter - 10 * 3600)).toBeLessThanOrEqual(2); + expect(bossSend.mock.calls[0]![1]).toMatchObject({ + data: { slice: 8 }, + }); + }); + + it("does not stamp over a parked collection when the retry cannot be queued", async () => { + resumeState.doneBefore = GOOGLE_HEALTH_COLLECTION_KEYS.filter( + (k) => k !== "fetchHeartRate", + ); + resumeState.parks = ["fetchHeartRate"]; + resumeState.retryDueAt = Date.now() + 3_600_000; + bossSend.mockResolvedValue(null); + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: false }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 7), + ).resolves.toEqual({ imported: 0 }); + expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); + }); + it("continues a slice that only moved a collection's edge", async () => { resumeState.advances = ["fetchHeartRate"]; syncUserGoogleHealthMock.mockResolvedValue({ imported: 900, failed: true }); diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index 4c143ff16..b5e94201f 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -26,8 +26,10 @@ import { readGoogleHealthBackfillDone, readGoogleHealthBackfillEdgeFilterUnsupported, readGoogleHealthBackfillEnded, + googleHealthBackfillRetryPending, runWithGoogleHealthBackfillResume, } from "@/lib/google-health/sync-core"; +import { googleHealthBackfillEndIsFinal } from "@/lib/google-health/backfill-bounds"; import { isGoogleHealthParked, syncUserGoogleHealth, @@ -204,11 +206,20 @@ export async function runGoogleHealthBackfillForUser( 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. + // A collection that ended short for good has read all it is going to, so + // it counts as finished alongside the ones read to the end. One parked + // after a first end without progress does not: it waits for another try. const finishedCount = (): number => - new Set([...done, ...endedBefore.keys()]).size; + new Set([ + ...done, + ...[...endedBefore] + .filter(([, e]) => googleHealthBackfillEndIsFinal(e)) + .map(([key]) => key), + ]).size; const doneBefore = finishedCount(); + const parkedBefore = new Set( + googleHealthBackfillRetryPending(endedBefore)?.keys ?? [], + ); const budget = googleHealthSliceBudget( sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS, @@ -274,28 +285,75 @@ export async function runGoogleHealthBackfillForUser( ); } - if (failed) { - // 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( - boss, - { - kind: "google-health-backfill", - data: { - userId, - enqueuedAt: new Date().toISOString(), - slice: slice + 1, - }, + const sendNextSlice = async (delaySeconds: number): Promise => { + if (slice + 1 > GOOGLE_HEALTH_BACKFILL_MAX_SLICE) return false; + const boss = getGlobalBoss(); + const sent = boss + ? await enqueueIntegrationBackfillAdmission( + boss, + { + kind: "google-health-backfill", + data: { + userId, + enqueuedAt: new Date().toISOString(), + slice: slice + 1, }, - budget.stopReason === "rate_limited" - ? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS - : INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, - ) - : null; + }, + delaySeconds, + ) + : null; + return Boolean(sent); + }; + + // Collections parked after a first end without progress + // (`backfill-bounds.ts`). Once nothing else is left, the next slice waits + // until the earliest of them is due; the marker is never stamped over one. + const pending = googleHealthBackfillRetryPending(ended); + const onlyRetriesLeft = + pending !== null && + GOOGLE_HEALTH_COLLECTION_KEYS.every( + (key) => done.has(key) || pending.keys.includes(key), + ); + if (pending && (onlyRetriesLeft || !failed)) { + const delaySeconds = Math.max( + INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + Math.ceil((pending.dueAt - Date.now()) / 1000), + ); + if (await sendNextSlice(delaySeconds)) { + annotate({ + action: { + name: "google_health.backfill.retry_scheduled", + details: { slice, collections: pending.keys, delaySeconds }, + }, + }); + return { imported, continued: true }; + } + // Not queued (most often one is already waiting under the same slice + // key, sent by an earlier boot's slice). Nothing is lost: the marker is + // not stamped, so the next boot offers the backfill again. + annotate({ + action: { + name: "google_health.backfill.retry_not_queued", + details: { slice, collections: pending.keys }, + }, + }); + return { imported }; + } + + if (failed) { + // Progress is what this slice put on disk: collections finished, edges + // moved, or a collection parked for another attempt. Any is enough to go + // on; none means stuck. + const newlyParked = (pending?.keys ?? []).filter( + (key) => !parkedBefore.has(key), + ).length; + const progressed = finished > 0 || advanced.size > 0 || newlyParked > 0; + if (progressed) { + const sent = await sendNextSlice( + budget.stopReason === "rate_limited" + ? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS + : INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + ); if (sent) { annotate({ action: { diff --git a/tests/integration/google-health-backfill-progress-bound.test.ts b/tests/integration/google-health-backfill-progress-bound.test.ts index d9116e5d8..df945ae9f 100644 --- a/tests/integration/google-health-backfill-progress-bound.test.ts +++ b/tests/integration/google-health-backfill-progress-bound.test.ts @@ -17,8 +17,11 @@ * - 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; + * bounded number of requests; the data type waits a day for a second + * attempt (a restart before then leaves it alone), and only when that + * one stalls too does the backfill complete with the type noted as + * ending there; + * - a walk that stalls before writing anything notes no end at all; * - 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 @@ -46,6 +49,8 @@ const fake = vi.hoisted(() => ({ perPage: 1000, /** Answer every heart-rate request with the newest page and a next token. */ loop: false, + /** Answer every heart-rate request with no readings and a next token. */ + empty: false, /** Hand the history out oldest first instead. */ asc: false, heartRateRequests: 0, @@ -66,6 +71,9 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { return respond({}); } fake.heartRateRequests += 1; + if (fake.empty) { + return respond({ dataPoints: [], nextPageToken: "1" }); + } const filter = u.searchParams.get("filter") ?? ""; const upTo = /physical_time <= "([^"]+)"/.exec(filter)?.[1]; const from = /physical_time >= "([^"]+)"/.exec(filter)?.[1]; @@ -109,13 +117,21 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { }; }); -/** The next slices the backfill queues, recorded. */ -const sent = vi.hoisted(() => ({ slices: [] as number[] })); +/** The next slices the backfill queues, recorded, with their delays. */ +const sent = vi.hoisted(() => ({ + slices: [] as number[], + delays: [] as number[], +})); vi.mock("@/lib/jobs/boss-instance", async (importOriginal) => ({ ...(await importOriginal()), getGlobalBoss: () => ({ - send: async (_queue: string, data: { data?: { slice?: number } }) => { + send: async ( + _queue: string, + data: { data?: { slice?: number } }, + options?: { startAfter?: number }, + ) => { sent.slices.push(data.data?.slice ?? -1); + sent.delays.push(options?.startAfter ?? 0); return `job-${sent.slices.length}`; }, }), @@ -176,9 +192,11 @@ async function seed(opts: { fake.spacing = opts.spacing; fake.perPage = opts.perPage; fake.loop = false; + fake.empty = false; fake.asc = false; fake.heartRateRequests = 0; sent.slices = []; + sent.delays = []; } /** Pages a slice reads here: a fifth of production, so slices stay cheap. */ @@ -186,12 +204,13 @@ 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. + * until one ends the chain or sends the next one an hour or more out (a + * collection waiting for another attempt). Returns how many ran. */ -async function runChain(maxSlices: number): Promise { +async function runChain(maxSlices: number, from = 0): Promise { const { runGoogleHealthBackfillForUser } = await import("@/lib/jobs/google-health-backfill"); - let slice = 0; + let slice = from; let ran = 0; for (; ran < maxSlices; ran++) { const before = sent.slices.length; @@ -204,6 +223,7 @@ async function runChain(maxSlices: number): Promise { if (!outcome.continued) return ran + 1; expect(sent.slices.length).toBe(before + 1); slice = sent.slices.at(-1)!; + if (sent.delays.at(-1)! >= 3600) return ran + 1; } throw new Error(`backfill did not finish within ${maxSlices} slices`); } @@ -311,29 +331,78 @@ describe("Google Health backfill bounded by progress, not pages (#1220)", () => 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 () => { + it("parks a walk Google keeps answering with the same page, tries it again a day later, and only then ends it", 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(); + const walk = 1 + GOOGLE_HEALTH_NO_PROGRESS_PAGES; + + // First attempt: the first page and the twenty that repeat it. Not + // complete, nothing published, and the next slice a day out. + await runChain(5); + expect(fake.heartRateRequests).toBe(walk); + let c = await connection(); + expect(c.backfillCompletedAt).toBeNull(); + expect(googleHealthBackfillEndedList(c.backfillProgress)).toEqual([]); + expect(sent.delays.at(-1)).toBeGreaterThan(23 * 3600); + + // A slice that comes early (a restart) leaves it alone and waits again. + await runChain(1, sent.slices.at(-1)!); + expect(fake.heartRateRequests).toBe(walk); + expect((await connection()).backfillCompletedAt).toBeNull(); + + // A day later the walk goes on from its edge, gets one page further, + // meets the same wall, and this time the history ends there. + const prisma = getPrismaClient(); + const progress = c.backfillProgress as { + ended: Record; + }; + progress.ended.fetchHeartRate!.endedAt = new Date( + Date.now() - 25 * HOUR, + ).toISOString(); + await prisma.googleHealthConnection.update({ + where: { userId: USER }, + data: { backfillProgress: progress }, + }); + await runChain(5, sent.slices.at(-1)!); + + expect(fake.heartRateRequests).toBe(2 * walk); + 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(), + // The oldest reading the second attempt's one real page wrote. + endsAt: new Date(NOW - 1998 * 60_000).toISOString(), reason: "no_progress", }, ]); - expect(await pulseRows()).toBe(1000); + expect(await pulseRows()).toBe(1999); + }, 600_000); + + it("notes no end for a walk that wrote nothing before it stalled", async () => { + // Pages without a single reading, forever: there is no place to say the + // history ends at, and "today" would be false. The slice fails as stuck. + await seed({ total: 5_000, spacing: 60_000, perPage: 1000 }); + fake.empty = true; + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + const { GOOGLE_HEALTH_NO_PROGRESS_PAGES } = + await import("@/lib/google-health/backfill-bounds"); + + await expect( + runGoogleHealthBackfillForUser(USER, () => false, 0, { + slicePages: SLICE_PAGES, + }), + ).rejects.toThrow(/no progress in slice 0/); + expect(fake.heartRateRequests).toBe(GOOGLE_HEALTH_NO_PROGRESS_PAGES); + const c = await connection(); + expect(c.backfillCompletedAt).toBeNull(); + expect(c.backfillProgress).not.toHaveProperty("ended"); + expect(await pulseRows()).toBe(0); }, 600_000); it("reads back to the horizon and notes where the history ends", async () => { From 5bb6ee1f427ea974d4f0ce6457956df53a657287 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 19 Oct 2026 20:10:16 +0200 Subject: [PATCH 2/2] ci: correct the pull-request sizes named in the long integration workflow --- .github/workflows/integration-long.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/integration-long.yml b/.github/workflows/integration-long.yml index e249156b1..f5bdadc42 100644 --- a/.github/workflows/integration-long.yml +++ b/.github/workflows/integration-long.yml @@ -2,7 +2,7 @@ name: Integration tests (long) # The integration tests that take their size from the environment run small # on every pull request and at full size here, once a night. Today that is -# the Google Health multi-year backfill: 45 and 60 days of heart rate on a +# the Google Health multi-year backfill: 120 and 160 days of heart rate on a # pull request, three and four years (some 3.7 million readings, over twenty # minutes) here. See the header of # tests/integration/google-health-backfill-multi-year.test.ts.