fix(core): validate writeBatch result contract in bulkWrite (#3151)

bulkWrite correlated writeBatch's return to input rows purely by
position, with no check that it returned one record per row. A driver
that returned fewer rows silently backfilled `record: undefined` and
reported phantom successes; a longer return dropped records; a non-array
return marked the whole batch ok with no records.

Guard the return right after the batch write succeeds: a short / long /
non-array result now throws ERR_BULK_RESULT_MISMATCH, which falls into
the existing per-row degradation path (each row re-attempted via
writeOne) instead of being trusted. The error is thrown outside
withRetry and its message avoids any transient signature, so the batch
is never retried on it.

Also document bulkWrite's at-least-once delivery semantics in the module
header, ahead of the idempotency work in #3149.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga

Author: claude

packages/core/src/utils/bulk-write.test.ts

@@ -112,6 +112,50 @@ describe('bulkWrite', () => {
     expect(writeOne).not.toHaveBeenCalled();
     expect(results[0]).toMatchObject({ index: 0, ok: false });
   });
+
+  it('rejects a short writeBatch return as a failed batch and degrades to per-row (#3151)', async () => {
+    // Driver dropped a row from its RETURNING set: 2-row batch, 1 record back.
+    const writeBatch = vi.fn(async (batch: { n: number }[]) => batch.slice(1).map((r) => ({ id: `r${r.n}` })));
+    const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
+
+    const results = await bulkWrite([{ n: 1 }, { n: 2 }], { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeBatch).toHaveBeenCalledTimes(1); // mismatch is NOT transient — no batch retry
+    expect(writeOne).toHaveBeenCalledTimes(2);   // degraded to per-row instead of phantom success
+    expect(results.every((r) => r.ok)).toBe(true);
+    expect(results.map((r) => r.record)).toEqual([{ id: 'r1' }, { id: 'r2' }]);
+  });
+
+  it('rejects a non-array writeBatch return and degrades to per-row (#3151)', async () => {
+    const writeBatch = vi.fn(async () => undefined as unknown as { id: string }[]);
+    const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
+
+    const results = await bulkWrite([{ n: 1 }, { n: 2 }], { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeOne).toHaveBeenCalledTimes(2);
+    expect(results.every((r) => r.ok)).toBe(true);
+  });
+
+  it('rejects an over-long writeBatch return and degrades to per-row (#3151)', async () => {
+    const writeBatch = vi.fn(async (batch: { n: number }[]) => [...batch, { n: 999 }].map((r) => ({ id: `r${r.n}` })));
+    const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
+
+    const results = await bulkWrite([{ n: 1 }, { n: 2 }], { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeOne).toHaveBeenCalledTimes(2);
+    expect(results.every((r) => r.ok)).toBe(true);
+  });
+
+  it('surfaces ERR_BULK_RESULT_MISMATCH on a single-row batch with the wrong return count (#3151)', async () => {
+    const writeBatch = vi.fn(async () => [] as { id: string }[]); // empty for a 1-row batch
+    const writeOne = vi.fn(async () => ({ id: 'unused' }));
+
+    const results = await bulkWrite([{ n: 1 }], { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeOne).not.toHaveBeenCalled(); // single-row batch failure IS the row's final result
+    expect(results[0].ok).toBe(false);
+    expect((results[0].error as { code?: string })?.code).toBe('ERR_BULK_RESULT_MISMATCH');
+  });
 });
 
 describe('withTransientRetry', () => {

packages/core/src/utils/bulk-write.ts

@@ -26,6 +26,17 @@
  *    can reassemble output in input order even though rows are processed in
  *    batches (and a batch's flush may be interleaved with other, immediate,
  *    per-row work such as updates).
+ *
+ * Delivery semantics: **at-least-once**. Transient retry and per-row
+ * degradation both RE-RUN a write whose outcome was unknown — e.g. a turso
+ * `fetch failed` that arrived *after* the row was already committed
+ * (framework#3149), or a result-count mismatch that voids the batch
+ * (framework#3151). A caller that needs exactly-once must make its
+ * `writeBatch`/`writeOne` idempotent; both receive an `attempt` counter for
+ * exactly this — see the natural-key recheck the seed loader and import
+ * runner perform on `attempt > 1`. `writeBatch` MUST also resolve exactly one
+ * record per input row, in input order: a short / long / non-array return is
+ * rejected as a failed batch (framework#3151), never silently backfilled.
  */
 
 export interface BulkWriteRowResult<TRecord = any> {
@@ -162,6 +173,25 @@ export async function bulkWrite<TRow, TRecord = any>(
     const batch = rows.slice(start, start + batchSize);
     try {
       const records = await withRetry(() => opts.writeBatch(batch), retryOpts);
+      // Contract guard (framework#3151): `writeBatch` must resolve one record
+      // per input row. A short / long / non-array return breaks the positional
+      // correlation below, so backfilling it would report phantom successes
+      // (`record: undefined`) or drop records. Treat the whole batch as failed
+      // and fall through to per-row degradation (each row re-attempted via
+      // `writeOne`, which under an idempotent caller rechecks before writing).
+      // The message deliberately avoids any transient signature so this never
+      // reads as a retryable blip — and it is thrown *outside* `withRetry`, so
+      // the batch is not retried on it.
+      if (!Array.isArray(records) || records.length !== batch.length) {
+        throw Object.assign(
+          new Error(
+            `bulkWrite: writeBatch returned ${
+              Array.isArray(records) ? `${records.length} record(s)` : String(typeof records)
+            } for a ${batch.length}-row batch — treating batch as failed`,
+          ),
+          { code: 'ERR_BULK_RESULT_MISMATCH' },
+        );
+      }
       for (let i = 0; i < batch.length; i++) {
         results[start + i] = { index: start + i, ok: true, record: records[i] };
       }