From d8f623aea31567e0e9a2beb47186bd6423ac1757 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 3 Jun 2026 09:16:38 +0200 Subject: [PATCH] fix(measurements): close residual P2002 in the dense intra-day fold Resolve the canonical daily row deterministically: prefer the row already carrying the target stats: externalId before falling back to the row on the canonical local-noon instant, with a stable orderBy tiebreaker. Adopting a sibling and stamping the target externalId onto it could otherwise collide on the (userId, type, source, externalId) index with the real canonical row. Re-anchor the adopted row to local-noon only when that slot is free, so an externalId row at a tz/DST-shifted instant cannot collide on the measured_at composite during adopt. Guard the create against a concurrent insert on the same canonical composite: catch P2002 outside the transaction (a caught error inside leaves the tx in the aborted state) and retry the fold once, which then resolves and adopts the winning row instead of creating a duplicate. Add a real-Postgres case covering a row already on the canonical instant under the target stats: externalId plus a fresh out-of-window batch: the fold adopts in place, no P2002, one canonical row with the correctly folded mean. --- .../measurements/dense-intraday-retention.ts | 212 ++++++++++++------ .../dense-intraday-retention.test.ts | 83 +++++++ 2 files changed, 225 insertions(+), 70 deletions(-) diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index 070ca8fdd..fd072a1f4 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -77,6 +77,20 @@ export const DENSE_INTRADAY_RETENTION_DAYS = 14; /** The daily-stats externalId prefix marks an already-collapsed row. */ const DAILY_STATS_PREFIX = "stats:"; +/** + * True for a Prisma unique-constraint violation (P2002). Mirrors the + * `consolidate-legacy-steps.ts` precedent verbatim so the two drains + * classify the same error the same way. + */ +function isUniqueConstraintViolation(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code?: unknown }).code === "P2002" + ); +} + export interface DenseIntradayRetentionSummary { dryRun: boolean; totals: { @@ -181,87 +195,145 @@ export async function runDenseIntradayRetention( sourceRowIds, }): Promise => { const unit = dayRows[0]?.unit ?? "unknown"; - let removed = 0; - await pc.$transaction(async (tx) => { - // Fold the day to one canonical daily-mean row at local-noon. - // - // Unlike `consolidate-daily-mean` (whose types never share the - // canonical local-noon instant with another path), the dense-tier - // types HRV / PULSE are dense enough that a per-sample row can land - // exactly on the canonical timestamp, and a previously-minted daily - // row can already occupy it. An externalId-keyed upsert misses the - // lookup (the colliding row carries a different externalId) and the - // INSERT then violates the OTHER unique index - // `(userId, type, measuredAt, source, sleepStage)` with P2002. - // - // The canonical daily row is identified by BOTH unique keys at once: - // its measuredAt is always `canonicalTimestamp` (local-noon) and its - // externalId is always the `stats:` id, so the two indexes point at - // the same logical row. Resolve the row by the measured_at composite - // — the one the collision is on (`sleepStage` is NULL for these - // continuous types, matched via the NULLS-NOT-DISTINCT index) — and - // either adopt it in place or create it. Built field-by-field (no - // spread) per the no-mass-assignment convention. - const existing = await tx.measurement.findFirst({ - where: { - userId, - type, - source: "APPLE_HEALTH", - measuredAt: canonicalTimestamp, - sleepStage: null, - }, - select: { id: true }, - }); - let canonicalRowId: string; - if (existing) { - // Adopt the row already sitting on the canonical instant: refresh - // its value, stamp the `stats:` externalId, and un-tombstone it. - // This coexists with a pre-existing daily row (a prior fold, a - // manual daily entry, or a per-sample row that happened at local - // noon) instead of colliding with it. - await tx.measurement.update({ - where: { id: existing.id }, - data: { - value: reducedValue, - unit, - externalId, - deletedAt: null, - }, + // Fold the day to one canonical daily-mean row at local-noon. + // + // Unlike `consolidate-daily-mean` (whose types never share the + // canonical local-noon instant with another path), the dense-tier + // types HRV / PULSE are dense enough that a per-sample row can land + // exactly on the canonical timestamp, and a previously-minted daily + // row can already occupy it. TWO unique indexes can collide: + // A) `(userId, type, source, externalId)` — the row already + // carrying the target `stats:` externalId. + // B) `(userId, type, measuredAt, source, sleepStage)` (NULLS NOT + // DISTINCT) — any row sitting on the canonical local-noon + // instant, which may carry a DIFFERENT externalId (a sibling + // consolidation path, a manual daily entry, or a per-sample row + // that happened at local noon). + // + // Determinism (VECTOR 1): resolve the canonical row by index A FIRST — + // the row that already holds the target `stats:` externalId IS the + // canonical daily row by construction. Only when no such row exists do + // we fall back to the index-B row physically occupying the canonical + // instant. A stable `orderBy: id` pins the pick. Adopting the wrong + // sibling and stamping the target externalId onto it would otherwise + // collide with the real `stats:` row on index A. + // + // Concurrency (VECTOR 2): the resolve→create is a check-then-act not + // serialised by the unique index under READ COMMITTED. If a concurrent + // writer wins the canonical slot between the lookup and the INSERT, the + // `create` throws P2002 and Postgres aborts the whole transaction. We + // therefore catch P2002 OUTSIDE the transaction (a caught error inside + // would leave the tx in the aborted `25P02` state) and retry the fold + // once: on the retry the deterministic lookup finds the now-existing + // row and ADOPTS it, so the create path never fires twice. Mirrors the + // `consolidate-legacy-steps.ts` P2002 classification, but recovers in + // place rather than skipping, because the colliding row IS the + // canonical daily row this fold owns. Built field-by-field (no spread) + // per the no-mass-assignment convention. + const foldOnce = async (): Promise => { + let removed = 0; + await pc.$transaction(async (tx) => { + // Index-A row: already carries the target `stats:` externalId. By + // construction this drain only ever mints `stats:` rows at the + // canonical instant, so in the common case this row IS the + // canonical daily row already sitting at local-noon. + const eidRow = await tx.measurement.findFirst({ + where: { userId, type, source: "APPLE_HEALTH", externalId }, + select: { id: true, measuredAt: true }, + orderBy: { id: "asc" }, }); - canonicalRowId = existing.id; - } else { - const created = await tx.measurement.create({ - data: { + // Index-B row: occupies the canonical local-noon instant + // (`sleepStage` is NULL for these continuous types, matched via the + // NULLS-NOT-DISTINCT index). At most one such row can exist. + const slotRow = await tx.measurement.findFirst({ + where: { userId, type, - value: reducedValue, - unit, source: "APPLE_HEALTH", measuredAt: canonicalTimestamp, - externalId, + sleepStage: null, }, select: { id: true }, + orderBy: { id: "asc" }, }); - canonicalRowId = created.id; - } - // Soft-delete the out-of-window per-sample rows in the same - // transaction — tombstone, never hard-delete; they remain as an - // audit trail and drop off the live read + this pass's re-run - // discovery. EXCLUDE the canonical row itself: a per-sample row that - // happened to fall on the canonical local-noon instant is the row we - // just adopted as the daily mean, so tombstoning it would erase the - // fold. - const del = await tx.measurement.updateMany({ - where: { - id: { in: sourceRowIds, not: canonicalRowId }, - deletedAt: null, - }, - data: { deletedAt: new Date() }, + // Prefer the externalId-carrying row (index A) — adopting a sibling + // and stamping the target externalId onto it would collide with it. + const adoptTarget = eidRow ?? slotRow; + + let canonicalRowId: string; + if (adoptTarget) { + // Pin the adopted row to local-noon, but only when the canonical + // slot is free or already this row — a DIFFERENT row occupying + // the slot (a per-sample row that fell on local-noon, now being + // folded) would otherwise collide on index B. In that rare case + // (only reachable on a tz/DST shift between mints) the row keeps + // its existing instant; the `stats:` externalId is the identity, + // measuredAt is secondary. + const slotIsFreeForTarget = + slotRow === null || slotRow.id === adoptTarget.id; + // Adopt: refresh value, stamp the `stats:` externalId, optionally + // re-anchor to local-noon, and un-tombstone. This coexists with a + // pre-existing daily row instead of colliding with it. + await tx.measurement.update({ + where: { id: adoptTarget.id }, + data: { + value: reducedValue, + unit, + externalId, + deletedAt: null, + ...(slotIsFreeForTarget + ? { measuredAt: canonicalTimestamp } + : {}), + }, + }); + canonicalRowId = adoptTarget.id; + } else { + const created = await tx.measurement.create({ + data: { + userId, + type, + value: reducedValue, + unit, + source: "APPLE_HEALTH", + measuredAt: canonicalTimestamp, + externalId, + }, + select: { id: true }, + }); + canonicalRowId = created.id; + } + + // Soft-delete the out-of-window per-sample rows in the same + // transaction — tombstone, never hard-delete; they remain as an + // audit trail and drop off the live read + this pass's re-run + // discovery. EXCLUDE the canonical row itself: a per-sample row + // that happened to fall on the canonical local-noon instant is the + // row we just adopted as the daily mean, so tombstoning it would + // erase the fold. + const del = await tx.measurement.updateMany({ + where: { + id: { in: sourceRowIds, not: canonicalRowId }, + deletedAt: null, + }, + data: { deletedAt: new Date() }, + }); + removed = del.count; }); - removed = del.count; - }); + return removed; + }; + + let removed: number; + try { + removed = await foldOnce(); + } catch (err) { + if (!isUniqueConstraintViolation(err)) throw err; + // A concurrent writer won the canonical slot mid-transaction. Retry + // once: the deterministic lookup now resolves the winning row and + // adopts it, so this pass cannot create a duplicate. + removed = await foldOnce(); + } // Recompute the affected (user, type, day) rollup buckets after the // fold commits — the rollup tier reads only `deleted_at IS NULL`, so diff --git a/tests/integration/dense-intraday-retention.test.ts b/tests/integration/dense-intraday-retention.test.ts index 1301f1989..8a78be8c0 100644 --- a/tests/integration/dense-intraday-retention.test.ts +++ b/tests/integration/dense-intraday-retention.test.ts @@ -163,6 +163,89 @@ describe("runDenseIntradayRetention (real Postgres)", () => { expect(tombstoned).toHaveLength(2); }); + it("adopts the row already carrying the target stats externalId on the canonical instant (VECTOR 1)", async () => { + const prisma = getPrismaClient(); + // BOTH-present collision (VECTOR 1). A prior fold already minted the + // canonical daily row carrying the TARGET `stats:` externalId and it + // sits on the canonical local-noon instant. A fresh batch of + // out-of-window per-sample rows then arrives for the same day. The fold + // must adopt the EXISTING canonical row in place — refreshing its value + // and re-anchoring it — never INSERT a second row at local-noon (that + // would collide on the measured_at composite) and never stamp the target + // externalId onto a sibling (that would collide on the externalId + // composite with this very row). The pre-fix `findFirst` had no + // externalId discrimination and no deterministic `orderBy`, so it could + // pick the wrong row and trip P2002 on the externalId index. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 999, // stale — the fold overwrites it with the day's mean + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: CANONICAL, + externalId: HRV_STATS_ID, // the TARGET canonical externalId + }, + }); + // Out-of-window per-sample rows that fold into the canonical row. + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 40, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T06:00:00.000Z"), + externalId: "hk-hrv-1", + }, + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 60, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T18:00:00.000Z"), + externalId: "hk-hrv-2", + }, + ], + }); + + // Must not throw P2002. + const summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + expect(summary.totals.daysConsolidated).toBe(1); + + const live = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + }); + // Exactly ONE live canonical row remains — the row that already carried + // the target `stats:` externalId, adopted in place and folded. + expect(live).toHaveLength(1); + expect(live[0].externalId).toBe(HRV_STATS_ID); + expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); + // MEAN of the two folded per-sample values (40, 60) = 50, overwriting + // the stale 999. + expect(live[0].value).toBeCloseTo(50, 6); + + // Both scanned per-sample rows are tombstoned. + const tombstoned = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: { not: null }, + }, + }); + expect(tombstoned).toHaveLength(2); + }); + it("does not tombstone a per-sample row that happens to fall on the canonical instant", async () => { const prisma = getPrismaClient(); // One per-sample row sits exactly on the canonical local-noon instant;