fix(rollups): treat a concurrent same-partition P2002 as benign

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) <noreply@anthropic.com>
This commit is contained in:
Marc-André BombeckandClaude Opus 4.8 committed 2026-06-04 13:01:54 +02:00
1 parent 7923d75afe
commit 07c444f938
2 files changed
+106 -5

No files matched your search

@@ -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.
+38 -5
View File
@@ -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.