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 &&