diff --git a/src/lib/google-health/__tests__/backfill-resume-progress.test.ts b/src/lib/google-health/__tests__/backfill-resume-progress.test.ts new file mode 100644 index 000000000..aa80d605d --- /dev/null +++ b/src/lib/google-health/__tests__/backfill-resume-progress.test.ts @@ -0,0 +1,65 @@ +/** + * A collection counts toward a backfill's progress only once its resume note + * is written. The job reads the size of `done` as progress and queues another + * slice on it, so a key added before a write that failed would let a chain of + * slices "progress" over the same collection without end. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { prismaMock, warnings } = vi.hoisted(() => ({ + prismaMock: { + googleHealthConnection: { update: vi.fn() }, + }, + warnings: [] as string[], +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/logging/context", () => ({ + annotate: () => {}, + getEvent: () => ({ addWarning: (w: string) => warnings.push(w) }), +})); + +import { + runGoogleHealthCollection, + runWithGoogleHealthBackfillResume, +} from "../sync-core"; + +beforeEach(() => { + vi.clearAllMocks(); + warnings.length = 0; +}); + +describe("runGoogleHealthCollection under a backfill", () => { + it("adds a clean collection to done once its note is written", async () => { + prismaMock.googleHealthConnection.update.mockResolvedValue({}); + const done = new Set(["fetchSteps"]); + + await runWithGoogleHealthBackfillResume("u1", done, () => + runGoogleHealthCollection("sleep", async () => 3), + ); + + expect([...done]).toEqual(["fetchSteps", "sleep"]); + expect(prismaMock.googleHealthConnection.update).toHaveBeenCalledWith({ + where: { userId: "u1" }, + data: { backfillProgress: { done: ["fetchSteps", "sleep"] } }, + }); + }); + + it("counts no progress when the note cannot be written", async () => { + prismaMock.googleHealthConnection.update.mockRejectedValue( + new Error("connection reset"), + ); + const done = new Set(["fetchSteps"]); + + const { result } = await runWithGoogleHealthBackfillResume("u1", done, () => + runGoogleHealthCollection("sleep", async () => 3), + ); + + // The rows are in, the collection did not fail, but it is not finished. + expect(result).toBe(3); + expect([...done]).toEqual(["fetchSteps"]); + expect(warnings.some((w) => w.includes("progress write failed"))).toBe( + true, + ); + }); +}); diff --git a/src/lib/google-health/collection-keys.ts b/src/lib/google-health/collection-keys.ts new file mode 100644 index 000000000..f6304891c --- /dev/null +++ b/src/lib/google-health/collection-keys.ts @@ -0,0 +1,29 @@ +/** + * Every collection a Google Health backfill walks, by the key its resume note + * stores (`runGoogleHealthCollection`). The key type is what that function + * accepts, so a collection added without an entry here does not compile, and + * the backfill's slice ceiling (`GOOGLE_HEALTH_BACKFILL_MAX_SLICE`) is derived + * from this list rather than counted by hand. + */ +export const GOOGLE_HEALTH_COLLECTION_KEYS = [ + "fetchSteps", + "fetchDistance", + "fetchActiveEnergy", + "fetchFloors", + "fetchVo2Max", + "fetchWeight", + "fetchBodyFat", + "fetchSpo2", + "fetchHrv", + "fetchRhr", + "fetchRespiratoryRate", + "fetchHeartRate", + "fetchBloodGlucose", + "fetchCoreBodyTemperature", + "fetchSleepTemperature", + "fetchExercise", + "sleep", +] as const; + +export type GoogleHealthCollectionKey = + (typeof GOOGLE_HEALTH_COLLECTION_KEYS)[number]; diff --git a/src/lib/google-health/sync-activity.ts b/src/lib/google-health/sync-activity.ts index 32344e993..0c17e2209 100644 --- a/src/lib/google-health/sync-activity.ts +++ b/src/lib/google-health/sync-activity.ts @@ -46,6 +46,7 @@ import { type GoogleHealthResourceSyncOptions, } from "./sync-core"; import { annotate } from "@/lib/logging/context"; +import type { GoogleHealthCollectionKey } from "./collection-keys"; import { resolveUserTimezone } from "@/lib/tz/resolver"; /** One mappable activity metric: its data-type encoding + the mapper + a verb. */ @@ -55,7 +56,7 @@ interface ActivityResource { point: Record, tz: string, ) => GoogleHealthMappedMeasurement[]; - verb: string; + verb: GoogleHealthCollectionKey; } /** The four cumulative daily totals — read via `POST :dailyRollUp`. */ diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index de64e573d..dc215266d 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -20,6 +20,7 @@ import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; import { encrypt, decrypt } from "@/lib/crypto"; import { annotate, getEvent } from "@/lib/logging/context"; +import type { GoogleHealthCollectionKey } from "./collection-keys"; import { userDayKey } from "@/lib/tz/format"; import { isP2002 } from "@/lib/prisma-errors"; import { @@ -464,7 +465,7 @@ export async function runWithGoogleHealthBackfillResume( * budget asking it to stop). Anything less is walked again next attempt. */ export async function runGoogleHealthCollection( - key: string, + key: GoogleHealthCollectionKey, fn: () => Promise, ): Promise { const resume = backfillResumeStorage.getStore(); @@ -483,19 +484,27 @@ export async function runGoogleHealthCollection( !truncatedBefore && !googleHealthOutcomeTruncated(); if (clean) { - resume.done.add(key); - await prisma.googleHealthConnection - .update({ + // 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 + // slice on it; a key added before a write that then failed would be + // 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] } }, - }) - .catch((err) => - // Losing the note costs a re-walk of this collection next attempt, - // nothing more; it must not fail a collection that did complete. - getEvent()?.addWarning( - `google-health: backfill progress write failed: ${err}`, - ), + data: { backfillProgress: { done: [...resume.done, key] } }, + }); + resume.done.add(key); + } catch (err) { + // Not finished, then: the collection is walked again next attempt. It + // must not fail a collection whose rows are written. + getEvent()?.addWarning( + `google-health: backfill progress write failed for ${key}: ${err}`, ); + annotate({ + meta: { "googleHealth.backfill.progressWriteFailed": key }, + }); + } } return imported; } diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index 59deed1d7..a86dfb59e 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -55,6 +55,7 @@ import { type GoogleHealthResourceSyncOptions, } from "./sync-core"; import { getEvent } from "@/lib/logging/context"; +import type { GoogleHealthCollectionKey } from "./collection-keys"; import { prisma } from "@/lib/db"; import { resolveUserTimezone } from "@/lib/tz/resolver"; @@ -65,7 +66,7 @@ interface MetricResource { point: Record, tz: string, ) => GoogleHealthMappedMeasurement[]; - verb: string; + verb: GoogleHealthCollectionKey; /** Page ceiling for the walk; the client default when omitted. */ maxPages?: number; /** diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index 8afde8dd0..4e5e5a269 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -69,7 +69,10 @@ vi.mock("@/lib/google-health/sync", () => ({ syncUserGoogleHealth: (...a: unknown[]) => syncUserGoogleHealthMock(...a), })); -import { runGoogleHealthBackfillForUser } from "../google-health-backfill"; +import { + GOOGLE_HEALTH_BACKFILL_MAX_SLICE, + runGoogleHealthBackfillForUser, +} from "../google-health-backfill"; import { GOOGLE_HEALTH_DATA_TYPES, forEachDataPointPage, @@ -229,6 +232,39 @@ describe("runGoogleHealthBackfillForUser — slices", () => { ); }); + it("fails a slice past the ceiling without walking or queueing", async () => { + syncUserGoogleHealthMock.mockResolvedValue({ imported: 1, failed: true }); + resumeState.finishes = ["sleep"]; + + await expect( + runGoogleHealthBackfillForUser( + "u1", + undefined, + GOOGLE_HEALTH_BACKFILL_MAX_SLICE + 1, + ), + ).rejects.toThrow(/past the ceiling/); + expect(syncUserGoogleHealthMock).not.toHaveBeenCalled(); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("does not queue a slice past the ceiling even when this one progressed", async () => { + resumeState.finishes = ["sleep"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 1, failed: true }); + + await expect( + runGoogleHealthBackfillForUser( + "u1", + undefined, + GOOGLE_HEALTH_BACKFILL_MAX_SLICE, + ), + ).rejects.toThrow(/incomplete/); + 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("stamps the marker on the slice that finishes the history", async () => { resumeState.doneBefore = ["steps", "sleep"]; resumeState.finishes = ["weight"]; diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index bb3e66b42..4f3a00294 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -30,6 +30,7 @@ import { } 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 { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys"; export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill"; @@ -40,6 +41,17 @@ 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. + */ +export const GOOGLE_HEALTH_BACKFILL_MAX_SLICE = + GOOGLE_HEALTH_COLLECTION_KEYS.length + 2; + export interface GoogleHealthBackfillPayload { userId: string; enqueuedAt: string; @@ -78,7 +90,10 @@ export interface GoogleHealthBackfillPayload { * `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 - * bounded by the number of collections, so the chain of slices is too. + * 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. * * 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 @@ -90,6 +105,17 @@ export async function runGoogleHealthBackfillForUser( shouldStop: () => boolean = () => false, slice: number = 0, ): Promise<{ imported: number; continued?: boolean }> { + if (slice > GOOGLE_HEALTH_BACKFILL_MAX_SLICE) { + annotate({ + action: { + name: "google_health.backfill.slice_ceiling", + details: { slice, maxSlice: GOOGLE_HEALTH_BACKFILL_MAX_SLICE }, + }, + }); + throw new Error( + `google-health backfill for user ${userId} reached slice ${slice}, past the ceiling of ${GOOGLE_HEALTH_BACKFILL_MAX_SLICE}`, + ); + } if (await isGoogleHealthParked(userId)) { annotate({ action: { @@ -143,7 +169,7 @@ export async function runGoogleHealthBackfillForUser( // `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) { + if (finished > 0 && slice + 1 <= GOOGLE_HEALTH_BACKFILL_MAX_SLICE) { const boss = getGlobalBoss(); const sent = boss ? await enqueueIntegrationBackfillAdmission(