perf(measurements): insert fresh batch samples set-based; answer health while starting

POST /api/measurements/batch ran one interactive reconciler transaction
per fresh sample: a savepoint, three advisory locks, two candidate reads
and the insert, about ten round trips. A 500-entry upload of new samples
was 500 of those, the 2 to 15 second outliers. Fresh samples now land
through INSERT ... ON CONFLICT DO NOTHING under the same per-account
advisory lock. Both measurement identities are full unique indexes, so
the statement inserts exactly what the reconciler would have inserted,
and every skipped row still goes through the reconciler for its own
verdict. Only rows whose external and natural keys each occur once in
the batch are buffered, and the buffer is flushed before any other row
is reconciled, so the order of effects stays the sequential one. The
contract and every per-entry status are unchanged; a golden test runs
each batch through both paths against the same table.

/api/health took seconds after a start because register() awaited the
whole pg-boss worker start (queues, schedules, boot discovery) before
the server took a request. The worker, or a web process's producer, now
starts in the background after the blocking boot checks; the probe
answers at once with reason "starting" until it is up, and gives the
database two seconds before calling it not answering.

src/app/api/health/__tests__/route.test.ts

@@ -163,3 +163,53 @@ describe("GET /api/health under ENCRYPTION_KEY_CHECK=warn", () => {
     }
   });
 });
+
+describe("GET /api/health while the background role starts", () => {
+  it("says `starting` at once, and `ok` when the role is up", async () => {
+    const { markStartup } = await import("@/lib/boot/startup-state");
+    vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web");
+    markStartup("starting");
+    try {
+      const starting = await health();
+      expect(starting.response.status).toBe(503);
+      expect(starting.body).toEqual({ status: "degraded", reason: "starting" });
+
+      getGlobalBoss.mockReturnValue({});
+      markStartup("ready");
+      const ready = await health();
+      expect(ready.response.status).toBe(200);
+      expect(ready.body).toEqual({ status: "ok" });
+    } finally {
+      markStartup("ready");
+    }
+  });
+
+  it("does not call a failed start `starting`", async () => {
+    const { markStartup } = await import("@/lib/boot/startup-state");
+    vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web");
+    markStartup("failed");
+    try {
+      const { response, body } = await health();
+      expect(response.status).toBe(503);
+      expect(body).toEqual({ status: "degraded" });
+    } finally {
+      markStartup("ready");
+    }
+  });
+
+  it("does not wait on a database that does not answer", async () => {
+    vi.useFakeTimers();
+    try {
+      vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web");
+      getGlobalBoss.mockReturnValue({});
+      queryRaw.mockImplementation(() => new Promise(() => undefined));
+      const pending = health();
+      await vi.advanceTimersByTimeAsync(2_000);
+      const { response, body } = await pending;
+      expect(response.status).toBe(503);
+      expect(body).toEqual({ status: "degraded" });
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});

src/app/api/health/route.ts

@@ -11,19 +11,41 @@ import {
   isKeyMismatch,
   KEY_MISMATCH_HEALTH_REASON,
 } from "@/lib/boot/key-mismatch-state";
+import { startupPhase } from "@/lib/boot/startup-state";
+
+/**
+ * How long the probe waits for `SELECT 1`. During a boot storm the pool can
+ * be saturated by the worker's start-up work; a probe that waits its turn
+ * held a monitor for seconds. Past this the database counts as not answering
+ * and the probe says `degraded` at once.
+ */
+const DB_PROBE_TIMEOUT_MS = 2_000;
+
+/** Reported while the background role is still coming up. */
+const STARTING_HEALTH_REASON = "starting";
+
+async function probeDatabase(): Promise<boolean> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      prisma.$queryRaw`SELECT 1`.then(() => true),
+      new Promise<boolean>((resolve) => {
+        timer = setTimeout(() => resolve(false), DB_PROBE_TIMEOUT_MS);
+      }),
+    ]);
+  } catch {
+    return false;
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
 
 export const dynamic = "force-dynamic";
 
 export const GET = apiHandler(async () => {
   annotate({ action: { name: "health.check" } });
 
-  let dbOk = true;
-
-  try {
-    await prisma.$queryRaw`SELECT 1`;
-  } catch {
-    dbOk = false;
-  }
+  const dbOk = await probeDatabase();
 
   const worker = getWorkerStatus();
   const producerReady = getGlobalBoss() !== null;
@@ -31,20 +53,23 @@ export const GET = apiHandler(async () => {
   // other route; the probe says so in a word every caller may see, because
   // it is the one place a catalog install's status page reads.
   const keyMismatch = isKeyMismatch();
-  const reason = keyMismatch ? KEY_MISMATCH_HEALTH_REASON : undefined;
+  // Otherwise, while the background role is still starting, the 503 below
+  // says so: a transient state, not a fault, and not a hang.
+  const roleDown =
+    (shouldRunWeb() && !producerReady) ||
+    (shouldRunWorker() && !worker.running);
+  const reason = keyMismatch
+    ? KEY_MISMATCH_HEALTH_REASON
+    : dbOk && roleDown && startupPhase() === "starting"
+      ? STARTING_HEALTH_REASON
+      : undefined;
   // Under ENCRYPTION_KEY_CHECK=warn the same finding does not refuse; it is
   // named as a warning and leaves the status alone.
   const warning =
     !keyMismatch && getKeyMismatchWarning()
       ? KEY_MISMATCH_HEALTH_REASON
       : undefined;
-  const status =
-    !keyMismatch &&
-    dbOk &&
-    (!shouldRunWeb() || producerReady) &&
-    (!shouldRunWorker() || worker.running)
-      ? "ok"
-      : "degraded";
+  const status = !keyMismatch && dbOk && !roleDown ? "ok" : "degraded";
   const statusCode = status === "ok" ? 200 : 503;
 
   const cacheHeaders = {

src/app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts

@@ -9,6 +9,11 @@
  * prefilter's read answering nothing, which is the pre-v1.42 path where every
  * row went through the reconciler. Statuses, reasons and counters must match
  * exactly; only the transaction count may differ.
+ *
+ * v1.44 — the golden run is also blind to the set-based insert: its
+ * `INSERT … ON CONFLICT DO NOTHING` stand-in lands nothing, so every row
+ * reaches the reconciler one at a time, the sequential path. The shipped run
+ * takes the set insert, and the two must still leave the same table.
  */
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { NextRequest } from "next/server";
@@ -29,6 +34,8 @@ const store = vi.hoisted(() => ({
   rows: [] as StoredRow[],
   /** Golden mode: the prefilter's read answers nothing. */
   prefilterBlind: false,
+  /** Golden mode: the set-based insert lands nothing. */
+  setInsertBlind: false,
   nextId: 1,
 }));
 
@@ -81,12 +88,70 @@ vi.mock("@/lib/db", () => {
         },
       ),
     },
-    $transaction: vi.fn(async (fn: (tx: unknown) => unknown) => fn({})),
+    $transaction: vi.fn(async (fn: (tx: unknown) => unknown) =>
+      fn({
+        measurement: {
+          // `INSERT … ON CONFLICT DO NOTHING` over both unique identities,
+          // in VALUES order, returning the rows that landed.
+          createManyAndReturn: async (args: {
+            data: Array<{
+              userId: string;
+              type: string;
+              source: string;
+              externalId: string;
+              measuredAt: Date;
+              sleepStage: string | null;
+              value: number;
+            }>;
+          }) => {
+            if (store.setInsertBlind) return [];
+            const landed: Array<{
+              type: string;
+              source: string;
+              externalId: string;
+            }> = [];
+            for (const d of args.data) {
+              const scope = store.rows.filter(
+                (r) =>
+                  r.userId === d.userId &&
+                  r.type === d.type &&
+                  r.source === d.source,
+              );
+              const conflict = scope.some(
+                (r) =>
+                  r.externalId === d.externalId ||
+                  (sameInstant(r.measuredAt, d.measuredAt) &&
+                    r.sleepStage === (d.sleepStage ?? null)),
+              );
+              if (conflict) continue;
+              store.rows.push({
+                id: `m${store.nextId++}`,
+                userId: d.userId,
+                type: d.type,
+                source: d.source,
+                externalId: d.externalId,
+                measuredAt: d.measuredAt,
+                sleepStage: d.sleepStage ?? null,
+                value: d.value,
+                deletedAt: null,
+              });
+              landed.push({
+                type: d.type,
+                source: d.source,
+                externalId: d.externalId,
+              });
+            }
+            return landed;
+          },
+        },
+      }),
+    ),
   };
   return { prisma };
 });
 
 vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  lockMeasurementUser: async () => undefined,
   reconcileExternalMeasurement: async (
     _tx: unknown,
     desired: {
@@ -283,6 +348,7 @@ async function golden(setup: () => void, entries: Entry[]) {
   store.nextId = 1;
   setup();
   store.prefilterBlind = true;
+  store.setInsertBlind = true;
   vi.mocked(prisma.$transaction).mockClear();
   const old = await post(entries);
   const oldTransactions = vi.mocked(prisma.$transaction).mock.calls.length;
@@ -292,6 +358,7 @@ async function golden(setup: () => void, entries: Entry[]) {
   store.nextId = 1;
   setup();
   store.prefilterBlind = false;
+  store.setInsertBlind = false;
   vi.mocked(prisma.$transaction).mockClear();
   const next = await post(entries);
   const transactions = vi.mocked(prisma.$transaction).mock.calls.length;
@@ -304,6 +371,7 @@ beforeEach(() => {
   vi.clearAllMocks();
   store.rows = [];
   store.prefilterBlind = false;
+  store.setInsertBlind = false;
   vi.mocked(getSession).mockResolvedValue({
     session: { id: "s", expiresAt: new Date(NOW + HOUR) },
     user: { id: USER, username: "u", role: "USER" as const },
@@ -334,7 +402,9 @@ describe("exact-duplicate prefilter (v1.42)", () => {
       entries,
     );
     expect(next.duplicates).toBe(500);
-    expect(oldTransactions).toBe(500);
+    // Sequentially: two set-insert statements that land nothing, then one
+    // reconciler transaction per row.
+    expect(oldTransactions).toBe(502);
     expect(transactions).toBe(0);
   });
 
@@ -366,7 +436,8 @@ describe("exact-duplicate prefilter (v1.42)", () => {
       weight("uuid-move", new Date(NOW - DAY)),
     ]);
     expect(next.entries[0].status).toBe("updated");
-    expect(transactions).toBe(1);
+    // The set insert skips it on the external identity, the reconciler moves it.
+    expect(transactions).toBe(2);
   });
 
   it("4: the same id with another sleep stage still reconciles", async () => {
@@ -389,7 +460,7 @@ describe("exact-duplicate prefilter (v1.42)", () => {
       sleepStage: 3,
     };
     const { transactions } = await golden(setup, [entry]);
-    expect(transactions).toBe(1);
+    expect(transactions).toBe(2);
   });
 
   it("5: `stats:` rows are never prefiltered; a second snapshot is superseded", async () => {
@@ -615,3 +686,88 @@ describe("folded_window guard (v1.42)", () => {
     expect(data.entries[0].status).toBe("inserted");
   });
 });
+
+describe("set-based insert (v1.44)", () => {
+  it("500 fresh samples land in two statements, as the reconciler would have", async () => {
+    const entries = Array.from({ length: 500 }, (_, i) =>
+      pulse(`fresh-${i}`, new Date(NOW - DAY - i * 60_000), 60 + (i % 30)),
+    );
+    const { next, oldTransactions, transactions } = await golden(
+      () => undefined,
+      entries,
+    );
+    expect(next.inserted).toBe(500);
+    expect(oldTransactions).toBe(502);
+    expect(transactions).toBe(2);
+  });
+
+  it("a mix of fresh, moved and resurrected rows leaves the same table", async () => {
+    const at = (h: number) => new Date(NOW - DAY - h * HOUR);
+    const setup = () => {
+      // Moves: the stored row sits at another instant.
+      seed({ type: "WEIGHT", externalId: "moved", measuredAt: at(9) });
+      // Resurrects: a tombstone at a different instant under the same id.
+      seed({
+        type: "WEIGHT",
+        externalId: "back",
+        measuredAt: at(8),
+        deletedAt: new Date(NOW - HOUR),
+      });
+      // Natural-identity collision under another id: reconciled, not inserted.
+      seed({ type: "WEIGHT", externalId: "other", measuredAt: at(3) });
+    };
+    const { next, transactions } = await golden(setup, [
+      weight("fresh-a", at(1)),
+      weight("moved", at(2)),
+      weight("collides", at(3)),
+      weight("fresh-b", at(4)),
+      weight("back", at(5)),
+      weight("fresh-c", at(6)),
+    ]);
+    expect(next.entries.map((e) => e.status)).toEqual([
+      "inserted",
+      "updated",
+      "updated",
+      "inserted",
+      "updated",
+      "inserted",
+    ]);
+    // One set statement, then the three rows it skipped.
+    expect(transactions).toBe(4);
+  });
+
+  it("a row sharing an identity in the batch flushes the buffer first", async () => {
+    const at = new Date(NOW - DAY);
+    const { next } = await golden(
+      () => undefined,
+      [
+        weight("solo", new Date(at.getTime() - HOUR)),
+        weight("twice", at),
+        weight("twice", at),
+      ],
+    );
+    expect(next.entries.map((e) => e.status)).toEqual([
+      "inserted",
+      "inserted",
+      "duplicate",
+    ]);
+  });
+
+  it("keeps the order when a later reconcile would claim a buffered row's slot", async () => {
+    // Stored X at t1. C moves X to t5; D and E share t1, so neither is
+    // buffered. Sequentially C frees t1 before D arrives: D inserts, E
+    // merges into D. Reconciling D before C would instead take X's row for
+    // D and leave C to insert, a different table with the same rows.
+    const t1 = new Date(NOW - DAY - HOUR);
+    const t5 = new Date(NOW - DAY - 5 * HOUR);
+    const { next } = await golden(
+      () => seed({ type: "WEIGHT", externalId: "X", measuredAt: t1 }),
+      [weight("X", t5), weight("Y", t1), weight("Z", t1)],
+    );
+    expect(next.entries.map((e) => e.status)).toEqual([
+      "updated",
+      "inserted",
+      "updated",
+    ]);
+  });
+});

src/app/api/measurements/batch/route.ts

@@ -66,7 +66,11 @@ import {
   oppositeMergeSource,
   type MergeCandidate,
 } from "@/lib/measurements/cross-source-merge";
-import { reconcileExternalMeasurement } from "@/lib/measurements/reconcile-external-measurement";
+import {
+  lockMeasurementUser,
+  reconcileExternalMeasurement,
+  type MeasurementReconciliationVerdict,
+} from "@/lib/measurements/reconcile-external-measurement";
 import { classifyRangeRejection } from "@/lib/measurements/plausibility-gate";
 import {
   FOLDED_WINDOW_REASON,
@@ -134,6 +138,12 @@ const MAX_BATCH_ENTRIES = 500;
 // The cap is a DoS ceiling, not a tight bound.
 const MAX_BODY_BYTES = 4 * 1024 * 1024;
 
+/**
+ * Rows per set-based insert statement. 500 entries fit two statements; each
+ * row binds about fifteen parameters, far under PostgreSQL's 65 535.
+ */
+const SET_INSERT_CHUNK = 250;
+
 // v1.4.25 (security H-2): cap batch ingest at 60
 // batches per user per minute. Healthy iOS sync drains its
 // HealthKit observer queue in well under one batch per minute (the
@@ -635,6 +645,8 @@ async function postBatch(request: NextRequest): Promise<Response> {
   let updatedCount = 0;
   let duplicateCount = 0;
   let failedCount = 0;
+  // v1.44 — subset of `insertedCount` written by the set-based insert.
+  let setInsertedCount = 0;
   // v1.11.4 (iOS #2) — subset of `duplicateCount` collapsed by the
   // MANUAL↔APPLE_HEALTH same-reading merge (as opposed to a plain
   // composite-key duplicate). Surfaced as a dedicated wide-event count so
@@ -855,6 +867,188 @@ async function postBatch(request: NextRequest): Promise<Response> {
       [...foldedIndexes].map((index) => prepared[index]),
     );
 
+    /** The reconciler's input for one prepared row. */
+    const desiredFor = (p: Prepared) => {
+      const statsRow = isStatsExternalId(p.row.externalId as string);
+      return {
+        userId: user.id,
+        type: p.row.type as MeasurementType,
+        value: p.row.value as number,
+        valueMin: p.row.valueMin as number | null,
+        valueMax: p.row.valueMax as number | null,
+        unit: p.row.unit as string,
+        source: p.row
+          .source as Prisma.MeasurementUncheckedCreateInput["source"],
+        measuredAt: p.row.measuredAt as Date,
+        externalId: p.row.externalId as string,
+        externalSourceVersion: p.row.externalSourceVersion as string | null,
+        glucoseContext: p.row
+          .glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"],
+        sleepStage: p.row
+          .sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"],
+        rhythmClassification: p.row
+          .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"],
+        deviceType: p.row.deviceType as string | null,
+        aggregationProvenance: statsRow
+          ? ("HEALTHKIT_STATISTICS" as const)
+          : null,
+      };
+    };
+
+    const pushMergeCandidate = (p: Prepared) => {
+      if (isMergeableSource(p.row.source as string)) {
+        inBatchCandidates.push({
+          type: p.row.type as MeasurementType,
+          source: p.row.source as string,
+          value: p.row.value as number,
+          measuredAt: p.row.measuredAt as Date,
+        });
+      }
+    };
+
+    /**
+     * Record one row's verdict. `candidate` is false for a row whose merge
+     * candidate was already pushed when it was buffered (see below).
+     */
+    const applyVerdict = (
+      p: Prepared,
+      verdict: Pick<MeasurementReconciliationVerdict, "status"> &
+        Partial<{
+          dirtyIdentities: Array<{ type: MeasurementType; measuredAt: Date }>;
+          error: { code?: string };
+        }>,
+      candidate: boolean,
+    ) => {
+      switch (verdict.status) {
+        case "inserted":
+          results[p.index] = { index: p.index, status: "inserted" };
+          insertedCount++;
+          insertedPrepared.push(p);
+          writtenIdentities.push({
+            type: p.row.type as MeasurementType,
+            measuredAt: p.row.measuredAt as Date,
+          });
+          break;
+        case "updated":
+        case "resurrected":
+          results[p.index] = { index: p.index, status: "updated" };
+          updatedCount++;
+          writtenIdentities.push(
+            {
+              type: p.row.type as MeasurementType,
+              measuredAt: p.row.measuredAt as Date,
+            },
+            ...(verdict.dirtyIdentities ?? []),
+          );
+          break;
+        case "duplicate":
+          results[p.index] = { index: p.index, status: "duplicate" };
+          duplicateCount++;
+          break;
+        case "failed":
+          results[p.index] = {
+            index: p.index,
+            status: "failed",
+            reason: verdict.error?.code ?? "write_failed",
+          };
+          failedCount++;
+          return;
+        case "rejected_range":
+          // The reconciler carries the same plausibility gate this route
+          // already applied above, so reaching here means the two disagreed —
+          // report it exactly as the earlier guard would have, and let the
+          // wide-event tally record which end refused.
+          results[p.index] = outOfRangeResult(
+            p.index,
+            p.row.type as string,
+            p.row.value as number,
+          );
+          return;
+      }
+      if (candidate) pushMergeCandidate(p);
+    };
+
+    const reconcileRow = (p: Prepared) =>
+      prisma.$transaction((tx) =>
+        reconcileExternalMeasurement(tx, desiredFor(p), {
+          exactExternalMatch: isStatsExternalId(p.row.externalId as string)
+            ? "update"
+            : "duplicate",
+        }),
+      );
+
+    // v1.44 — set-based inserts. A fresh sample used to cost one interactive
+    // transaction of about ten round trips (savepoint, three advisory locks,
+    // two candidate reads, the insert), so a 500-entry batch of new samples
+    // was 500 of those in a row: the 2–15 s outliers.
+    //
+    // The reconciler inserts exactly when no row, live or tombstoned, holds
+    // the entry's external identity `(type, source, externalId)` or its
+    // natural identity `(type, measuredAt, source, sleepStage)`. Both are
+    // full unique indexes (the natural one NULLS NOT DISTINCT), so
+    // `INSERT … ON CONFLICT DO NOTHING` inserts exactly those rows and skips
+    // every other; a skipped row then goes through the reconciler as before
+    // and gets its own verdict (update, resurrect, duplicate). The insert
+    // runs under the same per-account advisory lock the reconciler takes, so
+    // the two serialise like two reconcilers would.
+    //
+    // Only a row whose external and natural key each occur once in the batch
+    // is buffered, and the buffer is flushed before any other row reaches the
+    // reconciler. Such a row shares no identity with another entry, so no
+    // other entry's reconcile can create, move or retire a row at its
+    // identity, and inserting it out of turn changes nothing. Rows the insert
+    // skips are reconciled in their original order. A buffered row's merge
+    // candidate is pushed when it is buffered, as the sequential loop pushed
+    // it after a successful write; the only case that differs is a write that
+    // fails outright, where the candidate stays.
+    const pendingInserts: Prepared[] = [];
+    const flushPendingInserts = async () => {
+      if (pendingInserts.length === 0) return;
+      const batch = pendingInserts.splice(0, pendingInserts.length);
+      const landed = new Set<Prepared>();
+      try {
+        for (let i = 0; i < batch.length; i += SET_INSERT_CHUNK) {
+          const slice = batch.slice(i, i + SET_INSERT_CHUNK);
+          const rows = await prisma.$transaction(async (tx) => {
+            await lockMeasurementUser(tx, user.id);
+            return tx.measurement.createManyAndReturn({
+              data: slice.map((p) => ({
+                ...desiredFor(p),
+                sleepStage: (p.row.sleepStage ??
+                  null) as Prisma.MeasurementUncheckedCreateInput["sleepStage"],
+                deletedAt: null,
+              })),
+              skipDuplicates: true,
+              select: { type: true, source: true, externalId: true },
+            });
+          });
+          const keys = new Set(
+            rows.map((r) => `${r.type}::${r.source}::${r.externalId}`),
+          );
+          for (const p of slice) {
+            if (keys.has(`${p.row.type}::${p.row.source}::${p.row.externalId}`))
+              landed.add(p);
+          }
+        }
+      } catch (err) {
+        // Whatever did not land goes through the reconciler below, which
+        // turns a real database fault into a per-entry `failed` verdict.
+        annotate({
+          action: { name: "measurement.batch.set_insert_fallback" },
+          meta: {
+            error_class: err instanceof Error ? err.name : "unknown",
+            landed: landed.size,
+            pending: batch.length,
+          },
+        });
+      }
+      setInsertedCount += landed.size;
+      for (const p of batch) {
+        if (landed.has(p)) applyVerdict(p, { status: "inserted" }, false);
+        else applyVerdict(p, await reconcileRow(p), false);
+      }
+    };
+
     for (const p of prepared) {
       const externalId = p.row.externalId as string;
       const statsRow = isStatsExternalId(externalId);
@@ -895,101 +1089,29 @@ async function postBatch(request: NextRequest): Promise<Response> {
           : { index: p.index, status: "duplicate" };
         duplicateCount++;
         if (folded) foldedWindowCount++;
-        if (isMergeableSource(p.row.source as string)) {
-          inBatchCandidates.push({
-            type: p.row.type as MeasurementType,
-            source: p.row.source as string,
-            value: p.row.value as number,
-            measuredAt: p.row.measuredAt as Date,
-          });
-        }
+        pushMergeCandidate(p);
         continue;
       }
 
-      const verdict = await prisma.$transaction((tx) =>
-        reconcileExternalMeasurement(
-          tx,
-          {
-            userId: user.id,
-            type: p.row.type as MeasurementType,
-            value: p.row.value as number,
-            valueMin: p.row.valueMin as number | null,
-            valueMax: p.row.valueMax as number | null,
-            unit: p.row.unit as string,
-            source: p.row
-              .source as Prisma.MeasurementUncheckedCreateInput["source"],
-            measuredAt: p.row.measuredAt as Date,
-            externalId,
-            externalSourceVersion: p.row.externalSourceVersion as string | null,
-            glucoseContext: p.row
-              .glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"],
-            sleepStage: p.row
-              .sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"],
-            rhythmClassification: p.row
-              .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"],
-            deviceType: p.row.deviceType as string | null,
-            aggregationProvenance: statsRow ? "HEALTHKIT_STATISTICS" : null,
-          },
-          { exactExternalMatch: statsRow ? "update" : "duplicate" },
-        ),
+      const naturalKey = naturalKeyOf(
+        p.row.type,
+        p.row.source,
+        p.row.measuredAt as Date,
+        p.row.sleepStage,
       );
-
-      switch (verdict.status) {
-        case "inserted":
-          results[p.index] = { index: p.index, status: "inserted" };
-          insertedCount++;
-          insertedPrepared.push(p);
-          writtenIdentities.push({
-            type: p.row.type as MeasurementType,
-            measuredAt: p.row.measuredAt as Date,
-          });
-          break;
-        case "updated":
-        case "resurrected":
-          results[p.index] = { index: p.index, status: "updated" };
-          updatedCount++;
-          writtenIdentities.push(
-            {
-              type: p.row.type as MeasurementType,
-              measuredAt: p.row.measuredAt as Date,
-            },
-            ...(verdict.dirtyIdentities ?? []),
-          );
-          break;
-        case "duplicate":
-          results[p.index] = { index: p.index, status: "duplicate" };
-          duplicateCount++;
-          break;
-        case "failed":
-          results[p.index] = {
-            index: p.index,
-            status: "failed",
-            reason: verdict.error.code ?? "write_failed",
-          };
-          failedCount++;
-          continue;
-        case "rejected_range":
-          // The reconciler carries the same plausibility gate this route
-          // already applied above, so reaching here means the two disagreed —
-          // report it exactly as the earlier guard would have, and let the
-          // wide-event tally record which end refused.
-          results[p.index] = outOfRangeResult(
-            p.index,
-            p.row.type as string,
-            p.row.value as number,
-          );
-          continue;
+      if (
+        externalKeyCounts.get(key) === 1 &&
+        naturalKeyCounts.get(naturalKey) === 1
+      ) {
+        pendingInserts.push(p);
+        pushMergeCandidate(p);
+        continue;
       }
 
-      if (isMergeableSource(p.row.source as string)) {
-        inBatchCandidates.push({
-          type: p.row.type as MeasurementType,
-          source: p.row.source as string,
-          value: p.row.value as number,
-          measuredAt: p.row.measuredAt as Date,
-        });
-      }
+      await flushPendingInserts();
+      applyVerdict(p, await reconcileRow(p), true);
     }
+    await flushPendingInserts();
   }
 
   // v1.43 — the hours, after the measurements. One write per slot: the last
@@ -1342,6 +1464,9 @@ async function postBatch(request: NextRequest): Promise<Response> {
       // v1.42 — duplicates refused because a live `stats:` row already
       // covers the sample's hour or day (`folded-window.ts`).
       folded_window: foldedWindowCount,
+      // v1.44 — inserts that took the set-based path rather than one
+      // reconciler transaction each.
+      set_inserted: setInsertedCount,
     },
   });