From 6d8fe06148fe23c4b518654eb26ab6e4fe2dd14b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 14 Jul 2026 11:56:29 +0200 Subject: [PATCH] =?UTF-8?q?chore(release):=20v1.28.33=20=E2=80=94=20re-imp?= =?UTF-8?q?orting=20a=20cumulative=20Apple=20Health=20export=20works=20aga?= =?UTF-8?q?in?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rollup writes were delete-then-createMany inside a transaction; under READ COMMITTED a concurrent recompute racing the import's full fold on the same (user, type, granularity, bucket, source) partition tripped the composite PK — and the P2002 swallow dropped the loser's freshly computed aggregate, leaving a stale bucket. persistRollupRows now writes chunked INSERT … ON CONFLICT DO UPDATE with deterministic (type, source, bucket) lock ordering inside one transaction: a race can neither fail the import nor lose an aggregate, and previously-staled buckets self-heal on their next recompute. The vanished-source delete and in-transaction guarantees are preserved; the now-unreachable P2002 swallow is removed. The masked ENOENT is gone with it: long imports outlived the queue's 15-minute expiry, pg-boss redelivered the still-running job, and the redelivery re-opened the consumed staging file — overwriting the real terminal state with a raw ENOENT. Import sends now carry retryLimit 0 + 6h expiry (a retry after the staging file is consumed can only mask), deliveries for terminal ImportJob rows are no-ops, a genuinely missing staging file reports an honest re-upload message, and the extracted XML is cleaned up on failure too. --- CHANGELOG.md | 17 + docs/api/openapi.yaml | 2 +- package.json | 2 +- public/sw.js | 2 +- .../admin/import-apple-health-export/route.ts | 10 +- .../api/import/apple-health-export/route.ts | 10 +- .../apple-health-import-send-policy.test.ts | 51 +++ src/lib/jobs/apple-health-import-worker.ts | 76 ++++- .../__tests__/measurement-rollups.test.ts | 163 ++++++---- src/lib/rollups/measurement-rollups.ts | 212 +++++++------ ...pple-health-import-worker-failures.test.ts | 221 +++++++++++++ .../rollup-concurrent-recompute.test.ts | 293 ++++++++++++++++++ 12 files changed, 902 insertions(+), 157 deletions(-) create mode 100644 src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts create mode 100644 tests/integration/apple-health-import-worker-failures.test.ts create mode 100644 tests/integration/rollup-concurrent-recompute.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 8380fa5f3..aa0aa539b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,23 @@ ## [Unreleased] +## [1.28.33] — 2026-07-14 — Re-importing a cumulative Apple Health export works again + +Re-uploading a fresh export.zip over existing data could fail with a duplicate-key +error in the rollup tier: a concurrent recompute (a live sync, the nightly drain) +racing the import's full fold could collide on the same aggregation bucket — and +the losing side silently dropped its freshly computed aggregate, leaving a stale +bucket. Rollup writes are now conflict-safe upserts with deterministic lock +ordering: a race can neither fail the import nor lose an aggregate, and buckets +left stale by earlier races self-heal on their next recompute. + +The misleading error is gone too. A long import could outlive its queue timeout; +the redelivery re-opened the already-consumed staging file and overwrote the real +failure — or even a successful completion — with a raw ENOENT. Import jobs now +finish terminally (no redelivery after the staging file is consumed, six-hour +timeout), a delivery for an already-finished import is a no-op, and a genuinely +missing staging file reports an honest "upload the export again" message. + ## [1.28.32] — 2026-07-12 — Fitbit: re-keyed re-imports can no longer be silently dropped Single-fix release for the Fitbit integration, mirroring the Google Health diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 98992b787..f81246937 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.32 + version: 1.28.33 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index e371aceea..ef82df582 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.28.32", + "version": "1.28.33", "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/public/sw.js b/public/sw.js index f4feb1b02..4d35ea43d 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.32"; + /* @sw-version-fallback */ "v1.28.33"; 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/route.ts b/src/app/api/admin/import-apple-health-export/route.ts index 2d59f23ec..6e5d4c5a5 100644 --- a/src/app/api/admin/import-apple-health-export/route.ts +++ b/src/app/api/admin/import-apple-health-export/route.ts @@ -22,6 +22,7 @@ import { checkRateLimit } from "@/lib/rate-limit"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk"; @@ -158,7 +159,14 @@ export const POST = apiHandler(async (request: NextRequest) => { uploadBytes: uploaded.bytes, enqueuedAt: new Date().toISOString(), }; - const bossJobId = await boss.send(APPLE_HEALTH_IMPORT_QUEUE, payload); + // No retries + wide expiration: the staged upload is consumed by the + // first run, so a redelivery could only fail on the deleted `/tmp` + // file and mask the real outcome (see the worker's send-options doc). + const bossJobId = await boss.send( + APPLE_HEALTH_IMPORT_QUEUE, + payload, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, + ); await prisma.importJob.update({ where: { id: importJob.id }, diff --git a/src/app/api/import/apple-health-export/route.ts b/src/app/api/import/apple-health-export/route.ts index 4ff136a93..25263623c 100644 --- a/src/app/api/import/apple-health-export/route.ts +++ b/src/app/api/import/apple-health-export/route.ts @@ -27,6 +27,7 @@ import { checkRateLimit } from "@/lib/rate-limit"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk"; @@ -148,7 +149,14 @@ export const POST = apiHandler(async (request: NextRequest) => { uploadBytes: uploaded.bytes, enqueuedAt: new Date().toISOString(), }; - const bossJobId = await boss.send(APPLE_HEALTH_IMPORT_QUEUE, payload); + // No retries + wide expiration: the staged upload is consumed by the + // first run, so a redelivery could only fail on the deleted `/tmp` + // file and mask the real outcome (see the worker's send-options doc). + const bossJobId = await boss.send( + APPLE_HEALTH_IMPORT_QUEUE, + payload, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, + ); await prisma.importJob.update({ where: { id: importJob.id }, diff --git a/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts b/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts new file mode 100644 index 000000000..c6659c456 --- /dev/null +++ b/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts @@ -0,0 +1,51 @@ +/** + * v1.28.33 (issue #486) — send-policy pins for the Apple Health import + * queue. + * + * The queue defaults (retryLimit 2, expireInSeconds 900) redelivered a + * long-running import after its staged `/tmp` upload was already + * consumed: the re-run failed on the deleted file and overwrote the + * first run's terminal state with a raw ENOENT. The kick-off sends must + * therefore disable retries and widen the expiration past the largest + * observed exports — and BOTH kick-off routes (user + admin) must + * actually pass the shared policy, or the default silently returns. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ + prisma: {}, + buildSessionOptions: () => null, + getPoolMax: () => 5, + toJson: (value: unknown) => value, +})); + +import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "../apple-health-import-worker"; + +const ROUTE_FILES = [ + "src/app/api/import/apple-health-export/route.ts", + "src/app/api/admin/import-apple-health-export/route.ts", +]; + +describe("apple health import — pg-boss send policy", () => { + it("disables retries and widens the expiration window", () => { + // retryLimit 0: the staged upload is unlinked by the first run, so + // a retry can only mask the original outcome behind an ENOENT. + expect(APPLE_HEALTH_IMPORT_SEND_OPTIONS.retryLimit).toBe(0); + // Expiration must exceed the 15-minute queue default by a wide + // margin — GB-scale exports parse for well over an hour. + expect( + APPLE_HEALTH_IMPORT_SEND_OPTIONS.expireInSeconds, + ).toBeGreaterThanOrEqual(60 * 60); + }); + + it("both kick-off routes send with the shared policy", () => { + for (const file of ROUTE_FILES) { + const source = readFileSync(join(process.cwd(), file), "utf8"); + expect(source).toMatch( + /boss\.send\(\s*APPLE_HEALTH_IMPORT_QUEUE,\s*payload,\s*APPLE_HEALTH_IMPORT_SEND_OPTIONS,?\s*\)/, + ); + } + }); +}); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index f3d224894..bfd16e008 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -44,6 +44,26 @@ export const APPLE_HEALTH_IMPORT_QUEUE = "apple-health-import"; /** Concurrency cap per worker host. */ export const APPLE_HEALTH_IMPORT_CONCURRENCY = 1; +/** + * v1.28.33 (issue #486) — job-level pg-boss overrides for the import + * sends. The queue defaults (retryLimit 2, expireInSeconds 900) are + * wrong for this job shape twice over: + * + * - A retry can never succeed: the first run consumes and unlinks the + * staged `/tmp` upload, so a redelivery re-opens a deleted file and + * its ENOENT masks the first run's real outcome. + * - A GB-scale export parses for well over 15 minutes; the default + * expiration marked the still-running job failed mid-run and + * scheduled exactly that doomed retry. + * + * `retryLimit: 0` makes the single run authoritative; the expiration + * leaves generous headroom over the largest observed exports. + */ +export const APPLE_HEALTH_IMPORT_SEND_OPTIONS = { + retryLimit: 0, + expireInSeconds: 6 * 60 * 60, +} as const; + /** Payload `boss.send` carries onto the queue. */ export interface AppleHealthImportPayload { /** Owner of the imported rows. */ @@ -59,6 +79,19 @@ export interface AppleHealthImportPayload { } let workerPrismaSingleton: PrismaClient | null = null; + +/** + * Test-only handle on the worker Prisma singleton. Mirrors the + * `_resetEnsureUserRollupsFreshInFlightForTests` pattern in + * `measurement-rollups.ts` — the integration suite injects the shared + * testcontainer client so the handler does not open a second pool that + * would dangle past the container teardown. Production code never calls + * this. + */ +export function _setWorkerPrismaForTests(client: PrismaClient | null): void { + workerPrismaSingleton = client; +} + function getWorkerPrisma(): PrismaClient { if (!workerPrismaSingleton) { // Inherit the web client's pool ceiling + per-session statement timeouts @@ -136,6 +169,29 @@ export async function handleAppleHealthImport( } const importJobId = importJob.id; + // v1.28.33 (issue #486) — refuse to re-run a job whose mirror row is + // already terminal. pg-boss redelivers after the queue's expiration + // window (a GB-scale import outlives the default 15 minutes), but the + // first run consumed and unlinked the staged upload, so a redelivery + // can only re-open the deleted `/tmp` file, fail with ENOENT, and + // OVERWRITE the first run's real outcome (a genuine failure reason — + // or a completed import flipped back to `failed`). The kick-off + // endpoints now send with `retryLimit: 0`; this guard keeps any + // residual redelivery (expiration sweep, operator requeue) from + // masking the terminal state. + if (importJob.status === "done" || importJob.status === "failed") { + console.warn( + `[apple-health-import] Ignoring duplicate delivery for ImportJob=${importJobId}` + + ` — row is already terminal (${importJob.status}); the staged upload` + + " was consumed by the first run and a re-run could only mask its outcome", + ); + return; + } + + // Extracted-XML path, hoisted so the failure path can clean it up — + // pre-v1.28.33 a parse failure stranded the multi-GB XML in `/tmp`. + let extractedXmlPath: string | null = null; + try { // Resolve the user's timezone — required for the cumulative // `stats:` day-key bucketing. Default to Europe/Berlin if the @@ -159,6 +215,7 @@ export async function handleAppleHealthImport( }); const unzip = extractExportXml(uploadPath); + extractedXmlPath = unzip.xmlPath; // Phase 2 + 3: parsing + upserting (the parser tracks both // phases internally via the onProgress hook). @@ -241,7 +298,23 @@ export async function handleAppleHealthImport( safeUnlink(unzip.xmlPath); safeUnlink(uploadPath); } catch (err) { - const reason = err instanceof Error ? err.message : String(err); + // v1.28.33 (issue #486) — a missing staging file is an operational + // condition, not a parse failure: `/tmp` is wiped on a container + // restart and a previous attempt unlinks the upload on its own + // failure path. Surface an honest, actionable reason instead of the + // raw `ENOENT: no such file or directory, open '/tmp/…'` string the + // status endpoint used to hand the UI. + const missingStagingFile = + typeof err === "object" && + err !== null && + "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" + : err instanceof Error + ? err.message + : String(err); await prisma.importJob.update({ where: { id: importJobId }, data: { @@ -250,6 +323,7 @@ export async function handleAppleHealthImport( completedAt: new Date(), }, }); + if (extractedXmlPath) safeUnlink(extractedXmlPath); safeUnlink(uploadPath); throw err; } diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index 6c458dff2..21d23ab4b 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -28,6 +28,7 @@ const mocks = vi.hoisted(() => ({ findFirst: vi.fn(), findMany: vi.fn(), findFirstMeasurement: vi.fn(), + txExecuteRawUnsafe: vi.fn(), bossSend: vi.fn(), getGlobalBossMock: vi.fn(), })); @@ -72,6 +73,7 @@ const { deleteMany, findFirst, findFirstMeasurement, + txExecuteRawUnsafe, bossSend, getGlobalBossMock, } = mocks; @@ -105,18 +107,18 @@ beforeEach(() => { findFirst.mockReset(); mocks.findMany.mockReset(); findFirstMeasurement.mockReset(); + txExecuteRawUnsafe.mockReset(); bossSend.mockReset(); getGlobalBossMock.mockReset(); // v1.4.38 — clear the per-userId in-flight map so a previous test's // resolved promise does not short-circuit the next test's recompute // assertion. _resetEnsureUserRollupsFreshInFlightForTests(); - // Default: $transaction supports both client shapes the populator - // uses — the batched array form (`[deleteMany(...), createMany(...)]` - // on the ≤CHUNK path) resolves the pre-built promises like the real - // client, and the interactive callback form (large path) invokes the - // callback with a tx proxy wired to the same delegate mocks so the - // assertions can observe the in-transaction calls. + // Default: $transaction invokes the interactive callback with a tx + // proxy wired to the same delegate mocks so the assertions can + // observe the in-transaction calls; the batched array form (legacy — + // no persist path uses it any more) resolves the pre-built promises + // like the real client. transaction.mockImplementation(async (arg: unknown) => { if (typeof arg === "function") { return (arg as (tx: unknown) => Promise)({ @@ -124,11 +126,14 @@ beforeEach(() => { deleteMany: mocks.deleteMany, createMany: mocks.createMany, }, + $executeRawUnsafe: mocks.txExecuteRawUnsafe, }); } return Promise.all(arg as unknown[]); }); - // v1.11.1 — persistRollupRows writes via createMany; default the count. + // v1.28.33 — persistRollupRows writes via a raw ON CONFLICT upsert; + // default one affected row per chunk call. + txExecuteRawUnsafe.mockResolvedValue(1); createMany.mockResolvedValue({ count: 1 }); }); @@ -194,24 +199,34 @@ describe("recomputeBucketsForMeasurement", () => { new Date("2026-05-10T14:30:00.000Z"), ); - // DAY pass — v1.11.1 delete-then-insert the day partition in one tx. + // DAY pass — v1.11.1 delete-then-upsert the day partition in one tx. expect(transaction).toHaveBeenCalledTimes(1); expect(deleteMany).toHaveBeenCalledTimes(1); expect(deleteMany.mock.calls[0][0].where.userId).toBe("user-1"); expect(deleteMany.mock.calls[0][0].where.granularity).toBe("DAY"); expect(deleteMany.mock.calls[0][0].where.type.in).toContain("WEIGHT"); - expect(createMany).toHaveBeenCalledTimes(1); - const created = createMany.mock.calls[0][0].data; - expect(created).toHaveLength(1); - expect(created[0].userId).toBe("user-1"); - expect(created[0].type).toBe("WEIGHT"); - expect(created[0].granularity).toBe("DAY"); - expect(created[0].source).toBe("MANUAL"); - expect(created[0].count).toBe(3); - expect(created[0].mean).toBe(82.5); + // v1.28.33 (issue #486) — the insert leg is a raw ON CONFLICT DO + // UPDATE upsert so a concurrent same-partition writer overwrites + // instead of raising (or being dropped). Param layout per row: + // [userId, type, granularity, bucketStart, source, count, mean, + // min, max, sum, sd, slope, r2, sumX, sumXy, sumXx, sumYy, + // computedAt]. + expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1); + const [sql, ...params] = txExecuteRawUnsafe.mock.calls[0]; + expect(sql).toContain( + "ON CONFLICT (user_id, type, granularity, bucket_start, source)", + ); + expect(sql).toContain("DO UPDATE SET"); + expect(params).toHaveLength(18); + expect(params[0]).toBe("user-1"); + expect(params[1]).toBe("WEIGHT"); + expect(params[2]).toBe("DAY"); + expect(params[4]).toBe("MANUAL"); + expect(params[5]).toBe(3); + expect(params[6]).toBe(82.5); // v1.4.39 W-SUM — sumValue flows through. Cumulative read paths // (steps, flights, distance, daylight, active-energy) consume it. - expect(created[0].sumValue).toBe(247.5); + expect(params[9]).toBe(247.5); // WEEK / MONTH / YEAR — three enqueues against the worker queue. expect(bossSend).toHaveBeenCalledTimes(3); @@ -223,13 +238,14 @@ describe("recomputeBucketsForMeasurement", () => { } }); - it("swallows a P2002 from a concurrent same-partition recompute", async () => { - // Two writes on the same (user, type, day) can interleave the - // delete-then-insert (no pg-boss singleton on the sync DAY hook), so - // the losing createMany trips the per-source unique index. The peer - // wrote the identical source-collapsed partition, so the rollup is - // already correct — persistRollupRows must treat the P2002 as benign - // rather than failing the recompute. + it("upserts every stat column so a racing recompute overwrites last-write-wins", async () => { + // v1.28.33 (issue #486) — two writes on the same (user, type, day) + // can interleave the delete-then-insert (no pg-boss singleton on + // the sync DAY hook). The pre-fix shape swallowed the loser's + // P2002 and DROPPED its freshly computed aggregate — leaving the + // bucket stale whenever the loser had seen newer measurements than + // the winner. The upsert must instead carry EVERY stat column in + // its DO UPDATE clause so the later recompute lands its values. queryRawUnsafe.mockResolvedValueOnce([ { type: "WEIGHT", @@ -247,23 +263,34 @@ describe("recomputeBucketsForMeasurement", () => { ]); getGlobalBossMock.mockReturnValue({ send: bossSend }); bossSend.mockResolvedValue("job-id"); - // The DAY transaction loses the race and throws a unique violation. - transaction.mockRejectedValueOnce({ code: "P2002" }); - await expect( - recomputeBucketsForMeasurement( - "user-1", - "WEIGHT", - new Date("2026-05-10T14:30:00.000Z"), - ), - ).resolves.not.toThrow(); + await recomputeBucketsForMeasurement( + "user-1", + "WEIGHT", + new Date("2026-05-10T14:30:00.000Z"), + ); - // The downstream WEEK/MONTH/YEAR enqueues still fire — the benign DAY - // race does not abort the rest of the recompute. - expect(bossSend).toHaveBeenCalledTimes(3); + const [sql] = txExecuteRawUnsafe.mock.calls[0]; + for (const column of [ + "count", + "mean", + "min_value", + "max_value", + "sum_value", + "sd", + "slope", + "r2", + "sum_x", + "sum_xy", + "sum_xx", + "sum_yy", + "computed_at", + ]) { + expect(sql).toMatch(new RegExp(`${column}\\s*= EXCLUDED.${column}`)); + } }); - it("rethrows a non-P2002 transaction error", async () => { + it("rethrows a transaction error", async () => { queryRawUnsafe.mockResolvedValueOnce([ { type: "WEIGHT", @@ -318,13 +345,13 @@ describe("recomputeBucketsForMeasurement", () => { new Date("2026-05-10T14:30:00.000Z"), ); - expect(createMany).toHaveBeenCalledTimes(1); - const arg = createMany.mock.calls[0][0].data[0]; - expect(arg.sumValue).toBe(12480); + expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1); + const [, ...params] = txExecuteRawUnsafe.mock.calls[0]; + expect(params[9]).toBe(12480); // mean × count algebraic equivalence — the writer carries SUM // directly because the aggregator computed it once. Asserting // both gives parity coverage for the consumer-side fallback. - expect(arg.mean * arg.count).toBe(12480); + expect((params[6] as number) * (params[5] as number)).toBe(12480); }); it("passes through null sum_value when the aggregator returns NULL", async () => { @@ -355,8 +382,8 @@ describe("recomputeBucketsForMeasurement", () => { new Date("2026-05-10T14:30:00.000Z"), ); - const arg = createMany.mock.calls[0][0].data[0]; - expect(arg.sumValue).toBeNull(); + const [, ...params] = txExecuteRawUnsafe.mock.calls[0]; + expect(params[9]).toBeNull(); }); it("deletes the DAY row when the post-mutation aggregate is empty", async () => { @@ -374,6 +401,7 @@ describe("recomputeBucketsForMeasurement", () => { expect(deleteMany.mock.calls[0][0].where.userId).toBe("user-1"); expect(deleteMany.mock.calls[0][0].where.granularity).toBe("DAY"); expect(upsert).not.toHaveBeenCalled(); + expect(txExecuteRawUnsafe).not.toHaveBeenCalled(); }); }); @@ -395,10 +423,13 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { })); } - it("wraps the delete + every insert chunk in ONE interactive transaction with skipDuplicates", async () => { - // 1200 rows → large path (> 500), 3 insert chunks (500/500/200). + it("wraps the delete + every upsert chunk in ONE interactive transaction", async () => { + // 1200 rows → 3 upsert chunks (500/500/200). queryRawUnsafe.mockResolvedValueOnce(syntheticRows(1200)); - createMany.mockResolvedValue({ count: 400 }); + txExecuteRawUnsafe + .mockResolvedValueOnce(500) + .mockResolvedValueOnce(500) + .mockResolvedValueOnce(200); const result = await recomputeUserRollups("user-1", { granularities: ["DAY"], @@ -407,8 +438,8 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { // ONE interactive transaction — the callback form, not the batched // array. The committed-delete-without-inserts window of the // pre-v1.16.10 shape (delete commits, a concurrent write-hook - // P2002 aborts the chunk loop, the types lose ALL buckets) cannot - // exist when both legs share the transaction. + // unique violation aborts the chunk loop, the types lose ALL + // buckets) cannot exist when both legs share the transaction. expect(transaction).toHaveBeenCalledTimes(1); expect(typeof transaction.mock.calls[0][0]).toBe("function"); // A generous interactive timeout — the chunk loop on a multi-year @@ -422,13 +453,16 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { expect(deleteMany.mock.calls[0][0].where.userId).toBe("user-1"); expect(deleteMany.mock.calls[0][0].where.granularity).toBe("DAY"); - // 3 chunks, every one tolerant of a mid-transaction hook insert. - expect(createMany).toHaveBeenCalledTimes(3); - expect(createMany.mock.calls[0][0].data).toHaveLength(500); - expect(createMany.mock.calls[1][0].data).toHaveLength(500); - expect(createMany.mock.calls[2][0].data).toHaveLength(200); - for (const call of createMany.mock.calls) { - expect(call[0].skipDuplicates).toBe(true); + // 3 chunks of 18 params per row, every one an ON CONFLICT DO UPDATE + // upsert — tolerant of a mid-transaction hook insert AND of a + // concurrent fold on the same partition (v1.28.33, issue #486). + expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(3); + expect(txExecuteRawUnsafe.mock.calls[0]).toHaveLength(1 + 500 * 18); + expect(txExecuteRawUnsafe.mock.calls[1]).toHaveLength(1 + 500 * 18); + expect(txExecuteRawUnsafe.mock.calls[2]).toHaveLength(1 + 200 * 18); + for (const call of txExecuteRawUnsafe.mock.calls) { + expect(call[0]).toContain("ON CONFLICT"); + expect(call[0]).toContain("DO UPDATE SET"); } expect(result.rowsUpserted).toBe(1200); @@ -436,8 +470,8 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { it("a failing chunk rejects through the transaction so the delete rolls back with it", async () => { queryRawUnsafe.mockResolvedValueOnce(syntheticRows(700)); - createMany - .mockResolvedValueOnce({ count: 500 }) + txExecuteRawUnsafe + .mockResolvedValueOnce(500) .mockRejectedValueOnce(new Error("connection reset")); // The error must bubble (pg-boss retries the fold); the structural @@ -453,14 +487,19 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { expect(deleteMany).toHaveBeenCalledTimes(1); }); - it("the ≤500-row path keeps the batched two-statement transaction", async () => { + it("the ≤500-row path runs through the same interactive upsert transaction", async () => { queryRawUnsafe.mockResolvedValueOnce(syntheticRows(2)); - createMany.mockResolvedValue({ count: 2 }); + txExecuteRawUnsafe.mockResolvedValueOnce(2); - await recomputeUserRollups("user-1", { granularities: ["DAY"] }); + const result = await recomputeUserRollups("user-1", { + granularities: ["DAY"], + }); expect(transaction).toHaveBeenCalledTimes(1); - expect(Array.isArray(transaction.mock.calls[0][0])).toBe(true); + expect(typeof transaction.mock.calls[0][0]).toBe("function"); + expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1); + expect(txExecuteRawUnsafe.mock.calls[0]).toHaveLength(1 + 2 * 18); + expect(result.rowsUpserted).toBe(2); }); }); diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 7747182bf..f3211c928 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -30,7 +30,6 @@ import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { annotate } from "@/lib/logging/context"; -import { isP2002 } from "@/lib/prisma-errors"; import { collapseRollupRowsBySource, loadUserSourcePriority, @@ -38,7 +37,6 @@ import { } from "@/lib/rollups/measurement-read"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import type { - MeasurementSource, MeasurementType, RollupGranularity, } from "@/generated/prisma/client"; @@ -504,9 +502,10 @@ async function runRollupAggregate(input: { /** * UPSERT the aggregate rows into `measurement_rollups`. Returns the - * number of rows touched. Uses a single chunked `$executeRaw` insert - * with `ON CONFLICT DO UPDATE` so the round-trip count is bounded - * regardless of fan-out width. + * number of rows touched. Uses a chunked raw insert with + * `ON CONFLICT DO UPDATE` so the round-trip count is bounded + * regardless of fan-out width AND a concurrent writer on the same + * partition can never raise a unique violation. */ async function persistRollupRows( userId: string, @@ -515,7 +514,7 @@ async function persistRollupRows( ): Promise { if (rows.length === 0) return 0; - // v1.11.1 — rows are now minted per source. Delete-then-insert the affected + // v1.11.1 — rows are now minted per source. Delete-then-upsert the affected // (type, bucket) partitions across ALL sources before writing, so a source // that disappeared from a bucket (the user disconnected a provider, or // deleted the last reading from one device) does not strand a stale @@ -537,85 +536,49 @@ async function persistRollupRows( bucketStart: { gte: minBucket, lte: maxBucket }, }; - const toData = (row: RollupRow) => ({ - userId, - type: row.type as MeasurementType, - granularity, - bucketStart: row.bucket_start, - source: row.source as MeasurementSource, - count: Number(row.count), - mean: row.mean ?? 0, - minValue: row.min_value ?? 0, - maxValue: row.max_value ?? 0, - // v1.4.39 W-SUM — populate for every type. Cumulative read paths (steps, - // flights, distance, daylight, active-energy) consume this column - // directly; spot metrics carry it for free because the underlying SUM - // aggregator runs in the same query as AVG / MIN / MAX. - sumValue: row.sum_value ?? null, - sd: row.sd, - slope: row.slope, - r2: row.r2, - // v1.20.0 F6 — persist the regression accumulators alongside the - // existing stats; they ride the same aggregate (no extra round-trip). - sumX: row.sum_x ?? null, - sumXy: row.sum_xy ?? null, - sumXx: row.sum_xx ?? null, - sumYy: row.sum_yy ?? null, - computedAt: new Date(), - }); + // v1.28.33 (issue #486) — the insert leg is a raw + // `INSERT … ON CONFLICT (pk) DO UPDATE` instead of `createMany`. The + // synchronous DAY hook carries no pg-boss singleton, so a write hook + // can race the end-of-import full fold (or a second hook) on the same + // (user, type, day, source) partition: under READ COMMITTED the + // loser's delete leg cannot see the winner's still-uncommitted rows, + // and its plain insert then tripped `measurement_rollups_pkey` + // (Postgres 23505). The previous shape swallowed that P2002 and + // DROPPED the loser's freshly computed aggregate — wrong whenever the + // loser had already seen newer measurements than the winner: the + // bucket kept the stale aggregate until the next unrelated write. + // `DO UPDATE` makes the write last-write-wins: every recompute lands + // its own freshly computed values, so the later writer overwrites the + // earlier one and no interleaving can raise or drop. The rows are + // sorted on (type, source, bucket) so two concurrent multi-row + // upserts take their row locks in the same order and cannot deadlock. + const ordered = [...rows].sort((a, b) => + a.type !== b.type + ? a.type < b.type + ? -1 + : 1 + : a.source !== b.source + ? a.source < b.source + ? -1 + : 1 + : a.bucket_start.getTime() - b.bucket_start.getTime(), + ); // Chunk to keep the parameter count below Postgres' 65k cap and the // round-trip count bounded for large backfills. const CHUNK = 500; - // Common path — the entire write fits one chunk (every incremental hook - // and most bucket-span recomputes). Atomic delete-then-insert so a - // concurrent read never sees a half-rewritten partition. - // - // The synchronous DAY hook carries no pg-boss singleton, so two writes - // landing on the same (user, type, day) can interleave: T1.delete, - // T2.delete, T1.create, T2.create — and T2's createMany then trips the - // (userId, type, granularity, bucketStart, source) unique index. Treat - // that P2002 as benign: the peer rewrote the identical source-collapsed - // partition (the aggregate is a pure function of the same underlying - // rows), so the rollup is already correct. Annotate and return 0 written - // rather than bubbling a recompute failure. - if (rows.length <= CHUNK) { - try { - const [, created] = await prisma.$transaction([ - prisma.measurementRollup.deleteMany({ where: deleteWhere }), - prisma.measurementRollup.createMany({ data: rows.map(toData) }), - ]); - return created.count; - } catch (err) { - if (isP2002(err)) { - annotate({ - meta: { - rollup_persist_race: true, - rollup_persist_granularity: granularity, - rollup_persist_types: types.length, - }, - }); - return 0; - } - throw err; - } - } - - // Large backfill — delete the partition range and insert the chunks - // inside ONE interactive transaction. Pre-v1.16.10 the deleteMany - // committed standalone before the chunk loop, and the chunks ran - // without the P2002 tolerance the ≤CHUNK path has: a concurrent - // write-hook insert landing mid-loop made a createMany throw, the - // loop aborted, the delete stayed committed — and every type in the - // window lost ALL its buckets until the next race-free fold (the - // coverage probe then cycled covered/uncovered as the boot backfill - // re-raced the hooks). The transaction keeps the delete invisible - // until every chunk landed; `skipDuplicates` (ON CONFLICT DO - // NOTHING) additionally absorbs a hook row that commits between two - // chunks — READ COMMITTED lets a later statement see it — without - // aborting the batch. The hook's row is the per-(type, day) loser of - // the same aggregate, so skipping the fold's duplicate is benign. + // Delete the partition range and upsert the chunks inside ONE + // interactive transaction. Pre-v1.16.10 the deleteMany committed + // standalone before the chunk loop: a concurrent write-hook insert + // landing mid-loop aborted the loop, the delete stayed committed — + // and every type in the window lost ALL its buckets until the next + // race-free fold (the coverage probe then cycled covered/uncovered + // as the boot backfill re-raced the hooks). The transaction keeps + // the delete invisible until every chunk landed; the ON CONFLICT + // upsert absorbs a hook row that commits between two chunks — READ + // COMMITTED lets a later statement see it — without aborting the + // batch. // // Transaction sizing: one call covers ONE granularity, so the worst // case is the full 5-year DAY fold of a multi-source power user — @@ -624,19 +587,16 @@ async function persistRollupRows( // transaction was considered and rejected: it would re-open the // committed-delete gap between types sharing the bucket range. The // 120 s timeout (Prisma default: 5 s) covers the chunk round-trips - // with headroom; the fold runs on the serial backfill worker, so a - // long-held transaction cannot pile up behind itself. + // with headroom; the incremental hook path is a single sub-500-row + // chunk and finishes far inside it. let touched = 0; await prisma.$transaction( async (tx) => { await tx.measurementRollup.deleteMany({ where: deleteWhere }); - for (let i = 0; i < rows.length; i += CHUNK) { - const slice = rows.slice(i, i + CHUNK); - const res = await tx.measurementRollup.createMany({ - data: slice.map(toData), - skipDuplicates: true, - }); - touched += res.count; + for (let i = 0; i < ordered.length; i += CHUNK) { + const slice = ordered.slice(i, i + CHUNK); + const { sql, params } = buildRollupUpsert(userId, granularity, slice); + touched += await tx.$executeRawUnsafe(sql, ...params); } }, { maxWait: 10_000, timeout: 120_000 }, @@ -644,6 +604,80 @@ async function persistRollupRows( return touched; } +/** + * Build the chunked `INSERT … ON CONFLICT DO UPDATE` statement for one + * slice of aggregate rows. Every value travels as a positional `$N` + * parameter — nothing from the rows is spliced into the SQL text; the + * only literals are the fixed column list and the enum type names. + */ +function buildRollupUpsert( + userId: string, + granularity: RollupGranularity, + rows: RollupRow[], +): { sql: string; params: unknown[] } { + const params: unknown[] = []; + const tuples: string[] = []; + const computedAt = new Date(); + for (const row of rows) { + const base = params.length; + params.push( + userId, + row.type, + granularity, + row.bucket_start, + row.source, + Number(row.count), + row.mean ?? 0, + row.min_value ?? 0, + row.max_value ?? 0, + // v1.4.39 W-SUM — populate for every type. Cumulative read paths + // (steps, flights, distance, daylight, active-energy) consume this + // column directly; spot metrics carry it for free because the + // underlying SUM aggregator runs in the same query as AVG/MIN/MAX. + row.sum_value ?? null, + row.sd, + row.slope, + row.r2, + // v1.20.0 F6 — persist the regression accumulators alongside the + // existing stats; they ride the same aggregate (no extra round-trip). + row.sum_x ?? null, + row.sum_xy ?? null, + row.sum_xx ?? null, + row.sum_yy ?? null, + computedAt, + ); + tuples.push( + `($${base + 1}, $${base + 2}::"measurement_type", $${base + 3}::"measurement_rollup_granularity", ` + + `$${base + 4}::timestamptz, $${base + 5}::"measurement_source", $${base + 6}::int, ` + + `$${base + 7}, $${base + 8}, $${base + 9}, $${base + 10}, $${base + 11}, $${base + 12}, ` + + `$${base + 13}, $${base + 14}, $${base + 15}, $${base + 16}, $${base + 17}, $${base + 18}::timestamptz)`, + ); + } + const sql = ` + INSERT INTO measurement_rollups + (user_id, type, granularity, bucket_start, source, count, mean, + min_value, max_value, sum_value, sd, slope, r2, + sum_x, sum_xy, sum_xx, sum_yy, computed_at) + VALUES ${tuples.join(", ")} + ON CONFLICT (user_id, type, granularity, bucket_start, source) + DO UPDATE SET + count = EXCLUDED.count, + mean = EXCLUDED.mean, + min_value = EXCLUDED.min_value, + max_value = EXCLUDED.max_value, + sum_value = EXCLUDED.sum_value, + sd = EXCLUDED.sd, + slope = EXCLUDED.slope, + r2 = EXCLUDED.r2, + sum_x = EXCLUDED.sum_x, + sum_xy = EXCLUDED.sum_xy, + sum_xx = EXCLUDED.sum_xx, + sum_yy = EXCLUDED.sum_yy, + computed_at = EXCLUDED.computed_at + `; + return { sql, params }; +} + /** * Span of the bucket containing `measuredAt` at the given granularity. * Postgres `date_trunc` is the canonical truncator; we mirror its diff --git a/tests/integration/apple-health-import-worker-failures.test.ts b/tests/integration/apple-health-import-worker-failures.test.ts new file mode 100644 index 000000000..ff5e2ebc2 --- /dev/null +++ b/tests/integration/apple-health-import-worker-failures.test.ts @@ -0,0 +1,221 @@ +/** + * v1.28.33 — failure-path contracts for the Apple Health import worker + * (issue #486). + * + * A cumulative re-import that outlives the queue's expiration window is + * redelivered by pg-boss after the first run already consumed (and + * unlinked) the staged upload. The redelivery used to re-open the + * deleted `/tmp/healthlog-apple-health-import-*.bin`, fail with a raw + * ENOENT, and OVERWRITE the first run's terminal state — masking the + * real outcome (a genuine failure reason, or even a completed import) + * behind "ENOENT: no such file or directory". Pinned here against the + * real Postgres: + * + * 1. A delivery for an ImportJob already in a terminal state + * (`done` / `failed`) is ignored — status, failureReason and + * result stay exactly as the first run left them. + * + * 2. A missing staging file on a live (non-terminal) job surfaces an + * honest operator-facing reason instead of the raw ENOENT string. + * + * 3. A genuine parse failure records ITS message as the failure + * reason and deterministically removes the staged upload. + */ +import { mkdtempSync, writeFileSync, existsSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job } from "pg-boss"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { + handleAppleHealthImport, + _setWorkerPrismaForTests, + type AppleHealthImportPayload, +} from "@/lib/jobs/apple-health-import-worker"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +// No pg-boss attached — the end-of-import rollup fold's enqueue helpers +// are silent no-ops without a boss. +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +const scratchDir = mkdtempSync(join(tmpdir(), "hl-import-worker-test-")); + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + // Route the worker's Prisma singleton onto the shared testcontainer + // client so the handler does not open a second pool. + _setWorkerPrismaForTests(getPrismaClient()); +}); + +afterAll(() => { + _setWorkerPrismaForTests(null); + rmSync(scratchDir, { recursive: true, force: true }); +}); + +async function createUser(username: string) { + return getPrismaClient().user.create({ + data: { + username, + email: `${username}@example.test`, + role: "USER", + }, + }); +} + +function bossJob( + id: string, + payload: AppleHealthImportPayload, +): Job { + return { + id, + name: "apple-health-import", + data: payload, + } as Job; +} + +describe("apple health import worker — failure paths (issue #486)", () => { + it("ignores a redelivery for a job already failed — the original reason survives", async () => { + const prisma = getPrismaClient(); + const user = await createUser("import-terminal-failed"); + const originalReason = "database write failed during upsert"; + await prisma.importJob.create({ + data: { + userId: user.id, + pgBossJobId: "boss-terminal-failed", + status: "failed", + failureReason: originalReason, + uploadBytes: 1234, + completedAt: new Date("2026-04-22T10:00:00.000Z"), + }, + }); + + // The staged upload is long gone — exactly the redelivery scenario. + await expect( + handleAppleHealthImport( + bossJob("boss-terminal-failed", { + userId: user.id, + uploadPath: join(scratchDir, "gone.bin"), + uploadBytes: 1234, + enqueuedAt: new Date().toISOString(), + }), + ), + ).resolves.toBeUndefined(); + + const row = await prisma.importJob.findUnique({ + where: { pgBossJobId: "boss-terminal-failed" }, + }); + expect(row!.status).toBe("failed"); + expect(row!.failureReason).toBe(originalReason); + expect(row!.failureReason).not.toContain("ENOENT"); + }); + + it("ignores a redelivery for a job already done — the completed import stays done", async () => { + const prisma = getPrismaClient(); + const user = await createUser("import-terminal-done"); + await prisma.importJob.create({ + data: { + userId: user.id, + pgBossJobId: "boss-terminal-done", + status: "done", + uploadBytes: 5678, + completedAt: new Date("2026-04-22T10:00:00.000Z"), + }, + }); + + await expect( + handleAppleHealthImport( + bossJob("boss-terminal-done", { + userId: user.id, + uploadPath: join(scratchDir, "gone-too.bin"), + uploadBytes: 5678, + enqueuedAt: new Date().toISOString(), + }), + ), + ).resolves.toBeUndefined(); + + const row = await prisma.importJob.findUnique({ + where: { pgBossJobId: "boss-terminal-done" }, + }); + expect(row!.status).toBe("done"); + expect(row!.failureReason).toBeNull(); + }); + + it("a missing staging file on a live job records an honest reason, not a raw ENOENT", async () => { + const prisma = getPrismaClient(); + const user = await createUser("import-missing-staging"); + await prisma.importJob.create({ + data: { + userId: user.id, + pgBossJobId: "boss-missing-staging", + status: "queued", + uploadBytes: 999, + }, + }); + + await expect( + handleAppleHealthImport( + bossJob("boss-missing-staging", { + userId: user.id, + uploadPath: join(scratchDir, "never-existed.bin"), + uploadBytes: 999, + enqueuedAt: new Date().toISOString(), + }), + ), + ).rejects.toThrow(); + + const row = await prisma.importJob.findUnique({ + where: { pgBossJobId: "boss-missing-staging" }, + }); + expect(row!.status).toBe("failed"); + expect(row!.failureReason).not.toContain("ENOENT"); + expect(row!.failureReason).toContain("no longer available"); + }); + + it("a genuine parse failure keeps ITS message and removes the staged upload", async () => { + const prisma = getPrismaClient(); + const user = await createUser("import-real-failure"); + await prisma.importJob.create({ + data: { + userId: user.id, + pgBossJobId: "boss-real-failure", + status: "queued", + uploadBytes: 64, + }, + }); + + // Garbage bytes — not a ZIP archive; the extractor throws its own + // descriptive error, which must reach the ImportJob row verbatim. + const uploadPath = join(scratchDir, "garbage.bin"); + writeFileSync(uploadPath, Buffer.from("this is not a zip archive")); + + let thrownMessage = ""; + try { + await handleAppleHealthImport( + bossJob("boss-real-failure", { + userId: user.id, + uploadPath, + uploadBytes: 64, + enqueuedAt: new Date().toISOString(), + }), + ); + expect.unreachable("handler must rethrow the parse failure"); + } catch (err) { + thrownMessage = err instanceof Error ? err.message : String(err); + } + + const row = await prisma.importJob.findUnique({ + where: { pgBossJobId: "boss-real-failure" }, + }); + expect(row!.status).toBe("failed"); + expect(row!.failureReason).toBe(thrownMessage.slice(0, 1000)); + expect(row!.failureReason).not.toContain("ENOENT"); + // Deterministic cleanup: the staged upload is gone after the failure. + expect(existsSync(uploadPath)).toBe(false); + }); +}); diff --git a/tests/integration/rollup-concurrent-recompute.test.ts b/tests/integration/rollup-concurrent-recompute.test.ts new file mode 100644 index 000000000..552ccdab3 --- /dev/null +++ b/tests/integration/rollup-concurrent-recompute.test.ts @@ -0,0 +1,293 @@ +/** + * v1.28.33 — concurrency contract for the measurement-rollup persist + * path (issue #486). + * + * A cumulative Apple Health `export.zip` re-import runs the end-of-import + * full fold while the regular write hooks (iOS batch sync, dashboard + * warm-up) keep firing their own per-day recomputes. Two writers on the + * same `(userId, type, granularity, bucketStart, source)` partition used + * to interleave delete-then-insert, and the losing INSERT tripped the + * `measurement_rollups_pkey` composite primary key (Postgres 23505). The + * previous shape swallowed the P2002 and DROPPED the loser's freshly + * computed aggregate — which is exactly wrong when the loser saw newer + * measurements than the winner: the bucket stayed stale until the next + * unrelated write. + * + * The persist path now writes through `INSERT … ON CONFLICT … DO UPDATE` + * so the later recompute overwrites with its freshly computed values + * (last-write-wins) instead of raising or dropping. Pinned here against + * the real Postgres: + * + * 1. **Racing peer holds the partition uncommitted.** A peer + * transaction inserts a stale aggregate for the bucket and holds + * its transaction open; the real write-hook recompute (which sees + * one more measurement) runs concurrently and blocks on the PK. + * When the peer commits, the hook's fresher aggregate must land — + * not vanish into a swallowed unique violation. + * + * 2. **Two full user folds racing.** `recomputeUserRollups` twice in + * parallel neither throws nor duplicates rows. + * + * 3. **N-way write-hook fan-out.** Concurrent + * `recomputeBucketsForMeasurement` calls for the same day all + * resolve and converge on the correct aggregate. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { + recomputeBucketsForMeasurement, + recomputeUserRollups, +} from "@/lib/rollups/measurement-rollups"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +// No pg-boss in the integration harness — the WEEK/MONTH/YEAR enqueue +// inside the write hook is a silent no-op when no boss is attached. +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +/** + * Poll `pg_stat_activity` until some backend is lock-waiting on a + * `measurement_rollups` statement (the hook's upsert queued behind the + * peer's uncommitted insert). Bounded so a scheduling hiccup degrades to + * a plain sleep rather than a hang. + */ +async function waitForRollupLockWaiter(timeoutMs = 5_000): Promise { + const prisma = getPrismaClient(); + const startedAt = Date.now(); + while (Date.now() - startedAt < timeoutMs) { + const rows = await prisma.$queryRaw>` + SELECT count(*)::bigint AS n + FROM pg_stat_activity + WHERE wait_event_type = 'Lock' + AND query ILIKE '%measurement_rollups%' + `; + if (Number(rows[0]?.n ?? 0) > 0) return true; + await new Promise((r) => setTimeout(r, 50)); + } + return false; +} + +describe("measurement rollups — concurrent recompute (issue #486)", () => { + it("a recompute racing a committed peer write keeps the fresher aggregate", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-race-user", + email: "rollup-race-user@example.test", + role: "USER", + }, + }); + const dayStart = new Date("2026-04-22T00:00:00.000Z"); + const m1At = new Date("2026-04-22T07:00:00.000Z"); + const m2At = new Date("2026-04-22T19:00:00.000Z"); + + await prisma.measurement.create({ + data: { + userId: user.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "APPLE_HEALTH", + measuredAt: m1At, + }, + }); + + // Peer transaction: mimics the end-of-import fold writing the bucket + // from an aggregate snapshot taken BEFORE the second measurement + // landed (count 1, mean 80). It holds its transaction open so the + // real write hook below queues behind the uncommitted PK row — + // the exact interleaving the reporter's Postgres log shows. + let releasePeer!: () => void; + const gate = new Promise((r) => { + releasePeer = r; + }); + let peerInserted!: () => void; + const peerReady = new Promise((r) => { + peerInserted = r; + }); + const peer = prisma.$transaction( + async (tx) => { + await tx.measurementRollup.deleteMany({ + where: { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: dayStart, + }, + }); + await tx.measurementRollup.createMany({ + data: [ + { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: dayStart, + source: "APPLE_HEALTH", + count: 1, + mean: 80, + minValue: 80, + maxValue: 80, + sumValue: 80, + sd: 0, + slope: null, + r2: null, + computedAt: new Date(), + }, + ], + }); + peerInserted(); + await gate; + }, + { maxWait: 10_000, timeout: 30_000 }, + ); + await peerReady; + + // Second measurement lands (committed) — the write hook's aggregate + // sees BOTH rows: count 2, mean 90. + await prisma.measurement.create({ + data: { + userId: user.id, + type: "WEIGHT", + value: 100, + unit: "kg", + source: "APPLE_HEALTH", + measuredAt: m2At, + }, + }); + + const hook = recomputeBucketsForMeasurement(user.id, "WEIGHT", m1At); + + // Let the hook's persist reach the PK wait behind the peer's + // uncommitted row, then commit the peer with its STALE aggregate. + await waitForRollupLockWaiter(); + releasePeer(); + await peer; + await hook; + + const row = await prisma.measurementRollup.findUnique({ + where: { + userId_type_granularity_bucketStart_source: { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: dayStart, + source: "APPLE_HEALTH", + }, + }, + }); + // The hook computed from the full day (both measurements). Its + // recompute must survive the race — a swallowed unique violation + // would leave the peer's stale count-1 / mean-80 aggregate behind. + expect(row).not.toBeNull(); + expect(row!.count).toBe(2); + expect(row!.mean).toBeCloseTo(90, 6); + expect(row!.minValue).toBeCloseTo(80, 6); + expect(row!.maxValue).toBeCloseTo(100, 6); + expect(row!.sumValue).toBeCloseTo(180, 6); + }); + + it("two full user folds racing neither throw nor duplicate rows", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-fold-race-user", + email: "rollup-fold-race-user@example.test", + role: "USER", + }, + }); + const days = 30; + const data = Array.from({ length: days }, (_, i) => ({ + userId: user.id, + type: "WEIGHT" as const, + value: 80 + (i % 5), + unit: "kg", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(Date.UTC(2026, 3, 1 + i, 8, 0, 0)), + })); + await prisma.measurement.createMany({ data }); + + const window = { + from: new Date("2026-03-31T00:00:00.000Z"), + to: new Date("2026-05-02T00:00:00.000Z"), + }; + // The end-of-import fold racing the boot backfill for the same user. + await expect( + Promise.all([ + recomputeUserRollups(user.id, window), + recomputeUserRollups(user.id, window), + ]), + ).resolves.toBeDefined(); + + const rows = await prisma.measurementRollup.findMany({ + where: { userId: user.id, type: "WEIGHT", granularity: "DAY" }, + }); + expect(rows).toHaveLength(days); + for (const row of rows) { + expect(row.count).toBe(1); + } + }); + + it("N concurrent write-hook recomputes for the same day all resolve", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-fanout-user", + email: "rollup-fanout-user@example.test", + role: "USER", + }, + }); + const measuredAt = new Date("2026-04-22T12:00:00.000Z"); + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-04-22T07:00:00.000Z"), + }, + { + userId: user.id, + type: "WEIGHT", + value: 84, + unit: "kg", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-04-22T20:00:00.000Z"), + }, + ], + }); + + await expect( + Promise.all( + Array.from({ length: 8 }, () => + recomputeBucketsForMeasurement(user.id, "WEIGHT", measuredAt), + ), + ), + ).resolves.toBeDefined(); + + const row = await prisma.measurementRollup.findUnique({ + where: { + userId_type_granularity_bucketStart_source: { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: new Date("2026-04-22T00:00:00.000Z"), + source: "APPLE_HEALTH", + }, + }, + }); + expect(row).not.toBeNull(); + expect(row!.count).toBe(2); + expect(row!.mean).toBeCloseTo(82, 6); + }); +});