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.

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);
   });

src/lib/import/apple-health-staging.ts

@@ -81,14 +81,27 @@ export async function discardStagedUpload(path: string): Promise<void> {
   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<string>;
+  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<number> {
   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;

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();
+  });
+});

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<StagingInUse | null> {
+  const prisma = getWorkerPrisma();
+  const rows = await prisma.importJob.findMany({
+    where: { status: { in: [...ACTIVE_IMPORT_STATUSES] } },
+    select: { status: true, pgBossJobId: true },
+  });
+  const paths = new Set<string>();
+  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