diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index 2773abef9..8afde8dd0 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -7,24 +7,37 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { prismaMock, syncUserGoogleHealthMock, isParkedMock } = vi.hoisted( - () => ({ - prismaMock: { - googleHealthConnection: { - findMany: vi.fn(), - findUnique: vi.fn(async () => null), - update: vi.fn(), - }, +const { + prismaMock, + syncUserGoogleHealthMock, + isParkedMock, + bossSend, + resumeState, +} = vi.hoisted(() => ({ + prismaMock: { + googleHealthConnection: { + findMany: vi.fn(), + findUnique: vi.fn( + async (): Promise<{ + backfillProgress: unknown; + backfillCompletedAt: Date | null; + } | null> => null, + ), + update: vi.fn(), }, - syncUserGoogleHealthMock: vi.fn(), - isParkedMock: vi.fn(async () => false), - }), -); + }, + syncUserGoogleHealthMock: vi.fn(), + isParkedMock: vi.fn(async () => false), + bossSend: vi.fn(), + // Collections the mocked walk "finishes" in this attempt, added to the + // resume set the way `runGoogleHealthCollection` adds them. + resumeState: { finishes: [] as string[], doneBefore: [] as string[] }, +})); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); vi.mock("@/lib/jobs/boss-instance", () => ({ - getGlobalBoss: () => ({ send: vi.fn() }), + getGlobalBoss: () => ({ send: bossSend }), })); vi.mock("@/lib/logging/context", () => ({ @@ -40,12 +53,16 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { vi.mock("@/lib/google-health/sync-core", () => ({ GOOGLE_HEALTH_INTEGRATION_KEY: "google-health", - readGoogleHealthBackfillDone: () => new Set(), + readGoogleHealthBackfillDone: () => new Set(resumeState.doneBefore), runWithGoogleHealthBackfillResume: async ( _userId: string, - _done: Set, + done: Set, fn: () => Promise, - ) => ({ result: await fn(), skipped: [] }), + ) => { + const result = await fn(); + for (const key of resumeState.finishes) done.add(key); + return { result, skipped: [] }; + }, })); vi.mock("@/lib/google-health/sync", () => ({ isGoogleHealthParked: isParkedMock, @@ -63,6 +80,10 @@ beforeEach(() => { vi.clearAllMocks(); isParkedMock.mockResolvedValue(false); prismaMock.googleHealthConnection.update.mockResolvedValue({}); + prismaMock.googleHealthConnection.findUnique.mockResolvedValue(null); + bossSend.mockResolvedValue("job-next"); + resumeState.finishes = []; + resumeState.doneBefore = []; }); describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { @@ -150,3 +171,92 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); }); }); + +/** + * A long history walks in slices. A slice that finished more collections than + * it started with is progressing, not failing: it queues the next slice and + * ends quietly, without the error line or the failure count. Only a slice that + * finished nothing new is stuck and keeps throwing. + */ +describe("runGoogleHealthBackfillForUser — slices", () => { + it("continues a progressing slice instead of failing it", async () => { + resumeState.doneBefore = ["steps"]; + resumeState.finishes = ["sleep", "weight"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 9, failed: true }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 2), + ).resolves.toEqual({ + imported: 9, + continued: true, + }); + + expect(bossSend).toHaveBeenCalledTimes(1); + const [queue, payload, options] = bossSend.mock.calls[0]!; + expect(queue).toBe("integration-backfill-admission"); + expect(payload).toMatchObject({ + kind: "google-health-backfill", + data: { userId: "u1", slice: 3 }, + }); + // Its own key, so the exclusive policy does not swallow it while this + // slice is still active. + expect(options).toMatchObject({ + singletonKey: "google-health-backfill|u1|slice-3", + startAfter: expect.any(Number), + }); + // Not complete: no stamp. + expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); + }); + + it("still throws for a slice that finished nothing new", async () => { + resumeState.doneBefore = ["steps"]; + resumeState.finishes = []; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 3, failed: true }); + + await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow( + /incomplete/, + ); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("throws rather than drop the work when the next slice cannot be queued", async () => { + resumeState.finishes = ["sleep"]; + bossSend.mockResolvedValue(null); + syncUserGoogleHealthMock.mockResolvedValue({ imported: 1, failed: true }); + + await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow( + /incomplete/, + ); + }); + + it("stamps the marker on the slice that finishes the history", async () => { + resumeState.doneBefore = ["steps", "sleep"]; + resumeState.finishes = ["weight"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 4, failed: false }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 5), + ).resolves.toEqual({ + imported: 4, + }); + expect(bossSend).not.toHaveBeenCalled(); + expect( + prismaMock.googleHealthConnection.update.mock.calls[0]![0].data + .backfillCompletedAt, + ).toBeInstanceOf(Date); + }); + + it("walks nothing again once an earlier slice completed the import", async () => { + prismaMock.googleHealthConnection.findUnique.mockResolvedValue({ + backfillProgress: null, + backfillCompletedAt: new Date(), + }); + + await expect( + runGoogleHealthBackfillForUser("u1", undefined, 4), + ).resolves.toEqual({ + imported: 0, + }); + expect(syncUserGoogleHealthMock).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index 03d428e5e..bb3e66b42 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -15,7 +15,11 @@ 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 { + enqueueIntegrationBackfillAdmission, + INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + integrationBackfillSourceOptions, +} from "@/lib/jobs/integration-backfill-admission"; import { readGoogleHealthBackfillDone, runWithGoogleHealthBackfillResume, @@ -68,6 +72,14 @@ export interface GoogleHealthBackfillPayload { * (`runWithGoogleHealthBackfillResume`). The note is cleared once the stamp * lands, and on every reconnect. * + * Runs in slices: an attempt that ends before the whole history is in but + * finished at least one more collection than it started with made progress. + * It is not a failure: it sends the next slice (`slice + 1`) and returns + * `continued`, without the error line or the queue's failure count. Only an + * attempt that finished nothing new throws, because that one is stuck: the + * pg-boss retry and, after it, the next boot take it from there. Progress is + * bounded by the number of collections, so the chain of slices is too. + * * 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 @@ -76,7 +88,8 @@ export interface GoogleHealthBackfillPayload { export async function runGoogleHealthBackfillForUser( userId: string, shouldStop: () => boolean = () => false, -): Promise<{ imported: number }> { + slice: number = 0, +): Promise<{ imported: number; continued?: boolean }> { if (await isGoogleHealthParked(userId)) { annotate({ action: { @@ -89,9 +102,21 @@ export async function runGoogleHealthBackfillForUser( const connection = await prisma.googleHealthConnection.findUnique({ where: { userId }, - select: { backfillProgress: true }, + select: { backfillProgress: true, backfillCompletedAt: true }, }); + if (connection?.backfillCompletedAt) { + // A slice queued behind the one that finished the import, or a boot + // delivery that raced it: the history is in, nothing to walk again. + annotate({ + action: { + name: "google_health.backfill.already_complete", + details: { slice }, + }, + }); + return { imported: 0 }; + } const done = readGoogleHealthBackfillDone(connection?.backfillProgress); + const doneBefore = done.size; const { result: { imported, failed, busy }, @@ -115,9 +140,46 @@ export async function runGoogleHealthBackfillForUser( } if (failed) { - // Surface through the pg-boss retry path (retryLimit 3, backoff). If the - // failure parked the connection at error_reauth meanwhile, the next - // attempt takes the reauth return above instead of failing again. + // `done` is the resume set this attempt added to as it finished + // collections; a larger set than it started with is progress. + const finished = done.size - doneBefore; + if (finished > 0) { + const boss = getGlobalBoss(); + const sent = boss + ? await enqueueIntegrationBackfillAdmission( + boss, + { + kind: "google-health-backfill", + data: { + userId, + enqueuedAt: new Date().toISOString(), + slice: slice + 1, + }, + }, + INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + ) + : null; + if (sent) { + annotate({ + action: { + name: "google_health.backfill.continued", + details: { + imported, + slice, + collectionsFinished: finished, + collectionsDone: done.size, + }, + }, + }); + return { imported, continued: true }; + } + // No follow-up could be queued; fall through to the retry path so the + // work is not dropped. + } + // No collection finished in this attempt: the backfill is stuck, not + // slow. Surface through the pg-boss retry path (retryLimit 3, backoff). + // If the failure parked the connection at error_reauth meanwhile, the + // next attempt takes the reauth return above instead of failing again. throw new Error( `google-health backfill incomplete for user ${userId} — marker not stamped`, ); diff --git a/src/lib/jobs/integration-backfill-admission.ts b/src/lib/jobs/integration-backfill-admission.ts index 2e528ca55..382a65374 100644 --- a/src/lib/jobs/integration-backfill-admission.ts +++ b/src/lib/jobs/integration-backfill-admission.ts @@ -28,6 +28,14 @@ export type IntegrationBackfillData = { userId: string; enqueuedAt: string; provider?: "WHOOP" | "WITHINGS"; + /** + * The continuation number of an import that runs in slices: absent (0) for + * the job discovery or a connect sends, 1, 2, ... for each follow-up a + * slice that made progress sent for the rest. Part of the job's singleton + * key, so the follow-up is not swallowed by the `exclusive` policy while the + * slice that sends it is still active; never part of the import identity. + */ + slice?: number; }; export interface IntegrationBackfillAdmissionPayload { @@ -88,7 +96,12 @@ export function integrationBackfillSourceOptions( }; } -export function integrationBackfillAdmissionSingletonKey( +/** + * The import a job belongs to: one kind for one account (and provider). The + * import lock keys on it, so no two slices or deliveries of one import ever + * run at once. + */ +export function integrationBackfillIdentity( payload: IntegrationBackfillAdmissionPayload, ): string { if (payload.kind === "sleep-timeline-backfill") { @@ -100,17 +113,29 @@ export function integrationBackfillAdmissionSingletonKey( return `${payload.kind}|${payload.data.userId}`; } +/** The job's singleton key: the import identity, plus its slice number. */ +export function integrationBackfillAdmissionSingletonKey( + payload: IntegrationBackfillAdmissionPayload, +): string { + const identity = integrationBackfillIdentity(payload); + const slice = payload.data.slice ?? 0; + return slice > 0 ? `${identity}|slice-${slice}` : identity; +} + export async function enqueueIntegrationBackfillAdmission( boss: BossSender, payload: IntegrationBackfillAdmissionPayload, + startAfterSeconds: number = 0, ): Promise { - return boss.send( - INTEGRATION_BACKFILL_ADMISSION_QUEUE, - payload, - integrationBackfillAdmissionSendOptions(payload), - ); + return boss.send(INTEGRATION_BACKFILL_ADMISSION_QUEUE, payload, { + ...integrationBackfillAdmissionSendOptions(payload), + ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), + }); } +/** Breathing room between one slice of an import and the next. */ +export const INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS = 60; + /** * The send options of one admission job. The options ride on the send, not on * `createQueue`, whose `ON CONFLICT DO NOTHING` leaves an existing queue's diff --git a/src/lib/jobs/integration-backfill-drain.ts b/src/lib/jobs/integration-backfill-drain.ts index 76f91c4d6..68d7f7174 100644 --- a/src/lib/jobs/integration-backfill-drain.ts +++ b/src/lib/jobs/integration-backfill-drain.ts @@ -25,7 +25,7 @@ import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill"; import { runSleepTimelineBackfillForUser } from "@/lib/jobs/sleep-timeline-backfill"; import { runLabBiomarkerBackfillForUser } from "@/lib/jobs/lab-biomarker-backfill"; import { - integrationBackfillAdmissionSingletonKey, + integrationBackfillIdentity, type IntegrationBackfillAdmissionPayload, } from "@/lib/jobs/integration-backfill-admission"; import { withIntegrationBackfillLock } from "@/lib/jobs/integration-backfill-lock"; @@ -36,6 +36,8 @@ import { withBackgroundEvent } from "@/lib/logging/background"; interface RunCounts { imported: number; + /** Imports that ended a slice with progress and sent the next one. */ + continued: number; removed: number; deleted: number; markers: number; @@ -66,14 +68,16 @@ async function runAdmitted( return; } case "google-health-backfill": { - const { imported } = await runGoogleHealthBackfillForUser( + const { imported, continued } = await runGoogleHealthBackfillForUser( userId, shouldStop, + data.slice ?? 0, ); totals.imported += imported; + if (continued) totals.continued += 1; workerLog( "info", - `[google-health-backfill] user=${userId} imported=${imported}`, + `[google-health-backfill] user=${userId} imported=${imported}${continued ? " continued" : ""}`, ); return; } @@ -152,6 +156,7 @@ export async function drainIntegrationBackfillAdmission( const shouldStop = jobBudget(jobs); const totals: RunCounts = { imported: 0, + continued: 0, removed: 0, deleted: 0, markers: 0, @@ -159,7 +164,7 @@ export async function drainIntegrationBackfillAdmission( }; let alreadyRunning = 0; for (const job of jobs) { - const identity = integrationBackfillAdmissionSingletonKey(job.data); + const identity = integrationBackfillIdentity(job.data); const run = await withIntegrationBackfillLock(identity, () => runAdmitted(job.data, shouldStop, totals), ); @@ -183,6 +188,7 @@ export async function drainIntegrationBackfillAdmission( jobs: jobs.length, already_running: alreadyRunning, imported: totals.imported, + continued: totals.continued, removed: totals.removed, deleted: totals.deleted, markers: totals.markers, diff --git a/tests/integration/google-health-backfill-rate-limit.test.ts b/tests/integration/google-health-backfill-rate-limit.test.ts index 6c2d3e04a..2cf7bd9a1 100644 --- a/tests/integration/google-health-backfill-rate-limit.test.ts +++ b/tests/integration/google-health-backfill-rate-limit.test.ts @@ -168,6 +168,20 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { }; }); +/** The follow-up slices a progressing backfill queues, recorded. */ +const sent = vi.hoisted(() => ({ + jobs: [] as Array<{ queue: string; data: unknown; options: unknown }>, +})); +vi.mock("@/lib/jobs/boss-instance", async (importOriginal) => ({ + ...(await importOriginal()), + getGlobalBoss: () => ({ + send: async (queue: string, data: unknown, options: unknown) => { + sent.jobs.push({ queue, data, options }); + return `job-${sent.jobs.length}`; + }, + }), +})); + vi.mock("@/lib/db-compat", () => ({ ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), })); @@ -212,6 +226,7 @@ beforeEach(async () => { fake.requestsByPath = new Map(); fake.status429 = 0; fake.status401 = 0; + sent.jobs = []; await seed(); }, 120_000); @@ -353,9 +368,19 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () fake.requestsByPath.get("heart-rate/dataPoints") ?? 0; return heartRatePagesSeen >= 3; }; + // It finished four collections, so it is progressing, not failing: it + // ends quietly and queues the next slice of the import. await expect( runGoogleHealthBackfillForUser(USER, stopAfterThreeHeartRatePages), - ).rejects.toThrow(/incomplete/); + ).resolves.toMatchObject({ continued: true }); + expect(sent.jobs).toHaveLength(1); + expect(sent.jobs[0]).toMatchObject({ + queue: "integration-backfill-admission", + data: { + kind: "google-health-backfill", + data: { userId: USER, slice: 1 }, + }, + }); const afterFirst = await getPrismaClient().googleHealthConnection.findUnique({ @@ -373,9 +398,10 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () ); expect(done).not.toContain("fetchHeartRate"); - // Second attempt: the finished collections are not read again. + // Second attempt, the queued slice: the finished collections are not + // read again. fake.requestsByPath = new Map(); - await runGoogleHealthBackfillForUser(USER); + await runGoogleHealthBackfillForUser(USER, undefined, 1); expect(fake.requestsByPath.get("exercise/dataPoints")).toBeUndefined(); expect(fake.requestsByPath.get("sleep/dataPoints")).toBeUndefined(); @@ -392,6 +418,29 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", () expect(afterSecond?.backfillProgress).toBeNull(); }); + it("keeps failing a backfill whose slice finished nothing new", async () => { + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + + // A first slice finishes everything but the heart-rate stream. + const stopAtThirdPage = () => + (fake.requestsByPath.get("heart-rate/dataPoints") ?? 0) >= 3; + await runGoogleHealthBackfillForUser(USER, stopAtThirdPage); + expect(sent.jobs).toHaveLength(1); + + // The next slice cannot get through the heart-rate stream either: no + // collection finishes, so this one is stuck and says so. + fake.requestsByPath = new Map(); + await expect( + runGoogleHealthBackfillForUser(USER, stopAtThirdPage, 1), + ).rejects.toThrow(/incomplete/); + expect(sent.jobs).toHaveLength(1); + const row = await getPrismaClient().googleHealthConnection.findUnique({ + where: { userId: USER }, + }); + expect(row?.backfillCompletedAt).toBeNull(); + }); + it("never runs the hourly poll or a manual sync beside a running backfill", async () => { fake.holdAtPage = 2; fake.held = new Promise((resolve) => {