diff --git a/src/lib/import/__tests__/unzip-export-xml.test.ts b/src/lib/import/__tests__/unzip-export-xml.test.ts
index 3151558c5..f006a3d45 100644
--- a/src/lib/import/__tests__/unzip-export-xml.test.ts
+++ b/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(
``,
);
@@ -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(
+ ``,
+ );
+ 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/);
});
});
diff --git a/src/lib/import/unzip-export-xml.ts b/src/lib/import/unzip-export-xml.ts
index 7333f2d09..6379ce0ec 100644
--- a/src/lib/import/unzip-export-xml.ts
+++ b/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 {
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 {
+ 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 {
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;
}
/**
diff --git a/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts
new file mode 100644
index 000000000..13e240628
--- /dev/null
+++ b/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();
+ });
+});
diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts
index 662b144c2..f3919d5bf 100644
--- a/src/lib/jobs/apple-health-import-worker.ts
+++ b/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