diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 09264477d..04ac50ba9 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -19847,6 +19847,14 @@ paths: schema: $ref: "#/components/schemas/ErrorEnvelope" "401": *a1 + "409": + description: Another sync of this account is already running (the history backfill after a reconnect, or the hourly + poll), so none was started. `meta.errorCode` = `sync_in_progress`. `GET /api/google-health/sync/status` + reports the running one. + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorEnvelope" "413": description: Body exceeds 64 KiB. content: @@ -34400,8 +34408,8 @@ components: health_score_config_conflict, insights_layout_conflict, invalid_base_updated_at, json_carries_no_intake_time, json_not_an_array, medication_layout_conflict, missing_required_columns, modules_conflict, mood_tag_layout_conflict, not_configured, notification_prefs_conflict, oidc_only, - private_origin_not_approved, private_origin_not_grantable, rate_limited_self, too_many_rows, - unreadable_json, upstream_invalid_json, vapid_not_configured" + private_origin_not_approved, private_origin_not_grantable, rate_limited_self, sync_in_progress, + too_many_rows, unreadable_json, upstream_invalid_json, vapid_not_configured" type: string capability: description: "On an AI refusal: the capability that is unavailable, one of the keys of `AiCapabilities` (`coach`, diff --git a/docs/integrations/google-health.md b/docs/integrations/google-health.md index ca623a9d0..7fc61e208 100644 --- a/docs/integrations/google-health.md +++ b/docs/integrations/google-health.md @@ -120,6 +120,24 @@ GOOGLE_HEALTH_REDIRECT_URI="https://your-instance.example.com/api/google-health/ 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. + +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. + +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. + ## The 7-day re-consent caveat While the OAuth consent screen stays in **Testing** publishing mode, Google diff --git a/prisma/migrations/0393_google_health_backfill_progress/migration.sql b/prisma/migrations/0393_google_health_backfill_progress/migration.sql new file mode 100644 index 000000000..f061f5961 --- /dev/null +++ b/prisma/migrations/0393_google_health_backfill_progress/migration.sql @@ -0,0 +1,4 @@ +-- The collections an in-flight Google Health history backfill has already +-- walked to the end, so a retried attempt resumes with the collections it did +-- not reach instead of starting over. Null when no backfill is in flight. +ALTER TABLE "google_health_connections" ADD COLUMN "backfill_progress" JSONB; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 49e96f28f..99eb867c2 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -4369,6 +4369,11 @@ 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 + /// instead of starting over. Null once the backfill completes, and reset + /// on every reconnect. + backfillProgress Json? @map("backfill_progress") /// One bounded, owner-scoped current sync run. This is replaced at the start /// of every run and never retains provider payloads or historical runs. syncProgress Json? @map("sync_progress") diff --git a/src/app/api/google-health/callback/route.ts b/src/app/api/google-health/callback/route.ts index a278ffd9e..6c8c8d8f2 100644 --- a/src/app/api/google-health/callback/route.ts +++ b/src/app/api/google-health/callback/route.ts @@ -154,6 +154,9 @@ export const GET = apiHandler(async (request: NextRequest) => { // reconnect CTA and the sync path stops short-circuiting. needsReauth: false, backfillCompletedAt: null, + // A new grant walks history from the start: what an earlier + // backfill finished was read under the old one. + backfillProgress: Prisma.DbNull, }, create: { userId: user.id, diff --git a/src/app/api/google-health/sync/route.ts b/src/app/api/google-health/sync/route.ts index 1987a5170..77f505786 100644 --- a/src/app/api/google-health/sync/route.ts +++ b/src/app/api/google-health/sync/route.ts @@ -104,6 +104,15 @@ export const POST = apiHandler(async (request: NextRequest) => { } const result = await syncUserGoogleHealth(user.id, { fullSync }); + // Another run of this account holds it (the history backfill after a + // reconnect, or the hourly poll). Starting a second one beside it is what + // pushed accounts over Google's per-minute quota; the card follows the + // running one through the status route instead. + if (result.busy) { + return apiError("A Google Health sync is already running", 409, { + errorCode: "sync_in_progress", + }); + } // A run that failed AND wrote nothing is an error, as it always was. A run // that failed after writing some of its resources is a partial, and answering // 502 for it threw away the honest half of the result. diff --git a/src/lib/google-health/__tests__/rate-limit-and-token.test.ts b/src/lib/google-health/__tests__/rate-limit-and-token.test.ts new file mode 100644 index 000000000..4a80869fb --- /dev/null +++ b/src/lib/google-health/__tests__/rate-limit-and-token.test.ts @@ -0,0 +1,213 @@ +/** + * #1194 — how one Google Health data request answers a 429 and a 401, and + * which of two readings sharing a natural key a write keeps (#1195). + * + * A 429 is Google's per-user, per-minute quota. It is waited out and the + * request sent again: the delay Google asks for when it asks (`Retry-After`, + * or a `RetryInfo` detail in the body), otherwise a doubling backoff, and a + * request still throttled after the last retry fails as the transient it is. + * Never as a reauth. + * + * A 401 with a token source is answered by asking the source once for a + * fresh token and repeating the request; the token had expired, not the + * grant. A second 401, or a 401 on a fixed token, still fails as a reauth, + * because then it is one. + */ +import { afterEach, 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 }; +}); + +import { + GOOGLE_HEALTH_DATA_TYPES, + GOOGLE_HEALTH_RATE_LIMIT, + fetchDataPoints, + googleHealthRetryDelayMs, +} from "../client"; +import { collapseNaturalKeyTwins } from "../sync-core"; +import { GoogleHealthApiError } from "../response-classifier"; + +function response( + status: number, + body: unknown = {}, + headers: Record = {}, +): Response { + return new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json", ...headers }, + }); +} + +const throttled = () => + response(429, { + error: { code: 429, status: "RESOURCE_EXHAUSTED", message: "Quota" }, + }); + +beforeEach(() => { + safeFetchMock.mockReset(); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("a 429 from a data read", () => { + it("waits with a doubling backoff and repeats the request", async () => { + vi.useFakeTimers(); + safeFetchMock + .mockResolvedValueOnce(throttled()) + .mockResolvedValueOnce(throttled()) + .mockResolvedValueOnce(response(200, { dataPoints: [{ a: 1 }] })); + + const read = fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.weight, + "token", + "fetchWeight", + ); + await vi.advanceTimersByTimeAsync(GOOGLE_HEALTH_RATE_LIMIT.baseDelayMs); + expect(safeFetchMock).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(GOOGLE_HEALTH_RATE_LIMIT.baseDelayMs * 2); + + await expect(read).resolves.toEqual([{ a: 1 }]); + expect(safeFetchMock).toHaveBeenCalledTimes(3); + }); + + it("fails as transient, never as a reauth, once the retries run out", async () => { + vi.useFakeTimers(); + safeFetchMock.mockImplementation(async () => throttled()); + + const read = fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.weight, + "token", + "fetchWeight", + ).catch((err: unknown) => err); + await vi.advanceTimersByTimeAsync(10 * 60 * 1000); + const err = await read; + + expect(err).toBeInstanceOf(GoogleHealthApiError); + expect((err as GoogleHealthApiError).classification).toBe("transient"); + expect(safeFetchMock).toHaveBeenCalledTimes( + GOOGLE_HEALTH_RATE_LIMIT.maxRetries + 1, + ); + }); + + it("does not sleep out a delay longer than the cap", async () => { + safeFetchMock.mockResolvedValue( + response(429, {}, { "retry-after": "3600" }), + ); + + await expect( + fetchDataPoints(GOOGLE_HEALTH_DATA_TYPES.weight, "token", "fetchWeight"), + ).rejects.toMatchObject({ classification: "transient" }); + expect(safeFetchMock).toHaveBeenCalledTimes(1); + }); + + it("reads the delay Google asks for from the header or the body", () => { + const now = Date.parse("2026-10-10T12:00:00.000Z"); + expect( + googleHealthRetryDelayMs(response(429, {}, { "retry-after": "7" }), {}), + ).toBe(7000); + expect( + googleHealthRetryDelayMs( + response(429, {}, { "retry-after": "Sat, 10 Oct 2026 12:00:30 GMT" }), + {}, + now, + ), + ).toBe(30_000); + expect( + googleHealthRetryDelayMs(response(429), { + error: { + details: [ + { + "@type": "type.googleapis.com/google.rpc.RetryInfo", + retryDelay: "12.5s", + }, + ], + }, + }), + ).toBe(12_500); + expect(googleHealthRetryDelayMs(response(429), {})).toBeUndefined(); + }); +}); + +describe("a 401 from a data read", () => { + it("asks a token source once for a fresh token and repeats the request", async () => { + safeFetchMock + .mockResolvedValueOnce(response(401)) + .mockResolvedValueOnce(response(200, { dataPoints: [{ a: 1 }] })); + const source = vi.fn(async ({ forceRefresh }: { forceRefresh: boolean }) => + forceRefresh ? "fresh" : "stale", + ); + + await expect( + fetchDataPoints(GOOGLE_HEALTH_DATA_TYPES.weight, source, "fetchWeight"), + ).resolves.toEqual([{ a: 1 }]); + expect(source.mock.calls.map(([o]) => o.forceRefresh)).toEqual([ + false, + true, + ]); + const auth = (n: number) => + new Headers((safeFetchMock.mock.calls[n]![1] as RequestInit).headers).get( + "authorization", + ); + expect(auth(0)).toBe("Bearer stale"); + expect(auth(1)).toBe("Bearer fresh"); + }); + + it("is a reauth when the fresh token is refused too", async () => { + safeFetchMock.mockResolvedValue(response(401)); + + await expect( + fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.weight, + async () => "token", + "fetchWeight", + ), + ).rejects.toMatchObject({ classification: "reauth_required" }); + expect(safeFetchMock).toHaveBeenCalledTimes(2); + }); + + it("is a reauth at once on a fixed token", async () => { + safeFetchMock.mockResolvedValue(response(401)); + + await expect( + fetchDataPoints(GOOGLE_HEALTH_DATA_TYPES.weight, "token", "fetchWeight"), + ).rejects.toMatchObject({ classification: "reauth_required" }); + expect(safeFetchMock).toHaveBeenCalledTimes(1); + }); +}); + +describe("collapseNaturalKeyTwins (#1195)", () => { + const end = new Date("2026-10-09T02:00:00.000Z"); + const reading = (externalId: string, value: number, stage = "CORE") => ({ + type: "SLEEP_DURATION", + value, + unit: "minutes", + measuredAt: end, + externalId, + sleepStage: stage as "CORE" | "DEEP", + }); + + it("keeps the longer of two segments sharing a stage and an end", () => { + const { kept, dropped } = collapseNaturalKeyTwins([ + reading("s:sleep:00:30", 90), + reading("s:sleep:00:00", 120), + reading("s:sleep:01:00", 60, "DEEP"), + ]); + expect(kept.map((r) => r.externalId)).toEqual([ + "s:sleep:00:00", + "s:sleep:01:00", + ]); + expect(dropped).toBe(1); + }); + + it("picks the same reading whatever order the pair arrives in", () => { + const a = reading("phone:sleep:00:00", 60); + const b = reading("watch:sleep:00:00", 60); + expect(collapseNaturalKeyTwins([a, b]).kept).toEqual([a]); + expect(collapseNaturalKeyTwins([b, a]).kept).toEqual([a]); + }); +}); diff --git a/src/lib/google-health/__tests__/sync-activity-failsoft.test.ts b/src/lib/google-health/__tests__/sync-activity-failsoft.test.ts index 1d4c766d4..7c08c9bcd 100644 --- a/src/lib/google-health/__tests__/sync-activity-failsoft.test.ts +++ b/src/lib/google-health/__tests__/sync-activity-failsoft.test.ts @@ -28,6 +28,10 @@ vi.mock("../sync-core", () => ({ connection: { id: "c1", googleUserId: "g1" }, })), handleCollectionFetchError: handleErrorMock, + googleHealthTokenSource: + (_userId: string, info: { accessToken: string }) => async () => + info.accessToken, + runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), upsertGoogleHealthMeasurements: vi.fn(async () => ({ imported: 0, touched: [], diff --git a/src/lib/google-health/__tests__/sync-failsoft.test.ts b/src/lib/google-health/__tests__/sync-failsoft.test.ts index 1e7ea8d53..1e587d398 100644 --- a/src/lib/google-health/__tests__/sync-failsoft.test.ts +++ b/src/lib/google-health/__tests__/sync-failsoft.test.ts @@ -28,6 +28,7 @@ const { recordSyncSuccessMock: vi.fn(async () => {}), invalidateUserMeasurementsMock: vi.fn(), prismaMock: { + integrationStatus: { findUnique: vi.fn(async () => null) }, googleHealthConnection: { findUnique: vi.fn(), update: vi.fn(async () => ({})), @@ -47,6 +48,18 @@ const { })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +// The account lock is a Postgres advisory lock on a connection of its own; +// these cases run without a database, so the lock is always free. +vi.mock("../sync-lock", () => ({ + withGoogleHealthSyncLock: async ( + _userId: string, + run: () => Promise, + ) => ({ + ran: true, + result: await run(), + }), +})); vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() })); vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn(async () => false), diff --git a/src/lib/google-health/__tests__/sync-fairness.test.ts b/src/lib/google-health/__tests__/sync-fairness.test.ts index 3ea7554fc..8f3aaf35f 100644 --- a/src/lib/google-health/__tests__/sync-fairness.test.ts +++ b/src/lib/google-health/__tests__/sync-fairness.test.ts @@ -14,6 +14,7 @@ const { calls, prismaMock, recordSyncSuccess, resources } = vi.hoisted(() => { return { calls, prismaMock: { + integrationStatus: { findUnique: vi.fn(async () => null) }, googleHealthConnection: { findUnique: vi.fn(async () => ({ lastSyncedAt: null })), update: vi.fn(async () => ({})), @@ -45,6 +46,18 @@ const { calls, prismaMock, recordSyncSuccess, resources } = vi.hoisted(() => { }); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +// The account lock is a Postgres advisory lock on a connection of its own; +// these cases run without a database, so the lock is always free. +vi.mock("../sync-lock", () => ({ + withGoogleHealthSyncLock: async ( + _userId: string, + run: () => Promise, + ) => ({ + ran: true, + result: await run(), + }), +})); vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn(async () => false), recordSyncSuccess, diff --git a/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts b/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts index 05b0bee44..92e68d395 100644 --- a/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts +++ b/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts @@ -11,6 +11,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const { prismaMock, metrics } = vi.hoisted(() => ({ prismaMock: { + integrationStatus: { findUnique: vi.fn(async () => null) }, googleHealthConnection: { findUnique: vi.fn(), update: vi.fn(async () => ({})), @@ -20,6 +21,18 @@ const { prismaMock, metrics } = vi.hoisted(() => ({ })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +// The account lock is a Postgres advisory lock on a connection of its own; +// these cases run without a database, so the lock is always free. +vi.mock("../sync-lock", () => ({ + withGoogleHealthSyncLock: async ( + _userId: string, + run: () => Promise, + ) => ({ + ran: true, + result: await run(), + }), +})); vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn(async () => false), recordSyncSuccess: vi.fn(async () => {}), diff --git a/src/lib/google-health/__tests__/sync-intraday-overlap.test.ts b/src/lib/google-health/__tests__/sync-intraday-overlap.test.ts index 4bf475bbb..0a09d4a36 100644 --- a/src/lib/google-health/__tests__/sync-intraday-overlap.test.ts +++ b/src/lib/google-health/__tests__/sync-intraday-overlap.test.ts @@ -21,7 +21,10 @@ vi.mock("../sync-core", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, - getValidToken: vi.fn(async () => ({ accessToken: "token" })), + getValidToken: vi.fn(async () => ({ + accessToken: "token", + expiresAt: new Date(Date.now() + 60 * 60 * 1000), + })), handleCollectionFetchError: vi.fn(async () => 0), upsertGoogleHealthMeasurements: vi.fn(async () => ({ imported: 0, 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 7eee64efc..b4bf66c8e 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 @@ -28,6 +28,10 @@ vi.mock("../sync-core", () => ({ getValidToken: vi.fn(async () => ({ accessToken: "token" })), handleCollectionFetchError: vi.fn(async () => 0), upsertGoogleHealthMeasurements: upsertMock, + googleHealthTokenSource: + (_userId: string, info: { accessToken: string }) => async () => + info.accessToken, + runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), })); 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 8d48929df..0a4c021e0 100644 --- a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts +++ b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts @@ -54,6 +54,10 @@ vi.mock("../sync-core", () => ({ connection: { id: "c1", googleUserId: "g1" }, })), handleCollectionFetchError: handleErrorMock, + googleHealthTokenSource: + (_userId: string, info: { accessToken: string }) => async () => + info.accessToken, + runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), upsertGoogleHealthMeasurements: vi.fn(async () => ({ imported: 0, touched: [], diff --git a/src/lib/google-health/__tests__/sync-outcomes.test.ts b/src/lib/google-health/__tests__/sync-outcomes.test.ts index db252921a..1aae53140 100644 --- a/src/lib/google-health/__tests__/sync-outcomes.test.ts +++ b/src/lib/google-health/__tests__/sync-outcomes.test.ts @@ -112,6 +112,7 @@ const resourceOutcomes = [ const { prismaMock, runCycle } = vi.hoisted(() => ({ prismaMock: { + integrationStatus: { findUnique: vi.fn(async () => null) }, googleHealthConnection: { findUnique: vi.fn(async () => ({ lastSyncedAt: null })), update: vi.fn(async () => ({})), @@ -121,6 +122,18 @@ const { prismaMock, runCycle } = vi.hoisted(() => ({ })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +// The account lock is a Postgres advisory lock on a connection of its own; +// these cases run without a database, so the lock is always free. +vi.mock("../sync-lock", () => ({ + withGoogleHealthSyncLock: async ( + _userId: string, + run: () => Promise, + ) => ({ + ran: true, + result: await run(), + }), +})); vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn(async () => false), recordSyncSuccess: vi.fn(async () => {}), @@ -134,6 +147,7 @@ vi.mock("@/lib/insights/comprehensive-generate", () => ({ })); vi.mock("../sync-core", () => ({ GOOGLE_HEALTH_INTEGRATION_KEY: "google-health", + getValidToken: vi.fn(), incrementalStart: vi.fn(() => undefined), intradayOverlapMs: vi.fn(() => 2 * 60 * 60 * 1000), markSynced: vi.fn(async () => {}), diff --git a/src/lib/google-health/__tests__/sync-sleep-morning-refresh.test.ts b/src/lib/google-health/__tests__/sync-sleep-morning-refresh.test.ts index 4a9d99813..4ca2f771b 100644 --- a/src/lib/google-health/__tests__/sync-sleep-morning-refresh.test.ts +++ b/src/lib/google-health/__tests__/sync-sleep-morning-refresh.test.ts @@ -36,6 +36,10 @@ vi.mock("../client", () => ({ vi.mock("../sync-core", () => ({ getValidToken: vi.fn(async () => ({ accessToken: "tok" })), handleCollectionFetchError: vi.fn(() => 0), + googleHealthTokenSource: + (_userId: string, info: { accessToken: string }) => async () => + info.accessToken, + runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), noteHardFailure: vi.fn(), replaceStaleGoogleHealthSleep: vi.fn(async () => 0), upsertGoogleHealthMeasurements: upsertMock, diff --git a/src/lib/google-health/__tests__/sync-workout-insert-identity.test.ts b/src/lib/google-health/__tests__/sync-workout-insert-identity.test.ts index a0a61a56c..b92e7fb97 100644 --- a/src/lib/google-health/__tests__/sync-workout-insert-identity.test.ts +++ b/src/lib/google-health/__tests__/sync-workout-insert-identity.test.ts @@ -38,6 +38,10 @@ vi.mock("../client", () => ({ vi.mock("../sync-core", () => ({ getValidToken, handleCollectionFetchError: vi.fn(), + googleHealthTokenSource: + (_userId: string, info: { accessToken: string }) => async () => + info.accessToken, + runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), noteHardFailure: vi.fn(), })); diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index e784bb5af..3bdab3aec 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -106,6 +106,11 @@ export function noteGoogleHealthOutcomeFailure( if (tracker && !tracker.reasonCode) tracker.reasonCode = reasonCode; } +/** Whether a walk in the current resource scope has ended short so far. */ +export function googleHealthOutcomeTruncated(): boolean { + return clientOutcomeStorage.getStore()?.truncated === true; +} + export async function runWithGoogleHealthClientOutcome( fn: () => Promise, ): Promise<{ result: T; outcome: GoogleHealthClientOutcome }> { @@ -729,6 +734,167 @@ export function extractGoogleApiErrorDetail(json: unknown): string | undefined { return [head, tail].filter(Boolean).join(" ").slice(0, 500); } +// ─── Token source + one data request ─────────────────────────── + +/** + * Where a walk gets its access token. A plain string is used as given. A + * function is asked before every request, so a walk that runs for longer than + * the token lives (an access token lasts an hour; a full-history heart-rate + * walk can take several) picks up the refreshed one instead of carrying an + * expired token into a 401. `forceRefresh` asks for a new token even though + * the current one has not reached its expiry, after Google refused it. + */ +export type GoogleHealthTokenSource = (opts: { + forceRefresh: boolean; +}) => Promise; + +export type GoogleHealthAccessToken = string | GoogleHealthTokenSource; + +/** + * How a data request sits out Google's per-user, per-minute quota (429 + * `RESOURCE_EXHAUSTED`). The quota refills within the minute, so a short wait + * turns a throttled request into a completed one, where failing it used to + * cost the whole collection. The waits follow `Retry-After` (or the + * `google.rpc.RetryInfo` delay in the body) when Google sends one, and + * otherwise double from `baseDelayMs`: 2, 4, 8, 16, 32 s, a minute and a + * bit in all, which spans a full quota window. A requested delay above + * `maxDelayMs` is not slept out; the request fails as the transient it is + * and the next run picks the window up. + */ +export const GOOGLE_HEALTH_RATE_LIMIT = { + maxRetries: 5, + baseDelayMs: 2_000, + maxDelayMs: 60_000, +} as const; + +/** + * The delay Google asked for on a 429, in ms: the `Retry-After` header in + * either of its forms, else a `google.rpc.RetryInfo` detail (`"30s"`) in the + * error body. Undefined when it asked for nothing readable. + */ +export function googleHealthRetryDelayMs( + res: Response, + json: unknown, + now: number = Date.now(), +): number | undefined { + const header = res.headers.get("retry-after")?.trim(); + if (header) { + if (/^\d+$/.test(header)) return Number.parseInt(header, 10) * 1000; + const at = Date.parse(header); + if (Number.isFinite(at)) return Math.max(0, at - now); + } + const details = (json as { error?: { details?: unknown } } | null)?.error + ?.details; + if (Array.isArray(details)) { + for (const d of details) { + const delay = (d as { retryDelay?: unknown } | null)?.retryDelay; + if (typeof delay === "string") { + const m = /^(\d+(?:\.\d+)?)s$/.exec(delay.trim()); + if (m) return Math.round(Number.parseFloat(m[1]!) * 1000); + } + } + } + return undefined; +} + +function waitMs(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +interface GoogleHealthDataResponse { + status: number; + json: T | null; + verdict: ReturnType; + apiErrorDetail: string | undefined; +} + +/** + * Send one data request and answer with the response it finally got. + * + * Two 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. + * - 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 + * fixed string token, is returned like any other answer. + * + * Every attempt is recorded as its own external call, so the wide event + * shows the throttling and the waits rather than only their outcome. + */ +async function googleHealthDataRequest( + token: GoogleHealthAccessToken, + label: string, + send: (accessToken: string) => Promise, + isExpectedFallback: (status: number) => boolean = () => false, +): Promise> { + let rateLimited = 0; + let refreshed = false; + let forceRefresh = false; + for (;;) { + const accessToken = + typeof token === "string" ? token : await token({ forceRefresh }); + forceRefresh = false; + const started = performance.now(); + const res = await send(accessToken); + const json = (await res.json().catch(() => null)) as T | null; + const verdict = classifyGoogleHealthResponse(res.status); + const apiErrorDetail = + verdict.classification === "success" + ? undefined + : extractGoogleApiErrorDetail(json); + + const retryAfterMs = + res.status === 429 ? googleHealthRetryDelayMs(res, json) : undefined; + const backoffMs = + res.status === 429 + ? (retryAfterMs ?? + GOOGLE_HEALTH_RATE_LIMIT.baseDelayMs * 2 ** rateLimited) + : 0; + const willWait = + res.status === 429 && + rateLimited < GOOGLE_HEALTH_RATE_LIMIT.maxRetries && + backoffMs <= GOOGLE_HEALTH_RATE_LIMIT.maxDelayMs && + !stopRequested(); + const willRefresh = + res.status === 401 && typeof token !== "string" && !refreshed; + + getEvent()?.addExternalCall({ + service: "google-health", + method: label, + duration_ms: Math.round(performance.now() - started), + status: res.status, + ...(isExpectedFallback(res.status) + ? { note: "date_filter_rejected_retrying_snake_case" } + : willWait + ? { 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, + }), + }); + + if (willWait) { + rateLimited += 1; + await waitMs(backoffMs); + continue; + } + if (willRefresh) { + refreshed = true; + forceRefresh = true; + continue; + } + return { status: res.status, json, verdict, apiErrorDetail }; + } +} + /** * Walk every `DataPoint` for one data type since the incremental cursor via * `dataPoints.list` with `nextPageToken` pagination. The data-type id is @@ -748,7 +914,7 @@ export function extractGoogleApiErrorDetail(json: unknown): string | undefined { */ export async function fetchDataPoints( dataType: GoogleHealthDataType, - accessToken: string, + accessToken: GoogleHealthAccessToken, verb: string, query: DataPointQuery = {}, ): Promise { @@ -776,7 +942,7 @@ export async function fetchDataPoints( */ export async function forEachDataPointPage( dataType: GoogleHealthDataType, - accessToken: string, + accessToken: GoogleHealthAccessToken, verb: string, query: DataPointQuery, onPage: (points: GoogleHealthDataPoint[]) => void | Promise, @@ -812,50 +978,31 @@ export async function forEachDataPointPage( if (pageToken) params.set("pageToken", pageToken); requestCount += 1; - const pageStart = performance.now(); - const res = await safeFetch( - `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`, - { - method: "GET", - headers: { Authorization: `Bearer ${accessToken}` }, - }, - ); - - const json = (await res - .json() - .catch(() => null)) as GoogleHealthDataPointPage | null; - const verdict = classifyGoogleHealthResponse(res.status); - const apiErrorDetail = - verdict.classification === "success" - ? undefined - : extractGoogleApiErrorDetail(json); // v1.42 (#1023) — the first filtered daily-summary read answering 400 // is the expected trigger for the snake_case retry below, not a // failure: some accounts reject the documented camelCase prefix on // some types, and the very next request answers 200. It was logged as // an error next to the 200 that followed it. It is recorded as a note // now; any other 400 still carries its error. - const expectedFallback = - canFallBack && - dateStyle === "camel" && - requestCount === 1 && - res.status === 400; - getEvent()?.addExternalCall({ - service: "google-health", - method: `${verb}(page=${pageCount})`, - duration_ms: Math.round(performance.now() - pageStart), - status: res.status, - ...(expectedFallback - ? { note: "date_filter_rejected_retrying_snake_case" } - : { - error: - verdict.classification === "success" - ? undefined - : apiErrorDetail - ? `${verdict.reason} ${apiErrorDetail}` - : verdict.reason, - }), - }); + const firstRequest = requestCount === 1; + const { json, verdict, apiErrorDetail } = + await googleHealthDataRequest( + accessToken, + `${verb}(page=${pageCount})`, + (token) => + safeFetch( + `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`, + { + method: "GET", + headers: { Authorization: `Bearer ${token}` }, + }, + ), + (status) => + canFallBack && + dateStyle === "camel" && + firstRequest && + status === 400, + ); if (verdict.classification !== "success") { throw new GoogleHealthApiError({ verb, @@ -1078,7 +1225,7 @@ export function buildDailyRollUpBody( */ export async function fetchDailyRollUp( dataType: GoogleHealthDataType, - accessToken: string, + accessToken: GoogleHealthAccessToken, verb: string, query: RollupQuery = {}, ): Promise { @@ -1108,38 +1255,23 @@ export async function fetchDailyRollUp( const body = buildDailyRollUpBody(chunk, pageToken ?? undefined); requestCount += 1; - const reqStart = performance.now(); - const res = await safeFetch( - `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`, - { - method: "POST", - headers: { - Authorization: `Bearer ${accessToken}`, - "Content-Type": "application/json", - }, - body: JSON.stringify(body), - }, - ); - const json = (await res - .json() - .catch(() => null)) as GoogleHealthRollupPage | null; - const verdict = classifyGoogleHealthResponse(res.status); - const apiErrorDetail = - verdict.classification === "success" - ? undefined - : extractGoogleApiErrorDetail(json); - getEvent()?.addExternalCall({ - service: "google-health", - method: `${verb}(chunk=${chunkIndex},page=${pageCount})`, - duration_ms: Math.round(performance.now() - reqStart), - status: res.status, - error: - verdict.classification === "success" - ? undefined - : apiErrorDetail - ? `${verdict.reason} ${apiErrorDetail}` - : verdict.reason, - }); + const { json, verdict, apiErrorDetail } = + await googleHealthDataRequest( + accessToken, + `${verb}(chunk=${chunkIndex},page=${pageCount})`, + (token) => + safeFetch( + `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`, + { + method: "POST", + headers: { + Authorization: `Bearer ${token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(body), + }, + ), + ); if (verdict.classification !== "success") { throw new GoogleHealthApiError({ verb, diff --git a/src/lib/google-health/sync-activity.ts b/src/lib/google-health/sync-activity.ts index 80cd1be80..32344e993 100644 --- a/src/lib/google-health/sync-activity.ts +++ b/src/lib/google-health/sync-activity.ts @@ -38,7 +38,9 @@ import { } from "./client"; import { getValidToken, + googleHealthTokenSource, handleCollectionFetchError, + runGoogleHealthCollection, upsertGoogleHealthMeasurements, type GoogleHealthMeasurementUpsert, type GoogleHealthResourceSyncOptions, @@ -96,6 +98,7 @@ export async function syncUserActivity( ): Promise { const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; + const token = googleHealthTokenSource(userId, tokenInfo); // The dailyRollUp request range is civil and user-local; resolve the user's // stored zone so the range bounds land on the correct civil days rather than @@ -118,51 +121,51 @@ export async function syncUserActivity( const rollupEmptyResponse: string[] = []; const rollupDroppedAll: string[] = []; for (const resource of ROLLUP_RESOURCES) { - let points: Record[]; - try { - points = await fetchDailyRollUp( - resource.dataType, - tokenInfo.accessToken, - resource.verb, - { start, tz }, - ); - } catch (err) { - imported += await handleCollectionFetchError(resource.verb, userId, err); - continue; - } - - // The mapper runs INSIDE a per-type catch too (mirrors sync-metrics): a - // single malformed point whose `resource.map(point)` throws must not - // escape the ROLLUP_RESOURCES loop and abort every sibling type ordered - // after it. Route a map throw through the same ledger as a fetch failure — - // record it, fail the cycle, and move on to the next type. - const readings: GoogleHealthMeasurementUpsert[] = []; - try { - for (const point of points) { - for (const m of resource.map(point, tz)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: externalIdFor(m), - }); - } + imported += await runGoogleHealthCollection(resource.verb, async () => { + let points: Record[]; + try { + points = await fetchDailyRollUp( + resource.dataType, + token, + resource.verb, + { start, tz }, + ); + } catch (err) { + return handleCollectionFetchError(resource.verb, userId, err); } - } catch (err) { - imported += await handleCollectionFetchError(resource.verb, userId, err); - continue; - } - if (points.length === 0) { - rollupEmptyResponse.push(resource.verb); - } else if (readings.length === 0) { - rollupDroppedAll.push(resource.verb); - } - imported += ( - await upsertGoogleHealthMeasurements(userId, readings, { - deferRollup: opts.deferRollup, - }) - ).imported; + + // The mapper runs INSIDE a per-type catch too (mirrors sync-metrics): a + // single malformed point whose `resource.map(point)` throws must not + // escape the ROLLUP_RESOURCES loop and abort every sibling type ordered + // after it. Route a map throw through the same ledger as a fetch failure — + // record it, fail the cycle, and move on to the next type. + const readings: GoogleHealthMeasurementUpsert[] = []; + try { + for (const point of points) { + for (const m of resource.map(point, tz)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: externalIdFor(m), + }); + } + } + } catch (err) { + return handleCollectionFetchError(resource.verb, userId, err); + } + if (points.length === 0) { + rollupEmptyResponse.push(resource.verb); + } else if (readings.length === 0) { + rollupDroppedAll.push(resource.verb); + } + return ( + await upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + }); } if (rollupEmptyResponse.length > 0 || rollupDroppedAll.length > 0) { annotate({ @@ -174,33 +177,35 @@ export async function syncUserActivity( } // VO2 max — a daily summary (list + `.date` filter), not a rollup type. - try { - const points = await fetchDataPoints( - GOOGLE_HEALTH_DATA_TYPES.vo2Max, - tokenInfo.accessToken, - "fetchVo2Max", - { start }, - ); - const readings: GoogleHealthMeasurementUpsert[] = []; - for (const point of points) { - for (const m of mapVo2Max(point, tz)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: externalIdFor(m), - }); + imported += await runGoogleHealthCollection("fetchVo2Max", async () => { + try { + const points = await fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.vo2Max, + token, + "fetchVo2Max", + { start }, + ); + const readings: GoogleHealthMeasurementUpsert[] = []; + for (const point of points) { + for (const m of mapVo2Max(point, tz)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: externalIdFor(m), + }); + } } + return ( + await upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + } catch (err) { + return handleCollectionFetchError("fetchVo2Max", userId, err); } - imported += ( - await upsertGoogleHealthMeasurements(userId, readings, { - deferRollup: opts.deferRollup, - }) - ).imported; - } catch (err) { - imported += await handleCollectionFetchError("fetchVo2Max", userId, err); - } + }); // `markSynced` is owned by the orchestrator (`syncUserGoogleHealth`). annotate({ diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index 204368f6c..de64e573d 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -19,8 +19,13 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; import { encrypt, decrypt } from "@/lib/crypto"; -import { getEvent } from "@/lib/logging/context"; +import { annotate, getEvent } from "@/lib/logging/context"; import { userDayKey } from "@/lib/tz/format"; +import { isP2002 } from "@/lib/prisma-errors"; +import { + reconcileExternalMeasurement, + type MeasurementReconciliationVerdict, +} from "@/lib/measurements/reconcile-external-measurement"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status"; @@ -30,10 +35,12 @@ import { } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { + googleHealthOutcomeTruncated, noteGoogleHealthMapped, noteGoogleHealthOutcomeFailure, noteGoogleHealthWritten, refreshAccessToken, + type GoogleHealthTokenSource, } from "./client"; import { getUserGoogleHealthCredentials } from "./credentials"; import { @@ -99,6 +106,8 @@ export function intradayOverlapMs( export interface GoogleHealthTokenInfo { accessToken: string; + /** When `accessToken` stops being accepted. */ + expiresAt: Date; connection: { id: string; googleUserId: string }; } @@ -115,6 +124,7 @@ export interface GoogleHealthTokenInfo { */ export async function getValidToken( userId: string, + opts: { forceRefresh?: boolean } = {}, ): Promise { const connection = await prisma.googleHealthConnection.findUnique({ where: { userId }, @@ -125,8 +135,8 @@ export async function getValidToken( const refreshToken = decrypt(connection.refreshToken); if ( - connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < - Date.now() + opts.forceRefresh === true || + connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < Date.now() ) { try { const creds = await getUserGoogleHealthCredentials(userId); @@ -170,6 +180,7 @@ export async function getValidToken( return { accessToken: newTokens.access_token, + expiresAt, connection: { id: connection.id, googleUserId: connection.googleUserId, @@ -209,6 +220,7 @@ export async function getValidToken( return { accessToken, + expiresAt: connection.tokenExpiresAt, connection: { id: connection.id, googleUserId: connection.googleUserId, @@ -216,6 +228,51 @@ export async function getValidToken( }; } +/** + * Thrown by a token source when no valid token can be had any more. The + * failure is already on the ledger and the cycle's verdict by then + * (`getValidToken` records it), so the collection that hit it only has to + * stop. + */ +export class GoogleHealthTokenUnavailableError extends Error { + constructor() { + super("Google Health access token unavailable"); + this.name = "GoogleHealthTokenUnavailableError"; + } +} + +/** + * The token source a resource hands to its walks (see + * `GoogleHealthTokenSource`). It starts from the token the resource resolved + * and renews it whenever it comes within the refresh buffer of expiring, or + * when a request reports it refused. + * + * Before this the resource resolved one token and every walk carried it to + * the end. An access token lives an hour, a full-history heart-rate walk runs + * for several, and once the hour was up every remaining request answered 401. + * A 401 classifies as a revoked grant, so the account was parked at + * `error_reauth` over a token that had simply expired, with a refresh token + * that was still perfectly good, and stayed parked until it was reconnected + * by hand (#1194). + */ +export function googleHealthTokenSource( + userId: string, + initial: GoogleHealthTokenInfo, +): GoogleHealthTokenSource { + let current = initial; + return async ({ forceRefresh }) => { + if ( + forceRefresh || + current.expiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < Date.now() + ) { + const next = await getValidToken(userId, { forceRefresh }); + if (!next) throw new GoogleHealthTokenUnavailableError(); + current = next; + } + return current.accessToken; + }; +} + /** * True when a caught error is a per-data-class 403 (forbidden). The Google * Health Restricted bundles are granted independently in the consent flow, so a @@ -343,6 +400,106 @@ export async function runWithGoogleHealthSyncCycle( }; } +/** + * Collections a full-history backfill already walked to the end, 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. + */ +interface BackfillResumeTracker { + userId: string; + done: Set; + /** Collections this attempt skipped because an earlier one finished them. */ + skipped: string[]; +} +const backfillResumeStorage = new AsyncLocalStorage(); + +/** 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; + return new Set( + Array.isArray(done) + ? done.filter( + (key): key is string => typeof key === "string" && key.length <= 64, + ) + : [], + ); +} + +/** + * 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. + */ +export async function runWithGoogleHealthBackfillResume( + userId: string, + done: Set, + fn: () => Promise, +): Promise<{ result: T; skipped: string[] }> { + const tracker: BackfillResumeTracker = { userId, done, skipped: [] }; + const result = await backfillResumeStorage.run(tracker, fn); + return { result, skipped: tracker.skipped }; +} + +/** + * Walk one collection, under the backfill's resume bookkeeping when there is + * 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. + */ +export async function runGoogleHealthCollection( + key: string, + fn: () => Promise, +): Promise { + const resume = backfillResumeStorage.getStore(); + if (!resume) return fn(); + if (resume.done.has(key)) { + resume.skipped.push(key); + return 0; + } + 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(); + if (clean) { + resume.done.add(key); + 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}`, + ), + ); + } + return imported; +} + /** * Single-source the per-resource collection-fetch error handling. A 403 on one * data class soft-skips it (warn + return 0) so sibling resources still sync; a @@ -362,6 +519,14 @@ export async function handleCollectionFetchError( userId: string, err: unknown, ): Promise { + if (err instanceof GoogleHealthTokenUnavailableError) { + // `getValidToken` already recorded the failure and failed the cycle; + // recording it again would count one dead token twice. + getEvent()?.addWarning( + `google-health ${resource} stopped for ${userId}: no valid access token`, + ); + return 0; + } if (isCollectionForbidden(err)) { getEvent()?.addWarning( `google-health ${resource} sync skipped for ${userId}: collection 403 (soft-skip)`, @@ -445,6 +610,51 @@ export interface GoogleHealthMeasurementUpsert { sleepStage?: "IN_BED" | "AWAKE" | "ASLEEP" | "REM" | "CORE" | "DEEP" | null; } +/** + * Keep one reading per natural key — `(type, measuredAt, sleepStage)`, the + * second unique index on measurements — out of a batch. + * + * Google reports overlapping sleep segments: two sessions for one night, or + * two segments of one session with the same stage and the same end. Both + * readings carry their own externalId, but only one row can hold their + * natural key. Writing both was never possible; what happened instead + * depended on the order of the batch. A fresh pair lost one reading to + * `skipDuplicates` without a trace, and a re-score that moved one row onto + * the other's slot threw P2002 and failed the whole sync (#1195). + * + * The rule picks the same reading every time Google sends the same pair, so + * the row that holds the slot does not change hands from one sync to the + * next: the longer one (for a sleep segment, the span that covers the + * shorter one ending at the same instant), then the smaller externalId. + * Readings that share an externalId as well are the same reading fetched + * twice, and collapse the same way. + */ +export function collapseNaturalKeyTwins< + T extends GoogleHealthMeasurementUpsert, +>(readings: readonly T[]): { kept: T[]; dropped: number } { + const byKey = new Map(); + for (const r of readings) { + const key = `${r.type}|${r.measuredAt.getTime()}|${r.sleepStage ?? ""}`; + const held = byKey.get(key); + if ( + !held || + r.value > held.value || + (r.value === held.value && r.externalId < held.externalId) + ) { + byKey.set(key, r); + } + } + return { + kept: readings.filter( + (r) => + byKey.get( + `${r.type}|${r.measuredAt.getTime()}|${r.sleepStage ?? ""}`, + ) === r, + ), + dropped: readings.length - byKey.size, + }; +} + /** Chunk size for the batched `createMany` insert of fresh readings. */ const GOOGLE_HEALTH_CREATE_CHUNK = 500; @@ -454,7 +664,10 @@ export interface GoogleHealthSleepReplaceWindow { windowStart: Date | null; /** Latest segment end (UTC). */ windowEnd: Date | null; - /** The fresh externalIds for THIS session — never soft-deleted. */ + /** + * The fresh externalIds that must survive — every session's of the fetch, + * since sessions can overlap (see `replaceStaleGoogleHealthSleep`). + */ keepIds: string[]; } @@ -466,11 +679,12 @@ export interface GoogleHealthSleepReplaceWindow { * night-total silently double-counted (a 7h35 night read as 10h+). For each * just-fetched session this soft-deletes any LIVE `GOOGLE_HEALTH` * `SLEEP_DURATION` row whose `measuredAt` falls inside the session's - * `[windowStart, windowEnd]` but was NOT re-produced by this fetch (`keepIds` - * are this session's fresh externalIds). Sleep sessions do not overlap in time, - * so a scan bounded to one session's window only ever touches that session's own - * rows — a fresh row is protected by `keepIds`, and any leftover (old volatile - * key, or a segment Google dropped) is cleared. Rows OUTSIDE every returned + * `[windowStart, windowEnd]` but was NOT re-produced by this fetch. Sessions + * CAN overlap in time (Google reports two sessions over one stretch of a night + * from two sources), so the caller passes every fresh externalId of the fetch + * in `keepIds`, not only this session's: a window then never clears a row + * another session just re-produced, and any leftover (old volatile key, or a + * segment Google dropped) is cleared. Rows OUTSIDE every returned * window are never touched, so a night Google did not re-report this tick stays * intact — no data loss, only stale duplicates and re-score orphans go. This is * also the repair path: a full backfill re-fetches history and cleans each night @@ -554,11 +768,15 @@ export async function upsertGoogleHealthMeasurements( // before they can become rows. A refused reading is dropped and tallied on // the ambient event, never clamped and never written for a later reader to // discover. - const readings = dropImplausibleMeasurements( - "googleHealth", - incoming, - (r) => ({ type: r.type, value: r.value }), + const { kept: readings, dropped: naturalKeyTwins } = collapseNaturalKeyTwins( + dropImplausibleMeasurements("googleHealth", incoming, (r) => ({ + type: r.type, + value: r.value, + })), ); + if (naturalKeyTwins > 0) { + annotate({ meta: { "googleHealth.naturalKeyTwins": naturalKeyTwins } }); + } if (readings.length === 0) return { imported: 0, touched: [], inserted: [] }; // Probe EVERY existing row (live AND tombstoned) for the batch's externalIds @@ -712,6 +930,13 @@ export async function upsertGoogleHealthMeasurements( // value, the NEW externalId (the migration itself), `deletedAt: null`. if (toCreate.length > 0) { try { + // A row this batch already matched by externalId belongs to that + // reading. When the reading moves it (a re-scored segment end), the + // slot it leaves is free by the time the inserts run (updates go + // first, below), so a fresh reading for that slot is an insert, not a + // rescue: re-keying the moving row would hand it to two readings at + // once and drop one of them. + const claimedIds = new Set([...rowByKey.values()].map((row) => row.id)); const rescued = new Set(); const naturalKeyOf = ( type: string, @@ -749,7 +974,7 @@ export async function upsertGoogleHealthMeasurements( const id = byNaturalKey.get( naturalKeyOf(c.type, c.measuredAt, c.sleepStage), ); - if (id) { + if (id && !claimedIds.has(id)) { rescued.add(i + j); toUpdate.push({ id, @@ -789,6 +1014,121 @@ export async function upsertGoogleHealthMeasurements( measuredAt: Date; }> = []; + // Existing-row overwrites first, inserts second: a re-scored segment that + // moves its row frees the slot the row held, and a fresh reading for that + // slot can only land once the move has happened. + // + // Per-row update (differing values) on the matched id, so the re-fetched + // daily summary overwrites in place and bumps `syncVersion`. + // `deletedAt: null` rides along unconditionally — a no-op on a live row, a + // deliberate RESURRECTION on a tombstoned one (Google is the source of truth + // for its own rows; see TOMBSTONES RESURRECT above). + // + // A move can still land on a slot another row holds: Google reports + // overlapping sleep segments, and a re-score can put one segment's end on + // the end another segment of the same stage already has. The natural-key + // index refuses that write (P2002). It is not a failed sync: the row that + // holds the slot is a stale copy of the same reading, so the move is handed + // to `reconcileExternalMeasurement`, which retires the stale copy and lands + // the reading, the same way every other provider settles the two + // identities. Only a reconcile that itself fails holds the watermark. + const conflicts: typeof toUpdate = []; + for (const entry of toUpdate) { + const { id, r, reKeyTo, movedFrom } = entry; + try { + await prisma.measurement.update({ + where: { id }, + data: { + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + sleepStage: r.sleepStage ?? null, + deletedAt: null, + // Key-format migration: adopt the fresh externalId in place. + ...(reKeyTo ? { externalId: reKeyTo } : {}), + // Surface the server-side mutation to the iOS LWW reconciler. + syncVersion: { increment: 1 }, + }, + }); + touched.push({ + type: r.type as MeasurementType, + measuredAt: r.measuredAt, + }); + // A row that moved (a daily anchor re-cut on the user's local noon) + // leaves its old day too; that day's rollup needs the refold. + if (movedFrom) { + touched.push({ + type: r.type as MeasurementType, + measuredAt: movedFrom, + }); + } + imported++; + } catch (err) { + if (isP2002(err)) { + conflicts.push(entry); + continue; + } + getEvent()?.addWarning( + `google-health: failed to update measurement: ${err}`, + ); + // Same rule as the create catch: hold the watermark so the overwrite is + // retried next tick rather than silently dropped. + noteHardFailure("measurements:update"); + } + } + + if (conflicts.length > 0) { + let resolved = 0; + for (const { r, movedFrom } of conflicts) { + const verdict = await prisma + .$transaction((tx) => + reconcileExternalMeasurement(tx, { + userId, + type: r.type as MeasurementType, + source: "GOOGLE_HEALTH", + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + externalId: r.externalId, + sleepStage: r.sleepStage ?? null, + }), + ) + .catch((err): MeasurementReconciliationVerdict => ({ + status: "failed", + error: { message: String(err) }, + })); + if (verdict.status === "failed") { + getEvent()?.addWarning( + `google-health: failed to reconcile measurement: ${verdict.error.message}`, + ); + noteHardFailure("measurements:update"); + continue; + } + if (verdict.status === "rejected_range") continue; + resolved++; + imported++; + touched.push({ + type: r.type as MeasurementType, + measuredAt: r.measuredAt, + }); + if (movedFrom) { + touched.push({ + type: r.type as MeasurementType, + measuredAt: movedFrom, + }); + } + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push({ type: dirty.type, measuredAt: dirty.measuredAt }); + } + } + annotate({ + meta: { + "googleHealth.naturalKeyConflicts": conflicts.length, + "googleHealth.naturalKeyConflictsResolved": resolved, + }, + }); + } + // Fresh inserts: chunked `createMany` (server-owned rows, field-by-field). // `skipDuplicates` guards the partial-unique index in the rare race where a // concurrent run inserted the same live key between the probe and the write. @@ -831,50 +1171,6 @@ export async function upsertGoogleHealthMeasurements( } } - // Existing-row overwrites: per-row update (differing values) on the matched - // id, so the re-fetched daily summary overwrites in place and bumps - // `syncVersion`. `deletedAt: null` rides along unconditionally — a no-op on - // a live row, a deliberate RESURRECTION on a tombstoned one (Google is the - // source of truth for its own rows; see TOMBSTONES RESURRECT above). - for (const { id, r, reKeyTo, movedFrom } of toUpdate) { - try { - await prisma.measurement.update({ - where: { id }, - data: { - value: r.value, - unit: r.unit, - measuredAt: r.measuredAt, - sleepStage: r.sleepStage ?? null, - deletedAt: null, - // Key-format migration: adopt the fresh externalId in place. - ...(reKeyTo ? { externalId: reKeyTo } : {}), - // Surface the server-side mutation to the iOS LWW reconciler. - syncVersion: { increment: 1 }, - }, - }); - touched.push({ - type: r.type as MeasurementType, - measuredAt: r.measuredAt, - }); - // A row that moved (a daily anchor re-cut on the user's local noon) - // leaves its old day too; that day's rollup needs the refold. - if (movedFrom) { - touched.push({ - type: r.type as MeasurementType, - measuredAt: movedFrom, - }); - } - imported++; - } catch (err) { - getEvent()?.addWarning( - `google-health: failed to update measurement: ${err}`, - ); - // Same rule as the create catch: hold the watermark so the overwrite is - // retried next tick rather than silently dropped. - noteHardFailure("measurements:update"); - } - } - // On a `fullSync` backfill the caller defers the rollup fold: it collapses // every resource's touched type-days into a SINGLE range-recompute at the end // of the cycle. The incremental path keeps the inline per-day hook here. @@ -919,14 +1215,21 @@ export async function upsertGoogleHealthMeasurements( } /** - * Stamp `lastSyncedAt = now`. Called ONCE per cycle by `syncUserGoogleHealth` - * after a non-degenerate run — never per resource, so the watermark can't move - * mid-cycle and shrink a later resource's fetch window. + * Stamp `lastSyncedAt`. Called ONCE per cycle by `syncUserGoogleHealth` after + * a non-degenerate run — never per resource, so the watermark can't move + * mid-cycle and shrink a later resource's fetch window. The orchestrator + * passes the instant the cycle STARTED: a sample written to Google while the + * cycle was still walking is newer than what the walk read, and stamping the + * end would put the watermark past it. That gap is seconds for an hourly run + * and hours for a full-history backfill. */ -export async function markSynced(userId: string): Promise { +export async function markSynced( + userId: string, + at: Date = new Date(), +): Promise { await prisma.googleHealthConnection.update({ where: { userId }, - data: { lastSyncedAt: new Date() }, + data: { lastSyncedAt: at }, }); } diff --git a/src/lib/google-health/sync-lock.ts b/src/lib/google-health/sync-lock.ts new file mode 100644 index 000000000..db8ca51c8 --- /dev/null +++ b/src/lib/google-health/sync-lock.ts @@ -0,0 +1,48 @@ +/** + * One Google Health run per account at a time, across processes. + * + * The hourly poll, the manual trigger, the full-history backfill and the + * sleep repair all read the same account's collections with the same grant. + * Run side by side they share Google's per-user, per-minute request quota: + * a backfill deep in a dense heart-rate walk and an hourly sync starting + * beside it pushed the account over the quota, and every request either of + * them made in that minute answered 429. They also write the same rows, so + * two runs racing one night's sleep segments can each move a row the other + * is about to move. + * + * So every run for one account holds the same advisory lock for as long as + * it reads and writes (`withJobLock`, a connection of its own, released when + * the run ends or the process dies). A run that finds it taken does no work: + * 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. + */ +import { withJobLock, type GuardedRun } from "@/lib/jobs/job-lock"; + +/** The lock key for one account's Google Health runs. */ +export function googleHealthSyncLockKey(userId: string): string { + return `google-health-sync:${userId}`; +} + +/** + * How long a backfill or repair waits for an hourly or manual run to finish. + * Those runs read a day of history at most and take a minute or two, so ten + * minutes covers one that is also sitting out a rate limit. + */ +export const GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS = 10 * 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 + * not let go within `waitMs` (zero: give up at once). + */ +export function withGoogleHealthSyncLock( + userId: string, + run: () => Promise, + options: { waitMs?: number } = {}, +): Promise> { + return withJobLock(googleHealthSyncLockKey(userId), run, undefined, { + waitMs: options.waitMs ?? 0, + }); +} diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index 473a06147..59deed1d7 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -30,6 +30,7 @@ import { GOOGLE_HEALTH_DENSE_MAX_PAGES, type GoogleHealthDataType, type GoogleHealthMappedMeasurement, + type GoogleHealthTokenSource, fetchDataPoints, forEachDataPointPage, mapBloodGlucose, @@ -46,7 +47,9 @@ import { } from "./client"; import { getValidToken, + googleHealthTokenSource, handleCollectionFetchError, + runGoogleHealthCollection, upsertGoogleHealthMeasurements, type GoogleHealthMeasurementUpsert, type GoogleHealthResourceSyncOptions, @@ -137,6 +140,7 @@ export async function syncUserMetrics( ): Promise { const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; + const token = googleHealthTokenSource(userId, tokenInfo); // The incremental lower bound is the cycle-wide watermark snapshotted once by // `syncUserGoogleHealth` — never re-read here, so a sibling resource's stamp @@ -167,38 +171,9 @@ export async function syncUserMetrics( // the failure stay written: every write is keyed, so the retry that the // failed verdict causes overwrites them in place. for (const resource of METRIC_RESOURCES) { - try { - await forEachDataPointPage( - resource.dataType, - tokenInfo.accessToken, - resource.verb, - { - start: resource.intraday ? (opts.intradayStart ?? start) : start, - maxPages: resource.maxPages, - }, - async (points) => { - const readings: GoogleHealthMeasurementUpsert[] = []; - for (const point of points) { - for (const m of resource.map(point, tz)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: m.fieldTag, - }); - } - } - imported += ( - await upsertGoogleHealthMeasurements(userId, readings, { - deferRollup: opts.deferRollup, - }) - ).imported; - }, - ); - } catch (err) { - imported += await handleCollectionFetchError(resource.verb, userId, err); - } + imported += await runGoogleHealthCollection(resource.verb, () => + syncMetricCollection(userId, resource, token, tz, opts), + ); } // Height → User.heightCm, only when the user has no height yet. Never mint a @@ -208,7 +183,7 @@ export async function syncUserMetrics( try { heightPoints = await fetchDataPoints( GOOGLE_HEALTH_DATA_TYPES.height, - tokenInfo.accessToken, + token, "fetchHeight", { start }, ); @@ -255,3 +230,53 @@ export async function syncUserMetrics( // mid-cycle. return imported; } + +/** + * Walk one metric collection page by page and write each page as it arrives. + * A failure (fetch, map or write) goes through the collection ledger and + * stops only this collection; the siblings still run. + */ +async function syncMetricCollection( + userId: string, + resource: MetricResource, + token: GoogleHealthTokenSource, + tz: string, + opts: GoogleHealthResourceSyncOptions, +): Promise { + let imported = 0; + try { + await forEachDataPointPage( + resource.dataType, + token, + resource.verb, + { + start: resource.intraday + ? (opts.intradayStart ?? opts.start) + : opts.start, + maxPages: resource.maxPages, + }, + async (points) => { + const readings: GoogleHealthMeasurementUpsert[] = []; + for (const point of points) { + for (const m of resource.map(point, tz)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.fieldTag, + }); + } + } + imported += ( + await upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + }, + ); + } catch (err) { + imported += await handleCollectionFetchError(resource.verb, userId, err); + } + return imported; +} diff --git a/src/lib/google-health/sync-sleep.ts b/src/lib/google-health/sync-sleep.ts index 7bd1bf4be..4d79e797a 100644 --- a/src/lib/google-health/sync-sleep.ts +++ b/src/lib/google-health/sync-sleep.ts @@ -28,8 +28,10 @@ import { } from "./client"; import { getValidToken, + googleHealthTokenSource, handleCollectionFetchError, noteHardFailure, + runGoogleHealthCollection, replaceStaleGoogleHealthSleep, upsertGoogleHealthMeasurements, type GoogleHealthMeasurementUpsert, @@ -43,6 +45,15 @@ import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger" export async function syncUserSleep( userId: string, opts: GoogleHealthResourceSyncOptions = {}, +): Promise { + return runGoogleHealthCollection("sleep", () => + syncSleepSessions(userId, opts), + ); +} + +async function syncSleepSessions( + userId: string, + opts: GoogleHealthResourceSyncOptions, ): Promise { const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; @@ -60,7 +71,7 @@ export async function syncUserSleep( try { points = await fetchDataPoints( GOOGLE_HEALTH_DATA_TYPES.sleep, - tokenInfo.accessToken, + googleHealthTokenSource(userId, tokenInfo), "fetchSleep", { start, pageSize: GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE }, ); @@ -116,7 +127,16 @@ export async function syncUserSleep( }); } - await replaceStaleGoogleHealthSleep(userId, replaceWindows); + // Every session's fresh rows are protected in every window, not only in + // their own. Google can report two sessions over the same stretch of a + // night (two sources, or a nap inside the main sleep), and a window that + // protected only its own session's rows tombstoned the other session's rows + // on every sync, for the upsert to bring them straight back. + const allKeepIds = readings.map((r) => r.externalId); + await replaceStaleGoogleHealthSleep( + userId, + replaceWindows.map((w) => ({ ...w, keepIds: allKeepIds })), + ); const { imported, inserted } = await upsertGoogleHealthMeasurements( userId, diff --git a/src/lib/google-health/sync-workout.ts b/src/lib/google-health/sync-workout.ts index fd559fb3e..24671cf79 100644 --- a/src/lib/google-health/sync-workout.ts +++ b/src/lib/google-health/sync-workout.ts @@ -25,8 +25,10 @@ import { } from "./client"; import { getValidToken, + googleHealthTokenSource, handleCollectionFetchError, noteHardFailure, + runGoogleHealthCollection, type GoogleHealthResourceSyncOptions, } from "./sync-core"; import { prisma } from "@/lib/db"; @@ -38,6 +40,15 @@ import { resolveUserTimezone } from "@/lib/tz/resolver"; export async function syncUserWorkout( userId: string, opts: GoogleHealthResourceSyncOptions = {}, +): Promise { + return runGoogleHealthCollection("fetchExercise", () => + syncExerciseSessions(userId, opts), + ); +} + +async function syncExerciseSessions( + userId: string, + opts: GoogleHealthResourceSyncOptions, ): Promise { const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; @@ -57,7 +68,7 @@ export async function syncUserWorkout( try { points = await fetchDataPoints( GOOGLE_HEALTH_DATA_TYPES.exercise, - tokenInfo.accessToken, + googleHealthTokenSource(userId, tokenInfo), "fetchExercise", { start, pageSize: GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE, tz }, ); diff --git a/src/lib/google-health/sync.ts b/src/lib/google-health/sync.ts index 5876b81a8..b0a72601f 100644 --- a/src/lib/google-health/sync.ts +++ b/src/lib/google-health/sync.ts @@ -1,7 +1,7 @@ import pLimit from "p-limit"; import { prisma } from "@/lib/db"; import { annotate, getEvent } from "@/lib/logging/context"; -import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status"; +import { recordSyncSuccess } from "@/lib/integrations/status"; import { collapseToTypeDayKeys, recomputeUserRollups, @@ -17,8 +17,10 @@ import { syncUserActivity } from "./sync-activity"; import { syncUserMetrics } from "./sync-metrics"; import { syncUserSleep } from "./sync-sleep"; import { syncUserWorkout } from "./sync-workout"; +import { withGoogleHealthSyncLock } from "./sync-lock"; import { GOOGLE_HEALTH_INTEGRATION_KEY, + getValidToken, incrementalStart, intradayOverlapMs, markSynced, @@ -40,6 +42,64 @@ export interface GoogleHealthSyncResult { imported: number; failed: boolean; resources?: GoogleHealthResourceOutcome[]; + /** + * Set when nothing ran because another run for this account holds the + * account's sync lock (a backfill, the hourly poll, a manual trigger). + * `failed` is true alongside it so a caller that ignores the flag cannot + * mistake the skipped run for a clean one. + */ + busy?: true; +} + +/** + * Whether an account's Google Health connection is parked for a reason no + * retry can fix, so the run is not attempted. + * + * `parked` (a persistent failure older than a day) is one: it waits for the + * operator or the user to resume it. `error_reauth` is one only when the + * grant itself was refused: the token endpoint answered `invalid_grant` or + * 401, which `getValidToken` marks on the connection as `needsReauth`. The + * ledger also reaches `error_reauth` from a 401 on a data request, and that + * one says nothing certain about the grant: an access token that expired + * part-way through a long walk produced it, with a refresh token that was + * still valid, and the account stayed parked until it was reconnected by + * hand (#1194). Such a connection is not parked here; the next run refreshes + * the token first, a refresh that works clears the state when the run + * completes, and a refresh that is refused marks `needsReauth` and parks it + * properly. + */ +export async function isGoogleHealthParked(userId: string): Promise { + const state = await ledgerState(userId); + if (state === "parked") return true; + if (state !== "error_reauth") return false; + const connection = await prisma.googleHealthConnection.findUnique({ + where: { userId }, + select: { needsReauth: true }, + }); + return connection?.needsReauth !== false; +} + +async function ledgerState(userId: string): Promise { + const row = await prisma.integrationStatus.findUnique({ + where: { + userId_integration: { + userId, + integration: GOOGLE_HEALTH_INTEGRATION_KEY, + }, + }, + select: { state: true }, + }); + return row?.state ?? null; +} + +/** Options for one `syncUserGoogleHealth` run. */ +export interface GoogleHealthSyncOptions { + fullSync?: boolean; + /** + * How long to wait for another run of this account to let go of the + * account's sync lock. Zero (the default) answers `busy` at once. + */ + waitForLockMs?: number; } const RESOURCE_STATUSES = new Set([ @@ -135,15 +195,39 @@ function terminalResourceOutcome( */ export async function syncUserGoogleHealth( userId: string, - opts: { fullSync?: boolean } = {}, + opts: GoogleHealthSyncOptions = {}, ): Promise { - if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) { + if (await isGoogleHealthParked(userId)) { getEvent()?.addWarning( `google-health sync skipped for ${userId}: parked at error_reauth`, ); return { state: "failed", imported: 0, failed: true, resources: [] }; } + const run = await withGoogleHealthSyncLock( + userId, + () => runGoogleHealthSyncCycle(userId, opts), + { waitMs: opts.waitForLockMs ?? 0 }, + ); + if (!run.ran) { + annotate({ meta: { "googleHealth.sync.busy": true } }); + return { + state: "in_progress", + imported: 0, + failed: true, + resources: [], + busy: true, + }; + } + return run.result; +} + +/** One sync cycle, run while holding the account's sync lock. */ +async function runGoogleHealthSyncCycle( + userId: string, + opts: GoogleHealthSyncOptions, +): Promise { + const cycleStartedAt = new Date(); const connection = await prisma.googleHealthConnection.findUnique({ where: { userId }, select: { lastSyncedAt: true }, @@ -152,6 +236,19 @@ export async function syncUserGoogleHealth( return { state: "failed", imported: 0, failed: true, resources: [] }; } + // A connection held at `error_reauth` by a data request's 401 (see + // `isGoogleHealthParked`) gets a fresh token before anything else: the + // refresh is the one call that can tell an expired access token from a + // revoked grant. Refused, it records the reauth and the run stops here. + if ((await ledgerState(userId)) === "error_reauth") { + const probe = await runWithGoogleHealthSyncCycle(() => + getValidToken(userId, { forceRefresh: true }), + ); + if (!probe.result) { + return { state: "failed", imported: 0, failed: true, resources: [] }; + } + } + const progress = await startGoogleHealthSyncProgress(userId); const persistProgress = async ( state: GoogleHealthSyncState, @@ -287,7 +384,7 @@ export async function syncUserGoogleHealth( const failed = anyFailed || allSoftSkipped || truncated; if (!failed) { - await markSynced(userId); + await markSynced(userId, cycleStartedAt); await recordSyncSuccess(userId, GOOGLE_HEALTH_INTEGRATION_KEY); } diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index 09a50ff29..2773abef9 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -7,14 +7,19 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { prismaMock, syncUserGoogleHealthMock, isReauthRequiredMock } = - vi.hoisted(() => ({ +const { prismaMock, syncUserGoogleHealthMock, isParkedMock } = vi.hoisted( + () => ({ prismaMock: { - googleHealthConnection: { findMany: vi.fn(), update: vi.fn() }, + googleHealthConnection: { + findMany: vi.fn(), + findUnique: vi.fn(async () => null), + update: vi.fn(), + }, }, syncUserGoogleHealthMock: vi.fn(), - isReauthRequiredMock: vi.fn(async () => false), - })); + isParkedMock: vi.fn(async () => false), + }), +); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); @@ -33,14 +38,17 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { return { ...actual, safeFetch: safeFetchMock }; }); -vi.mock("@/lib/integrations/status", () => ({ - isReauthRequired: isReauthRequiredMock, -})); - vi.mock("@/lib/google-health/sync-core", () => ({ GOOGLE_HEALTH_INTEGRATION_KEY: "google-health", + readGoogleHealthBackfillDone: () => new Set(), + runWithGoogleHealthBackfillResume: async ( + _userId: string, + _done: Set, + fn: () => Promise, + ) => ({ result: await fn(), skipped: [] }), })); vi.mock("@/lib/google-health/sync", () => ({ + isGoogleHealthParked: isParkedMock, syncUserGoogleHealth: (...a: unknown[]) => syncUserGoogleHealthMock(...a), })); @@ -53,7 +61,7 @@ import { beforeEach(() => { vi.clearAllMocks(); - isReauthRequiredMock.mockResolvedValue(false); + isParkedMock.mockResolvedValue(false); prismaMock.googleHealthConnection.update.mockResolvedValue({}); }); @@ -66,6 +74,7 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { expect(imported).toBe(42); expect(syncUserGoogleHealthMock).toHaveBeenCalledWith("u1", { fullSync: true, + waitForLockMs: expect.any(Number), }); const updateArg = prismaMock.googleHealthConnection.update.mock .calls[0]![0] as { @@ -74,6 +83,21 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { }; expect(updateArg.where).toEqual({ userId: "u1" }); expect(updateArg.data.backfillCompletedAt).toBeInstanceOf(Date); + // The resume note goes with the stamp: a later backfill starts over. + expect(updateArg.data).toHaveProperty("backfillProgress"); + }); + + it("throws without stamping when another sync of the account holds it", async () => { + syncUserGoogleHealthMock.mockResolvedValue({ + imported: 0, + failed: true, + busy: true, + }); + + await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow( + /another sync of the account is running/, + ); + expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); }); it("a failed verdict THROWS without stamping — pg-boss retries become real", async () => { @@ -86,7 +110,7 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { }); it("a connection parked at error_reauth returns WITHOUT running the sync or stamping", async () => { - isReauthRequiredMock.mockResolvedValue(true); + isParkedMock.mockResolvedValue(true); await expect(runGoogleHealthBackfillForUser("u1")).resolves.toEqual({ imported: 0, diff --git a/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts b/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts index b64c9619d..35d232e7f 100644 --- a/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts +++ b/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts @@ -30,6 +30,20 @@ vi.mock("@/lib/google-health/sync-sleep", () => ({ syncUserSleep: (...a: unknown[]) => syncUserSleep(...a), })); +vi.mock("@/lib/google-health/sync", () => ({ + isGoogleHealthParked: isReauthRequiredMock, +})); + +// The account lock is a Postgres advisory lock on a connection of its own; +// these cases run without a database, so the lock is always free. +vi.mock("@/lib/google-health/sync-lock", () => ({ + GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS: 0, + withGoogleHealthSyncLock: async ( + _userId: string, + run: () => Promise, + ) => ({ ran: true, result: await run() }), +})); + vi.mock("@/lib/logging/context", () => ({ annotate: () => {}, getEvent: () => null, diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index e37268a2b..03d428e5e 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -11,13 +11,20 @@ * provisions it and the boot enqueue silently never drains (the v1.4.37 * dead-queue class). */ +import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; -import { isReauthRequired } from "@/lib/integrations/status"; -import { GOOGLE_HEALTH_INTEGRATION_KEY } from "@/lib/google-health/sync-core"; -import { syncUserGoogleHealth } from "@/lib/google-health/sync"; +import { + readGoogleHealthBackfillDone, + runWithGoogleHealthBackfillResume, +} from "@/lib/google-health/sync-core"; +import { + isGoogleHealthParked, + 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"; export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill"; @@ -53,14 +60,24 @@ export interface GoogleHealthBackfillPayload { * 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; the retry or the next boot walks - * again and the key-stable upserts overwrite them. + * the stamp. The rows already written stay. + * + * 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. + * + * 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 + * lock does not come free the attempt throws without doing anything, and the + * retry tries again. */ export async function runGoogleHealthBackfillForUser( userId: string, shouldStop: () => boolean = () => false, ): Promise<{ imported: number }> { - if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) { + if (await isGoogleHealthParked(userId)) { annotate({ action: { name: "google_health.backfill.skipped_reauth", @@ -70,9 +87,32 @@ export async function runGoogleHealthBackfillForUser( return { imported: 0 }; } - const { imported, failed } = await runWithGoogleHealthStop(shouldStop, () => - syncUserGoogleHealth(userId, { fullSync: true }), + const connection = await prisma.googleHealthConnection.findUnique({ + where: { userId }, + select: { backfillProgress: true }, + }); + const done = readGoogleHealthBackfillDone(connection?.backfillProgress); + + const { + result: { imported, failed, busy }, + skipped, + } = await runWithGoogleHealthBackfillResume(userId, done, () => + runWithGoogleHealthStop(shouldStop, () => + syncUserGoogleHealth(userId, { + fullSync: true, + waitForLockMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS, + }), + ), ); + if (skipped.length > 0) { + annotate({ meta: { "googleHealth.backfill.resumedPast": skipped } }); + } + + if (busy) { + throw new Error( + `google-health backfill for user ${userId} deferred: another sync of the account is running`, + ); + } if (failed) { // Surface through the pg-boss retry path (retryLimit 3, backoff). If the @@ -85,7 +125,7 @@ export async function runGoogleHealthBackfillForUser( await prisma.googleHealthConnection.update({ where: { userId }, - data: { backfillCompletedAt: new Date() }, + data: { backfillCompletedAt: new Date(), backfillProgress: Prisma.DbNull }, }); annotate({ diff --git a/src/lib/jobs/google-health-sleep-repair.ts b/src/lib/jobs/google-health-sleep-repair.ts index 96a232f45..8c01adc9b 100644 --- a/src/lib/jobs/google-health-sleep-repair.ts +++ b/src/lib/jobs/google-health-sleep-repair.ts @@ -27,13 +27,16 @@ import { prisma } from "@/lib/db"; import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; -import { isReauthRequired } from "@/lib/integrations/status"; import { - GOOGLE_HEALTH_INTEGRATION_KEY, GOOGLE_HEALTH_TOKEN_HARD_FAIL, runWithGoogleHealthHardFailLedger, } from "@/lib/google-health/sync-core"; import { syncUserSleep } from "@/lib/google-health/sync-sleep"; +import { isGoogleHealthParked } from "@/lib/google-health/sync"; +import { + GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS, + withGoogleHealthSyncLock, +} from "@/lib/google-health/sync-lock"; export const GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE = "google-health-sleep-repair"; @@ -71,7 +74,7 @@ export interface GoogleHealthSleepRepairPayload { export async function runGoogleHealthSleepRepairForUser( userId: string, ): Promise<{ imported: number }> { - if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) { + if (await isGoogleHealthParked(userId)) { annotate({ action: { name: "googleHealth.sleepRepair.skipped_reauth", @@ -81,10 +84,22 @@ export async function runGoogleHealthSleepRepairForUser( return { imported: 0 }; } - const { result: imported, failures } = - await runWithGoogleHealthHardFailLedger(() => - syncUserSleep(userId, { deferRollup: false }), + // The repair re-reads the account's sleep, so it takes the account's lock + // like every other run (`withGoogleHealthSyncLock`). + const run = await withGoogleHealthSyncLock( + userId, + () => + runWithGoogleHealthHardFailLedger(() => + syncUserSleep(userId, { deferRollup: false }), + ), + { waitMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS }, + ); + if (!run.ran) { + throw new Error( + `google-health sleep repair for user ${userId} deferred: another sync of the account is running`, ); + } + const { result: imported, failures } = run.result; if (failures.length > 0) { if (failures.every((f) => f === GOOGLE_HEALTH_TOKEN_HARD_FAIL)) { diff --git a/src/lib/jobs/job-lock.ts b/src/lib/jobs/job-lock.ts index a7a49e737..711fe471d 100644 --- a/src/lib/jobs/job-lock.ts +++ b/src/lib/jobs/job-lock.ts @@ -24,15 +24,31 @@ import { caughtAs } from "@/lib/logging/signal"; /** What a guarded run came to. */ export type GuardedRun = { ran: true; result: T } | { ran: false }; +export interface JobLockOptions { + /** + * How long to keep asking for a lock another run holds before giving up. + * Zero (the default) gives up at once. A caller that would rather wait out + * a short run than drop its own work sets this; a run whose holder can take + * hours never should. + */ + waitMs?: number; + /** The pause between two asks while waiting. */ + pollMs?: number; +} + +const DEFAULT_LOCK_POLL_MS = 2_000; + /** * Run `run` while holding the advisory lock for `key`. Resolves - * `{ ran: false }` at once, without calling `run`, when another run holds it. + * `{ ran: false }` without calling `run` when another run holds it, at once or, + * with `waitMs`, once that long has passed without the lock coming free. * Rejects exactly as `run` does otherwise. */ export async function withJobLock( key: string, run: () => Promise, connectionString: string | undefined = process.env.DATABASE_URL, + options: JobLockOptions = {}, ): Promise> { const client = new Client({ connectionString, keepAlive: true }); // An idle client whose server goes away emits `error`; unhandled, that @@ -41,11 +57,17 @@ export async function withJobLock( client.on("error", () => {}); await client.connect(); try { - const { rows } = await client.query<{ held: boolean }>( - "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held", - [key], - ); - if (!rows[0]?.held) return { ran: false }; + const deadline = Date.now() + Math.max(0, options.waitMs ?? 0); + const pollMs = Math.max(1, options.pollMs ?? DEFAULT_LOCK_POLL_MS); + for (;;) { + const { rows } = await client.query<{ held: boolean }>( + "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held", + [key], + ); + if (rows[0]?.held) break; + if (Date.now() + pollMs > deadline) return { ran: false }; + await new Promise((resolve) => setTimeout(resolve, pollMs)); + } try { return { ran: true, result: await run() }; } finally { diff --git a/src/lib/jobs/reminder/google-health-sync.ts b/src/lib/jobs/reminder/google-health-sync.ts index 57a7bfa19..f58329bc0 100644 --- a/src/lib/jobs/reminder/google-health-sync.ts +++ b/src/lib/jobs/reminder/google-health-sync.ts @@ -12,9 +12,9 @@ import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { GOOGLE_HEALTH_POLL_CONCURRENCY, + isGoogleHealthParked, syncUserGoogleHealth, } from "@/lib/google-health/sync"; -import { isReauthRequired } from "@/lib/integrations/status"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup"; import { @@ -64,16 +64,19 @@ export async function handleGoogleHealthSync( const verdicts = await Promise.all( targets.map(({ userId }) => limit(async (): Promise => { - if (await isReauthRequired(userId, "google-health")) { + if (await isGoogleHealthParked(userId)) { return { status: "parked", imported: 0 }; } try { const result = await syncUserGoogleHealth(userId); + // Another run of this account (its backfill, a manual sync) + // holds the account; this tick leaves it to that run. + if (result.busy) return { status: "skipped", imported: 0 }; const parked = result.failed && result.imported === 0 && - (await isReauthRequired(userId, "google-health")); + (await isGoogleHealthParked(userId)); const status = parked ? "parked" : result.failed diff --git a/src/lib/openapi/error-codes.ts b/src/lib/openapi/error-codes.ts index 95f87bb50..e0ae588ad 100644 --- a/src/lib/openapi/error-codes.ts +++ b/src/lib/openapi/error-codes.ts @@ -433,6 +433,7 @@ export const ERROR_CODE_CATALOGUE: Readonly> = "private_origin_not_approved", "private_origin_not_grantable", "rate_limited_self", + "sync_in_progress", "too_many_rows", "unreadable_json", "upstream_invalid_json", diff --git a/src/lib/openapi/routes/integrations.ts b/src/lib/openapi/routes/integrations.ts index ad7793b75..0967516c9 100644 --- a/src/lib/openapi/routes/integrations.ts +++ b/src/lib/openapi/routes/integrations.ts @@ -1955,6 +1955,11 @@ export const integrationPaths: NonNullable = { "The run failed and wrote nothing. A run that wrote something before failing answers 200 instead.", content: { "application/json": { schema: errorEnvelope } }, }, + "409": { + description: + "Another sync of this account is already running (the history backfill after a reconnect, or the hourly poll), so none was started. `meta.errorCode` = `sync_in_progress`. `GET /api/google-health/sync/status` reports the running one.", + content: { "application/json": { schema: errorEnvelope } }, + }, ...stdResponses, "422": { description: diff --git a/tests/integration/google-health-backfill-rate-limit.test.ts b/tests/integration/google-health-backfill-rate-limit.test.ts new file mode 100644 index 000000000..6c2d3e04a --- /dev/null +++ b/tests/integration/google-health-backfill-rate-limit.test.ts @@ -0,0 +1,436 @@ +/** + * #1194 — a Google Health history backfill that meets Google's per-minute + * quota and an access token that runs out part-way, against Postgres. + * + * What went wrong. After a reconnect the backfill walked the account's whole + * history while the hourly sync ran beside it, and the two together went over + * Google's per-user, per-minute quota: every request in that minute answered + * 429. The backfill resolved one access token at the start of each resource + * and carried it to the end, and the heart-rate walk outlived it: an hour + * after the reconnect every request answered 401, a 401 classifies as a + * revoked grant, and the account was parked at `error_reauth` with a refresh + * token that was still good. Every later run skipped the parked account, so + * nothing synced until it was reconnected by hand, and the next backfill + * parked it again. + * + * What this pins, each against the real sync, ledger and lock: + * - a 429 is waited out and the request repeated; it never parks anything; + * - a token that reaches its expiry mid-walk is renewed before the next + * request, and one Google refuses mid-walk is renewed once and the + * request repeated, so the walk finishes and the ledger stays connected; + * - an account held at `error_reauth` by a data request's 401 recovers on + * the next run, while one whose grant the token endpoint refused stays + * parked; + * - a backfill that stops part-way resumes with the collections it did not + * finish, and finishes; + * - while one run of an account holds it, the hourly poll and a manual sync + * do nothing for that account. + * + * Google is replaced at the `safeFetch` seam by a small fake that issues + * tokens, answers the data reads, and can be told to throttle, revoke or + * hold a request. + */ +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER = "gh-rate-limit-owner"; + +const fake = vi.hoisted(() => ({ + /** Heart-rate points in the account's history; 10 per page. */ + heartRatePoints: 60, + /** The access token Google currently accepts; others answer 401. */ + validToken: "access-0", + /** Tokens issued so far by the token endpoint. */ + issued: 0, + /** `expires_in` the token endpoint answers with. */ + expiresIn: 3600, + /** + * How many data requests one token serves before Google treats it as + * expired (the stand-in for an hour passing). Unlimited when 0. + */ + requestsPerToken: 0, + servedByCurrentToken: 0, + /** Answer the token endpoint with `invalid_grant`. */ + refuseGrant: false, + /** Heart-rate page indexes that answer 429 (once each) before succeeding. */ + throttledPages: new Set(), + /** Revoke the current access token when this heart-rate page is asked. */ + revokeAtPage: -1, + /** Hold the request for this heart-rate page until `release` is called. */ + holdAtPage: -1, + release: null as null | (() => void), + held: null as null | Promise, + /** Counters. */ + requestsByPath: new Map(), + status429: 0, + status401: 0, + now: Date.parse("2026-10-09T12:00:00.000Z"), +})); + +function json( + body: unknown, + status = 200, + headers: Record = {}, +) { + return new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json", ...headers }, + }); +} + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + safeFetch: async (url: string, init?: RequestInit) => { + const u = new URL(url); + if (u.hostname === "oauth2.googleapis.com") { + if (fake.refuseGrant) { + return json({ error: "invalid_grant" }, 400); + } + fake.issued += 1; + fake.validToken = `access-${fake.issued}`; + fake.servedByCurrentToken = 0; + return json({ + access_token: fake.validToken, + expires_in: fake.expiresIn, + token_type: "Bearer", + }); + } + + const path = u.pathname.replace(/^\/v4\/users\/me\/dataTypes\//, ""); + fake.requestsByPath.set(path, (fake.requestsByPath.get(path) ?? 0) + 1); + const auth = new Headers(init?.headers).get("authorization"); + if ( + fake.requestsPerToken > 0 && + fake.servedByCurrentToken >= fake.requestsPerToken + ) { + fake.validToken = "expired"; + } + fake.servedByCurrentToken += 1; + if (auth !== `Bearer ${fake.validToken}`) { + fake.status401 += 1; + return json({ error: { code: 401, status: "UNAUTHENTICATED" } }, 401); + } + + if (path === "heart-rate/dataPoints") { + const page = Number(u.searchParams.get("pageToken") ?? 0); + if (page === fake.revokeAtPage) { + fake.revokeAtPage = -1; + fake.validToken = "revoked"; + fake.status401 += 1; + return json({ error: { code: 401, status: "UNAUTHENTICATED" } }, 401); + } + if (fake.throttledPages.has(page)) { + fake.throttledPages.delete(page); + fake.status429 += 1; + return json( + { + error: { + code: 429, + status: "RESOURCE_EXHAUSTED", + message: + "Quota exceeded for quota metric 'Requests per minute per user'", + }, + }, + 429, + { "retry-after": "0" }, + ); + } + if (page === fake.holdAtPage) { + fake.holdAtPage = -1; + await fake.held; + } + const end = Math.min(fake.heartRatePoints, (page + 1) * 10); + const dataPoints = []; + for (let g = page * 10; g < end; g++) { + dataPoints.push({ + name: `users/me/dataTypes/heart-rate/dataPoints/${1_000_000 + g}`, + heartRate: { + beatsPerMinute: String(60 + (g % 20)), + sampleTime: { + physicalTime: new Date(fake.now - g * 60_000).toISOString(), + }, + }, + }); + } + return json({ + dataPoints, + ...(end < fake.heartRatePoints + ? { nextPageToken: String(page + 1) } + : {}), + }); + } + // Every other collection is empty for this account. + return json({}); + }, + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +async function seed(): Promise { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + const { encrypt } = await import("@/lib/crypto"); + await prisma.user.create({ + data: { + id: USER, + username: USER, + timezone: "Europe/Berlin", + googleHealthClientIdEncrypted: encrypt("client-id"), + googleHealthClientSecretEncrypted: encrypt("client-secret"), + }, + }); + await prisma.googleHealthConnection.create({ + data: { + userId: USER, + googleUserId: `google-${USER}`, + accessToken: encrypt("access-0"), + refreshToken: encrypt("refresh"), + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + }, + }); +} + +beforeEach(async () => { + fake.heartRatePoints = 60; + fake.validToken = "access-0"; + fake.issued = 0; + fake.expiresIn = 3600; + fake.requestsPerToken = 0; + fake.servedByCurrentToken = 0; + fake.refuseGrant = false; + fake.throttledPages = new Set(); + fake.revokeAtPage = -1; + fake.holdAtPage = -1; + fake.release = null; + fake.held = null; + fake.requestsByPath = new Map(); + fake.status429 = 0; + fake.status401 = 0; + await seed(); +}, 120_000); + +afterAll(async () => { + await truncateAllTables(getPrismaClient()); +}); + +async function ledgerState(): Promise { + const row = await getPrismaClient().integrationStatus.findUnique({ + where: { + userId_integration: { userId: USER, integration: "google-health" }, + }, + select: { state: true }, + }); + return row?.state ?? null; +} + +async function pulseRows(): Promise { + return getPrismaClient().measurement.count({ + where: { userId: USER, source: "GOOGLE_HEALTH", type: "PULSE" }, + }); +} + +describe("Google Health backfill under rate limits and token expiry (#1194)", () => { + it("waits out 429s, never parks, and finishes the backfill", async () => { + fake.throttledPages = new Set([1, 2, 4]); + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + + const { imported } = await runGoogleHealthBackfillForUser(USER); + + expect(fake.status429).toBe(3); + expect(imported).toBeGreaterThanOrEqual(60); + expect(await pulseRows()).toBe(60); + expect(await ledgerState()).toBe("connected"); + const connection = + await getPrismaClient().googleHealthConnection.findUnique({ + where: { userId: USER }, + }); + expect(connection?.backfillCompletedAt).toBeInstanceOf(Date); + expect(connection?.backfillProgress).toBeNull(); + expect(connection?.needsReauth).toBe(false); + }); + + it("renews a token that expires mid-walk before Google refuses it", async () => { + // Every token issued expires a second inside the refresh buffer, so it + // is due for renewal before each request, and Google stops taking it + // after three requests: a walk that carried one token to the end would + // hit 401 on its fourth. + fake.expiresIn = 299; + fake.requestsPerToken = 3; + await getPrismaClient().googleHealthConnection.update({ + where: { userId: USER }, + data: { tokenExpiresAt: new Date(Date.now() - 1000) }, + }); + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + + await runGoogleHealthBackfillForUser(USER); + + expect(fake.status401).toBe(0); + expect(fake.issued).toBeGreaterThan(1); + expect(await pulseRows()).toBe(60); + expect(await ledgerState()).toBe("connected"); + }); + + it("renews a token Google refuses mid-walk once and repeats the request", async () => { + fake.revokeAtPage = 3; + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + + await runGoogleHealthBackfillForUser(USER); + + expect(fake.status401).toBe(1); + expect(fake.issued).toBe(1); + expect(await pulseRows()).toBe(60); + expect(await ledgerState()).toBe("connected"); + }); + + it("recovers an account held at error_reauth by a data request's 401", async () => { + const prisma = getPrismaClient(); + await prisma.integrationStatus.create({ + data: { + userId: USER, + integration: "google-health", + state: "error_reauth", + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 8, + persistent: 0, + }, + }, + }); + const { syncUserGoogleHealth } = await import("@/lib/google-health/sync"); + + const result = await syncUserGoogleHealth(USER); + + expect(result.failed).toBe(false); + expect(fake.issued).toBe(1); + expect(await ledgerState()).toBe("connected"); + }); + + it("keeps an account parked when the token endpoint refuses the grant", async () => { + const prisma = getPrismaClient(); + await prisma.integrationStatus.create({ + data: { + userId: USER, + integration: "google-health", + state: "error_reauth", + }, + }); + fake.refuseGrant = true; + const { syncUserGoogleHealth } = await import("@/lib/google-health/sync"); + + const first = await syncUserGoogleHealth(USER); + expect(first.failed).toBe(true); + expect(await ledgerState()).toBe("error_reauth"); + const connection = await prisma.googleHealthConnection.findUnique({ + where: { userId: USER }, + select: { needsReauth: true }, + }); + expect(connection?.needsReauth).toBe(true); + + // Now parked for good: the next run does not even ask the token endpoint. + fake.requestsByPath = new Map(); + const second = await syncUserGoogleHealth(USER); + expect(second.failed).toBe(true); + expect(fake.requestsByPath.size).toBe(0); + }); + + it("resumes a backfill that stopped part-way with what it did not finish", async () => { + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + + // First attempt: the budget runs out three heart-rate pages in. + let heartRatePagesSeen = 0; + const stopAfterThreeHeartRatePages = () => { + heartRatePagesSeen = + fake.requestsByPath.get("heart-rate/dataPoints") ?? 0; + return heartRatePagesSeen >= 3; + }; + await expect( + runGoogleHealthBackfillForUser(USER, stopAfterThreeHeartRatePages), + ).rejects.toThrow(/incomplete/); + + const afterFirst = + await getPrismaClient().googleHealthConnection.findUnique({ + where: { userId: USER }, + }); + expect(afterFirst?.backfillCompletedAt).toBeNull(); + const done = (afterFirst?.backfillProgress as { done: string[] }).done; + expect(done).toEqual( + expect.arrayContaining([ + "fetchExercise", + "sleep", + "fetchSteps", + "fetchWeight", + ]), + ); + expect(done).not.toContain("fetchHeartRate"); + + // Second attempt: the finished collections are not read again. + fake.requestsByPath = new Map(); + await runGoogleHealthBackfillForUser(USER); + + expect(fake.requestsByPath.get("exercise/dataPoints")).toBeUndefined(); + expect(fake.requestsByPath.get("sleep/dataPoints")).toBeUndefined(); + expect( + fake.requestsByPath.get("steps/dataPoints:dailyRollUp"), + ).toBeUndefined(); + expect(fake.requestsByPath.get("heart-rate/dataPoints")).toBe(6); + expect(await pulseRows()).toBe(60); + const afterSecond = + await getPrismaClient().googleHealthConnection.findUnique({ + where: { userId: USER }, + }); + expect(afterSecond?.backfillCompletedAt).toBeInstanceOf(Date); + expect(afterSecond?.backfillProgress).toBeNull(); + }); + + it("never runs the hourly poll or a manual sync beside a running backfill", async () => { + fake.holdAtPage = 2; + fake.held = new Promise((resolve) => { + fake.release = resolve; + }); + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + const { syncUserGoogleHealth } = await import("@/lib/google-health/sync"); + const { handleGoogleHealthSync } = + await import("@/lib/jobs/reminder/google-health-sync"); + + const backfill = runGoogleHealthBackfillForUser(USER); + // Wait until the backfill is inside its heart-rate walk, holding the lock. + for (let i = 0; i < 200 && fake.holdAtPage !== -1; i++) { + await new Promise((r) => setTimeout(r, 25)); + } + expect(fake.holdAtPage).toBe(-1); + + const requestsBefore = [...fake.requestsByPath.values()].reduce( + (a, b) => a + b, + 0, + ); + 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 }, + }); + const requestsAfter = [...fake.requestsByPath.values()].reduce( + (a, b) => a + b, + 0, + ); + expect(requestsAfter).toBe(requestsBefore); + + fake.release?.(); + await backfill; + expect(await pulseRows()).toBe(60); + expect(await ledgerState()).toBe("connected"); + }); +}); diff --git a/tests/integration/google-health-sleep-overlap.test.ts b/tests/integration/google-health-sleep-overlap.test.ts new file mode 100644 index 000000000..ebc1e1e6a --- /dev/null +++ b/tests/integration/google-health-sleep-overlap.test.ts @@ -0,0 +1,258 @@ +/** + * #1195 — overlapping Google Health sleep segments, synced against Postgres. + * + * What went wrong. A measurement row has two unique identities: its + * externalId, and its natural key `(type, measuredAt, source, sleepStage)`. + * A sleep segment is stored at its END, so two segments of the same stage + * that end at the same instant share a natural key even though Google gives + * them different externalIds. Google reports such segments: two sessions over + * one stretch of a night, and segments it re-scores so that one ends where a + * sibling already ends. The sync updated rows by externalId one by one, the + * update that moved a row onto a slot another row held threw P2002, and the + * whole sync was marked failed, every hour, for a sync that had written + * everything else. + * + * What this pins: + * - a re-score that moves a segment onto a sibling's slot is a clean sync, + * with one row left on that slot carrying the longer segment; + * - two sessions reporting the same segments do not double the night, and + * re-syncing them leaves every row untouched (no tombstone churn, no + * `syncVersion` bump), so the slot does not change hands every hour; + * - the night's asleep total is the union of the overlapping segments, + * never their sum. + */ +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER = "gh-sleep-overlap-owner"; +const TZ = "Europe/Berlin"; + +const fake = vi.hoisted(() => ({ + sessions: [] as unknown[], +})); + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + safeFetch: async (url: string) => { + const u = new URL(url); + const body = u.pathname.endsWith("/dataTypes/sleep/dataPoints") + ? { dataPoints: fake.sessions } + : {}; + return new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }, + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +/** The night under test: 22:00 UTC two days ago to 06:00 UTC yesterday. */ +const NIGHT_START = (() => { + const d = new Date(); + d.setUTCHours(22, 0, 0, 0); + return d.getTime() - 2 * 24 * 60 * 60 * 1000; +})(); +const at = (minutes: number): string => + new Date(NIGHT_START + minutes * 60_000).toISOString(); + +function segment(type: string, from: number, to: number) { + return { type, startTime: at(from), endTime: at(to) }; +} + +function session( + id: string, + stages: Array>, +): unknown { + const starts = stages.map((s) => s.startTime).sort(); + const ends = stages.map((s) => s.endTime).sort(); + return { + name: `users/me/dataTypes/sleep/dataPoints/${id}`, + sleep: { + interval: { startTime: starts[0], endTime: ends[ends.length - 1] }, + stages, + }, + }; +} + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + const { encrypt } = await import("@/lib/crypto"); + await prisma.user.create({ + data: { id: USER, username: USER, timezone: TZ }, + }); + await prisma.googleHealthConnection.create({ + data: { + userId: USER, + googleUserId: `google-${USER}`, + accessToken: encrypt("access"), + refreshToken: encrypt("refresh"), + tokenExpiresAt: new Date(Date.now() + 24 * 60 * 60 * 1000), + }, + }); + fake.sessions = []; +}, 120_000); + +afterAll(async () => { + await truncateAllTables(getPrismaClient()); +}); + +async function sync() { + const { syncUserGoogleHealth } = await import("@/lib/google-health/sync"); + return syncUserGoogleHealth(USER); +} + +async function sleepRows() { + return getPrismaClient().measurement.findMany({ + where: { userId: USER, type: "SLEEP_DURATION" }, + select: { + id: true, + value: true, + measuredAt: true, + sleepStage: true, + source: true, + externalId: true, + deletedAt: true, + syncVersion: true, + }, + orderBy: [{ measuredAt: "asc" }, { id: "asc" }], + }); +} + +async function asleepMinutes(): Promise { + const { reconstructSleepNights } = + await import("@/lib/analytics/sleep-night"); + const live = (await sleepRows()).filter((r) => r.deletedAt === null); + const nights = reconstructSleepNights(live, TZ); + return nights.reduce((sum, n) => sum + n.asleepMinutes, 0); +} + +describe("Google Health overlapping sleep segments (#1195)", () => { + it("lands a re-score that moves a segment onto a sibling's slot without failing the sync", async () => { + // First scoring: two LIGHT segments, the second starting inside the + // first, ending at different instants. + fake.sessions = [ + session("night", [ + segment("LIGHT", 0, 60), + segment("LIGHT", 30, 120), + segment("DEEP", 120, 180), + ]), + ]; + const first = await sync(); + expect(first.failed).toBe(false); + + // Re-score: the first segment now runs to 120, onto the end the second + // segment already holds. Same stage, same end: one natural key. + fake.sessions = [ + session("night", [ + segment("LIGHT", 0, 120), + segment("LIGHT", 30, 120), + segment("DEEP", 120, 180), + ]), + ]; + const second = await sync(); + expect(second.failed).toBe(false); + + const live = (await sleepRows()).filter((r) => r.deletedAt === null); + const lightAt120 = live.filter( + (r) => r.sleepStage === "CORE" && r.measuredAt.toISOString() === at(120), + ); + expect(lightAt120).toHaveLength(1); + // The longer segment holds the slot. + expect(lightAt120[0]!.value).toBe(120); + expect(lightAt120[0]!.externalId).toContain(`:sleep:${at(0)}`); + // 120 minutes of light sleep (00:00-02:00) and 60 of deep: the overlap + // is counted once. + expect(await asleepMinutes()).toBe(180); + + // The same answer again is a no-op: the slot keeps its row. + const before = await sleepRows(); + const third = await sync(); + expect(third.failed).toBe(false); + expect(await sleepRows()).toEqual(before); + }); + + it("keeps both readings when a segment moves off a slot a new segment takes", async () => { + fake.sessions = [ + session("night", [segment("LIGHT", 0, 60), segment("DEEP", 90, 150)]), + ]; + expect((await sync()).failed).toBe(false); + + // Re-score: the light segment now runs to 90, and a new light segment + // ends at 60, the slot the first one is leaving. + fake.sessions = [ + session("night", [ + segment("LIGHT", 0, 90), + segment("LIGHT", 30, 60), + segment("DEEP", 90, 150), + ]), + ]; + expect((await sync()).failed).toBe(false); + + const light = (await sleepRows()) + .filter((r) => r.deletedAt === null && r.sleepStage === "CORE") + .map((r) => [r.measuredAt.toISOString(), r.value, r.externalId]); + expect(light).toEqual([ + [at(60), 30, expect.stringContaining(`:sleep:${at(30)}`)], + [at(90), 90, expect.stringContaining(`:sleep:${at(0)}`)], + ]); + }); + + it("keeps one row per segment when two sessions report the same night", async () => { + const stages = [ + segment("DEEP", 60, 120), + segment("REM", 120, 150), + segment("LIGHT", 150, 240), + ]; + fake.sessions = [session("phone", stages), session("watch", stages)]; + + const first = await sync(); + expect(first.failed).toBe(false); + const afterFirst = await sleepRows(); + expect(afterFirst.filter((r) => r.deletedAt === null)).toHaveLength(3); + expect(await asleepMinutes()).toBe(180); + + // Hour after hour the same two sessions come back. Nothing moves: + // neither session's window clears the other's rows, and the slot is not + // handed from one session's externalId to the other's. + for (let i = 0; i < 2; i++) { + const again = await sync(); + expect(again.failed).toBe(false); + expect(await sleepRows()).toEqual(afterFirst); + } + expect(await asleepMinutes()).toBe(180); + }); + + it("settles overlapping sessions that disagree, without failing and without inflating the night", async () => { + // The watch's session ends its light segment later than the phone's. + fake.sessions = [ + session("phone", [segment("LIGHT", 0, 90), segment("DEEP", 90, 150)]), + session("watch", [segment("LIGHT", 10, 90), segment("DEEP", 90, 160)]), + ]; + expect((await sync()).failed).toBe(false); + + // Re-score on the phone: its deep segment now ends where the watch's + // does, so both sessions put a DEEP segment on one slot. + fake.sessions = [ + session("phone", [segment("LIGHT", 0, 90), segment("DEEP", 90, 160)]), + session("watch", [segment("LIGHT", 10, 90), segment("DEEP", 90, 160)]), + ]; + expect((await sync()).failed).toBe(false); + expect((await sync()).failed).toBe(false); + + const live = (await sleepRows()).filter((r) => r.deletedAt === null); + const keys = live.map( + (r) => `${r.sleepStage}|${r.measuredAt.toISOString()}`, + ); + expect(new Set(keys).size).toBe(keys.length); + // Light 00:00-01:30 and deep 01:30-02:40: 160 minutes asleep. + expect(await asleepMinutes()).toBe(160); + }); +}); diff --git a/tests/integration/google-health-workout-terminal.test.ts b/tests/integration/google-health-workout-terminal.test.ts index 916cb17b1..f3cc5189e 100644 --- a/tests/integration/google-health-workout-terminal.test.ts +++ b/tests/integration/google-health-workout-terminal.test.ts @@ -27,6 +27,10 @@ vi.mock("@/lib/google-health/sync-core", () => ({ getValidToken: mocks.getValidToken, handleCollectionFetchError: mocks.handleCollectionFetchError, noteHardFailure: mocks.noteHardFailure, + googleHealthTokenSource: + (_userId: string, info: { accessToken: string }) => async () => + info.accessToken, + runGoogleHealthCollection: (_key: string, fn: () => Promise) => fn(), })); vi.mock("@/lib/arrivals/workout-emit", () => ({