From 382c44ee88cbcd151f51e1a7e3f5e491fe9351a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 26 Sep 2026 22:21:46 +0200 Subject: [PATCH] fix(import): keep a queued Apple Health upload when sweeping staged files The staging sweep removed any staged upload older than seven hours. An import can wait in the queue behind another account's import for longer than that, and when its turn came the file was gone and the import failed. The sweep now keeps every upload a queued or running import still names, and the extracted XML while an import is past the queue. When the queue cannot be read, it removes nothing that tick. --- .../__tests__/apple-health-staging.test.ts | 22 ++++++ src/lib/import/apple-health-staging.ts | 19 +++++- .../apple-health-import-reconcile.test.ts | 67 ++++++++++++++++++- src/lib/jobs/apple-health-import-worker.ts | 52 +++++++++++++- 4 files changed, 155 insertions(+), 5 deletions(-) diff --git a/src/lib/import/__tests__/apple-health-staging.test.ts b/src/lib/import/__tests__/apple-health-staging.test.ts index a8132d17e..3907c27d5 100644 --- a/src/lib/import/__tests__/apple-health-staging.test.ts +++ b/src/lib/import/__tests__/apple-health-staging.test.ts @@ -46,6 +46,28 @@ describe("sweepStaleImportStaging", () => { ); }); + it("keeps what a queued or running import still owns, whatever its age", async () => { + const dir = mkdtempSync(join(tmpdir(), "healthlog-sweep-")); + const uuid = "0b7f3c2a-1d4e-4f5a-9b8c-7d6e5f4a3b2c"; + const queued = `healthlog-apple-health-import-${uuid}.bin`; + const orphan = `healthlog-upload-${uuid}.bin`; + const xml = `healthlog-import-${"cd".repeat(12)}.xml`; + const now = Date.now(); + const old = new Date(now - STAGING_MAX_AGE_MS - 60_000); + for (const name of [queued, orphan, xml]) { + writeFileSync(join(dir, name), "x"); + utimesSync(join(dir, name), old, old); + } + + expect( + await sweepStaleImportStaging(dir, now, { + paths: new Set([join(dir, queued)]), + xmlInUse: true, + }), + ).toBe(1); + expect(readdirSync(dir).sort()).toEqual([queued, xml].sort()); + }); + it("answers zero for a directory that is not there", async () => { expect(await sweepStaleImportStaging("/nonexistent/healthlog")).toBe(0); }); diff --git a/src/lib/import/apple-health-staging.ts b/src/lib/import/apple-health-staging.ts index 08819a490..c38dacfa4 100644 --- a/src/lib/import/apple-health-staging.ts +++ b/src/lib/import/apple-health-staging.ts @@ -81,14 +81,27 @@ export async function discardStagedUpload(path: string): Promise { await rm(path, { force: true }).catch(() => {}); } +/** + * What a queued or running import still owns: the staged uploads its queued + * jobs name, and whether an import is extracting or reading its XML now. + */ +export interface StagingInUse { + paths: ReadonlySet; + xmlInUse: boolean; +} + /** * Remove staged uploads and extracted XML files older than - * `STAGING_MAX_AGE_MS` from `dir`. Only names this app writes are touched. - * Answers how many were removed. + * `STAGING_MAX_AGE_MS` from `dir`, except what `inUse` names. Age alone is + * not enough: an upload can wait in the queue behind another account's + * import for longer than that, and removing it failed the import when its + * turn came. Only names this app writes are touched. Answers how many were + * removed. */ export async function sweepStaleImportStaging( dir: string = tmpdir(), now: number = Date.now(), + inUse: StagingInUse = { paths: new Set(), xmlInUse: false }, ): Promise { let names: string[]; try { @@ -100,6 +113,8 @@ export async function sweepStaleImportStaging( for (const name of names) { if (!STAGING_NAME.test(name)) continue; const path = join(dir, name); + if (inUse.paths.has(path)) continue; + if (inUse.xmlInUse && name.endsWith(".xml")) continue; try { const info = await stat(path); if (!info.isFile()) continue; diff --git a/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts index cff18e33f..ad055b9a2 100644 --- a/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts +++ b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts @@ -25,6 +25,7 @@ const mocks = vi.hoisted(() => ({ })); vi.mock("@/lib/import/apple-health-staging", () => ({ + ACTIVE_IMPORT_STATUSES: ["queued", "unpacking", "parsing", "upserting"], sweepStaleImportStaging: vi.fn(async () => 2), })); @@ -49,7 +50,9 @@ import { IMPORT_JOB_RECONCILE_QUEUE, handleImportJobReconcileTick, reconcileOrphanImportJobs, + stagedImportFilesInUse, } from "../apple-health-import-worker"; +import { sweepStaleImportStaging } from "@/lib/import/apple-health-staging"; const maintenanceSource = readFileSync( join(process.cwd(), "src/lib/jobs/reminder/register-maintenance.ts"), @@ -178,7 +181,8 @@ describe("handleImportJobReconcileTick", () => { ok: true, did: { import_staging_swept: 2 }, }); - expect(mocks.findMany).toHaveBeenCalledTimes(1); + // Once for what the sweep must keep, once for the reconcile. + expect(mocks.findMany).toHaveBeenCalledTimes(2); }); it("reports a failed outcome when the reconcile pass throws", async () => { @@ -198,3 +202,64 @@ describe("handleImportJobReconcileTick", () => { }); }); }); + +describe("stagedImportFilesInUse", () => { + beforeEach(() => { + vi.mocked(sweepStaleImportStaging).mockResolvedValue(0); + }); + + it("keeps the upload of an import still waiting in the queue", async () => { + // An upload can wait behind another account's import for longer than + // the sweep's age limit; removing it failed the import on its turn. + const rows = [ + { status: "queued", pgBossJobId: "boss-1" }, + { status: "parsing", pgBossJobId: "boss-2" }, + { status: "queued", pgBossJobId: "boss-3" }, + ]; + // The sweep's read, then (in the tick) the reconcile's. + mocks.findMany + .mockResolvedValueOnce(rows) + .mockResolvedValueOnce(rows) + .mockResolvedValueOnce([]); + mocks.getGlobalBoss.mockReturnValue({ getJobById: mocks.getJobById }); + mocks.getJobById.mockImplementation(async (queue: string, id: string) => { + if (queue !== APPLE_HEALTH_IMPORT_V2_QUEUE) return null; + if (id === "boss-1") { + return { state: "created", data: { uploadPath: "/tmp/a.bin" } }; + } + if (id === "boss-2") { + return { state: "active", data: { uploadPath: "/tmp/b.bin" } }; + } + // A job pg-boss already finished holds nothing. + return { state: "failed", data: { uploadPath: "/tmp/c.bin" } }; + }); + + const inUse = await stagedImportFilesInUse(); + expect(inUse).not.toBeNull(); + expect([...(inUse?.paths ?? [])].sort()).toEqual([ + "/tmp/a.bin", + "/tmp/b.bin", + ]); + expect(inUse?.xmlInUse).toBe(true); + + await handleImportJobReconcileTick([] as never); + expect(sweepStaleImportStaging).toHaveBeenCalledWith( + undefined, + undefined, + expect.objectContaining({ xmlInUse: true }), + ); + }); + + it("sweeps nothing when the queue cannot be read", async () => { + const rows = [{ status: "queued", pgBossJobId: "boss-1" }]; + mocks.findMany + .mockResolvedValueOnce(rows) + .mockResolvedValueOnce(rows) + .mockResolvedValueOnce([]); + mocks.getGlobalBoss.mockReturnValue(null); + expect(await stagedImportFilesInUse()).toBeNull(); + + await handleImportJobReconcileTick([] as never); + expect(sweepStaleImportStaging).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index ee4d6af78..c803627e7 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -49,7 +49,11 @@ import { } from "@/lib/rollups/measurement-rollups"; import { withBackgroundEvent } from "@/lib/logging/background"; import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; -import { sweepStaleImportStaging } from "@/lib/import/apple-health-staging"; +import { + ACTIVE_IMPORT_STATUSES, + sweepStaleImportStaging, + type StagingInUse, +} from "@/lib/import/apple-health-staging"; /** * Queue + cron for the periodic orphan-ImportJob sweep. v1.32.1 @@ -94,7 +98,14 @@ export async function handleImportJobReconcileTick( // Staged uploads and extracted XML that no running import can still // own: whatever exit left them (a crash, a restart mid-import, a // deploy), the person's export does not stay in /tmp past this. - const swept = await sweepStaleImportStaging().catch(() => 0); + // A queued or running import keeps its files whatever their age; + // when that cannot be read, nothing is swept this tick. + const inUse = await stagedImportFilesInUse().catch(() => null); + const swept = inUse + ? await sweepStaleImportStaging(undefined, undefined, inUse).catch( + () => 0, + ) + : 0; evt.addMeta("import_staging_swept", swept); try { await reconcileOrphanImportJobs(); @@ -655,6 +666,43 @@ const IMPORT_HEARTBEAT_STALE_MS = 30 * 60 * 1000; /** pg-boss job states from which a mid-run import can still make progress. */ const LIVE_PG_BOSS_STATES = new Set(["active", "created", "retry"]); +/** + * The staged files a queued or running import still owns: the upload each + * live pg-boss job names, and whether any import is past the queue (it has + * an extracted XML open). `null` when that cannot be read, so the caller + * sweeps nothing rather than guess. + */ +export async function stagedImportFilesInUse(): Promise { + const prisma = getWorkerPrisma(); + const rows = await prisma.importJob.findMany({ + where: { status: { in: [...ACTIVE_IMPORT_STATUSES] } }, + select: { status: true, pgBossJobId: true }, + }); + const paths = new Set(); + if (rows.length === 0) return { paths, xmlInUse: false }; + const boss = getGlobalBoss(); + if (!boss) return null; + for (const row of rows) { + if (!row.pgBossJobId) continue; + let job: { state: string; data: unknown } | null = null; + for (const queue of [ + APPLE_HEALTH_IMPORT_V2_QUEUE, + APPLE_HEALTH_IMPORT_LEGACY_QUEUE, + ]) { + job = await boss.getJobById(queue, row.pgBossJobId); + if (job) break; + } + if (!job || !LIVE_PG_BOSS_STATES.has(job.state)) continue; + const uploadPath = (job.data as { uploadPath?: unknown } | null) + ?.uploadPath; + if (typeof uploadPath === "string") paths.add(uploadPath); + } + return { + paths, + xmlInUse: rows.some((row) => row.status !== "queued"), + }; +} + /** * Reconcile orphan `ImportJob` rows on worker startup. A row stuck in * `unpacking` / `parsing` / `upserting` means a worker was mid-run when