diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index b38e95a1b..f9cc7ea90 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -121,6 +121,28 @@ export async function runWithGoogleHealthClientOutcome( return { result, outcome }; } +/** + * A stop condition for the walks below, set by a caller that runs under a + * time limit (the full-history backfill, which runs under its pg-boss job's + * budget and abort signal). A walk asks it before each request and, when it + * says stop, ends there and reports itself truncated, the same outcome as + * reaching the page ceiling: the resource is incomplete, the cycle is not + * stamped, and the reason shown is that only part of the history was + * processed. Outside such a scope the walks never stop early. + */ +const stopStorage = new AsyncLocalStorage<() => boolean>(); + +function stopRequested(): boolean { + return stopStorage.getStore()?.() === true; +} + +export async function runWithGoogleHealthStop( + shouldStop: () => boolean, + fn: () => Promise, +): Promise { + return stopStorage.run(shouldStop, fn); +} + export const GOOGLE_HEALTH_API_BASE = "https://health.googleapis.com/v4"; const GOOGLE_OAUTH_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"; const GOOGLE_OAUTH_TOKEN_URL = "https://oauth2.googleapis.com/token"; @@ -769,8 +791,13 @@ export async function forEachDataPointPage( let fetched = 0; let pageToken: string | null | undefined; let pageCount = 0; + let stopped = false; do { + if (stopRequested()) { + stopped = true; + break; + } const params = new URLSearchParams({ pageSize: String(pageSize) }); if (query.start) { const { field, bound } = incrementalFilter( @@ -830,7 +857,7 @@ export async function forEachDataPointPage( await onPage(page); } while (pageToken && pageCount < maxPages); - notePagination(pageCount, fetched, Boolean(pageToken)); + notePagination(pageCount, fetched, stopped || Boolean(pageToken)); }; // The fallback only exists for filtered daily-summary reads — every other @@ -1055,6 +1082,10 @@ export async function fetchDailyRollUp( let totalPages = 0; let truncated = false; for (const chunk of chunkCivilRange(startCivil, endCivil, maxDays)) { + if (stopRequested()) { + truncated = true; + break; + } let pageToken: string | null | undefined; let pageCount = 0; do { diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index fd33085d4..09a50ff29 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -27,6 +27,12 @@ vi.mock("@/lib/logging/context", () => ({ getEvent: () => null, })); +const { safeFetchMock } = vi.hoisted(() => ({ safeFetchMock: vi.fn() })); +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, safeFetch: safeFetchMock }; +}); + vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: isReauthRequiredMock, })); @@ -39,6 +45,11 @@ vi.mock("@/lib/google-health/sync", () => ({ })); import { runGoogleHealthBackfillForUser } from "../google-health-backfill"; +import { + GOOGLE_HEALTH_DATA_TYPES, + forEachDataPointPage, + runWithGoogleHealthClientOutcome, +} from "@/lib/google-health/client"; beforeEach(() => { vi.clearAllMocks(); @@ -83,4 +94,35 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => { expect(syncUserGoogleHealthMock).not.toHaveBeenCalled(); expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); }); + + it("stops the walk when its budget runs out and throws without stamping", async () => { + // An endless heart-rate stream: every page names a next one. + safeFetchMock.mockImplementation(async () => ({ + status: 200, + ok: true, + json: async () => ({ dataPoints: [{}], nextPageToken: "more" }), + })); + // The real walk under the sync mock, reporting what the real sync reports + // for a truncated resource. + syncUserGoogleHealthMock.mockImplementation(async () => { + const { outcome } = await runWithGoogleHealthClientOutcome(() => + forEachDataPointPage( + GOOGLE_HEALTH_DATA_TYPES.heartRate, + "token", + "fetchHeartRate", + { maxPages: 10_000 }, + () => {}, + ), + ); + return { imported: outcome.fetched, failed: outcome.truncated }; + }); + let asked = 0; + const shouldStop = () => ++asked > 3; + + await expect( + runGoogleHealthBackfillForUser("u1", shouldStop), + ).rejects.toThrow(/incomplete/); + expect(safeFetchMock).toHaveBeenCalledTimes(3); + expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled(); + }); }); diff --git a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts index ae009c783..71eededb5 100644 --- a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts +++ b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts @@ -6,6 +6,7 @@ import { INTEGRATION_BACKFILL_ADMISSION_GROUP, INTEGRATION_BACKFILL_ADMISSION_QUEUE, INTEGRATION_BACKFILL_BOOT_ORDER, + INTEGRATION_BACKFILL_EXPIRE_SECONDS, INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY, INTEGRATION_BACKFILL_STAGGER_SECONDS, bootStaggerSecondsFor, @@ -109,12 +110,20 @@ describe("shared full-history backfill admission", () => { expect.objectContaining({ singletonKey: "sleep-timeline-backfill|WHOOP|u1", group: INTEGRATION_BACKFILL_ADMISSION_GROUP, + expireInSeconds: INTEGRATION_BACKFILL_EXPIRE_SECONDS, }), ); expect(first).toBe("job-1"); expect(duplicate).toBeNull(); }); + it("gives the admission job an expiry sized for a long import, under pg-boss's 24-hour ceiling", () => { + // pg-boss's default is 900 s; a full-history heart-rate import runs for + // hours, and past the expiry pg-boss retries the job beside the run. + expect(INTEGRATION_BACKFILL_EXPIRE_SECONDS).toBe(16 * 60 * 60); + expect(INTEGRATION_BACKFILL_EXPIRE_SECONDS).toBeLessThan(24 * 60 * 60); + }); + it("keeps source enqueue singleton keys while applying stagger only to boot discovery", () => { expect(integrationBackfillSourceOptions("whoop-backfill|u1", 60)).toEqual( expect.objectContaining({ diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index 0dde4dcf2..e37268a2b 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -18,6 +18,7 @@ import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfil 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 { runWithGoogleHealthStop } from "@/lib/google-health/client"; export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill"; @@ -47,9 +48,17 @@ export interface GoogleHealthBackfillPayload { * just burn the retry budget on the same no-op: return WITHOUT stamping — the * boot discovery re-enqueues the account on the next boot, and the stamp lands * once the user reconnects and a clean walk completes. + * + * `shouldStop` is the admission job's budget (`jobBudget`). A per-minute + * 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. */ export async function runGoogleHealthBackfillForUser( userId: string, + shouldStop: () => boolean = () => false, ): Promise<{ imported: number }> { if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) { annotate({ @@ -61,9 +70,9 @@ export async function runGoogleHealthBackfillForUser( return { imported: 0 }; } - const { imported, failed } = await syncUserGoogleHealth(userId, { - fullSync: true, - }); + const { imported, failed } = await runWithGoogleHealthStop(shouldStop, () => + syncUserGoogleHealth(userId, { fullSync: true }), + ); if (failed) { // Surface through the pg-boss retry path (retryLimit 3, backoff). If the diff --git a/src/lib/jobs/integration-backfill-admission.ts b/src/lib/jobs/integration-backfill-admission.ts index e00f2d06c..2e528ca55 100644 --- a/src/lib/jobs/integration-backfill-admission.ts +++ b/src/lib/jobs/integration-backfill-admission.ts @@ -49,6 +49,27 @@ const RETRY_OPTIONS = { retryBackoff: true, } as const; +/** + * The admission job's expiry: sixteen hours, so a legitimately long import + * is never declared dead underneath itself. + * + * Without it every admission job took pg-boss's default of fifteen minutes. + * A full-history import of a per-minute heart-rate stream is thousands of + * pages (a page is 1 000 readings, about seventeen hours of a watch's + * history, so five years is some 2 600 pages), and at a few seconds a page on + * a slow link that is hours. Past the expiry pg-boss fails the job and retries + * it while the handler keeps running (`integration-backfill-lock.ts`), and the + * import that could not finish in fifteen minutes started over beside itself. + * + * The runs that honour the job's budget (`jobBudget`, three quarters of this) + * stop on their own at twelve hours, which is still some 8 000 pages at five + * seconds each, and are retried or re-offered at the next boot. The expiry is + * also what ends a run that hangs: the lane is freed for the next import, and + * the import lock keeps a retry from starting beside the stuck run. pg-boss + * refuses an expiry of 24 hours or more. + */ +export const INTEGRATION_BACKFILL_EXPIRE_SECONDS = 16 * 60 * 60; + export function bootStaggerSecondsFor(kind: IntegrationBackfillKind): number { return ( (INTEGRATION_BACKFILL_BOOT_ORDER.indexOf(kind) + 1) * @@ -83,9 +104,25 @@ export async function enqueueIntegrationBackfillAdmission( boss: BossSender, payload: IntegrationBackfillAdmissionPayload, ): Promise { - return boss.send(INTEGRATION_BACKFILL_ADMISSION_QUEUE, payload, { + return boss.send( + INTEGRATION_BACKFILL_ADMISSION_QUEUE, + payload, + integrationBackfillAdmissionSendOptions(payload), + ); +} + +/** + * 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 + * defaults as they were. + */ +export function integrationBackfillAdmissionSendOptions( + payload: IntegrationBackfillAdmissionPayload, +): SendOptions { + return { ...RETRY_OPTIONS, + expireInSeconds: INTEGRATION_BACKFILL_EXPIRE_SECONDS, singletonKey: integrationBackfillAdmissionSingletonKey(payload), group: INTEGRATION_BACKFILL_ADMISSION_GROUP, - }); + }; } diff --git a/src/lib/jobs/integration-backfill-drain.ts b/src/lib/jobs/integration-backfill-drain.ts new file mode 100644 index 000000000..e2cacf471 --- /dev/null +++ b/src/lib/jobs/integration-backfill-drain.ts @@ -0,0 +1,188 @@ +/** + * The handler of the shared full-history admission queue. + * + * Every provider's backfill and repair reaches this one queue, and the worker + * runs it one job at a time across the whole instance + * (`register-integration-sync.ts`). Each job runs under two guards: + * + * - The import lock (`withIntegrationBackfillLock`), keyed on the job's + * identity. A retry that pg-boss starts after the job's expiry while the + * first run is still going finds the lock taken and does nothing, so one + * account's import of one kind never runs twice at once. + * - The job's budget (`jobBudget`), handed to the runners that can stop + * part-way. A run that stops on its budget throws without stamping its + * completion marker, so it is retried, and after that the next boot's + * discovery offers it again. + */ +import type { Job } from "pg-boss"; + +import { runWhoopBackfillForUser } from "@/lib/jobs/whoop-backfill"; +import { runFitbitBackfillForUser } from "@/lib/jobs/fitbit-backfill"; +import { runGoogleHealthBackfillForUser } from "@/lib/jobs/google-health-backfill"; +import { runGoogleHealthSleepRepairForUser } from "@/lib/jobs/google-health-sleep-repair"; +import { runFitbitSleepRepairForUser } from "@/lib/jobs/fitbit-sleep-repair"; +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, + type IntegrationBackfillAdmissionPayload, +} from "@/lib/jobs/integration-backfill-admission"; +import { withIntegrationBackfillLock } from "@/lib/jobs/integration-backfill-lock"; +import { jobBudget } from "@/lib/jobs/job-budget"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { workerLog } from "@/lib/jobs/reminder/shared"; +import { withBackgroundEvent } from "@/lib/logging/background"; + +interface RunCounts { + imported: number; + removed: number; + deleted: number; + markers: number; + linked: number; +} + +async function runAdmitted( + payload: IntegrationBackfillAdmissionPayload, + shouldStop: () => boolean, + totals: RunCounts, +): Promise { + const { kind, data } = payload; + const { userId } = data; + switch (kind) { + case "whoop-backfill": { + const { imported } = await runWhoopBackfillForUser(userId); + totals.imported += imported; + workerLog("info", `[whoop-backfill] user=${userId} imported=${imported}`); + return; + } + case "fitbit-backfill": { + const { imported } = await runFitbitBackfillForUser(userId); + totals.imported += imported; + workerLog( + "info", + `[fitbit-backfill] user=${userId} imported=${imported}`, + ); + return; + } + case "google-health-backfill": { + const { imported } = await runGoogleHealthBackfillForUser( + userId, + shouldStop, + ); + totals.imported += imported; + workerLog( + "info", + `[google-health-backfill] user=${userId} imported=${imported}`, + ); + return; + } + case "google-health-sleep-repair": { + const { imported } = await runGoogleHealthSleepRepairForUser(userId); + totals.imported += imported; + workerLog( + "info", + `[google-health-sleep-repair] user=${userId} imported=${imported}`, + ); + return; + } + case "fitbit-sleep-repair": { + const { imported, removed } = await runFitbitSleepRepairForUser(userId); + totals.imported += imported; + totals.removed += removed; + workerLog( + "info", + `[fitbit-sleep-repair] user=${userId} imported=${imported} removed=${removed}`, + ); + return; + } + case "sleep-timeline-backfill": { + if (!data.provider) { + throw new Error( + "sleep-timeline backfill admission requires a provider", + ); + } + const { deleted, imported } = await runSleepTimelineBackfillForUser( + userId, + data.provider, + ); + totals.deleted += deleted; + totals.imported += imported; + workerLog( + "info", + `[sleep-timeline-backfill] user=${userId} provider=${data.provider} deleted=${deleted} imported=${imported}`, + ); + return; + } + case "lab-biomarker-backfill": { + const { markers, linked } = await runLabBiomarkerBackfillForUser(userId); + totals.markers += markers; + totals.linked += linked; + workerLog( + "info", + `[lab-biomarker-backfill] user=${userId} markers=${markers} linked=${linked}`, + ); + return; + } + case "strava-backfill": { + const { imported } = await runStravaBackfillForUser(userId); + totals.imported += imported; + workerLog( + "info", + `[strava-backfill] user=${userId} imported=${imported}`, + ); + return; + } + } +} + +/** + * Drain a batch of admission jobs. There is no per-job error isolation: a + * runner that throws fails the job and the admission retry policy applies. So + * reaching the end means every admitted job either did its work or found its + * import already running, and the counts each runner returns are what the + * pass reports. + */ +export async function drainIntegrationBackfillAdmission( + jobs: Job[], +): Promise { + const shouldStop = jobBudget(jobs); + const totals: RunCounts = { + imported: 0, + removed: 0, + deleted: 0, + markers: 0, + linked: 0, + }; + let alreadyRunning = 0; + for (const job of jobs) { + const identity = integrationBackfillAdmissionSingletonKey(job.data); + const run = await withIntegrationBackfillLock(identity, () => + runAdmitted(job.data, shouldStop, totals), + ); + if (!run.ran) { + alreadyRunning += 1; + await withBackgroundEvent( + "integration_backfill.already_running", + async (event) => { + event.addMeta("identity", identity); + event.addMeta("job_id", job.id); + event.addMeta( + "reason", + "an earlier delivery of this import still holds its lock; this delivery does nothing", + ); + event.elevateLevel("warn"); + }, + ); + } + } + return jobDone({ + jobs: jobs.length, + already_running: alreadyRunning, + imported: totals.imported, + removed: totals.removed, + deleted: totals.deleted, + markers: totals.markers, + linked: totals.linked, + }); +} diff --git a/src/lib/jobs/integration-backfill-lock.ts b/src/lib/jobs/integration-backfill-lock.ts new file mode 100644 index 000000000..1e6a90905 --- /dev/null +++ b/src/lib/jobs/integration-backfill-lock.ts @@ -0,0 +1,70 @@ +/** + * One full-history import per account and kind at a time, across processes. + * + * The admission queue runs one import at a time (`groupConcurrency: 1`), but + * pg-boss only holds that line while the job is inside its expiry. Once the + * expiry passes, pg-boss fails the job, fires its abort signal, frees the + * worker slot and the group slot, and retries the job after its retry delay. + * It does not stop the handler. Measured against pg-boss 12.34 + * (`tests/integration/integration-backfill-lane-expiry.test.ts`): a handler + * that ignores the signal was still running when the retry of the same job + * started beside it in the same worker, so the lane's "one at a time" did not + * hold for the one case it exists for, an import too long for its expiry. + * + * So each run holds a session-level advisory lock keyed on the job's identity + * (kind, provider where there is one, account) for as long as it runs. A + * delivery that finds the lock taken does no work: the run that holds it is + * still importing, and it either stamps its completion marker or leaves the + * account for the next boot's discovery. The lock lives on a connection of its + * own, outside the Prisma pool, because a session lock has to stay on one + * connection for the whole run and a pooled one is handed back between + * statements. Ending the connection releases the lock, and so does the + * process dying, so a crash never leaves an account locked. + * + * One extra connection per running import; the lane admits one import per + * process at a time, so that is one connection. + */ +import { Client } from "pg"; + +/** The lock key for one admission job's identity. */ +function integrationBackfillLockKey(identity: string): string { + return `integration-backfill:${identity}`; +} + +/** What a guarded run came to. */ +type GuardedRun = { ran: true; result: T } | { ran: false }; + +/** + * Run `run` while holding the import lock for `identity`. Resolves + * `{ ran: false }` at once, without calling `run`, when another run holds it. + * Rejects exactly as `run` does otherwise. + */ +export async function withIntegrationBackfillLock( + identity: string, + run: () => Promise, + connectionString: string | undefined = process.env.DATABASE_URL, +): Promise> { + const client = new Client({ connectionString, keepAlive: true }); + // An idle client whose server goes away emits `error`; unhandled, that + // takes the whole worker down. The run itself notices the database is + // gone through its own queries. + client.on("error", () => {}); + await client.connect(); + try { + const key = integrationBackfillLockKey(identity); + 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 }; + try { + return { ran: true, result: await run() }; + } finally { + await client + .query("SELECT pg_advisory_unlock(hashtextextended($1, 0))", [key]) + .catch(() => {}); + } + } finally { + await client.end().catch(() => {}); + } +} diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 9a1c5e231..bd55c5c2d 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -43,6 +43,9 @@ export type JobFacts = Readonly>; */ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "access_tokens_deleted", + // An admission delivery that found its import still running from an + // earlier delivery of the same job and did nothing. + "already_running", "appointment_addresses_cleared", "assessments_warmed", "auto_resolved", diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index 54e753948..f241b4b62 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -17,56 +17,48 @@ import { PgBoss } from "pg-boss"; import { WHOOP_BACKFILL_QUEUE, WHOOP_BACKFILL_CONCURRENCY, - runWhoopBackfillForUser, enqueueBootTimeWhoopBackfill, type WhoopBackfillPayload, } from "@/lib/jobs/whoop-backfill"; import { FITBIT_BACKFILL_QUEUE, FITBIT_BACKFILL_CONCURRENCY, - runFitbitBackfillForUser, enqueueBootTimeFitbitBackfill, type FitbitBackfillPayload, } from "@/lib/jobs/fitbit-backfill"; import { GOOGLE_HEALTH_BACKFILL_QUEUE, GOOGLE_HEALTH_BACKFILL_CONCURRENCY, - runGoogleHealthBackfillForUser, enqueueBootTimeGoogleHealthBackfill, type GoogleHealthBackfillPayload, } from "@/lib/jobs/google-health-backfill"; import { GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE, GOOGLE_HEALTH_SLEEP_REPAIR_CONCURRENCY, - runGoogleHealthSleepRepairForUser, enqueueBootTimeGoogleHealthSleepRepair, type GoogleHealthSleepRepairPayload, } from "@/lib/jobs/google-health-sleep-repair"; import { FITBIT_SLEEP_REPAIR_QUEUE, FITBIT_SLEEP_REPAIR_CONCURRENCY, - runFitbitSleepRepairForUser, enqueueBootTimeFitbitSleepRepair, type FitbitSleepRepairPayload, } from "@/lib/jobs/fitbit-sleep-repair"; import { STRAVA_BACKFILL_QUEUE, STRAVA_BACKFILL_CONCURRENCY, - runStravaBackfillForUser, enqueueBootTimeStravaBackfill, type StravaBackfillPayload, } from "@/lib/jobs/strava-backfill"; import { SLEEP_TIMELINE_BACKFILL_QUEUE, SLEEP_TIMELINE_BACKFILL_CONCURRENCY, - runSleepTimelineBackfillForUser, enqueueBootTimeSleepTimelineBackfill, type SleepTimelineBackfillPayload, } from "@/lib/jobs/sleep-timeline-backfill"; import { LAB_BIOMARKER_BACKFILL_QUEUE, LAB_BIOMARKER_BACKFILL_CONCURRENCY, - runLabBiomarkerBackfillForUser, enqueueBootTimeLabBiomarkerBackfill, type LabBiomarkerBackfillPayload, } from "@/lib/jobs/lab-biomarker-backfill"; @@ -77,6 +69,7 @@ import { enqueueIntegrationBackfillAdmission, type IntegrationBackfillAdmissionPayload, } from "@/lib/jobs/integration-backfill-admission"; +import { drainIntegrationBackfillAdmission } from "@/lib/jobs/integration-backfill-drain"; import { workerLog } from "./shared"; import { jobDone } from "@/lib/jobs/job-outcome"; import { @@ -482,117 +475,7 @@ export async function registerIntegrationSyncQueues( localConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY, groupConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY, }, - async (jobs) => { - // The drain has no per-job error isolation: a runner that throws fails - // the job and the admission retry policy applies. So reaching the end - // means every admitted job did its work, and the counts each runner - // already returns are what the pass reports. - let importedTotal = 0; - let removedTotal = 0; - let deletedTotal = 0; - let markersTotal = 0; - let linkedTotal = 0; - for (const job of jobs) { - const { kind, data } = job.data; - const { userId } = data; - switch (kind) { - case "whoop-backfill": { - const { imported } = await runWhoopBackfillForUser(userId); - importedTotal += imported; - workerLog( - "info", - `[whoop-backfill] user=${userId} imported=${imported}`, - ); - break; - } - case "fitbit-backfill": { - const { imported } = await runFitbitBackfillForUser(userId); - importedTotal += imported; - workerLog( - "info", - `[fitbit-backfill] user=${userId} imported=${imported}`, - ); - break; - } - case "google-health-backfill": { - const { imported } = await runGoogleHealthBackfillForUser(userId); - importedTotal += imported; - workerLog( - "info", - `[google-health-backfill] user=${userId} imported=${imported}`, - ); - break; - } - case "google-health-sleep-repair": { - const { imported } = - await runGoogleHealthSleepRepairForUser(userId); - importedTotal += imported; - workerLog( - "info", - `[google-health-sleep-repair] user=${userId} imported=${imported}`, - ); - break; - } - case "fitbit-sleep-repair": { - const { imported, removed } = - await runFitbitSleepRepairForUser(userId); - importedTotal += imported; - removedTotal += removed; - workerLog( - "info", - `[fitbit-sleep-repair] user=${userId} imported=${imported} removed=${removed}`, - ); - break; - } - case "sleep-timeline-backfill": { - if (!data.provider) { - throw new Error( - "sleep-timeline backfill admission requires a provider", - ); - } - const { deleted, imported } = await runSleepTimelineBackfillForUser( - userId, - data.provider, - ); - deletedTotal += deleted; - importedTotal += imported; - workerLog( - "info", - `[sleep-timeline-backfill] user=${userId} provider=${data.provider} deleted=${deleted} imported=${imported}`, - ); - break; - } - case "lab-biomarker-backfill": { - const { markers, linked } = - await runLabBiomarkerBackfillForUser(userId); - markersTotal += markers; - linkedTotal += linked; - workerLog( - "info", - `[lab-biomarker-backfill] user=${userId} markers=${markers} linked=${linked}`, - ); - break; - } - case "strava-backfill": { - const { imported } = await runStravaBackfillForUser(userId); - importedTotal += imported; - workerLog( - "info", - `[strava-backfill] user=${userId} imported=${imported}`, - ); - break; - } - } - } - return jobDone({ - jobs: jobs.length, - imported: importedTotal, - removed: removedTotal, - deleted: deletedTotal, - markers: markersTotal, - linked: linkedTotal, - }); - }, + drainIntegrationBackfillAdmission, ); await createAndWork( diff --git a/tests/integration/integration-backfill-lane-expiry.test.ts b/tests/integration/integration-backfill-lane-expiry.test.ts new file mode 100644 index 000000000..0de70c998 --- /dev/null +++ b/tests/integration/integration-backfill-lane-expiry.test.ts @@ -0,0 +1,169 @@ +/** + * The full-history admission lane against a real pg-boss on a real Postgres: + * what happens to an import that runs past its job's expiry. + * + * pg-boss 12.34, measured with the lane's worker shape (localConcurrency 1, + * groupConcurrency 1, exclusive policy, retries with backoff), a two-second + * expiry, a one-second retry delay and a handler that ignores the abort + * signal and runs for eight seconds: + * + * +5ms start #1 (job 1709c8a6…) running=1 + * +2008ms abort signal #1 + * +4051ms start #2 (same job id) running=2 + * +6054ms abort signal #2 + * +8007ms end #1 running=1 + * +9593ms start #3 (same job id) running=2 + * +12053ms end #2 + * +17595ms end #3 + * +19166ms start #4 (same job id) running=1 + * + * At the expiry pg-boss fails the job, fires the signal, frees the worker and + * group slots, and retries the job after its delay; the handler keeps going. + * The retry then starts in the same worker beside the run it replaced, and + * the lane's one-at-a-time holds for everything except the import too long + * for its expiry. The first test keeps that as an executable statement about + * pg-boss; the second runs the lane's own handler under the same conditions. + */ +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { PgBoss, type Job } from "pg-boss"; + +const runner = vi.hoisted(() => ({ + running: 0, + maxRunning: 0, + calls: 0, + holdMs: 8_000, +})); + +vi.mock("@/lib/jobs/google-health-backfill", () => ({ + // Ignores the budget and the signal on purpose: the case the lock covers + // is a run that does not stop when pg-boss gives up on it. + runGoogleHealthBackfillForUser: async () => { + runner.calls += 1; + runner.running += 1; + runner.maxRunning = Math.max(runner.maxRunning, runner.running); + await new Promise((resolve) => setTimeout(resolve, runner.holdMs)); + runner.running -= 1; + return { imported: 1 }; + }, +})); + +import { + integrationBackfillAdmissionSendOptions, + type IntegrationBackfillAdmissionPayload, +} from "@/lib/jobs/integration-backfill-admission"; +import { drainIntegrationBackfillAdmission } from "@/lib/jobs/integration-backfill-drain"; +import { withIntegrationBackfillLock } from "@/lib/jobs/integration-backfill-lock"; +import { runJob } from "@/lib/jobs/run-job"; + +let boss: PgBoss; +const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`; + +/** Short enough to cross in a test, the lane's shape otherwise. */ +const EXPIRY_OVERRIDE = { expireInSeconds: 2, retryDelay: 1 } as const; + +const WORK_OPTIONS = { + localConcurrency: 1, + groupConcurrency: 1, + pollingIntervalSeconds: 0.5, +} as const; + +const payload: IntegrationBackfillAdmissionPayload = { + kind: "google-health-backfill", + data: { userId: `lane-user-${suffix}`, enqueuedAt: new Date().toISOString() }, +}; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +beforeAll(async () => { + boss = new PgBoss({ connectionString: process.env.DATABASE_URL }); + await boss.start(); +}, 120_000); + +afterAll(async () => { + await boss?.stop({ graceful: false }); +}); + +describe("an import that outlives its expiry", () => { + it("pg-boss starts the retry of the same job beside the run it expired", async () => { + const queue = `lane-expiry-raw-${suffix}`; + await boss.createQueue(queue, { policy: "exclusive" }); + const ids: string[] = []; + let running = 0; + let maxRunning = 0; + await boss.work(queue, WORK_OPTIONS, async ([job]: Job[]) => { + ids.push(job!.id); + running += 1; + maxRunning = Math.max(maxRunning, running); + await sleep(8_000); + running -= 1; + }); + await boss.send(queue, payload, { + ...integrationBackfillAdmissionSendOptions(payload), + ...EXPIRY_OVERRIDE, + }); + + await sleep(12_000); + await boss.offWork(queue); + + expect(ids.length).toBeGreaterThanOrEqual(2); + expect(new Set(ids).size).toBe(1); + expect(maxRunning).toBeGreaterThanOrEqual(2); + }, 60_000); + + it("the lane's handler runs the import once while its retries find it running", async () => { + const queue = `lane-expiry-drain-${suffix}`; + await boss.createQueue(queue, { policy: "exclusive" }); + const handler = runJob(queue, drainIntegrationBackfillAdmission); + const deliveries: string[] = []; + await boss.work( + queue, + WORK_OPTIONS, + async (jobs: Job[]) => { + deliveries.push(jobs[0]!.id); + return handler(jobs); + }, + ); + await boss.send(queue, payload, { + ...integrationBackfillAdmissionSendOptions(payload), + ...EXPIRY_OVERRIDE, + }); + + // The run holds for eight seconds; its retries arrive while it does. + await sleep(7_000); + await boss.offWork(queue); + await sleep(2_000); + + expect(deliveries.length).toBeGreaterThanOrEqual(2); + expect(runner.calls).toBe(1); + expect(runner.maxRunning).toBe(1); + }, 60_000); +}); + +describe("withIntegrationBackfillLock", () => { + it("admits one run per identity, others alongside, and the next once it ends", async () => { + let release!: () => void; + const held = new Promise((resolve) => (release = resolve)); + const first = withIntegrationBackfillLock("kind|a", () => held); + await sleep(300); + + const same = await withIntegrationBackfillLock("kind|a", async () => 1); + const other = await withIntegrationBackfillLock("kind|b", async () => 2); + expect(same).toEqual({ ran: false }); + expect(other).toEqual({ ran: true, result: 2 }); + + release(); + await first; + const after = await withIntegrationBackfillLock("kind|a", async () => 3); + expect(after).toEqual({ ran: true, result: 3 }); + }); + + it("releases the lock when the run throws", async () => { + await expect( + withIntegrationBackfillLock("kind|c", async () => { + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + const after = await withIntegrationBackfillLock("kind|c", async () => 4); + expect(after).toEqual({ ran: true, result: 4 }); + }); +});