fix(import): stream the Apple Health ZIP extraction and sweep orphaned import jobs

Closes #588. The ZIP extractor decompressed export.xml synchronously
into one in-memory buffer, then wrote it with a blocking writeFileSync
— on the same event loop the web server runs in the default
single-container deployment. A real multi-year export could pin the
Node main thread for minutes, freezing every other request, while
briefly holding roughly 2x the decompressed size in heap. A process
that died mid-extraction never reached the failure path, so the
ImportJob row stayed stuck in "unpacking" with "0 rows imported"
forever and no error ever surfaced.

Extraction now streams through zlib's async inflate into a file write
stream, moving the decompression work off the event loop and bounding
memory to the chunk size. The zip-bomb ceiling moves from
inflateRawSync's maxOutputLength to an equivalent byte-counting
transform on the streamed output.

Orphan-ImportJob reconciliation used to run once, at worker boot.
docker-compose's restart:unless-stopped does bring a crashed worker
back, but a fast restart can land before the stuck row's heartbeat
goes stale or pg-boss stops reporting the job as live — and because
the boot-time pass never re-runs on a healthy worker, that row was
never revisited again. A 15-minute cron now re-runs the same reconcile
so a genuinely abandoned import converges to a visible failure instead
of staying silently stuck.

src/lib/import/__tests__/unzip-export-xml.test.ts

@@ -3,16 +3,27 @@ import { writeFileSync, mkdtempSync, readFileSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 import { deflateRawSync } from "node:zlib";
+import { pipeline } from "node:stream/promises";
+import { Readable } from "node:stream";
 
-import { extractExportXml, readCentralDirectory } from "../unzip-export-xml";
+import {
+  createByteCap,
+  extractExportXml,
+  readCentralDirectory,
+} from "../unzip-export-xml";
 
 /**
- * Hand-build a single-entry deflate-compressed ZIP archive with the
- * supplied filename + payload. Enough to exercise the central-directory
- * walker + extractor without pulling in a full zip library.
+ * Hand-build a single-entry ZIP archive with the supplied filename +
+ * payload. Enough to exercise the central-directory walker + extractor
+ * without pulling in a full zip library. `method` selects compression
+ * method 8 (deflate, default) or 0 (stored — bytes copied verbatim).
  */
-function buildMinimalZip(filename: string, payload: Buffer): Buffer {
-  const compressed = deflateRawSync(payload);
+function buildMinimalZip(
+  filename: string,
+  payload: Buffer,
+  method: 0 | 8 = 8,
+): Buffer {
+  const compressed = method === 8 ? deflateRawSync(payload) : payload;
   const crc32 = (() => {
     // Pre-computed lookup table for CRC-32. node:zlib has crc32 in 22+
     // but we keep the function inline to stay version-agnostic.
@@ -38,7 +49,7 @@ function buildMinimalZip(filename: string, payload: Buffer): Buffer {
   localHeader.writeUInt32LE(0x04034b50, 0);
   localHeader.writeUInt16LE(20, 4); // version
   localHeader.writeUInt16LE(0, 6); // flags
-  localHeader.writeUInt16LE(8, 8); // method: deflate
+  localHeader.writeUInt16LE(method, 8); // method
   localHeader.writeUInt16LE(0, 10); // time
   localHeader.writeUInt16LE(0, 12); // date
   localHeader.writeUInt32LE(crc32, 14);
@@ -53,7 +64,7 @@ function buildMinimalZip(filename: string, payload: Buffer): Buffer {
   cdh.writeUInt16LE(20, 4); // version made by
   cdh.writeUInt16LE(20, 6); // version needed
   cdh.writeUInt16LE(0, 8); // flags
-  cdh.writeUInt16LE(8, 10); // method
+  cdh.writeUInt16LE(method, 10); // method
   cdh.writeUInt16LE(0, 12); // time
   cdh.writeUInt16LE(0, 14); // date
   cdh.writeUInt32LE(crc32, 16);
@@ -95,7 +106,7 @@ describe("readCentralDirectory", () => {
 });
 
 describe("extractExportXml", () => {
-  it("extracts the export.xml member to a temp file", () => {
+  it("extracts a deflated export.xml member to a temp file", async () => {
     const payload = Buffer.from(
       `<?xml version="1.0"?><HealthData><ExportDate value="2026-05-15"/></HealthData>`,
     );
@@ -104,7 +115,7 @@ describe("extractExportXml", () => {
     const zipPath = join(tmp, "export.zip");
     writeFileSync(zipPath, zip);
 
-    const out = extractExportXml(zipPath);
+    const out = await extractExportXml(zipPath);
     expect(out.xmlBytes).toBe(payload.length);
     expect(out.otherMembers).toHaveLength(0);
 
@@ -112,13 +123,65 @@ describe("extractExportXml", () => {
     expect(extracted.toString("utf8")).toBe(payload.toString("utf8"));
   });
 
-  it("throws when the export.xml member is missing", () => {
+  // v1.32.1 (issue #588) — the stored (uncompressed) path now runs
+  // through the same streaming pipeline as deflate; a stored member
+  // must still round-trip byte-exact through it.
+  it("extracts a stored (uncompressed) export.xml member to a temp file", async () => {
+    const payload = Buffer.from(
+      `<?xml version="1.0"?><HealthData><Record type="HKQuantityTypeIdentifierHeartRate"/></HealthData>`,
+    );
+    const zip = buildMinimalZip("apple_health_export/export.xml", payload, 0);
+    const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-"));
+    const zipPath = join(tmp, "export.zip");
+    writeFileSync(zipPath, zip);
+
+    const out = await extractExportXml(zipPath);
+    expect(out.xmlBytes).toBe(payload.length);
+
+    const extracted = readFileSync(out.xmlPath);
+    expect(extracted.toString("utf8")).toBe(payload.toString("utf8"));
+  });
+
+  it("throws when the export.xml member is missing", async () => {
     const payload = Buffer.from("noop");
     const zip = buildMinimalZip("other-file.txt", payload);
     const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-"));
     const zipPath = join(tmp, "export.zip");
     writeFileSync(zipPath, zip);
 
-    expect(() => extractExportXml(zipPath)).toThrow(/missing/);
+    await expect(extractExportXml(zipPath)).rejects.toThrow(/missing/);
+  });
+});
+
+// v1.32.1 (issue #588) — the streaming inflate Transform has no
+// `maxOutputLength` option, so the zip-bomb defence moved to a
+// hand-rolled byte-counting stage. Exercise it directly at a tiny
+// limit; the real 8 GiB ceiling is not practical to trip from a
+// fixture.
+describe("createByteCap", () => {
+  it("passes bytes through unchanged while under the limit", async () => {
+    const chunks: Buffer[] = [];
+    await pipeline(
+      Readable.from([Buffer.from("hello"), Buffer.from("world")]),
+      createByteCap(10),
+      async function collect(source) {
+        for await (const chunk of source) chunks.push(chunk as Buffer);
+      },
+    );
+    expect(Buffer.concat(chunks).toString("utf8")).toBe("helloworld");
+  });
+
+  it("rejects the stream once the running total exceeds the limit", async () => {
+    await expect(
+      pipeline(
+        Readable.from([Buffer.from("hello"), Buffer.from("world")]),
+        createByteCap(6),
+        async function drain(source) {
+          for await (const _chunk of source) {
+            /* drain */
+          }
+        },
+      ),
+    ).rejects.toThrow(/exceeds the 6-byte cap/);
   });
 });

src/lib/import/unzip-export-xml.ts

@@ -7,21 +7,49 @@
  * deflate-compressed ZIP archive. Pulling in `yauzl` / `adm-zip` / etc.
  * for one file we only ever read once would balloon the build graph
  * (and `yauzl` ships a non-trivial number of transitive deps). Node 22
- * ships `node:zlib` `inflateRawSync` already; the only missing piece
- * is a tiny ZIP central-directory walker.
+ * ships `node:zlib` already; the only missing piece is a tiny ZIP
+ * central-directory walker.
  *
  * Coverage scope:
  *   - Stored entries (compression method 0) — copy bytes verbatim.
- *   - Deflated entries (compression method 8) — `inflateRawSync`.
+ *   - Deflated entries (compression method 8) — streamed through
+ *     `zlib.createInflateRaw()`.
  *   - Encrypted entries → unsupported (Apple does not encrypt the
  *     export.zip; reject with a clear error).
  *   - Zip64 — supported for the central directory record (Apple
  *     exports easily push past the 4 GB limit on multi-year accounts).
  *
+ * v1.32.1 (issue #588) — extracting the member used to run through
+ * `inflateRawSync()` on a single in-memory buffer holding the WHOLE
+ * inflated `export.xml` (up to `MAX_DECOMPRESSED_BYTES`) at the same
+ * time the compressed-archive buffer was still resident, then
+ * `writeFileSync()`'d the result in one blocking call. `extractExportXml`
+ * runs on the same event loop as the web server in the default
+ * single-container deployment (`src/instrumentation.ts` starts the
+ * pg-boss worker in-process), so that synchronous path could pin the
+ * Node main thread for minutes on a real multi-year export — freezing
+ * every other request the app was serving — while briefly needing
+ * roughly 2x the decompressed size in JS heap. A user hitting this saw
+ * an import that never leaves "Unpacking the archive…": the phase
+ * label is written to `ImportJob.status` BEFORE extraction starts and
+ * only advances once `extractExportXml()` returns, so a process that
+ * dies mid-extraction (OOM, restart) before that return leaves the row
+ * silently stuck with no failure ever recorded. The member now streams
+ * through `zlib.createInflateRaw()` into a file write stream —
+ * decompression work moves off the JS event loop onto the libuv
+ * threadpool in bounded chunks, and peak memory is bounded by the
+ * chunk size rather than the whole decompressed payload. The
+ * `MAX_DECOMPRESSED_BYTES` zip-bomb defence moves with it: the
+ * streaming inflate API has no `maxOutputLength` option, so a
+ * byte-counting transform enforces the cap on the real output as it
+ * flows.
+ *
  * Locks per `.planning/research/v1434-r-1-xml-import.md` §6.1.
  */
-import { readFileSync, writeFileSync } from "node:fs";
-import { inflateRawSync } from "node:zlib";
+import { readFileSync, createWriteStream, statSync, unlinkSync } from "node:fs";
+import { createInflateRaw } from "node:zlib";
+import { Readable, Transform } from "node:stream";
+import { pipeline } from "node:stream/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 
@@ -44,11 +72,15 @@ const MAX_DECOMPRESSED_BYTES = 8 * 1024 * 1024 * 1024;
  * Pre-flight refusal threshold for the central-directory's advertised
  * ratio. Legitimate Apple Health XML compresses at maybe 10–20× under
  * DEFLATE; anything claiming a 200× expansion is a synthesized bomb.
- * This is a coarse signal — the `maxOutputLength` ceiling on the
- * inflate call below is the load-bearing defence.
+ * This is a coarse signal — the byte-counting cap on the streamed
+ * inflate output (`streamEntryToFile()` below) is the load-bearing
+ * defence.
  */
 const MAX_COMPRESSION_RATIO = 200;
 
+/** Chunk size fed into the inflate/write pipeline per `write()` call. */
+const STREAM_CHUNK_BYTES = 1 * 1024 * 1024;
+
 /** A single entry within the archive's central directory. */
 interface CentralDirectoryEntry {
   fileName: string;
@@ -78,7 +110,9 @@ export interface UnzipResult {
  * Throws when the member is missing, encrypted, or compressed with
  * an unsupported method.
  */
-export function extractExportXml(archivePath: string): UnzipResult {
+export async function extractExportXml(
+  archivePath: string,
+): Promise<UnzipResult> {
   const buf = readFileSync(archivePath);
   const entries = readCentralDirectory(buf);
 
@@ -105,8 +139,8 @@ export function extractExportXml(archivePath: string): UnzipResult {
   // Pre-flight zip-bomb defence. The central directory's advertised
   // `uncompressedSize` is attacker-controlled (a malicious archive can
   // lie), so this catches honest-but-oversized payloads early; the
-  // load-bearing defence is the `maxOutputLength` cap inside
-  // `extractEntry()` which trips on the actual inflate output.
+  // load-bearing defence is the byte-counting cap inside
+  // `streamEntryToFile()` which trips on the actual inflate output.
   if (exportXmlEntry.uncompressedSize > MAX_DECOMPRESSED_BYTES) {
     throw new Error(
       `export.xml declares an uncompressed size of ${exportXmlEntry.uncompressedSize} bytes` +
@@ -126,13 +160,11 @@ export function extractExportXml(archivePath: string): UnzipResult {
     );
   }
 
-  const xmlBytes = extractEntry(buf, exportXmlEntry);
-
   const xmlPath = join(
     tmpdir(),
     `healthlog-import-${Date.now()}-${Math.random().toString(36).slice(2, 10)}.xml`,
   );
-  writeFileSync(xmlPath, xmlBytes);
+  const xmlBytes = await streamEntryToFile(buf, exportXmlEntry, xmlPath);
 
   const otherMembers = entries
     .filter((e) => e !== exportXmlEntry)
@@ -140,7 +172,7 @@ export function extractExportXml(archivePath: string): UnzipResult {
 
   return {
     xmlPath,
-    xmlBytes: xmlBytes.length,
+    xmlBytes,
     otherMembers,
   };
 }
@@ -260,8 +292,59 @@ export function readCentralDirectory(buf: Buffer): CentralDirectoryEntry[] {
   return entries;
 }
 
-/** Extract the bytes of a single ZIP entry into a Buffer. */
-function extractEntry(buf: Buffer, entry: CentralDirectoryEntry): Buffer {
+/**
+ * Split a Buffer into fixed-size slices (no copying — `subarray` views
+ * share the backing memory) so the inflate/write pipeline processes the
+ * entry in bounded chunks instead of one giant `write()` call.
+ */
+function* chunkBuffer(buf: Buffer, size: number): Generator<Buffer> {
+  for (let offset = 0; offset < buf.length; offset += size) {
+    yield buf.subarray(offset, Math.min(offset + size, buf.length));
+  }
+}
+
+/**
+ * Byte-counting passthrough that refuses to forward more than `limit`
+ * bytes — the streaming equivalent of `inflateRawSync`'s
+ * `maxOutputLength`, since the Transform-based zlib API exposes no such
+ * option. Trips on the actual bytes flowing through even when the
+ * central-directory metadata lied about the uncompressed size. Exported
+ * so a unit test can exercise the cap directly against a small limit —
+ * the real `MAX_DECOMPRESSED_BYTES` ceiling (8 GiB) is not practical to
+ * trip from a test fixture.
+ */
+export function createByteCap(limit: number): Transform {
+  let total = 0;
+  return new Transform({
+    transform(chunk: Buffer, _enc, callback) {
+      total += chunk.length;
+      if (total > limit) {
+        callback(
+          new Error(
+            `Decompressed export.xml exceeds the ${limit}-byte cap` +
+              " — refusing as a suspected zip bomb.",
+          ),
+        );
+        return;
+      }
+      callback(null, chunk);
+    },
+  });
+}
+
+/**
+ * Stream a single ZIP entry's bytes out to `destPath`, decompressing
+ * through `zlib.createInflateRaw()` when needed. Runs entirely off the
+ * JS main thread (stream I/O + the libuv-threadpool-backed zlib
+ * binding), so a multi-GB member no longer blocks the event loop the
+ * way the old `inflateRawSync()` + `writeFileSync()` pair did. Returns
+ * the number of bytes written.
+ */
+async function streamEntryToFile(
+  buf: Buffer,
+  entry: CentralDirectoryEntry,
+  destPath: string,
+): Promise<number> {
   const local = entry.localHeaderOffset;
   if (buf.readUInt32LE(local) !== LOCAL_FILE_HEADER) {
     throw new Error(
@@ -271,18 +354,32 @@ function extractEntry(buf: Buffer, entry: CentralDirectoryEntry): Buffer {
   const localFileNameLen = buf.readUInt16LE(local + 26);
   const localExtraLen = buf.readUInt16LE(local + 28);
   const dataStart = local + 30 + localFileNameLen + localExtraLen;
-  const compressed = buf.slice(dataStart, dataStart + entry.compressedSize);
+  const compressed = buf.subarray(dataStart, dataStart + entry.compressedSize);
 
-  if (entry.compressionMethod === 0) {
-    return compressed;
+  const source = Readable.from(chunkBuffer(compressed, STREAM_CHUNK_BYTES));
+  const dest = createWriteStream(destPath);
+  const cap = createByteCap(MAX_DECOMPRESSED_BYTES);
+
+  try {
+    if (entry.compressionMethod === 0) {
+      await pipeline(source, cap, dest);
+    } else {
+      // Method 8 — DEFLATE, streamed through the async zlib Transform.
+      await pipeline(source, createInflateRaw(), cap, dest);
+    }
+  } catch (err) {
+    // A mid-stream failure (byte cap trip, corrupt deflate stream) can
+    // leave a partial file on disk — the old sync path only ever wrote
+    // once inflate had already fully succeeded. Clean up so a rejected
+    // archive doesn't leak a partial `/tmp` file.
+    try {
+      unlinkSync(destPath);
+    } catch {
+      // ignore — best-effort
+    }
+    throw err;
   }
-  // Method 8 — DEFLATE. `maxOutputLength` instructs `node:zlib` to
-  // refuse expansion past the cap with a thrown error, defeating
-  // zip-bomb amplification even when the central-directory metadata
-  // lies about the uncompressed size.
-  return inflateRawSync(compressed, {
-    maxOutputLength: MAX_DECOMPRESSED_BYTES,
-  });
+  return statSync(destPath).size;
 }
 
 /**

src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts

@@ -0,0 +1,186 @@
+/**
+ * v1.32.1 (issue #588) — periodic orphan-`ImportJob` reconcile.
+ *
+ * Before this change `reconcileOrphanImportJobs()` only ran once, at
+ * worker boot. A worker that crashed mid-extraction (OOM, restart) and
+ * came back up BEFORE the stuck row's heartbeat went stale (30 min) or
+ * pg-boss stopped reporting the backing job as live left that row
+ * stuck in `unpacking` / `parsing` / `upserting` forever — an import
+ * that never leaves "Unpacking the archive… 0 rows imported" with no
+ * failure ever surfaced, because nothing re-evaluated the row after
+ * that one boot-time pass. These tests cover both the periodic-tick
+ * wiring (queue provisioned, cron scheduled, handler bound) and the
+ * reconcile logic itself (a genuinely-abandoned row flips to `failed`;
+ * a row another worker is still actively running does not).
+ */
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  findMany: vi.fn(),
+  updateMany: vi.fn(),
+  getGlobalBoss: vi.fn(),
+  getJobById: vi.fn(),
+}));
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    importJob: {
+      findMany: mocks.findMany,
+      updateMany: mocks.updateMany,
+    },
+  },
+  toJson: (value: unknown) => value,
+}));
+
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: mocks.getGlobalBoss,
+}));
+
+import {
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
+  APPLE_HEALTH_IMPORT_V2_QUEUE,
+  IMPORT_JOB_RECONCILE_CRON,
+  IMPORT_JOB_RECONCILE_QUEUE,
+  handleImportJobReconcileTick,
+  reconcileOrphanImportJobs,
+} from "../apple-health-import-worker";
+
+const maintenanceSource = readFileSync(
+  join(process.cwd(), "src/lib/jobs/reminder/register-maintenance.ts"),
+  "utf8",
+);
+
+beforeEach(() => {
+  vi.resetAllMocks();
+  mocks.getGlobalBoss.mockReturnValue({ getJobById: mocks.getJobById });
+  mocks.updateMany.mockResolvedValue({ count: 0 });
+});
+
+describe("periodic reconcile — wiring", () => {
+  it("provisions the queue, schedules a 15-minute cron, and binds the handler", () => {
+    expect(IMPORT_JOB_RECONCILE_QUEUE).toBe("apple-health-import-reconcile");
+    expect(IMPORT_JOB_RECONCILE_CRON).toBe("*/15 * * * *");
+
+    const allQueues = maintenanceSource.match(
+      /const allQueues\s*=\s*\[([\s\S]*?)\];/,
+    );
+    expect(allQueues).not.toBeNull();
+    expect(allQueues![1]).toMatch(/\bIMPORT_JOB_RECONCILE_QUEUE\b/);
+
+    const schedules = maintenanceSource.match(
+      /const schedules[\s\S]*?=\s*\[([\s\S]*?)\];/,
+    );
+    expect(schedules).not.toBeNull();
+    expect(schedules![1]).toMatch(
+      /\[IMPORT_JOB_RECONCILE_QUEUE,\s*IMPORT_JOB_RECONCILE_CRON\]/,
+    );
+
+    expect(maintenanceSource).toMatch(
+      /boss\.work\(\s*IMPORT_JOB_RECONCILE_QUEUE[\s\S]{0,120}handleImportJobReconcileTick/,
+    );
+  });
+});
+
+describe("reconcileOrphanImportJobs", () => {
+  it("flips a row to failed once its heartbeat has gone stale", async () => {
+    mocks.findMany.mockResolvedValue([
+      {
+        id: "import-stale",
+        pgBossJobId: "boss-1",
+        updatedAt: new Date(Date.now() - 31 * 60 * 1000),
+      },
+    ]);
+    mocks.getJobById.mockResolvedValue({ state: "active" });
+
+    await reconcileOrphanImportJobs();
+
+    expect(mocks.updateMany).toHaveBeenCalledWith({
+      where: { id: { in: ["import-stale"] } },
+      data: {
+        status: "failed",
+        failureReason: "interrupted_by_restart",
+        completedAt: expect.any(Date),
+      },
+    });
+  });
+
+  it("leaves a row alone while its heartbeat is fresh and pg-boss reports the job active", async () => {
+    mocks.findMany.mockResolvedValue([
+      {
+        id: "import-live",
+        pgBossJobId: "boss-2",
+        updatedAt: new Date(Date.now() - 2 * 60 * 1000),
+      },
+    ]);
+    mocks.getJobById.mockResolvedValue({ state: "active" });
+
+    await reconcileOrphanImportJobs();
+
+    expect(mocks.updateMany).not.toHaveBeenCalled();
+    expect(mocks.getJobById).toHaveBeenCalledWith(
+      APPLE_HEALTH_IMPORT_V2_QUEUE,
+      "boss-2",
+    );
+  });
+
+  it("flips a row whose backing pg-boss job is gone even with a fresh heartbeat", async () => {
+    mocks.findMany.mockResolvedValue([
+      {
+        id: "import-gone",
+        pgBossJobId: "boss-3",
+        updatedAt: new Date(Date.now() - 60 * 1000),
+      },
+    ]);
+    mocks.getJobById.mockResolvedValue(null);
+
+    await reconcileOrphanImportJobs();
+
+    expect(mocks.updateMany).toHaveBeenCalledWith({
+      where: { id: { in: ["import-gone"] } },
+      data: {
+        status: "failed",
+        failureReason: "interrupted_by_restart",
+        completedAt: expect.any(Date),
+      },
+    });
+  });
+
+  it("only ever scopes the candidate query to the current parser revision", async () => {
+    mocks.findMany.mockResolvedValue([]);
+
+    await reconcileOrphanImportJobs();
+
+    expect(mocks.findMany).toHaveBeenCalledWith({
+      where: {
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+        status: { in: ["unpacking", "parsing", "upserting"] },
+      },
+      select: { id: true, pgBossJobId: true, updatedAt: true },
+    });
+  });
+});
+
+describe("handleImportJobReconcileTick", () => {
+  it("delegates to reconcileOrphanImportJobs and resolves on success", async () => {
+    mocks.findMany.mockResolvedValue([]);
+
+    await expect(
+      handleImportJobReconcileTick([] as never),
+    ).resolves.toBeUndefined();
+    expect(mocks.findMany).toHaveBeenCalledTimes(1);
+  });
+
+  it("swallows a reconcile failure instead of throwing", async () => {
+    mocks.findMany.mockRejectedValue(new Error("db unavailable"));
+
+    // A background sweep this cheap must never spam pg-boss's
+    // retry/backoff machinery — it just re-runs on the next 15-minute
+    // tick. The handler must resolve even when the underlying reconcile
+    // pass throws.
+    await expect(
+      handleImportJobReconcileTick([] as never),
+    ).resolves.toBeUndefined();
+  });
+});

src/lib/jobs/apple-health-import-worker.ts

@@ -37,6 +37,52 @@ import {
   type ImportJobResult,
 } from "@/lib/measurements/import-apple-health-export";
 import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups";
+import { withBackgroundEvent } from "@/lib/logging/background";
+
+/**
+ * Queue + cron for the periodic orphan-ImportJob sweep. v1.32.1
+ * (issue #588) — `reconcileOrphanImportJobs()` used to run only once,
+ * at worker boot. `restart: unless-stopped` in `docker-compose.yml`
+ * DOES bring a crashed/OOM-killed worker back up, so the boot-time
+ * pass usually catches a dead run — but only when that restart lands
+ * AFTER the row's heartbeat has already gone stale (30 minutes) or
+ * pg-boss no longer reports the backing job as live. A fast restart
+ * lands neither condition true yet, and because the pass runs
+ * exactly once per boot, a worker that then stays up and healthy
+ * NEVER re-evaluates that row — it is stuck in `unpacking` /
+ * `parsing` / `upserting` with "0 rows imported" forever and no
+ * failure ever surfaced to the status poll. A 15-minute cron closes
+ * that gap: within two ticks of the heartbeat going stale, the same
+ * idempotent reconcile flips the row to `failed` with an honest
+ * reason, so the UI's "Unpacking the archive…" spinner eventually
+ * turns into a visible, actionable error instead of an indefinite
+ * silent stall.
+ */
+export const IMPORT_JOB_RECONCILE_QUEUE = "apple-health-import-reconcile";
+export const IMPORT_JOB_RECONCILE_CRON = "*/15 * * * *";
+
+/**
+ * Cron handler for the periodic orphan-ImportJob sweep. Delegates to
+ * the same `reconcileOrphanImportJobs()` the boot path calls — the
+ * heartbeat + live pg-boss-job check already make it safe to re-run
+ * on a healthy worker (a live import's fresh heartbeat and `active`
+ * pg-boss state both keep its row untouched). Errors are logged, not
+ * rethrown — a reconcile miss must never spam pg-boss's retry/backoff
+ * machinery for a background sweep this cheap to just re-run in 15
+ * minutes.
+ */
+export async function handleImportJobReconcileTick(
+  jobs: Job<object>[],
+): Promise<void> {
+  void jobs;
+  await withBackgroundEvent("job.apple_health_import_reconcile", async (evt) => {
+    try {
+      await reconcileOrphanImportJobs();
+    } catch (err) {
+      evt.addWarning(`apple-health-import-reconcile failed: ${err}`);
+    }
+  });
+}
 
 /** Queue name isolated from revision-1 workers during rolling deployments. */
 export const APPLE_HEALTH_IMPORT_V2_QUEUE = "apple-health-import-v2";
@@ -260,7 +306,7 @@ export async function handleAppleHealthImport(
       elapsedMs: 0,
     });
 
-    const unzip = extractExportXml(uploadPath);
+    const unzip = await extractExportXml(uploadPath);
     extractedXmlPath = unzip.xmlPath;
 
     // Phase 2 + 3: parsing + upserting (the parser tracks both

src/lib/jobs/reminder/register-maintenance.ts

@@ -41,7 +41,10 @@ import {
   APPLE_HEALTH_IMPORT_V2_QUEUE,
   APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
   APPLE_HEALTH_IMPORT_CONCURRENCY,
+  IMPORT_JOB_RECONCILE_QUEUE,
+  IMPORT_JOB_RECONCILE_CRON,
   handleAppleHealthImport,
+  handleImportJobReconcileTick,
   migrateLegacyAppleHealthImport,
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
@@ -314,6 +317,11 @@ const allQueues = [
   INTAKE_AUTO_SKIP_QUEUE,
   APPLE_HEALTH_IMPORT_V2_QUEUE,
   APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
+  // v1.32.1 (issue #588) — periodic sweep for ImportJob rows orphaned by a
+  // worker crash/restart mid-run. Without this entry the 15-minute cron
+  // below silently no-ops and a stuck "unpacking"/"parsing"/"upserting" row
+  // that survives past the next worker boot is never revisited.
+  IMPORT_JOB_RECONCILE_QUEUE,
   MEDICATION_INTAKE_IMPORT_QUEUE,
   // v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery enqueues
   // one job per user holding two live intake rows that snap to the same
@@ -462,6 +470,13 @@ const schedules: ScheduleEntry[] = [
   // Document vault — daily 04:10 Europe/Berlin purge for tombstoned
   // documents past the 30-day undo grace.
   [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON],
+  // v1.32.1 (issue #588) — every-15-minute orphan-ImportJob sweep. Re-runs
+  // the same reconcile the boot path uses, so a stuck "unpacking" row
+  // whose worker crashed/restarted without the boot-time pass catching it
+  // (heartbeat not yet stale, pg-boss job not yet expired) still converges
+  // to a visible `failed` state within two ticks instead of staying stuck
+  // forever on an otherwise-healthy worker.
+  [IMPORT_JOB_RECONCILE_QUEUE, IMPORT_JOB_RECONCILE_CRON],
 ];
 
 /**
@@ -706,6 +721,14 @@ export async function registerMaintenanceQueues(
       }
     },
   );
+  // v1.32.1 (issue #588) — periodic orphan-ImportJob sweep. Single-flight:
+  // the underlying `updateMany` is idempotent and two ticks racing the same
+  // read-then-write pass would just be wasted work.
+  await boss.work(
+    IMPORT_JOB_RECONCILE_QUEUE,
+    { localConcurrency: 1 },
+    handleImportJobReconcileTick,
+  );
   await boss.work<MedicationIntakeImportQueuePayload>(
     MEDICATION_INTAKE_IMPORT_QUEUE,
     {

src/lib/multipart/__tests__/stream-to-disk.test.ts

@@ -431,7 +431,7 @@ describe("streamMultipartToDisk — large multi-chunk binary round-trip", () =>
         expect(written.equals(zip), `zip byte-exact for ${sizes}`).toBe(true);
         // The load-bearing assertion: the saved file is still a parseable
         // archive (EOCD intact) and the member round-trips.
-        const extracted = extractExportXml(result.filePath);
+        const extracted = await extractExportXml(result.filePath);
         xmlPath = extracted.xmlPath;
         expect(readFileSync(extracted.xmlPath).equals(xmlBuf)).toBe(true);
       } finally {