From 029d6d881c5c055dfdd7806eeb753bd2defe19cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 30 Sep 2026 23:01:38 +0200 Subject: [PATCH 1/2] fix(jobs): keep a full-history import from running beside its own retry The shared admission queue sent its jobs without an expiry, so every one took pg-boss's fifteen minutes. Against pg-boss 12.34, a handler that runs past its expiry is not stopped: the job is failed, its abort signal fires, the worker and group slots are freed, and the retry of the same job starts in the same worker beside the run it replaced. A per-minute heart-rate history is thousands of pages, so on a slow link one import turned into up to four copies of itself, on the one queue meant to run a single import at a time. - The admission job now carries a sixteen-hour expiry, sized for the longest legitimate import and under pg-boss's 24-hour ceiling. - Each run holds a session advisory lock on its identity (kind, provider, account) on a connection of its own. A delivery that finds it taken does nothing and says so in the log; the running import stamps its marker or is offered again at the next boot. Ending the connection or the process releases the lock. - The Google Health backfill runs under the job's budget and abort signal. The page walk asks before each request, stops there, and reports the run truncated, so it throws without the completion stamp and is retried. - The drain moves out of the registrar into its own module so it can be driven against a real pg-boss. The integration test pins the pg-boss behaviour and runs the lane's handler with a two-second expiry: without the lock the import ran twice. Refs #1023 --- src/lib/google-health/client.ts | 33 ++- .../__tests__/google-health-backfill.test.ts | 42 ++++ .../integration-backfill-admission.test.ts | 9 + src/lib/jobs/google-health-backfill.ts | 15 +- .../jobs/integration-backfill-admission.ts | 41 +++- src/lib/jobs/integration-backfill-drain.ts | 188 ++++++++++++++++++ src/lib/jobs/integration-backfill-lock.ts | 70 +++++++ src/lib/jobs/job-outcome.ts | 3 + .../reminder/register-integration-sync.ts | 121 +---------- .../integration-backfill-lane-expiry.test.ts | 169 ++++++++++++++++ 10 files changed, 566 insertions(+), 125 deletions(-) create mode 100644 src/lib/jobs/integration-backfill-drain.ts create mode 100644 src/lib/jobs/integration-backfill-lock.ts create mode 100644 tests/integration/integration-backfill-lane-expiry.test.ts 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 }); + }); +}); From a0c290e140af26a76ceef7643cbd5ca24bb23714 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 30 Sep 2026 23:06:53 +0200 Subject: [PATCH 2/2] fix(jobs): let the nightly tombstone purge step around a running restore A restore clears every row of the account, tombstones included, inside one long transaction under the account's exclusive restore lock. The nightly purge deleted expired tombstones across accounts, one statement per batch, without that lock: on an account under restore it waited on the restore's row locks for as long as the restore ran, past its statement timeout, or deadlocked with it once each side held rows the other needed. Each batch is now deleted one account at a time, in a transaction that first takes the restore lock in shared mode, the same order the consolidation passes keep. An account under restore is refused at once, kept out of the rest of the run, and purged the next night. Still one delete per account per batch, never one per row. The job reports how many accounts it deferred. The integration test holds a stand-in restore on one account while the purge runs: the purge returns without waiting, leaves that account alone and purges the other. With the lock call removed it waits out the hold. Refs #1031 --- .../measurement-tombstone-cleanup.test.ts | 94 ++++++++- src/lib/jobs/job-outcome.ts | 3 + src/lib/jobs/measurement-tombstone-cleanup.ts | 179 ++++++++++++++---- .../cleanup-handlers-visibility.test.ts | 18 +- src/lib/jobs/reminder/cleanup-handlers.ts | 8 + tests/integration/sync-changes.test.ts | 2 +- .../tombstone-cleanup-restore-lock.test.ts | 152 +++++++++++++++ 7 files changed, 403 insertions(+), 53 deletions(-) create mode 100644 tests/integration/tombstone-cleanup-restore-lock.test.ts diff --git a/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts b/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts index fa68a9100..84de68b99 100644 --- a/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts +++ b/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts @@ -21,19 +21,43 @@ import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; const DAY_MS = 86_400_000; +/** + * A client whose interactive transaction hands the callback a client with the + * same delegates, and whose restore-lock probe answers `held`. + */ +function mockClient( + measurement: { findMany: unknown; deleteMany: unknown }, + held: (userId: string) => boolean = () => true, +) { + const lockCalls: string[] = []; + const tx = { + measurement, + $queryRaw: vi.fn(async (_strings: TemplateStringsArray, key: string) => { + const userId = key.replace(/^backup-restore:/, ""); + lockCalls.push(userId); + return [{ held: held(userId) }]; + }), + }; + const prisma = { + measurement, + $transaction: vi.fn(async (fn: (client: typeof tx) => unknown) => fn(tx)), + } as unknown as Parameters[0]; + return { prisma, lockCalls }; +} + describe("cleanupExpiredMeasurementTombstones", () => { it("prunes only tombstones older than the retention horizon", async () => { const now = new Date("2026-06-01T00:00:00.000Z"); - const findMany = vi - .fn() - .mockResolvedValueOnce([{ id: "a" }, { id: "b" }, { id: "c" }]); + const findMany = vi.fn().mockResolvedValueOnce([ + { id: "a", userId: "u1" }, + { id: "b", userId: "u1" }, + { id: "c", userId: "u1" }, + ]); const deleteMany = vi.fn().mockResolvedValue({ count: 3 }); - const prisma = { - measurement: { findMany, deleteMany }, - } as unknown as Parameters[0]; + const { prisma } = mockClient({ findMany, deleteMany }); const outcome = await cleanupExpiredMeasurementTombstones(prisma, now); - expect(outcome).toEqual({ deleted: 3, drained: true }); + expect(outcome).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 }); expect(findMany).toHaveBeenCalledTimes(1); const where = findMany.mock.calls[0][0].where as { @@ -61,12 +85,11 @@ describe("cleanupExpiredMeasurementTombstones", () => { // to stop at the cap rather than loop forever. const fullBatch = Array.from({ length: PURGE_BATCH_SIZE }, (_, i) => ({ id: `row-${i}`, + userId: "u1", })); const findMany = vi.fn().mockResolvedValue(fullBatch); const deleteMany = vi.fn().mockResolvedValue({ count: PURGE_BATCH_SIZE }); - const prisma = { - measurement: { findMany, deleteMany }, - } as unknown as Parameters[0]; + const { prisma } = mockClient({ findMany, deleteMany }); const outcome = await cleanupExpiredMeasurementTombstones(prisma, now); @@ -81,6 +104,57 @@ describe("cleanupExpiredMeasurementTombstones", () => { }); }); +describe("cleanupExpiredMeasurementTombstones — accounts under restore", () => { + it("deletes one statement per account per batch, each under the account's restore lock", async () => { + const findMany = vi.fn().mockResolvedValueOnce([ + { id: "a1", userId: "a" }, + { id: "b1", userId: "b" }, + { id: "a2", userId: "a" }, + ]); + const deleteMany = vi.fn(async ({ where }) => ({ + count: (where.id.in as string[]).length, + })); + const { prisma, lockCalls } = mockClient({ findMany, deleteMany }); + + const outcome = await cleanupExpiredMeasurementTombstones(prisma); + + expect(outcome).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 }); + expect(lockCalls).toEqual(["a", "b"]); + expect(deleteMany.mock.calls.map((call) => call[0].where)).toEqual([ + { id: { in: ["a1", "a2"] } }, + { id: { in: ["b1"] } }, + ]); + }); + + it("leaves an account under restore and keeps it out of the rest of the run", async () => { + const full = (userId: string) => + Array.from({ length: PURGE_BATCH_SIZE }, (_, i) => ({ + id: `${userId}-${i}`, + userId, + })); + const findMany = vi + .fn() + .mockResolvedValueOnce(full("restoring")) + .mockResolvedValueOnce([{ id: "other-1", userId: "other" }]); + const deleteMany = vi.fn(async ({ where }) => ({ + count: (where.id.in as string[]).length, + })); + const { prisma } = mockClient( + { findMany, deleteMany }, + (userId) => userId !== "restoring", + ); + + const outcome = await cleanupExpiredMeasurementTombstones(prisma); + + expect(outcome).toEqual({ deleted: 1, drained: true, deferredAccounts: 1 }); + expect(findMany.mock.calls[0][0].where).not.toHaveProperty("userId"); + expect(findMany.mock.calls[1][0].where.userId).toEqual({ + notIn: ["restoring"], + }); + expect(deleteMany).toHaveBeenCalledTimes(1); + }); +}); + describe("reminder-worker — measurement-tombstone-cleanup schedule", () => { // v1.18.1 — the cleanup wiring moved out of the 2143-LOC reminder-worker // boot file into the maintenance registrar. The dead-queue guard follows it. diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index bd55c5c2d..fe6be9808 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -71,6 +71,9 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "days_skipped", "days_skipped_no_tombstones", "days_stored", + // Tombstone purge: accounts left for the next run because a restore of + // the account was running. + "deferred_accounts", "deleted", "dense_days_consolidated", "dense_rows_soft_deleted", diff --git a/src/lib/jobs/measurement-tombstone-cleanup.ts b/src/lib/jobs/measurement-tombstone-cleanup.ts index 43c2f5db5..306b2bb49 100644 --- a/src/lib/jobs/measurement-tombstone-cleanup.ts +++ b/src/lib/jobs/measurement-tombstone-cleanup.ts @@ -18,9 +18,17 @@ * (`NATIVE_REFRESH_TOKEN_DAYS` + margin) so it moves automatically if the * refresh-token lifetime changes — the two never drift. */ -import type { PrismaClient } from "@/generated/prisma/client"; +import type { Prisma, PrismaClient } from "@/generated/prisma/client"; import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; -import { purgeInBatches, type PurgeOutcome } from "@/lib/jobs/purge-batch"; +import { + AccountRestoreInProgressError, + holdAccountAgainstRestore, +} from "@/lib/export/restore-lock"; +import { + PURGE_BATCH_SIZE, + PURGE_MAX_BATCHES, + type PurgeOutcome, +} from "@/lib/jobs/purge-batch"; const DAY_MS = 86_400_000; @@ -28,6 +36,103 @@ function tombstoneCutoff(now: Date): Date { return new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS); } +/** A purge run, plus the accounts it left for the next run. */ +export interface TombstonePurgeOutcome extends PurgeOutcome { + /** + * Accounts whose tombstones were left in place because a restore of the + * account was running. The next night purges them. + */ + deferredAccounts: number; +} + +type TombstoneRow = { id: string; userId: string }; + +interface TombstonePurgeOptions { + prisma: PrismaClient; + /** Up to `take` expired tombstones, none of them owned by `skipUserIds`. */ + findRows: (take: number, skipUserIds: string[]) => Promise; + /** Remove exactly those ids inside `tx`; returns the rows removed. */ + deleteIds: (tx: Prisma.TransactionClient, ids: string[]) => Promise; + batchSize?: number; + maxBatches?: number; +} + +/** + * Walk the expired tombstones in bounded batches, deleting each batch one + * account at a time under the account's restore lock. + * + * A restore clears every row of the account, tombstones included, in one + * long transaction. The purge used to delete a batch spanning accounts in one + * statement, so on an account under restore it waited on the restore's row + * locks, for as long as the restore ran and well past `statement_timeout`, + * or, holding some of the account's rows while the restore held others, + * deadlocked with it. So each account's share of a batch is deleted in a + * transaction that first takes the restore lock in shared mode, the way the + * consolidation passes do (`restore-lock.ts`). An account under restore is + * refused at once, dropped from the rest of the run, and purged the next + * night, when its tombstones are whatever the restore left. + * + * Still one delete statement per account per batch, never one per row. + */ +async function purgeTombstonesByAccount({ + prisma, + findRows, + deleteIds, + batchSize = PURGE_BATCH_SIZE, + maxBatches = PURGE_MAX_BATCHES, +}: TombstonePurgeOptions): Promise { + let deleted = 0; + const deferred = new Set(); + const outcome = (drained: boolean): TombstonePurgeOutcome => ({ + deleted, + drained, + deferredAccounts: deferred.size, + }); + + for (let batch = 0; batch < maxBatches; batch++) { + const rows = await findRows(batchSize, [...deferred]); + if (rows.length === 0) return outcome(true); + + const byAccount = new Map(); + for (const row of rows) { + const ids = byAccount.get(row.userId); + if (ids) ids.push(row.id); + else byAccount.set(row.userId, [row.id]); + } + for (const [userId, ids] of byAccount) { + try { + deleted += await prisma.$transaction( + async (tx) => { + await holdAccountAgainstRestore(tx, userId); + return deleteIds(tx, ids); + }, + // A full batch of one account is one statement over thousands of + // rows; the interactive default of five seconds is too tight for + // it on a slow disk. The connection's statement_timeout bounds it. + { timeout: 60_000 }, + ); + } catch (err) { + if (!(err instanceof AccountRestoreInProgressError)) throw err; + deferred.add(userId); + } + } + + // A short batch means the predicate is exhausted, the deferred accounts + // aside; asking again would cost a round-trip to learn nothing. + if (rows.length < batchSize) return outcome(true); + } + + return outcome(false); +} + +/** The retention predicate, minus the accounts a run has deferred. */ +function expiredTombstones(cutoff: Date, skipUserIds: string[]) { + return { + deletedAt: { not: null, lt: cutoff }, + ...(skipUserIds.length > 0 ? { userId: { notIn: skipUserIds } } : {}), + }; +} + /** * Hard-delete soft-deleted measurement rows whose `deletedAt` is older than * the retention horizon. @@ -44,20 +149,18 @@ function tombstoneCutoff(now: Date): Date { export async function cleanupExpiredMeasurementTombstones( prisma: PrismaClient, now: Date = new Date(), -): Promise { +): Promise { const cutoff = tombstoneCutoff(now); - return purgeInBatches({ - findIds: async (take) => - ( - await prisma.measurement.findMany({ - where: { deletedAt: { not: null, lt: cutoff } }, - select: { id: true }, - take, - }) - ).map((row) => row.id), - deleteIds: async (ids) => - (await prisma.measurement.deleteMany({ where: { id: { in: ids } } })) - .count, + return purgeTombstonesByAccount({ + prisma, + findRows: (take, skipUserIds) => + prisma.measurement.findMany({ + where: expiredTombstones(cutoff, skipUserIds), + select: { id: true, userId: true }, + take, + }), + deleteIds: async (tx, ids) => + (await tx.measurement.deleteMany({ where: { id: { in: ids } } })).count, }); } @@ -65,19 +168,18 @@ export async function cleanupExpiredMeasurementTombstones( export async function cleanupExpiredMoodTombstones( prisma: PrismaClient, now: Date = new Date(), -): Promise { +): Promise { const cutoff = tombstoneCutoff(now); - return purgeInBatches({ - findIds: async (take) => - ( - await prisma.moodEntry.findMany({ - where: { deletedAt: { not: null, lt: cutoff } }, - select: { id: true }, - take, - }) - ).map((row) => row.id), - deleteIds: async (ids) => - (await prisma.moodEntry.deleteMany({ where: { id: { in: ids } } })).count, + return purgeTombstonesByAccount({ + prisma, + findRows: (take, skipUserIds) => + prisma.moodEntry.findMany({ + where: expiredTombstones(cutoff, skipUserIds), + select: { id: true, userId: true }, + take, + }), + deleteIds: async (tx, ids) => + (await tx.moodEntry.deleteMany({ where: { id: { in: ids } } })).count, }); } @@ -88,20 +190,19 @@ export async function cleanupExpiredMoodTombstones( export async function cleanupExpiredIntakeTombstones( prisma: PrismaClient, now: Date = new Date(), -): Promise { +): Promise { const cutoff = tombstoneCutoff(now); - return purgeInBatches({ - findIds: async (take) => + return purgeTombstonesByAccount({ + prisma, + findRows: (take, skipUserIds) => + prisma.medicationIntakeEvent.findMany({ + where: expiredTombstones(cutoff, skipUserIds), + select: { id: true, userId: true }, + take, + }), + deleteIds: async (tx, ids) => ( - await prisma.medicationIntakeEvent.findMany({ - where: { deletedAt: { not: null, lt: cutoff } }, - select: { id: true }, - take, - }) - ).map((row) => row.id), - deleteIds: async (ids) => - ( - await prisma.medicationIntakeEvent.deleteMany({ + await tx.medicationIntakeEvent.deleteMany({ where: { id: { in: ids } }, }) ).count, diff --git a/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts b/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts index 354169b08..20a3f479b 100644 --- a/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts +++ b/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts @@ -62,15 +62,16 @@ import { } from "../cleanup-handlers"; const DRAINED = { deleted: 1, drained: true }; +const TOMBSTONES_DRAINED = { ...DRAINED, deferredAccounts: 0 }; beforeEach(() => { vi.clearAllMocks(); for (const key of Object.keys(h.meta)) delete h.meta[key]; h.warnings.length = 0; h.cleanupOldAuditLogs.mockResolvedValue(DRAINED); - h.cleanupExpiredMeasurementTombstones.mockResolvedValue(DRAINED); - h.cleanupExpiredMoodTombstones.mockResolvedValue(DRAINED); - h.cleanupExpiredIntakeTombstones.mockResolvedValue(DRAINED); + h.cleanupExpiredMeasurementTombstones.mockResolvedValue(TOMBSTONES_DRAINED); + h.cleanupExpiredMoodTombstones.mockResolvedValue(TOMBSTONES_DRAINED); + h.cleanupExpiredIntakeTombstones.mockResolvedValue(TOMBSTONES_DRAINED); }); describe("audit-log cleanup handler", () => { @@ -115,6 +116,7 @@ describe("tombstone cleanup handler", () => { h.cleanupExpiredMoodTombstones.mockResolvedValue({ deleted: 200_000, drained: false, + deferredAccounts: 0, }); await handleMeasurementTombstoneCleanup([]); expect(h.meta.tombstone_cleanup_drained).toBe(false); @@ -122,6 +124,16 @@ describe("tombstone cleanup handler", () => { expect(h.warnings.join(" ")).toContain("backlog remains"); }); + it("reports the accounts it left because a restore was running", async () => { + h.cleanupExpiredMeasurementTombstones.mockResolvedValue({ + ...TOMBSTONES_DRAINED, + deferredAccounts: 2, + }); + const outcome = await handleMeasurementTombstoneCleanup([]); + expect(h.meta.tombstone_cleanup_deferred_accounts).toBe(2); + expect(outcome).toMatchObject({ did: { deferred_accounts: 2 } }); + }); + it("stays quiet when all three legs drained", async () => { await handleMeasurementTombstoneCleanup([]); expect(h.meta.tombstone_cleanup_drained).toBe(true); diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts index 98ad16be9..5795aae78 100644 --- a/src/lib/jobs/reminder/cleanup-handlers.ts +++ b/src/lib/jobs/reminder/cleanup-handlers.ts @@ -254,6 +254,13 @@ export async function handleMeasurementTombstoneCleanup( // outside, which is how a purge that never finished stayed invisible. const drained = measurements.drained && mood.drained && intakes.drained; evt.addMeta("tombstone_cleanup_drained", drained); + // Accounts under restore are left alone and purged the next night. + const deferredAccounts = Math.max( + measurements.deferredAccounts, + mood.deferredAccounts, + intakes.deferredAccounts, + ); + evt.addMeta("tombstone_cleanup_deferred_accounts", deferredAccounts); if (!drained) { evt.addWarning( "tombstone-cleanup stopped at the batch cap; a backlog remains for the next run", @@ -264,6 +271,7 @@ export async function handleMeasurementTombstoneCleanup( mood_pruned: mood.deleted, intakes_pruned: intakes.deleted, drained, + deferred_accounts: deferredAccounts, }); } catch (err) { // Rethrow. A retention purge that fails silently every night looks diff --git a/tests/integration/sync-changes.test.ts b/tests/integration/sync-changes.test.ts index 421aa7451..606933304 100644 --- a/tests/integration/sync-changes.test.ts +++ b/tests/integration/sync-changes.test.ts @@ -399,7 +399,7 @@ describe("GET /api/sync/changes (real Postgres)", () => { // pruned; the recent reachable tombstone survives. `drained` says the // batched walk reached the end of the backlog rather than stopping at // its per-run cap. - expect(pruned).toEqual({ deleted: 1, drained: true }); + expect(pruned).toEqual({ deleted: 1, drained: true, deferredAccounts: 0 }); const survivors = await prisma.measurement.findMany({ where: { userId: TEST_USER_ID }, select: { externalId: true }, diff --git a/tests/integration/tombstone-cleanup-restore-lock.test.ts b/tests/integration/tombstone-cleanup-restore-lock.test.ts new file mode 100644 index 000000000..bf49eee7e --- /dev/null +++ b/tests/integration/tombstone-cleanup-restore-lock.test.ts @@ -0,0 +1,152 @@ +/** + * #1031 — the nightly tombstone purge against a running restore. + * + * A restore clears every row of the account, tombstones included, inside one + * long transaction that holds the account's restore lock exclusively. The + * purge used to delete expired tombstones across accounts in one statement + * per batch, without the lock: on an account under restore it waited on the + * restore's row locks for as long as the restore ran, or deadlocked with it. + * + * Here a transaction stands in for the restore: it takes the lock, deletes the + * account's rows, and holds for three seconds before rolling back. The purge + * runs while it holds. What must hold: the purge does not wait, leaves the + * account under restore alone and says so, and still purges every other + * account. + * + * Mutation check: drop the `holdAccountAgainstRestore` call from + * `purgeTombstonesByAccount` and the purge waits out the hold, then deletes + * the restoring account's tombstones once the stand-in rolls back. + */ +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const RESTORING = "tombstone-restoring"; +const OTHER = "tombstone-other"; +const HOLD_MS = 3_000; +const LONG_AGO = new Date("2020-01-01T00:00:00.000Z"); + +async function seed() { + const prisma = getPrismaClient(); + for (const id of [RESTORING, OTHER]) { + await prisma.user.create({ + data: { id, username: id, role: "USER", timezone: "UTC" }, + }); + for (let i = 0; i < 3; i++) { + await prisma.measurement.create({ + data: { + id: `${id}-tomb-${i}`, + userId: id, + type: "WEIGHT", + value: 70 + i, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(`2020-01-0${i + 1}T08:00:00.000Z`), + deletedAt: LONG_AGO, + }, + }); + } + await prisma.moodEntry.create({ + data: { + id: `${id}-mood-tomb`, + userId: id, + date: "2020-01-01", + mood: "OKAY", + score: 3, + moodLoggedAt: new Date("2020-01-01T08:00:00.000Z"), + deletedAt: LONG_AGO, + }, + }); + } +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +afterAll(async () => { + await truncateAllTables(getPrismaClient()); +}); + +describe("tombstone purge against a running restore (#1031)", () => { + it("leaves the account under restore for the next run without waiting on it", async () => { + const prisma = getPrismaClient(); + await seed(); + const { takeRestoreLock } = await import("@/lib/export/restore-lock"); + const { + cleanupExpiredMeasurementTombstones, + cleanupExpiredMoodTombstones, + } = await import("@/lib/jobs/measurement-tombstone-cleanup"); + + let holding!: () => void; + const isHolding = new Promise((resolve) => (holding = resolve)); + const restore = prisma + .$transaction( + async (tx) => { + await takeRestoreLock(tx, RESTORING); + await tx.measurement.deleteMany({ where: { userId: RESTORING } }); + await tx.moodEntry.deleteMany({ where: { userId: RESTORING } }); + holding(); + await new Promise((resolve) => setTimeout(resolve, HOLD_MS)); + throw new Error("rolled back"); + }, + { timeout: 30_000 }, + ) + .catch(() => undefined); + await isHolding; + + const started = Date.now(); + const measurements = await cleanupExpiredMeasurementTombstones(prisma); + const moods = await cleanupExpiredMoodTombstones(prisma); + const elapsed = Date.now() - started; + await restore; + + expect(elapsed).toBeLessThan(HOLD_MS / 2); + expect(measurements).toEqual({ + deleted: 3, + drained: true, + deferredAccounts: 1, + }); + expect(moods).toEqual({ deleted: 1, drained: true, deferredAccounts: 1 }); + // The stand-in rolled back, so the restoring account still has its + // tombstones; the next run purges them. + expect( + await prisma.measurement.count({ where: { userId: RESTORING } }), + ).toBe(3); + expect(await prisma.measurement.count({ where: { userId: OTHER } })).toBe( + 0, + ); + expect(await prisma.moodEntry.count({ where: { userId: OTHER } })).toBe(0); + + const next = await cleanupExpiredMeasurementTombstones(prisma); + expect(next).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 }); + }, 60_000); + + it("a restore that starts during the purge waits only for the account's batch", async () => { + const prisma = getPrismaClient(); + await seed(); + const { takeRestoreLock } = await import("@/lib/export/restore-lock"); + const { cleanupExpiredMeasurementTombstones } = + await import("@/lib/jobs/measurement-tombstone-cleanup"); + + const purge = cleanupExpiredMeasurementTombstones(prisma); + const restore = prisma.$transaction(async (tx) => { + await takeRestoreLock(tx, RESTORING); + return tx.measurement.deleteMany({ where: { userId: RESTORING } }); + }); + const [purged] = await Promise.all([purge, restore]); + + // Whichever went first, both finished and nothing is left. + expect(purged.deferredAccounts).toBeLessThanOrEqual(1); + expect(await prisma.measurement.count({ where: { userId: OTHER } })).toBe( + 0, + ); + expect( + await prisma.measurement.count({ where: { userId: RESTORING } }), + ).toBe(0); + }, 60_000); +});