From 07c444f9387a38bc3407289a0ac460890c178778 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 4 Jun 2026 13:01:54 +0200 Subject: [PATCH] fix(rollups): treat a concurrent same-partition P2002 as benign MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The synchronous DAY rollup hook carries no pg-boss singleton, so two measurement writes landing on the same (user, type, day) can interleave their delete-then-insert and the losing createMany trips the per-source unique index. 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. Catch the P2002 in the single-chunk transactional path, annotate it, and return rather than failing the recompute. Any other Prisma error still bubbles. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../__tests__/measurement-rollups.test.ts | 68 +++++++++++++++++++ src/lib/rollups/measurement-rollups.ts | 43 ++++++++++-- 2 files changed, 106 insertions(+), 5 deletions(-) diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index f96c0018e..23e86443d 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -212,6 +212,74 @@ 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. + queryRawUnsafe.mockResolvedValueOnce([ + { + type: "WEIGHT", + source: "MANUAL", + bucket_start: new Date("2026-05-10T00:00:00.000Z"), + count: BigInt(1), + mean: 80.0, + min_value: 80.0, + max_value: 80.0, + sum_value: 80.0, + sd: 0, + slope: 0, + r2: 0, + }, + ]); + 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(); + + // The downstream WEEK/MONTH/YEAR enqueues still fire — the benign DAY + // race does not abort the rest of the recompute. + expect(bossSend).toHaveBeenCalledTimes(3); + }); + + it("rethrows a non-P2002 transaction error", async () => { + queryRawUnsafe.mockResolvedValueOnce([ + { + type: "WEIGHT", + source: "MANUAL", + bucket_start: new Date("2026-05-10T00:00:00.000Z"), + count: BigInt(1), + mean: 80.0, + min_value: 80.0, + max_value: 80.0, + sum_value: 80.0, + sd: 0, + slope: 0, + r2: 0, + }, + ]); + getGlobalBossMock.mockReturnValue({ send: bossSend }); + transaction.mockRejectedValueOnce({ code: "P2010" }); + + await expect( + recomputeBucketsForMeasurement( + "user-1", + "WEIGHT", + new Date("2026-05-10T14:30:00.000Z"), + ), + ).rejects.toMatchObject({ code: "P2010" }); + }); + it("writes sum_value for cumulative ACTIVITY_STEPS buckets", async () => { // 5 step samples in a day summing to 12480: emulates an iOS // batch of HealthKit slices reaching the rollup aggregator. diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index be973b5aa..6369483b6 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -437,6 +437,16 @@ async function runRollupAggregate(input: { * with `ON CONFLICT DO UPDATE` so the round-trip count is bounded * regardless of fan-out width. */ +/** Narrow a thrown Prisma error to the P2002 unique-constraint code. */ +function isP2002(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code: unknown }).code === "P2002" + ); +} + async function persistRollupRows( userId: string, granularity: RollupGranularity, @@ -494,12 +504,35 @@ async function persistRollupRows( // 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) { - const [, created] = await prisma.$transaction([ - prisma.measurementRollup.deleteMany({ where: deleteWhere }), - prisma.measurementRollup.createMany({ data: rows.map(toData) }), - ]); - return created.count; + 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 once, then insert in chunks.