From ad62fe229fe986305e2bd684ffeeeab71eec7d31 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 14 May 2026 17:01:07 +0200 Subject: [PATCH] =?UTF-8?q?fix(api):=20correct=20batch-ingest=20race=20rec?= =?UTF-8?q?onciliation=20=E2=80=94=20accurate=20inserted/duplicate=20count?= =?UTF-8?q?s=20under=20contention?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous race-reconciliation block compared post-insert rows in the DB against the rows we attempted to insert, downgrading any "inserted-but-not-stored" entries to "duplicate". Under standard Postgres semantics that branch never fires: `skipDuplicates` absorbs duplicate-key conflicts but the row is still present in the table (written by the other batch that won the race). So every attempted row was "stored" and the loop ran as a no-op, leaving the per-entry envelope inconsistent with the aggregate counts when two batches collide. The v1.5 iOS sync cursor depends on the envelope sums matching the aggregate counts to checkpoint correctly. Pragmatic fix: trust `createMany.count` for the aggregate `insertedCount` and downgrade exactly `toInsert.length - insertedCount` per-entry "inserted" statuses to "duplicate". The post-fix loop no longer touches `insertedCount` (it was already correct from createMany) so the count cannot drift negative. Order doesn't matter for the client — both statuses checkpoint past the row and the DB ends up with the same single-copy state. Concurrent-write integration test asserts the three invariants the iOS sync cursor relies on: per-entry sums equal aggregate counts, aggregate counts stay non-negative, and the DB row count matches the combined `inserted` total. --- src/app/api/measurements/batch/route.ts | 53 ++++++------ tests/integration/measurements-batch.test.ts | 87 ++++++++++++++++++++ 2 files changed, 111 insertions(+), 29 deletions(-) diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 5b8603cba..f1050a371 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -267,44 +267,39 @@ async function postBatch(request: NextRequest): Promise { } }); - // If `skipDuplicates` quietly absorbed a row (race with another - // batch), reconcile the per-entry status so the inserted + - // duplicate counts still equal `prepared.length`. + // v1.4.25 W10 reconcile (senior-dev H-1): the previous "stored + // vs not-stored" check was an effective no-op. Under standard + // Postgres semantics, `skipDuplicates` absorbs duplicate-key + // conflicts but the row is STILL present in the table (written + // by the other batch that won the race). So every row we + // attempted is in the DB after the call, and the + // `!stored.has(...)` branch never fired — leaving the aggregate + // `inserted` / `duplicate` counts inconsistent with the per- + // entry statuses under contention. + // + // Pragmatic fix: trust the `createMany` return count. The DB + // round-trip cannot distinguish a row this request wrote from + // a row the racing request wrote (the unique index sees both + // as the same key), so we cannot identify the SPECIFIC raced + // rows. What we CAN do is preserve count integrity for the + // iOS sync cursor: `insertedCount` is already the truth (it + // came from `createMany.count`); we only need to downgrade + // enough per-entry "inserted" statuses to "duplicate" so the + // per-entry envelope sums match. Order doesn't matter — the + // client checkpoints past both statuses, and the DB state for + // either outcome is identical (the row is now stored, + // single-copy). const racedDuplicates = toInsert.length - insertedCount; if (racedDuplicates > 0) { - // We can't tell *which* rows raced from the createMany return - // value — recheck the DB and downgrade matching `inserted` - // rows to `duplicate`. This is a rare path so the extra - // round-trip is acceptable. - const recheck = await prisma.measurement.findMany({ - where: { - userId: user.id, - source: "APPLE_HEALTH", - OR: toInsert.map((row) => ({ - type: row.type, - externalId: row.externalId as string, - })), - }, - select: { type: true, externalId: true }, - }); - const stored = new Set( - recheck.map((row) => `${row.type}::${row.externalId}`), - ); - // Anything `prepared` flagged as `inserted` but isn't in - // `stored` was never written and was raced by another batch — - // surface as duplicate. let downgraded = 0; for (const p of prepared) { - if ( - results[p.index]?.status === "inserted" && - !stored.has(`${p.row.type}::${p.row.externalId}`) - ) { + if (downgraded >= racedDuplicates) break; + if (results[p.index]?.status === "inserted") { results[p.index] = { index: p.index, status: "duplicate" }; duplicateCount += 1; downgraded += 1; } } - insertedCount -= downgraded; } } } diff --git a/tests/integration/measurements-batch.test.ts b/tests/integration/measurements-batch.test.ts index 6ee8a8b7e..b2ef109e3 100644 --- a/tests/integration/measurements-batch.test.ts +++ b/tests/integration/measurements-batch.test.ts @@ -325,6 +325,93 @@ describe("POST /api/measurements/batch (real Postgres)", () => { expect(stored.every((r) => r.unit === "minutes")).toBe(true); }); + // v1.4.25 W10 reconcile (senior-dev H-1): under contention the + // per-entry envelope MUST stay in sync with the aggregate + // `inserted` / `duplicate` counts. The previous reconciliation + // block was an effective no-op (it looked for rows that were + // "marked inserted but missing from the DB" — an impossibility + // because the raced row IS in the DB, written by the other + // batch). The corrected block downgrades enough "inserted" + // statuses to "duplicate" so the per-entry envelope and the + // aggregate counts agree, which the iOS sync cursor depends on + // to checkpoint correctly. + it("keeps per-entry status in sync with aggregate counts under a concurrent-write race", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + + // Two batches with overlapping externalIds posted in parallel. + // The composite unique index ensures only one row per + // (user, type, source, externalId) lands; `skipDuplicates` + // absorbs duplicate-key conflicts. The exact split between + // "won the race" and "got absorbed" depends on Postgres's + // commit order, so we assert only the invariant the iOS sync + // cursor relies on: per-entry envelope sums equal the aggregate + // counts for each response. + const sharedEntries = () => + Array.from({ length: 6 }, (_, i) => ({ + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 80 + i * 0.1, + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: `uuid-race-${i}`, + })); + + const [first, second] = await Promise.all([ + POST(makeRequest({ entries: sharedEntries() })), + POST(makeRequest({ entries: sharedEntries() })), + ]); + expect(first.status).toBe(200); + expect(second.status).toBe(200); + + const firstJson = (await first.json()) as { + data: { + processed: number; + inserted: number; + duplicates: number; + entries: Array<{ status: string }>; + }; + }; + const secondJson = (await second.json()) as typeof firstJson; + + // Invariant 1 — per-entry statuses sum to the aggregate counts. + // Before the fix this failed under contention because the + // no-op reconciliation left the envelope out of sync. + for (const json of [firstJson, secondJson]) { + const insertedEntries = json.data.entries.filter( + (e) => e.status === "inserted", + ).length; + const duplicateEntries = json.data.entries.filter( + (e) => e.status === "duplicate", + ).length; + const skippedEntries = json.data.entries.filter( + (e) => e.status === "skipped", + ).length; + expect(insertedEntries).toBe(json.data.inserted); + expect(duplicateEntries).toBe(json.data.duplicates); + expect( + insertedEntries + duplicateEntries + skippedEntries, + ).toBe(json.data.processed); + } + + // Invariant 2 — aggregate counts are non-negative. The previous + // logic could not produce a negative `inserted` count but a + // naive "downgrade and also decrement" implementation can. + expect(firstJson.data.inserted).toBeGreaterThanOrEqual(0); + expect(secondJson.data.inserted).toBeGreaterThanOrEqual(0); + + // Invariant 3 — the DB ends up with at most 6 rows because the + // composite unique index enforces single-copy, and the + // `inserted` counts across both requests cannot exceed the + // number of rows actually present. + const stored = await getPrismaClient().measurement.findMany({ + where: { userId: TEST_USER_ID }, + }); + expect(stored.length).toBeLessThanOrEqual(6); + expect(firstJson.data.inserted + secondJson.data.inserted).toBe( + stored.length, + ); + }); + it("rate-limits a user at 60 batches per minute (security H-2)", async () => { const { POST } = await import("@/app/api/measurements/batch/route");