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