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;