chore(release): v1.28.33 — re-importing a cumulative Apple Health export works again

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.

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 },

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 },

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*\)/,
+      );
+    }
+  });
+});

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;
   }

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<unknown>)({
@@ -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);
   });
 });
 

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<number> {
   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