fix(measurements): keep the mean-consolidation pass alive past unique collisions

The nightly mean-consolidation upsert targets the externalId arbiter, but
the measurement table carries a second full unique on (user, type,
measured_at, source, sleep_stage) that also spans tombstones. Any row
parked exactly on the canonical local-noon instant made the insert throw
P2002, and because the global pass had no per-bucket fence the first
poisoned day aborted consolidation for every remaining user and type.

writeDay now probes the canonical slot inside the transaction and
resolves the collision deterministically: rows the pass owns (tombstones,
per-sample rows about to fold) shift to the next free second, foreign
live rows keep their instant and the mean row yields instead. The runner
gains an optional per-bucket error fence so one bad day is logged,
counted as daysFailed, and skipped.

src/lib/jobs/reminder-worker.ts

@@ -3330,7 +3330,7 @@ export async function startReminderWorker() {
           });
           workerLog(
             "info",
-            `[mean-consolidation] triggeredAt=${job.data.triggeredAt} usersScanned=${meanSummary.totals.usersScanned} daysConsolidated=${meanSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${meanSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${meanSummary.totals.dailyRowsUpserted}`,
+            `[mean-consolidation] triggeredAt=${job.data.triggeredAt} usersScanned=${meanSummary.totals.usersScanned} daysConsolidated=${meanSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${meanSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${meanSummary.totals.dailyRowsUpserted} daysFailed=${meanSummary.totals.daysFailed}`,
           );
         } catch (err) {
           recordError();

src/lib/measurements/__tests__/consolidate-daily-mean.test.ts

@@ -12,6 +12,7 @@ import {
   consolidateDailyMean,
   meanBucketValue,
 } from "../consolidate-daily-mean";
+import { canonicalDailyTimestamp } from "../consolidation-tz";
 import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
 
@@ -85,12 +86,16 @@ describe("consolidateDailyMean — drain flow (mocked Prisma)", () => {
   function buildPrismaMock(rowsByType: Record<string, unknown[]>) {
     const upsert = vi.fn().mockResolvedValue({});
     const updateMany = vi.fn().mockResolvedValue({ count: 0 });
+    const update = vi.fn().mockResolvedValue({});
+    // Canonical-slot probe inside the write transaction — defaults to an
+    // unoccupied slot (the pre-collision-fix happy path).
+    const txFindFirst = vi.fn().mockResolvedValue(null);
     const findManyMeasurement = vi.fn(
       async (args: { where: { type: string } }) =>
         rowsByType[args.where.type] ?? [],
     );
     const tx = {
-      measurement: { upsert, updateMany },
+      measurement: { upsert, updateMany, update, findFirst: txFindFirst },
     };
     return {
       mock: {
@@ -108,6 +113,8 @@ describe("consolidateDailyMean — drain flow (mocked Prisma)", () => {
       },
       upsert,
       updateMany,
+      update,
+      txFindFirst,
       findManyMeasurement,
     };
   }
@@ -238,3 +245,183 @@ describe("consolidateDailyMean — drain flow (mocked Prisma)", () => {
     expect(call.where.measuredAt?.lt).toBeInstanceOf(Date);
   });
 });
+
+describe("consolidateDailyMean — canonical-slot collision (second unique index)", () => {
+  // Local-noon for 2026-05-01 in Europe/Berlin — the instant the mint
+  // targets and any colliding row occupies.
+  const noon = canonicalDailyTimestamp("2026-05-01", "Europe/Berlin");
+  const targetExternalId =
+    "stats:HKQuantityTypeIdentifierWalkingSpeed:2026-05-01";
+
+  function buildPrismaMock(rowsByType: Record<string, unknown[]>) {
+    const upsert = vi.fn().mockResolvedValue({});
+    const updateMany = vi.fn().mockResolvedValue({ count: 0 });
+    const update = vi.fn().mockResolvedValue({});
+    const txFindFirst = vi.fn().mockResolvedValue(null);
+    const findManyMeasurement = vi.fn(
+      async (args: { where: { type: string } }) =>
+        rowsByType[args.where.type] ?? [],
+    );
+    const tx = {
+      measurement: { upsert, updateMany, update, findFirst: txFindFirst },
+    };
+    return {
+      mock: {
+        user: {
+          findMany: vi
+            .fn()
+            .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]),
+        },
+        measurement: { findMany: findManyMeasurement },
+        $transaction: vi.fn(async (cb: (t: unknown) => Promise<unknown>) =>
+          cb(tx),
+        ),
+      } as unknown as PrismaClient,
+      upsert,
+      updateMany,
+      update,
+      txFindFirst,
+    };
+  }
+
+  /** Slot probe answering `occupant` at noon and "free" everywhere else. */
+  function occupySlot(
+    txFindFirst: ReturnType<typeof vi.fn>,
+    occupant: { id: string; externalId: string | null; deletedAt: Date | null },
+  ) {
+    txFindFirst.mockImplementation(
+      async (args: { where: { measuredAt: Date } }) =>
+        args.where.measuredAt.getTime() === noon.getTime() ? occupant : null,
+    );
+  }
+
+  it("steps a tombstone occupying the canonical instant aside and mints at noon (no P2002)", async () => {
+    const { mock, upsert, update, txFindFirst } = buildPrismaMock({
+      WALKING_SPEED: [
+        row("a", 1.0, "2026-05-01T08:00:00.000Z"),
+        row("b", 1.4, "2026-05-01T09:00:00.000Z"),
+      ],
+    });
+    occupySlot(txFindFirst, {
+      id: "tomb-1",
+      externalId: "uuid-foreign-sample",
+      deletedAt: new Date("2026-05-02T00:00:00.000Z"),
+    });
+
+    const summary = await consolidateDailyMean(mock, { log: () => {} });
+
+    // The tombstone is shifted off the canonical instant (+1s)…
+    expect(update).toHaveBeenCalledTimes(1);
+    const updArg = update.mock.calls[0]?.[0] as {
+      where: { id: string };
+      data: { measuredAt: Date };
+    };
+    expect(updArg.where.id).toBe("tomb-1");
+    expect(updArg.data.measuredAt.getTime()).toBe(noon.getTime() + 1000);
+
+    // …and the mean row keeps the local-noon anchor.
+    const upsertArg = upsert.mock.calls[0]?.[0] as {
+      create: { measuredAt: Date };
+      update: { measuredAt: Date };
+    };
+    expect(upsertArg.create.measuredAt.getTime()).toBe(noon.getTime());
+    expect(summary.totals.daysConsolidated).toBe(1);
+    expect(summary.totals.daysFailed).toBe(0);
+  });
+
+  it("shifts a per-sample row of the consolidation set off the canonical instant", async () => {
+    const { mock, update, txFindFirst } = buildPrismaMock({
+      WALKING_SPEED: [
+        // "a" sits exactly on local-noon and is part of the folded set.
+        row("a", 1.0, noon.toISOString(), "uuid-a"),
+        row("b", 1.4, "2026-05-01T09:00:00.000Z", "uuid-b"),
+      ],
+    });
+    occupySlot(txFindFirst, {
+      id: "a",
+      externalId: "uuid-a",
+      deletedAt: null,
+    });
+
+    const summary = await consolidateDailyMean(mock, { log: () => {} });
+
+    // The live sample is folded anyway, so it yields the instant.
+    expect(update).toHaveBeenCalledTimes(1);
+    const updArg = update.mock.calls[0]?.[0] as { where: { id: string } };
+    expect(updArg.where.id).toBe("a");
+    expect(summary.totals.daysFailed).toBe(0);
+  });
+
+  it("yields the mean row's instant when a live foreign row holds the canonical slot", async () => {
+    const { mock, upsert, update, txFindFirst } = buildPrismaMock({
+      WALKING_SPEED: [
+        row("a", 1.0, "2026-05-01T08:00:00.000Z"),
+        row("b", 1.4, "2026-05-01T09:00:00.000Z"),
+      ],
+    });
+    // Live row outside the consolidation set — not ours to move.
+    occupySlot(txFindFirst, {
+      id: "foreign-live",
+      externalId: "uuid-foreign-live",
+      deletedAt: null,
+    });
+
+    const summary = await consolidateDailyMean(mock, { log: () => {} });
+
+    expect(update).not.toHaveBeenCalled();
+    const upsertArg = upsert.mock.calls[0]?.[0] as {
+      create: { measuredAt: Date };
+    };
+    expect(upsertArg.create.measuredAt.getTime()).toBe(noon.getTime() + 1000);
+    expect(summary.totals.daysFailed).toBe(0);
+  });
+
+  it("re-run path: a slot row already carrying the target externalId is updated in place", async () => {
+    const { mock, upsert, update, txFindFirst } = buildPrismaMock({
+      WALKING_SPEED: [row("late", 1.2, "2026-05-01T18:00:00.000Z")],
+    });
+    occupySlot(txFindFirst, {
+      id: "mean-1",
+      externalId: targetExternalId,
+      deletedAt: null,
+    });
+
+    await consolidateDailyMean(mock, { log: () => {} });
+
+    // No shift, no yield — the upsert's update branch owns the row.
+    expect(update).not.toHaveBeenCalled();
+    const upsertArg = upsert.mock.calls[0]?.[0] as {
+      create: { measuredAt: Date };
+    };
+    expect(upsertArg.create.measuredAt.getTime()).toBe(noon.getTime());
+    // Only the single slot probe ran — no free-instant search.
+    expect(txFindFirst).toHaveBeenCalledTimes(1);
+  });
+
+  it("a failing day bucket does not abort the pass — later buckets still consolidate", async () => {
+    const { mock, upsert } = buildPrismaMock({
+      WALKING_SPEED: [row("a", 1.0, "2026-05-01T08:00:00.000Z")],
+      WALKING_ASYMMETRY: [
+        row("c", 2.0, "2026-05-01T08:00:00.000Z", null, "%", "WALKING_ASYMMETRY"),
+      ],
+    });
+    // First bucket's mint trips the unique index; the second succeeds.
+    upsert.mockRejectedValueOnce(
+      Object.assign(new Error("unique constraint failed"), { code: "P2002" }),
+    );
+    const lines: string[] = [];
+
+    const summary = await consolidateDailyMean(mock, {
+      log: (line) => lines.push(line),
+    });
+
+    expect(summary.totals.daysFailed).toBe(1);
+    expect(summary.totals.daysConsolidated).toBe(1);
+    // The surviving bucket reached the rollup recompute.
+    expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(1);
+    // The failed day is reported with user/type/day context.
+    expect(
+      lines.some((l) => l.includes("failed") && l.includes("user-1")),
+    ).toBe(true);
+  });
+});

src/lib/measurements/consolidate-daily-mean.ts

@@ -45,7 +45,11 @@
  * production standalone image strips `tsx`. Modelled on the
  * `step-consolidation` boot-time converging-backfill pattern.
  */
-import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+import type {
+  MeasurementType,
+  Prisma,
+  PrismaClient,
+} from "@/generated/prisma/client";
 
 import {
   dailyStatsExternalId,
@@ -95,6 +99,12 @@ export interface MeanConsolidationSummary {
     daysConsolidated: number;
     perSampleRowsSoftDeleted: number;
     dailyRowsUpserted: number;
+    /**
+     * Day buckets whose write failed and was stepped over. A failed day
+     * keeps its per-sample rows live, so the next nightly run retries it;
+     * every other bucket of the pass still consolidates.
+     */
+    daysFailed: number;
   };
 }
 
@@ -135,6 +145,51 @@ export function bucketMeanRows(
   return bucketRowsByDay(rows, tz, DAILY_STATS_PREFIX);
 }
 
+/**
+ * Find the first instant at/after `base + 1s` that the full unique
+ * `(userId, type, measuredAt, source, sleepStage)` — NULLS NOT DISTINCT,
+ * so tombstones occupy it too — does not already hold for this
+ * (user, type, APPLE_HEALTH, sleepStage NULL) tuple. A slot held by the
+ * row carrying `selfExternalId` counts as free: that row is the one the
+ * caller's upsert is about to update in place, so re-using its instant
+ * converges across re-runs instead of drifting one second per night.
+ * Bounded probe; the per-day failure boundary in `runConsolidation`
+ * absorbs the pathological exhaustion case.
+ */
+async function nextFreeInstant(
+  tx: Prisma.TransactionClient,
+  input: {
+    userId: string;
+    type: MeasurementType;
+    base: Date;
+    selfExternalId: string | null;
+  },
+): Promise<Date> {
+  for (let offsetSeconds = 1; offsetSeconds <= 60; offsetSeconds++) {
+    const candidate = new Date(input.base.getTime() + offsetSeconds * 1000);
+    const occupant = await tx.measurement.findFirst({
+      where: {
+        userId: input.userId,
+        type: input.type,
+        source: "APPLE_HEALTH",
+        measuredAt: candidate,
+        sleepStage: null,
+      },
+      select: { externalId: true },
+    });
+    if (
+      occupant === null ||
+      (input.selfExternalId !== null &&
+        occupant.externalId === input.selfExternalId)
+    ) {
+      return candidate;
+    }
+  }
+  throw new Error(
+    `no free measuredAt slot within 60s after ${input.base.toISOString()}`,
+  );
+}
+
 /**
  * Run the daily-mean consolidation. Idempotent — re-invocation after a
  * successful pass reports zero days because the per-sample rows are
@@ -156,10 +211,11 @@ export async function consolidateDailyMean(
       daysConsolidated: 0,
       perSampleRowsSoftDeleted: 0,
       dailyRowsUpserted: 0,
+      daysFailed: 0,
     },
   };
 
-  const { usersScanned } = await runConsolidation<MeasurementType>({
+  const { usersScanned, daysFailed } = await runConsolidation<MeasurementType>({
     prismaClient,
     options,
     types: HIGH_FREQUENCY_MEAN_TYPES,
@@ -205,7 +261,71 @@ export async function consolidateDailyMean(
       const unit = dayRows[0]?.unit ?? "unknown";
       let removed = 0;
       await pc.$transaction(async (tx) => {
-        // Mint / refresh the canonical daily-mean row first. The unique
+        // The upsert below arbiters on (userId, type, source, externalId),
+        // but the Measurement model carries a SECOND full unique —
+        // (userId, type, measuredAt, source, sleepStage), NULLS NOT
+        // DISTINCT, covering tombstones (schema.prisma, migration 0055).
+        // Any row sitting exactly on the canonical local-noon instant with
+        // a DIFFERENT externalId — live or soft-deleted — trips that index
+        // on the upsert's create path (P2002) and rolls the whole day
+        // back. Resolve the slot deterministically before minting:
+        //   1. Slot row carries the TARGET externalId → it IS the
+        //      canonical daily row (re-run path); the upsert updates it in
+        //      place, no conflict.
+        //   2. Slot row is a tombstone, or a per-sample row this very pass
+        //      is about to soft-delete → it has no live claim on the
+        //      instant; step it aside to the next free second so the
+        //      canonical row keeps the local-noon anchor.
+        //   3. Slot row is a live foreign row outside the consolidation
+        //      set (it stays user-visible and is not ours to move) → it
+        //      keeps its instant; the mean row yields and mints on the
+        //      next free second after local-noon instead.
+        const slotRow = await tx.measurement.findFirst({
+          where: {
+            userId,
+            type,
+            source: "APPLE_HEALTH",
+            measuredAt: canonicalTimestamp,
+            sleepStage: null,
+          },
+          select: { id: true, externalId: true, deletedAt: true },
+        });
+
+        let mintAt = canonicalTimestamp;
+        if (slotRow !== null && slotRow.externalId !== externalId) {
+          const yieldsToMean =
+            slotRow.deletedAt !== null || sourceRowIds.includes(slotRow.id);
+          if (yieldsToMean) {
+            // `selfExternalId: null` — the shifted row must land on a
+            // strictly free instant; sharing one with the canonical row
+            // (e.g. a mean row that yielded on an earlier run and still
+            // sits at +1s) would re-trip the unique.
+            await tx.measurement.update({
+              where: { id: slotRow.id },
+              data: {
+                measuredAt: await nextFreeInstant(tx, {
+                  userId,
+                  type,
+                  base: canonicalTimestamp,
+                  selfExternalId: null,
+                }),
+              },
+            });
+          } else {
+            // `selfExternalId: externalId` — an instant already held by
+            // the existing mean row (yielded on an earlier run) counts as
+            // free, so the upsert re-targets it instead of drifting one
+            // second further every night.
+            mintAt = await nextFreeInstant(tx, {
+              userId,
+              type,
+              base: canonicalTimestamp,
+              selfExternalId: externalId,
+            });
+          }
+        }
+
+        // Mint / refresh the canonical daily-mean row. The unique
         // index (userId, type, source, externalId) makes the upsert
         // idempotent across re-runs. Built field-by-field (no spread)
         // per the no-mass-assignment convention.
@@ -224,12 +344,12 @@ export async function consolidateDailyMean(
             value: reducedValue,
             unit,
             source: "APPLE_HEALTH",
-            measuredAt: canonicalTimestamp,
+            measuredAt: mintAt,
             externalId,
           },
           update: {
             value: reducedValue,
-            measuredAt: canonicalTimestamp,
+            measuredAt: mintAt,
             deletedAt: null,
           },
         });
@@ -285,12 +405,24 @@ export async function consolidateDailyMean(
         `[mean-consolidation] user=${userId} tz=${tz}${dryRun ? " (dry-run)" : ""}`,
       );
     },
+    // Per-day failure boundary: one poisoned day (e.g. a residual
+    // unique-index collision the slot resolution could not absorb) must
+    // not abort the global nightly pass — every later user / type / day
+    // still consolidates, and the failed day's per-sample rows stay live
+    // for the next run to retry.
+    onBucketError: ({ userId, type, dateKey, error }) => {
+      const reason = error instanceof Error ? error.message : String(error);
+      log(
+        `[mean-consolidation] user=${userId} type=${type} day=${dateKey} failed — ${reason}; continuing with the next bucket`,
+      );
+    },
   });
 
   summary.totals.usersScanned = usersScanned;
+  summary.totals.daysFailed = daysFailed;
 
   log(
-    `[mean-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`,
+    `[mean-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFailed=${summary.totals.daysFailed}${options.dryRun ? " (dry-run)" : ""}`,
   );
 
   return summary;

src/lib/measurements/consolidation-base.ts

@@ -231,6 +231,22 @@ export interface ConsolidationParams<TType extends MeasurementType> {
   }) => void;
   /** Per-user log line, COMPLETE — optional. */
   onUserComplete?: (input: { userId: string; tz: string; dryRun: boolean }) => void;
+  /**
+   * Per-day failure boundary — optional. When supplied, an error thrown
+   * while reducing / writing / recording ONE day bucket is reported here
+   * and the walk continues with the next bucket, so a single poisoned day
+   * (e.g. a unique-index collision on the mint) can no longer abort the
+   * whole global pass and strand every later user / type / day. When
+   * omitted, errors propagate exactly as before — drains whose `writeDay`
+   * classifies its own errors (and relies on the rethrow reaching pg-boss
+   * for a retry) keep their established semantics.
+   */
+  onBucketError?: (input: {
+    userId: string;
+    type: TType;
+    dateKey: string;
+    error: unknown;
+  }) => void;
 }
 
 const DEFAULT_SCAN_SELECT: Prisma.MeasurementSelect = {
@@ -245,7 +261,9 @@ const DEFAULT_SCAN_SELECT: Prisma.MeasurementSelect = {
  * Drive one consolidation pass. Walks `users → types → days`, scans live
  * source rows inside the grace window, buckets them, reduces each day,
  * and delegates the per-day mint + delete to `writeDay`. Returns the
- * number of users scanned so the caller can seed its summary totals.
+ * number of users scanned (so the caller can seed its summary totals)
+ * plus the number of day buckets absorbed by `onBucketError` (0 when the
+ * boundary is not supplied).
  *
  * Idempotency, the grace-window cutoff, and the "skip already-collapsed
  * rows" predicate are all owned here; the divergent reducer / delete /
@@ -253,11 +271,12 @@ const DEFAULT_SCAN_SELECT: Prisma.MeasurementSelect = {
  */
 export async function runConsolidation<TType extends MeasurementType>(
   params: ConsolidationParams<TType>,
-): Promise<{ usersScanned: number; dryRun: boolean }> {
+): Promise<{ usersScanned: number; dryRun: boolean; daysFailed: number }> {
   const { prismaClient, options } = params;
   const dryRun = options.dryRun ?? false;
   const cutoffAt = resolveCutoffInstant(options.cutoffHours);
   const scanSelect = params.scanSelect ?? DEFAULT_SCAN_SELECT;
+  let daysFailed = 0;
 
   const users = await loadConsolidationUsers(prismaClient, options.userId);
 
@@ -296,55 +315,68 @@ export async function runConsolidation<TType extends MeasurementType>(
       for (const [dateKey, dayRows] of byDay) {
         if (dayRows.length === 0) continue;
 
-        const reducedValue = params.reduce(dayRows);
-        const canonicalTimestamp = canonicalDailyTimestamp(dateKey, tz);
-        const externalId = params.dailyStatsExternalId(hkIdentifier, dateKey);
+        try {
+          const reducedValue = params.reduce(dayRows);
+          const canonicalTimestamp = canonicalDailyTimestamp(dateKey, tz);
+          const externalId = params.dailyStatsExternalId(hkIdentifier, dateKey);
 
-        const shouldMint = params.onBucket
-          ? await params.onBucket({
+          const shouldMint = params.onBucket
+            ? await params.onBucket({
+                prismaClient,
+                userId: user.id,
+                type,
+                dateKey,
+                dayRows,
+                reducedValue,
+                canonicalTimestamp,
+                externalId,
+              })
+            : true;
+
+          let outcome: DayWriteOutcome | null = null;
+          if (!dryRun) {
+            const sourceRowIds = dayRows.map((r) => r.id);
+            outcome = await params.writeDay({
               prismaClient,
               userId: user.id,
               type,
-              dateKey,
-              dayRows,
-              reducedValue,
-              canonicalTimestamp,
               externalId,
-            })
-          : true;
+              canonicalTimestamp,
+              reducedValue,
+              dayRows,
+              sourceRowIds,
+              shouldMint,
+            });
+          }
 
-        let outcome: DayWriteOutcome | null = null;
-        if (!dryRun) {
-          const sourceRowIds = dayRows.map((r) => r.id);
-          outcome = await params.writeDay({
-            prismaClient,
+          params.recordBucket({
             userId: user.id,
             type,
-            externalId,
-            canonicalTimestamp,
-            reducedValue,
+            dateKey,
             dayRows,
-            sourceRowIds,
+            reducedValue,
+            canonicalTimestamp,
+            externalId,
             shouldMint,
+            outcome,
+          });
+        } catch (err) {
+          // No boundary supplied → preserve the historical abort-the-run
+          // behaviour for the drains that classify errors in `writeDay`.
+          if (!params.onBucketError) throw err;
+          daysFailed += 1;
+          params.onBucketError({
+            userId: user.id,
+            type,
+            dateKey,
+            error: err,
           });
         }
-
-        params.recordBucket({
-          userId: user.id,
-          type,
-          dateKey,
-          dayRows,
-          reducedValue,
-          canonicalTimestamp,
-          externalId,
-          shouldMint,
-          outcome,
-        });
       }
     }
 
     params.onUserComplete?.({ userId: user.id, tz, dryRun });
   }
 
-  return { usersScanned: users.length, dryRun };
+  return { usersScanned: users.length, dryRun, daysFailed };
 }