diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c4b0c4f7..ba123c9de 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 1d99d8b6c..188d7c073 100644 --- a/docs/api/openapi.yaml +++ b/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. diff --git a/docs/self-hosting/scaling.md b/docs/self-hosting/scaling.md index 2ab4ea976..9d0527047 100644 --- a/docs/self-hosting/scaling.md +++ b/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. diff --git a/e2e/timezone-profile.spec.ts b/e2e/timezone-profile.spec.ts index e47d0bac4..50dd311d4 100644 --- a/e2e/timezone-profile.spec.ts +++ b/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 }); diff --git a/package.json b/package.json index 9a539cb18..4163b9cd5 100644 --- a/package.json +++ b/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", diff --git a/prisma/migrations/0242_import_job_heartbeat/migration.sql b/prisma/migrations/0242_import_job_heartbeat/migration.sql new file mode 100644 index 000000000..c9c8bf106 --- /dev/null +++ b/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; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 38274a2de..80848d888 100644 --- a/prisma/schema.prisma +++ b/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("{}") diff --git a/public/sw.js b/public/sw.js index 20da2a0a9..34d07b2f8 100644 --- a/public/sw.js +++ b/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/*` diff --git a/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts b/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts index 10b00c716..871b6fc77 100644 --- a/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts +++ b/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(); + 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); + }); }); diff --git a/src/app/api/admin/import-apple-health-export/route.ts b/src/app/api/admin/import-apple-health-export/route.ts index 6e5d4c5a5..e19739d2d 100644 --- a/src/app/api/admin/import-apple-health-export/route.ts +++ b/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( { diff --git a/src/app/api/import/apple-health-export/__tests__/route.test.ts b/src/app/api/import/apple-health-export/__tests__/route.test.ts index 226922c24..09fe09aa6 100644 --- a/src/app/api/import/apple-health-export/__tests__/route.test.ts +++ b/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(); + 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); + }); +}); diff --git a/src/app/api/import/apple-health-export/route.ts b/src/app/api/import/apple-health-export/route.ts index 25263623c..fcd505d03 100644 --- a/src/app/api/import/apple-health-export/route.ts +++ b/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( { diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index bfd16e008..888f80c29 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/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 { 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", diff --git a/tests/integration/apple-health-import-reconcile.test.ts b/tests/integration/apple-health-import-reconcile.test.ts new file mode 100644 index 000000000..541a42034 --- /dev/null +++ b/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); + }); +});