From 82aa66b400b48905764d1d3063bc0f6343810ef2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 15 Jul 2026 22:07:25 +0200 Subject: [PATCH 1/2] =?UTF-8?q?chore(release):=20v1.28.36=20=E2=80=94=20re?= =?UTF-8?q?-importing=20after=20a=20failed=20Apple=20Health=20import=20act?= =?UTF-8?q?ually=20retries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The v1.28.33 rollup + staging fixes were correct but never reached the reporter's case: the upload dedup keys on the content SHA-256 and the lookup matched a previously FAILED job too, so re-uploading the same export.zip returned the dead job's id and replayed its stale failureReason without ever staging or sending again. The dedup now excludes failed rows (short-circuits only on queued/unpacking/parsing/ upserting/done); the same bytes retry cleanly. A redundant staged upload is unlinked on a dedup hit instead of leaking in the staging dir. reconcileOrphanImportJobs no longer unconditionally fails every non-terminal ImportJob at boot: it now checks pg-boss job state plus a progress heartbeat (new import_jobs.updated_at, migration 0242), so a split or multi-worker deployment cannot kill an import running in another worker. The missing-staging message names the separate-container cause explicitly, with a matching note in docs/self-hosting/scaling.md. --- CHANGELOG.md | 17 ++ docs/api/openapi.yaml | 2 +- docs/self-hosting/scaling.md | 26 ++ package.json | 2 +- .../0242_import_job_heartbeat/migration.sql | 12 + prisma/schema.prisma | 6 + public/sw.js | 2 +- .../__tests__/route.test.ts | 73 +++++ .../admin/import-apple-health-export/route.ts | 10 +- .../__tests__/route.test.ts | 79 ++++++ .../api/import/apple-health-export/route.ts | 18 +- src/lib/jobs/apple-health-import-worker.ts | 98 ++++++- .../apple-health-import-reconcile.test.ts | 264 ++++++++++++++++++ 13 files changed, 590 insertions(+), 19 deletions(-) create mode 100644 prisma/migrations/0242_import_job_heartbeat/migration.sql create mode 100644 tests/integration/apple-health-import-reconcile.test.ts 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/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); + }); +}); From 7a6598f7c3ec49e89e173d74d351ee611127a1bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 15 Jul 2026 22:23:04 +0200 Subject: [PATCH 2/2] test(e2e): stabilise the profile-timezone sessions-card expand The sessions card toggle's onClick could be dropped when Playwright clicked before React hydrated the handler, leaving the region hidden for the whole wait and failing intermittently in CI. Retry the click until aria-expanded actually flips. --- e2e/timezone-profile.spec.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) 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 });