diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index ed1f997c7..06ae8ae4f 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/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(); diff --git a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts index 33e5f8d36..7577aa940 100644 --- a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts +++ b/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) { 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) { + 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) => + cb(tx), + ), + } as unknown as PrismaClient, + upsert, + updateMany, + update, + txFindFirst, + }; + } + + /** Slot probe answering `occupant` at noon and "free" everywhere else. */ + function occupySlot( + txFindFirst: ReturnType, + 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); + }); +}); diff --git a/src/lib/measurements/consolidate-daily-mean.ts b/src/lib/measurements/consolidate-daily-mean.ts index 555649734..b2d999e1c 100644 --- a/src/lib/measurements/consolidate-daily-mean.ts +++ b/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 { + 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({ + const { usersScanned, daysFailed } = await runConsolidation({ 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; diff --git a/src/lib/measurements/consolidation-base.ts b/src/lib/measurements/consolidation-base.ts index bd06f6d1b..94068600b 100644 --- a/src/lib/measurements/consolidation-base.ts +++ b/src/lib/measurements/consolidation-base.ts @@ -231,6 +231,22 @@ export interface ConsolidationParams { }) => 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( params: ConsolidationParams, -): 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( 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 }; }