Merge pull request #493 from MBombeck/release/v1.28.36

release/v1.28.36 — re-importing after a failed Apple Health import actually retries

CHANGELOG.md

@@ -2,6 +2,23 @@
 
 ## [Unreleased]
 
+## [1.28.36] — 2026-07-15 — Re-importing after a failed Apple Health import actually retries
+
+The v1.28.33 rollup and staging fixes were correct but never reached the
+reporter's case: the upload deduplicates by content hash, and the lookup
+matched a previously FAILED job too — so re-uploading the same export.zip
+returned the old failed job and replayed its stale error without ever running
+the import again. The dedup now ignores failed jobs; the same file retries
+cleanly. A redundant staged upload is also cleaned up on a dedup hit instead
+of lingering in the staging directory.
+
+Two related hardenings: the boot-time reconcile that marks interrupted imports
+failed now performs a real liveness check (queue state plus a progress
+heartbeat) instead of unconditionally failing every in-flight job — so a split
+or multi-worker deployment no longer kills an import running in another worker.
+And the "staging file not found" message now names the separate-container cause
+explicitly, with a matching note in the scaling guide.
+
 ## [1.28.35] — 2026-07-14 — Timestamps honor the profile timezone
 
 Client-rendered times previously fell back to a hardcoded display zone

docs/api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: 3.1.0
 info:
   title: HealthLog API
-  version: 1.28.35
+  version: 1.28.36
   description: >-
     Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest.
 

docs/self-hosting/scaling.md

@@ -30,6 +30,32 @@ but it doubles the DB load and muddies telemetry without adding
 throughput a personal instance needs (`src/lib/process-type.ts`).
 Run **one** worker container and scale the `web` containers instead.
 
+## Apple Health import staging (shared volume required for a split)
+
+The Apple Health `export.zip` import streams the upload to a staging
+file on local disk (under the container's temp dir) and hands the
+worker only the file **path**. In single-container mode
+(`HEALTHLOG_PROCESS_TYPE=all`, the default) the web handler and the
+worker share the same filesystem, so the handoff just works.
+
+In a web/worker split — or any topology where the container that
+accepts the upload is not the one that runs the job (a `web` +
+`worker` split, extra `all` replicas, or old+new containers coexisting
+during a rolling deploy) — the worker cannot see the web container's
+staging file. The import then fails with a staging-file-missing error
+("the import staging file is not visible to the worker"). Two things to
+know:
+
+- A pure `web` container has no pg-boss handle, so an import kicked off
+  there currently returns `503 Background worker is not running`. Run
+  imports from a container that also runs the worker until send-only
+  boss support lands.
+- If you run web and worker as separate containers, they **must share
+  the import staging directory** — mount the same named volume at the
+  temp path on both — or run imports in single-container (`all`) mode.
+  Without a shared staging volume, imports in split mode cannot
+  complete.
+
 ## Healthchecks
 
 - `app` (web) healthcheck: `wget /api/version` every 30s.

e2e/timezone-profile.spec.ts

@@ -77,12 +77,20 @@ test.describe("profile-timezone display (#490)", () => {
 
     await page.goto("/settings/privacy", { waitUntil: "domcontentloaded" });
 
-    // The sessions card is collapsed by default — expand it first.
+    // The sessions card is collapsed by default — expand it first. The
+    // toggle's onClick can be dropped if Playwright clicks before React has
+    // hydrated the handler, leaving the region `hidden` for the whole wait.
+    // Retry the click until `aria-expanded` actually flips.
     const toggle = page.locator(
       '[data-slot="settings-security-sessions-toggle"]',
     );
     await expect(toggle).toBeVisible({ timeout: 10_000 });
-    await toggle.click();
+    await expect(async () => {
+      await toggle.click();
+      await expect(toggle).toHaveAttribute("aria-expanded", "true", {
+        timeout: 2_000,
+      });
+    }).toPass({ timeout: 15_000 });
 
     const row = page.locator('[data-slot="security-session-row"]').first();
     await expect(row).toBeVisible({ timeout: 10_000 });

package.json

@@ -1,6 +1,6 @@
 {
   "name": "healthlog",
-  "version": "1.28.35",
+  "version": "1.28.36",
   "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.",
   "license": "PolyForm-Noncommercial-1.0.0",
   "homepage": "https://healthlog.dev",

prisma/migrations/0242_import_job_heartbeat/migration.sql

@@ -0,0 +1,12 @@
+-- Import-job liveness heartbeat (issue #486 follow-up).
+--
+-- `reconcileOrphanImportJobs` used to flip EVERY non-terminal ImportJob
+-- row to `failed` on any worker boot — killing an import that was still
+-- actively parsing in another worker (multi-replica / rolling deploy).
+-- The reconcile now gates on this heartbeat: a non-terminal row is only
+-- treated as orphaned when its pg-boss job is gone/terminal OR its
+-- heartbeat has gone stale. `@updatedAt` bumps this column on every
+-- write, including each mid-run progress tick, so a live import keeps it
+-- fresh. Existing rows backfill to the ALTER time via the DB default.
+ALTER TABLE "import_jobs"
+  ADD COLUMN "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;

prisma/schema.prisma

@@ -4089,6 +4089,12 @@ model ImportJob {
   exportedAt         DateTime? @map("exported_at")
   startedAt          DateTime  @default(now()) @map("started_at")
   completedAt        DateTime? @map("completed_at")
+  /// Bumped on every write, including each mid-run progress tick. Acts
+  /// as a liveness heartbeat: `reconcileOrphanImportJobs` treats a
+  /// non-terminal row whose heartbeat has gone stale as orphaned, so a
+  /// job still actively parsing in another worker (fresh heartbeat) is
+  /// never flipped to `failed` (issue #486 follow-up).
+  updatedAt          DateTime  @updatedAt @map("updated_at")
   /// JSON envelope mirroring `ImportJobProgress` (`src/lib/measurements/
   /// import-apple-health-export.ts`). Updated mid-run.
   progress           Json      @default("{}")

public/sw.js

@@ -36,7 +36,7 @@ try {
 // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild.
 const CACHE_VERSION =
   (typeof self !== "undefined" && self.__APP_VERSION__) ||
-  /* @sw-version-fallback */ "v1.28.35";
+  /* @sw-version-fallback */ "v1.28.36";
 const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`;
 const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`;
 // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*`

src/app/api/admin/import-apple-health-export/__tests__/route.test.ts

@@ -32,6 +32,8 @@ const mocks = vi.hoisted(() => ({
   importJobFindFirst: vi.fn().mockResolvedValue(null),
   importJobUpdate: vi.fn(),
   userFindUnique: vi.fn(),
+  streamToDisk: vi.fn(),
+  unlink: vi.fn().mockResolvedValue(undefined),
 }));
 
 vi.mock("@/lib/rate-limit", () => ({
@@ -42,6 +44,15 @@ vi.mock("@/lib/jobs/boss-instance", () => ({
   getGlobalBoss: mocks.getGlobalBoss,
 }));
 
+vi.mock("@/lib/multipart/stream-to-disk", () => ({
+  streamMultipartToDisk: mocks.streamToDisk,
+}));
+
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs/promises")>();
+  return { ...actual, unlink: mocks.unlink };
+});
+
 vi.mock("@/lib/db", () => ({
   prisma: {
     importJob: {
@@ -67,6 +78,8 @@ const USER_SESSION = {
   user: { id: "user-2", username: "alice", role: "USER" as const },
 };
 
+const STAGED_PATH = "/tmp/healthlog-admin-apple-health-import-abc.bin";
+
 beforeEach(() => {
   vi.resetAllMocks();
   mocks.checkRateLimit.mockResolvedValue({ allowed: true });
@@ -76,8 +89,27 @@ beforeEach(() => {
   mocks.importJobFindFirst.mockResolvedValue(null);
   mocks.importJobUpdate.mockResolvedValue({ id: "ij-1" });
   mocks.userFindUnique.mockResolvedValue({ id: "user-2", username: "alice" });
+  mocks.streamToDisk.mockResolvedValue({
+    filePath: STAGED_PATH,
+    bytes: 100,
+    sha256: "deadbeef",
+    originalFilename: "export.zip",
+    textFields: { userId: "user-2" },
+  });
+  mocks.unlink.mockResolvedValue(undefined);
 });
 
+function multipartReq(): NextRequest {
+  return new NextRequest(
+    "http://localhost/api/admin/import-apple-health-export",
+    {
+      method: "POST",
+      headers: { "content-type": "multipart/form-data; boundary=---x" },
+      body: "stub",
+    },
+  );
+}
+
 describe("POST /api/admin/import-apple-health-export", () => {
   it("returns 401 when unauthenticated", async () => {
     vi.mocked(getSession).mockResolvedValue(null);
@@ -139,4 +171,45 @@ describe("POST /api/admin/import-apple-health-export", () => {
     const res = await POST(req);
     expect(res.status).toBe(429);
   });
+
+  it("re-stages and enqueues when the only prior job for the same bytes is failed (issue #486)", async () => {
+    vi.mocked(getSession).mockResolvedValue(ADMIN_SESSION as never);
+    mocks.importJobFindFirst.mockImplementation(async (args: unknown) => {
+      const where = (args as { where?: { status?: { not?: string } } }).where;
+      if (where?.status?.not === "failed") return null;
+      return { id: "old-failed-job", status: "failed" };
+    });
+
+    const res = await POST(multipartReq());
+    expect(res.status).toBe(202);
+    const body = await res.json();
+    expect(body.data.status).toBe("queued");
+    expect(body.data.idempotent).toBeUndefined();
+    expect(body.data.jobId).not.toBe("old-failed-job");
+    expect(mocks.bossSend).toHaveBeenCalledTimes(1);
+    expect(mocks.importJobCreate).toHaveBeenCalledTimes(1);
+    expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({ status: { not: "failed" } }),
+      }),
+    );
+    expect(mocks.unlink).not.toHaveBeenCalled();
+  });
+
+  it("short-circuits to a viable prior job and unlinks the redundant upload", async () => {
+    vi.mocked(getSession).mockResolvedValue(ADMIN_SESSION as never);
+    mocks.importJobFindFirst.mockResolvedValue({
+      id: "live-job",
+      status: "parsing",
+    });
+
+    const res = await POST(multipartReq());
+    expect(res.status).toBe(202);
+    const body = await res.json();
+    expect(body.data.idempotent).toBe(true);
+    expect(body.data.jobId).toBe("live-job");
+    expect(mocks.bossSend).not.toHaveBeenCalled();
+    expect(mocks.importJobCreate).not.toHaveBeenCalled();
+    expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH);
+  });
 });

src/app/api/admin/import-apple-health-export/route.ts

@@ -26,6 +26,7 @@ import {
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
 import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
+import { unlink } from "node:fs/promises";
 
 export const dynamic = "force-dynamic";
 
@@ -112,15 +113,22 @@ export const POST = apiHandler(async (request: NextRequest) => {
 
   // Same content-hash idempotency as the user-facing route, but scoped
   // by the target user — an admin re-uploading the same file for the
-  // same user resolves to the previous job.
+  // same user resolves to the previous job. As on the user route, a
+  // prior `failed` job is excluded (issue #486) so the same export can
+  // be retried; only still-viable jobs (queued / in-flight / done)
+  // short-circuit.
   const existing = await prisma.importJob.findFirst({
     where: {
       userId: targetUser.id,
       uploadSha256: uploaded.sha256,
+      status: { not: "failed" },
     },
     orderBy: { startedAt: "desc" },
   });
   if (existing) {
+    // Redundant staging file — the existing viable job owns the bytes.
+    // Unlink so a deduped admin re-upload does not leak `/tmp`.
+    await unlink(uploaded.filePath).catch(() => {});
     annotate({ meta: { idempotent_hit: true, job_id: existing.id } });
     return apiSuccess(
       {

src/app/api/import/apple-health-export/__tests__/route.test.ts

@@ -33,6 +33,8 @@ const mocks = vi.hoisted(() => ({
   importJobCreate: vi.fn(),
   importJobFindFirst: vi.fn().mockResolvedValue(null),
   importJobUpdate: vi.fn(),
+  streamToDisk: vi.fn(),
+  unlink: vi.fn().mockResolvedValue(undefined),
 }));
 
 vi.mock("@/lib/rate-limit", () => ({
@@ -43,6 +45,15 @@ vi.mock("@/lib/jobs/boss-instance", () => ({
   getGlobalBoss: mocks.getGlobalBoss,
 }));
 
+vi.mock("@/lib/multipart/stream-to-disk", () => ({
+  streamMultipartToDisk: mocks.streamToDisk,
+}));
+
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs/promises")>();
+  return { ...actual, unlink: mocks.unlink };
+});
+
 vi.mock("@/lib/db", () => ({
   prisma: {
     importJob: {
@@ -61,6 +72,8 @@ const SESSION_OK = {
   user: { id: "user-1", username: "testuser", role: "USER" as const },
 };
 
+const STAGED_PATH = "/tmp/healthlog-apple-health-import-abc.bin";
+
 beforeEach(() => {
   vi.resetAllMocks();
   mocks.checkRateLimit.mockResolvedValue({ allowed: true });
@@ -69,8 +82,24 @@ beforeEach(() => {
   mocks.importJobCreate.mockResolvedValue({ id: "ij-1" });
   mocks.importJobFindFirst.mockResolvedValue(null);
   mocks.importJobUpdate.mockResolvedValue({ id: "ij-1" });
+  mocks.streamToDisk.mockResolvedValue({
+    filePath: STAGED_PATH,
+    bytes: 100,
+    sha256: "deadbeef",
+    originalFilename: "export.zip",
+    textFields: {},
+  });
+  mocks.unlink.mockResolvedValue(undefined);
 });
 
+function multipartReq(): NextRequest {
+  return new NextRequest("http://localhost/api/import/apple-health-export", {
+    method: "POST",
+    headers: { "content-type": "multipart/form-data; boundary=---x" },
+    body: "stub",
+  });
+}
+
 describe("POST /api/import/apple-health-export", () => {
   it("returns 401 when unauthenticated", async () => {
     vi.mocked(getSession).mockResolvedValue(null);
@@ -119,3 +148,53 @@ describe("POST /api/import/apple-health-export", () => {
     expect(res.status).toBe(429);
   });
 });
+
+describe("POST /api/import/apple-health-export — dedup by content hash (issue #486)", () => {
+  it("re-stages and enqueues when the only prior job for the same bytes is failed", async () => {
+    vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+    // The DB holds a FAILED prior job for this sha. The dedup query
+    // excludes failed rows, so it returns null and the upload must fall
+    // through to a fresh stage + boss.send instead of replaying the dead
+    // job's id/failureReason forever.
+    mocks.importJobFindFirst.mockImplementation(async (args: unknown) => {
+      const where = (args as { where?: { status?: { not?: string } } }).where;
+      if (where?.status?.not === "failed") return null;
+      return { id: "old-failed-job", status: "failed" };
+    });
+
+    const res = await POST(multipartReq());
+    expect(res.status).toBe(202);
+    const body = await res.json();
+    expect(body.data.status).toBe("queued");
+    expect(body.data.idempotent).toBeUndefined();
+    expect(body.data.jobId).not.toBe("old-failed-job");
+    expect(mocks.bossSend).toHaveBeenCalledTimes(1);
+    expect(mocks.importJobCreate).toHaveBeenCalledTimes(1);
+    // The dedup lookup must exclude failed rows.
+    expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({ status: { not: "failed" } }),
+      }),
+    );
+    // A fresh job was staged — nothing to unlink on this path.
+    expect(mocks.unlink).not.toHaveBeenCalled();
+  });
+
+  it("short-circuits to a viable prior job and unlinks the redundant upload", async () => {
+    vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+    mocks.importJobFindFirst.mockResolvedValue({
+      id: "live-job",
+      status: "parsing",
+    });
+
+    const res = await POST(multipartReq());
+    expect(res.status).toBe(202);
+    const body = await res.json();
+    expect(body.data.idempotent).toBe(true);
+    expect(body.data.jobId).toBe("live-job");
+    expect(mocks.bossSend).not.toHaveBeenCalled();
+    expect(mocks.importJobCreate).not.toHaveBeenCalled();
+    // The freshly staged upload is redundant and must be cleaned up.
+    expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH);
+  });
+});

src/app/api/import/apple-health-export/route.ts

@@ -12,7 +12,10 @@
  * the bytes) short-circuits to the previous `ImportJob` id without
  * re-queueing — the iOS client cannot reliably emit a stable
  * `Idempotency-Key` for a 1 GB file, so we hash the bytes and dedup
- * on content.
+ * on content. The dedup only covers still-viable jobs (queued,
+ * in-flight, or already `done`); a prior `failed` job is NOT matched,
+ * so the same export can always be retried with the same bytes
+ * (issue #486 — a failed row must never tombstone its own hash).
  *
  * Locks per `.planning/research/v1434-r-1-xml-import.md` §5.3.
  */
@@ -31,6 +34,7 @@ import {
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
 import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
+import { unlink } from "node:fs/promises";
 
 export const dynamic = "force-dynamic";
 
@@ -104,15 +108,25 @@ export const POST = apiHandler(async (request: NextRequest) => {
   }
 
   // Content-hash idempotency: a re-upload of the same bytes resolves
-  // to the previous ImportJob row instead of re-queueing.
+  // to the previous ImportJob row instead of re-queueing — but ONLY
+  // when that prior job is still viable (queued / in-flight / done). A
+  // `failed` prior job must NOT short-circuit (issue #486): matching it
+  // returned its stale failureReason forever and made retrying the same
+  // export impossible. Excluding `failed` lets the same bytes fall
+  // through to a fresh stage + enqueue.
   const existing = await prisma.importJob.findFirst({
     where: {
       userId: user.id,
       uploadSha256: uploaded.sha256,
+      status: { not: "failed" },
     },
     orderBy: { startedAt: "desc" },
   });
   if (existing) {
+    // The existing viable job owns the canonical bytes; the upload we
+    // just streamed to `/tmp` is redundant. Unlink it so a deduped
+    // re-upload does not leak a gigabyte-scale staging file.
+    await unlink(uploaded.filePath).catch(() => {});
     annotate({ meta: { idempotent_hit: true, job_id: existing.id } });
     return apiSuccess(
       {

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

@@ -31,6 +31,7 @@ import { PrismaPg } from "@prisma/adapter-pg";
 import type { Job } from "pg-boss";
 
 import { extractExportXml } from "@/lib/import/unzip-export-xml";
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import {
   streamParseExportXml,
   type ImportJobProgress,
@@ -310,8 +311,12 @@ export async function handleAppleHealthImport(
       "code" in err &&
       (err as NodeJS.ErrnoException).code === "ENOENT";
     const reason = missingStagingFile
-      ? "Import staging file is no longer available (the server restarted" +
-        " or a previous attempt cleaned it up) — upload the export again"
+      ? "Import staging file is no longer available — the server restarted," +
+        " a previous attempt cleaned it up, or (if you run separate web and" +
+        " worker containers) they do not share the import staging directory." +
+        " Upload the export again; split deployments must run single-container" +
+        " mode or mount a shared staging volume on both the web and worker" +
+        " containers."
       : err instanceof Error
         ? err.message
         : String(err);
@@ -338,22 +343,89 @@ function safeUnlink(path: string): void {
 }
 
 /**
- * Reconcile orphan `ImportJob` rows on worker startup. A row stuck
- * in `parsing` or `upserting` whose `pgBossJobId` is no longer alive
- * on the boss side means the worker was killed mid-run; flip it to
- * `failed` with `interrupted_by_restart` so the operator can re-run
- * the upload. Idempotent — re-running this on a clean startup is a
- * no-op.
+ * A non-terminal ImportJob whose heartbeat (`updatedAt`) has not moved
+ * for this long is treated as orphaned even when pg-boss still reports
+ * its job `active`. A live import bumps the heartbeat on every progress
+ * tick (every 1000 records → sub-second during the parse phase), and
+ * the only legitimately-quiet window is the `unpacking` unzip of a
+ * multi-GB archive, so 30 minutes clears the largest observed exports
+ * with headroom while still self-healing a genuinely stuck job.
+ */
+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"]);
+
+/**
+ * Reconcile orphan `ImportJob` rows on worker startup. A row stuck in
+ * `unpacking` / `parsing` / `upserting` means a worker was mid-run when
+ * it last shut down — flip it to `failed` with `interrupted_by_restart`
+ * so the operator can re-upload (and, post-issue-#486, so the kick-off
+ * dedup no longer short-circuits future re-uploads onto a dead job).
  *
- * Exported so the worker boot path in `reminder-worker.ts` can wire
- * it into the start-up sequence right after `boss.start()`.
+ * The reconcile is deliberately NOT unconditional. In a multi-replica
+ * or rolling-deploy topology a booting worker must not flip a row that
+ * another live worker is actively parsing. It therefore keeps a
+ * non-terminal row alive when BOTH:
+ *   - pg-boss still reports its backing job in a live state
+ *     (`active` / `created` / `retry`), AND
+ *   - the row's `updatedAt` heartbeat is fresher than
+ *     `IMPORT_HEARTBEAT_STALE_MS`.
+ * A row is reconciled to `failed` when its pg-boss job is gone (null,
+ * archived) or terminal (`completed` / `cancelled` / `failed`), OR when
+ * its heartbeat has gone stale (owner died but pg-boss has not yet
+ * expired the job). This keeps the single-worker default self-healing
+ * truly-stuck jobs while never racing a live import in another worker.
+ *
+ * If the boss handle is unavailable (should not happen — reconcile runs
+ * after `setGlobalBoss()`), it falls back to the heartbeat bound alone.
+ * Idempotent — re-running on a clean startup is a no-op.
+ *
+ * Exported so the worker boot path in `reminder-worker.ts` can wire it
+ * into the start-up sequence right after `boss.start()`.
  */
 export async function reconcileOrphanImportJobs(): Promise<void> {
   const prisma = getWorkerPrisma();
+  const candidates = await prisma.importJob.findMany({
+    where: { status: { in: ["unpacking", "parsing", "upserting"] } },
+    select: { id: true, pgBossJobId: true, updatedAt: true },
+  });
+  if (candidates.length === 0) return;
+
+  const boss = getGlobalBoss();
+  const staleBefore = Date.now() - IMPORT_HEARTBEAT_STALE_MS;
+  const orphanIds: string[] = [];
+
+  for (const row of candidates) {
+    const heartbeatStale = row.updatedAt.getTime() < staleBefore;
+
+    // No live-state source, or no backing job id, or a stale heartbeat:
+    // the row cannot be confirmed as running anywhere → reconcile it.
+    if (!boss || !row.pgBossJobId || heartbeatStale) {
+      orphanIds.push(row.id);
+      continue;
+    }
+
+    let live = false;
+    try {
+      const job = await boss.getJobById(
+        APPLE_HEALTH_IMPORT_QUEUE,
+        row.pgBossJobId,
+      );
+      live = job !== null && LIVE_PG_BOSS_STATES.has(job.state);
+    } catch {
+      // Lookup failed — the heartbeat is fresh (checked above), so leave
+      // the row alone rather than risk flipping a live import; a later
+      // boot re-evaluates it once the heartbeat goes stale.
+      live = true;
+    }
+
+    if (!live) orphanIds.push(row.id);
+  }
+
+  if (orphanIds.length === 0) return;
   await prisma.importJob.updateMany({
-    where: {
-      status: { in: ["unpacking", "parsing", "upserting"] },
-    },
+    where: { id: { in: orphanIds } },
     data: {
       status: "failed",
       failureReason: "interrupted_by_restart",

tests/integration/apple-health-import-reconcile.test.ts

@@ -0,0 +1,264 @@
+/**
+ * Issue #486 follow-up — `reconcileOrphanImportJobs` liveness contract +
+ * the kick-off dedup-by-status DB contract, pinned against real Postgres.
+ *
+ * Two defects fixed here:
+ *
+ *   1. Boot reconcile used to flip EVERY non-terminal ImportJob row to
+ *      `failed` unconditionally. In a multi-replica / rolling-deploy
+ *      topology that killed an import still actively parsing in another
+ *      worker. The reconcile now only flips a row whose pg-boss job is
+ *      gone/terminal OR whose `updatedAt` heartbeat has gone stale — a
+ *      job that is `active` in pg-boss with a fresh heartbeat survives.
+ *
+ *   2. The kick-off dedup matched ANY prior job with the same
+ *      `uploadSha256`, including a `failed` one, tombstoning the file's
+ *      hash forever. The lookup now excludes `failed` rows so the same
+ *      export always stays retryable; a still-viable job (queued /
+ *      in-flight / `done`) still short-circuits.
+ */
+import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import {
+  reconcileOrphanImportJobs,
+  _setWorkerPrismaForTests,
+} from "@/lib/jobs/apple-health-import-worker";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+// Controllable pg-boss handle: the reconcile consults `getJobById` for a
+// row's real queue state.
+const bossMock = vi.hoisted(() => ({
+  getJobById: vi.fn(),
+  handle: null as { getJobById: (n: string, id: string) => unknown } | null,
+}));
+
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: vi.fn(() => bossMock.handle),
+}));
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+  _setWorkerPrismaForTests(getPrismaClient());
+  bossMock.getJobById.mockReset();
+  bossMock.handle = { getJobById: bossMock.getJobById };
+});
+
+afterAll(() => {
+  _setWorkerPrismaForTests(null);
+});
+
+async function createUser(username: string) {
+  return getPrismaClient().user.create({
+    data: { username, email: `${username}@example.test`, role: "USER" },
+  });
+}
+
+/** Backdate the heartbeat past the staleness threshold (30 min). */
+async function backdateHeartbeat(rowId: string, minutesAgo: number) {
+  await getPrismaClient().$executeRawUnsafe(
+    `UPDATE import_jobs SET updated_at = NOW() - INTERVAL '${minutesAgo} minutes' WHERE id = $1`,
+    rowId,
+  );
+}
+
+describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => {
+  it("does NOT flip a row that is active in pg-boss with a fresh heartbeat", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("recon-active-fresh");
+    const row = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        pgBossJobId: "boss-active-fresh",
+        status: "parsing",
+        uploadBytes: 100,
+      },
+    });
+    bossMock.getJobById.mockResolvedValue({
+      id: "boss-active-fresh",
+      state: "active",
+    });
+
+    await reconcileOrphanImportJobs();
+
+    const after = await prisma.importJob.findUnique({ where: { id: row.id } });
+    expect(after!.status).toBe("parsing");
+    expect(after!.failureReason).toBeNull();
+  });
+
+  it("flips a row whose pg-boss job is gone (archived / null)", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("recon-gone");
+    const row = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        pgBossJobId: "boss-gone",
+        status: "upserting",
+        uploadBytes: 100,
+      },
+    });
+    bossMock.getJobById.mockResolvedValue(null);
+
+    await reconcileOrphanImportJobs();
+
+    const after = await prisma.importJob.findUnique({ where: { id: row.id } });
+    expect(after!.status).toBe("failed");
+    expect(after!.failureReason).toBe("interrupted_by_restart");
+  });
+
+  it("flips a row whose pg-boss job is terminal (completed / failed)", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("recon-terminal");
+    const row = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        pgBossJobId: "boss-terminal",
+        status: "parsing",
+        uploadBytes: 100,
+      },
+    });
+    bossMock.getJobById.mockResolvedValue({
+      id: "boss-terminal",
+      state: "completed",
+    });
+
+    await reconcileOrphanImportJobs();
+
+    const after = await prisma.importJob.findUnique({ where: { id: row.id } });
+    expect(after!.status).toBe("failed");
+  });
+
+  it("flips a stale-heartbeat row even when pg-boss still reports it active", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("recon-stale-active");
+    const row = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        pgBossJobId: "boss-stale-active",
+        status: "parsing",
+        uploadBytes: 100,
+      },
+    });
+    await backdateHeartbeat(row.id, 45);
+    bossMock.getJobById.mockResolvedValue({
+      id: "boss-stale-active",
+      state: "active",
+    });
+
+    await reconcileOrphanImportJobs();
+
+    const after = await prisma.importJob.findUnique({ where: { id: row.id } });
+    expect(after!.status).toBe("failed");
+  });
+
+  it("self-heals under the single-worker fallback when no boss handle is present", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("recon-no-boss");
+    const row = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        pgBossJobId: "boss-none",
+        status: "parsing",
+        uploadBytes: 100,
+      },
+    });
+    bossMock.handle = null; // getGlobalBoss() → null
+
+    await reconcileOrphanImportJobs();
+
+    const after = await prisma.importJob.findUnique({ where: { id: row.id } });
+    expect(after!.status).toBe("failed");
+  });
+
+  it("leaves terminal rows untouched", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("recon-already-terminal");
+    const done = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        pgBossJobId: "boss-done",
+        status: "done",
+        uploadBytes: 100,
+      },
+    });
+
+    await reconcileOrphanImportJobs();
+
+    const after = await prisma.importJob.findUnique({ where: { id: done.id } });
+    expect(after!.status).toBe("done");
+    expect(bossMock.getJobById).not.toHaveBeenCalled();
+  });
+});
+
+describe("kick-off dedup — status-scoped lookup (issue #486)", () => {
+  // Mirrors the exact `findFirst` the kick-off routes run.
+  async function dedupLookup(userId: string, sha: string) {
+    return getPrismaClient().importJob.findFirst({
+      where: { userId, uploadSha256: sha, status: { not: "failed" } },
+      orderBy: { startedAt: "desc" },
+    });
+  }
+
+  it("does NOT match a prior failed job for the same bytes", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("dedup-failed");
+    await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        status: "failed",
+        failureReason: "ENOENT: no such file or directory",
+        uploadBytes: 100,
+        uploadSha256: "sha-failed",
+        completedAt: new Date(),
+      },
+    });
+
+    expect(await dedupLookup(user.id, "sha-failed")).toBeNull();
+  });
+
+  it("still matches a viable prior job (queued / in-flight / done) for the same bytes", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("dedup-viable");
+    const done = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        status: "done",
+        uploadBytes: 100,
+        uploadSha256: "sha-viable",
+        completedAt: new Date(),
+      },
+    });
+
+    const hit = await dedupLookup(user.id, "sha-viable");
+    expect(hit!.id).toBe(done.id);
+  });
+
+  it("falls through to the viable row when a failed and a done row share a hash", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("dedup-mixed");
+    await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        status: "failed",
+        uploadBytes: 100,
+        uploadSha256: "sha-mixed",
+        completedAt: new Date(),
+      },
+    });
+    const done = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        status: "done",
+        uploadBytes: 100,
+        uploadSha256: "sha-mixed",
+        completedAt: new Date(),
+      },
+    });
+
+    const hit = await dedupLookup(user.id, "sha-mixed");
+    expect(hit!.id).toBe(done.id);
+  });
+});