From 1e07ecaaaa548882178c30c8744076e96c000a06 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 15 Oct 2026 16:16:07 +0200 Subject: [PATCH] fix(google-health): keep the hourly sync inside its job's expiry An account whose backfill slice held it made the hourly job wait up to eight minutes, and the sync after the wait ran as long as it needed: past the fifteen-minute expiry pg-boss failed the job and retried it beside the run still going. The pass now spends the job's budget: the lock wait is capped at what is left, the sync's walks stop when the budget is spent and leave the rest of the window to the next hour (the cycle is not stamped), and an account reached after that is skipped. A stop on the budget is not counted as a failure to retry. --- docs/integrations/google-health.md | 5 +- .../__tests__/sync-time-budget.test.ts | 85 ++++++++++++ src/lib/google-health/sync.ts | 19 ++- .../google-health-sync-budget.test.ts | 121 ++++++++++++++++++ src/lib/jobs/reminder/google-health-sync.ts | 37 +++++- 5 files changed, 263 insertions(+), 4 deletions(-) create mode 100644 src/lib/google-health/__tests__/sync-time-budget.test.ts create mode 100644 src/lib/jobs/reminder/__tests__/google-health-sync-budget.test.ts diff --git a/docs/integrations/google-health.md b/docs/integrations/google-health.md index 7aa0dd1e5..600a52fc7 100644 --- a/docs/integrations/google-health.md +++ b/docs/integrations/google-health.md @@ -157,7 +157,10 @@ still a failure. The hourly sync waits for the slice that is running (up to eight minutes) instead of skipping the account, so recent data keeps arriving while an old -history is still coming in. +history is still coming in. The hourly job stays inside its own fifteen-minute +limit: the wait is cut short when the job's time runs low, the sync after it +gets only what is left and leaves the rest of its window to the next hour, and +an account the job reaches after its time is up waits for the next hour. 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 diff --git a/src/lib/google-health/__tests__/sync-time-budget.test.ts b/src/lib/google-health/__tests__/sync-time-budget.test.ts new file mode 100644 index 000000000..8b2bd3b4b --- /dev/null +++ b/src/lib/google-health/__tests__/sync-time-budget.test.ts @@ -0,0 +1,85 @@ +/** + * A caller's time limit (`shouldStop`, the hourly job's budget) reaches every + * walk of the sync cycle, so a walk stops when it is spent; and a cycle that + * stopped on it is not stamped, so the next run reads the rest from the same + * watermark. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { prismaMock, recordSyncSuccess, resources } = vi.hoisted(() => { + return { + prismaMock: { + integrationStatus: { findUnique: vi.fn(async () => null) }, + googleHealthConnection: { + findUnique: vi.fn(async () => ({ lastSyncedAt: null })), + update: vi.fn(async () => ({})), + }, + }, + recordSyncSuccess: vi.fn(async () => {}), + resources: { + workout: vi.fn(async () => 0), + sleep: vi.fn(async () => 0), + activity: vi.fn(async () => 0), + metrics: vi.fn(async () => 0), + }, + }; +}); + +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.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: vi.fn(() => []), + recomputeUserRollups: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); +vi.mock("../sync-workout", () => ({ syncUserWorkout: resources.workout })); +vi.mock("../sync-sleep", () => ({ syncUserSleep: resources.sleep })); +vi.mock("../sync-activity", () => ({ syncUserActivity: resources.activity })); +vi.mock("../sync-metrics", () => ({ syncUserMetrics: resources.metrics })); + +import { googleHealthSliceExhausted } from "../client"; +import { syncUserGoogleHealth } from "../sync"; + +beforeEach(() => { + vi.clearAllMocks(); + prismaMock.googleHealthConnection.findUnique.mockResolvedValue({ + lastSyncedAt: null, + }); +}); + +describe("the caller's time limit inside the sync cycle", () => { + it("is what every resource's walks are asked, and an untouched one never stops", async () => { + let spent = false; + const asked: boolean[] = []; + resources.metrics.mockImplementation(async () => { + asked.push(googleHealthSliceExhausted()); + spent = true; + asked.push(googleHealthSliceExhausted()); + return 0; + }); + await syncUserGoogleHealth("budget-user", { shouldStop: () => spent }); + expect(asked).toEqual([false, true]); + + asked.length = 0; + spent = false; + await syncUserGoogleHealth("budget-user"); + expect(asked).toEqual([false, false]); + }); +}); diff --git a/src/lib/google-health/sync.ts b/src/lib/google-health/sync.ts index b0a72601f..85ba41422 100644 --- a/src/lib/google-health/sync.ts +++ b/src/lib/google-health/sync.ts @@ -11,6 +11,7 @@ import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { resolveUserTimezone } from "@/lib/tz/resolver"; import { runWithGoogleHealthClientOutcome, + runWithGoogleHealthStop, type GoogleHealthClientOutcome, } from "./client"; import { syncUserActivity } from "./sync-activity"; @@ -100,6 +101,14 @@ export interface GoogleHealthSyncOptions { * account's sync lock. Zero (the default) answers `busy` at once. */ waitForLockMs?: number; + /** + * The caller's time limit, asked before every request once the run holds + * the account (the hourly job's expiry). A walk that is told to stop ends + * there and reports itself truncated: the cycle is not stamped, so the next + * run reads the rest from the same watermark. Not for the backfill, which + * sets its own limits around the whole run. + */ + shouldStop?: () => boolean; } const RESOURCE_STATUSES = new Set([ @@ -292,7 +301,7 @@ async function runGoogleHealthSyncCycle( { name: "dense-heart-rate", fn: syncUserMetrics }, ]; - const cycle = await runWithGoogleHealthSyncCycle(async () => { + const runResources = async () => { let total = 0; let anyFailed = false; const resourceOutcomes: GoogleHealthResourceOutcome[] = []; @@ -334,7 +343,13 @@ async function runGoogleHealthSyncCycle( await persistProgress("in_progress", total, anyFailed, resourceOutcomes); } return { total, anyFailed, resources: resourceOutcomes }; - }); + }; + const shouldStop = opts.shouldStop; + const cycle = await runWithGoogleHealthSyncCycle(() => + shouldStop + ? runWithGoogleHealthStop(shouldStop, runResources) + : runResources(), + ); const total = cycle.result.total; let anyFailed = cycle.result.anyFailed || cycle.hardFailures.length > 0; diff --git a/src/lib/jobs/reminder/__tests__/google-health-sync-budget.test.ts b/src/lib/jobs/reminder/__tests__/google-health-sync-budget.test.ts new file mode 100644 index 000000000..24ffdcb44 --- /dev/null +++ b/src/lib/jobs/reminder/__tests__/google-health-sync-budget.test.ts @@ -0,0 +1,121 @@ +/** + * The hourly Google Health job runs under pg-boss's fifteen-minute expiry. + * An account whose backfill slice holds it can make the job wait eight of + * those minutes, and the sync after the wait used to run as long as it + * needed: past the expiry pg-boss marked the job failed and retried it beside + * the run still going. The pass now spends only the job's budget: the lock + * wait is capped at what is left, the sync is handed the budget and stops + * cleanly when it runs out, and an account reached after that is left for the + * next hour. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { syncUserGoogleHealth, connections } = vi.hoisted(() => ({ + syncUserGoogleHealth: vi.fn(), + connections: [] as Array<{ userId: string }>, +})); + +vi.mock("@/lib/google-health/sync", () => ({ + GOOGLE_HEALTH_POLL_CONCURRENCY: 1, + isGoogleHealthParked: vi.fn(async () => false), + syncUserGoogleHealth, +})); +vi.mock("../shared", () => ({ + getWorkerPrisma: () => ({ + googleHealthConnection: { findMany: vi.fn(async () => connections) }, + }), +})); +vi.mock("@/lib/jobs/reminder-satisfy", () => ({ + enqueueReminderSatisfy: vi.fn(async () => {}), +})); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); + +import type { Job } from "pg-boss"; +import { handleGoogleHealthSync } from "../google-health-sync"; +import { JOB_BUDGET_SHARE } from "@/lib/jobs/job-budget"; +import { GOOGLE_HEALTH_POLL_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock"; + +const START = new Date("2026-10-15T10:00:00.000Z").getTime(); +const EXPIRY_S = 15 * 60; +const BUDGET_MS = EXPIRY_S * 1000 * JOB_BUDGET_SHARE; + +function cronJob(): Job<{ userId?: string }> { + return { + id: "tick", + name: "google-health-sync", + data: {}, + expireInSeconds: EXPIRY_S, + } as unknown as Job<{ userId?: string }>; +} + +beforeEach(() => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(START); + syncUserGoogleHealth.mockReset(); + connections.length = 0; +}); +afterEach(() => { + vi.useRealTimers(); +}); + +describe("the hourly Google Health job and its expiry", () => { + it("hands the sync what is left of the budget after the lock wait, and stops it there", async () => { + connections.push({ userId: "waited" }, { userId: "late" }); + const seen: Array<{ beforeEnd: boolean; atEnd: boolean }> = []; + syncUserGoogleHealth.mockImplementation( + async (_userId: string, opts: { shouldStop: () => boolean }) => { + // Eight minutes waiting for the backfill slice to let go. + vi.setSystemTime(START + GOOGLE_HEALTH_POLL_LOCK_WAIT_MS); + const beforeEnd = opts.shouldStop(); + // The walk runs on until the budget is spent, and is told to stop. + vi.setSystemTime(START + BUDGET_MS); + seen.push({ beforeEnd, atEnd: opts.shouldStop() }); + return { + state: "truncated", + imported: 120, + failed: true, + resources: [], + }; + }, + ); + + const outcome = await handleGoogleHealthSync([cronJob()]); + + expect(syncUserGoogleHealth).toHaveBeenCalledTimes(1); + expect(syncUserGoogleHealth.mock.calls[0]![1]).toMatchObject({ + waitForLockMs: GOOGLE_HEALTH_POLL_LOCK_WAIT_MS, + }); + expect(seen).toEqual([{ beforeEnd: false, atEnd: true }]); + // A stop on the budget is the next hour's work, not a failure to retry; + // the account after it waits for the next hour. + expect(outcome.did).toMatchObject({ + users_partial: 1, + users_skipped: 1, + users_failed: 0, + users_retryable: 0, + }); + }); + + it("never waits for the lock past the budget", async () => { + connections.push({ userId: "a" }, { userId: "b" }); + syncUserGoogleHealth + .mockImplementationOnce(async () => { + // The first account takes most of the budget. + vi.setSystemTime(START + BUDGET_MS - 60_000); + return { state: "complete", imported: 1, failed: false, resources: [] }; + }) + .mockImplementationOnce(async () => ({ + state: "complete", + imported: 1, + failed: false, + resources: [], + })); + + await handleGoogleHealthSync([cronJob()]); + + expect(syncUserGoogleHealth).toHaveBeenCalledTimes(2); + expect(syncUserGoogleHealth.mock.calls[1]![1]).toMatchObject({ + waitForLockMs: 60_000, + }); + }); +}); diff --git a/src/lib/jobs/reminder/google-health-sync.ts b/src/lib/jobs/reminder/google-health-sync.ts index 7ac2e7d0f..0e49f4f64 100644 --- a/src/lib/jobs/reminder/google-health-sync.ts +++ b/src/lib/jobs/reminder/google-health-sync.ts @@ -16,6 +16,7 @@ import { syncUserGoogleHealth, } from "@/lib/google-health/sync"; import { GOOGLE_HEALTH_POLL_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock"; +import { jobBudget, jobDeadline } from "@/lib/jobs/job-budget"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup"; import { @@ -58,6 +59,18 @@ export async function handleGoogleHealthSync( }); } + // The job's own expiry bounds the pass (`jobBudget`). An account that + // waited for its backfill slice gets what is left of it, not a fresh + // hour: the lock wait is capped at the time remaining, the sync stops + // its walks once the budget is spent and leaves the rest of its window + // to the next hour (the cycle is not stamped), and an account reached + // only after the budget is gone is left for the next hour untouched. + // Before, a tick could run on past its expiry, and pg-boss then marked + // it failed and retried it beside the run still going. + const shouldStop = jobBudget(jobs); + const deadline = Math.min( + ...jobs.map((job) => jobDeadline(job) ?? Number.POSITIVE_INFINITY), + ); // Fan the cohort out with bounded concurrency + per-user error isolation: // one slow Google response can't stall the whole cohort, and a single // user's failure is warned without aborting the pass. @@ -65,6 +78,7 @@ export async function handleGoogleHealthSync( const verdicts = await Promise.all( targets.map(({ userId }) => limit(async (): Promise => { + if (shouldStop()) return { status: "skipped", imported: 0 }; if (await isGoogleHealthParked(userId)) { return { status: "parked", imported: 0 }; } @@ -74,12 +88,33 @@ export async function handleGoogleHealthSync( // the tick waits for the slice to end rather than skipping the // account for as long as the whole import runs. const result = await syncUserGoogleHealth(userId, { - waitForLockMs: GOOGLE_HEALTH_POLL_LOCK_WAIT_MS, + waitForLockMs: Math.max( + 0, + Math.min( + GOOGLE_HEALTH_POLL_LOCK_WAIT_MS, + deadline - Date.now(), + ), + ), + shouldStop, }); // Another run of this account (its backfill, a manual sync) // still held the account after the wait; this tick leaves it // to that run. if (result.busy) return { status: "skipped", imported: 0 }; + // Stopped on the job's budget: the rest of the window is the + // next hour's, which is not a failure to retry. + if (result.failed && shouldStop()) { + if (result.imported > 0) { + fireAndForget(enqueueReminderSatisfy(userId), { + action: "reminder.satisfy.enqueue", + }); + } + return { + status: result.imported > 0 ? "partial" : "skipped", + imported: result.imported, + retryable: false, + }; + } const parked = result.failed && result.imported === 0 &&