fix(core,seed,rest): idempotent batch retry via attempt-aware natural-key recheck (#3149)

bulkWrite's transient retry is at-least-once: when a driver commits a
batch but its response is lost (turso's fetch-based transport fails after
commit), the retry re-inserted the whole batch — 2 input rows became 4
stored rows, all reported success, each firing record-change and summary
recompute twice.

Rather than a generic dedup inside bulkWrite, thread a 1-based `attempt`
counter into writeBatch/writeOne and let each caller enforce batch-level
idempotency with its own natural key:

- core: withRetry passes the attempt number; writeBatch/writeOne gain a
  `{ attempt }` context (backward compatible — callers may ignore it).
- seed: on attempt > 1, recheck existing rows by externalId (reusing
  loadExistingRecords) and insert only the missing ones; reassemble the
  per-row result in order.
- import: on attempt > 1 with matchFields, recheck by matchFields and
  create only the missing rows. A short createManyData return is now
  surfaced as a failed batch (so degradation, which also rechecks, runs)
  instead of being padded. A pure-insert import (no matchFields) has no
  natural key and stays at-least-once by documented contract.

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

@@ -156,6 +156,37 @@ describe('bulkWrite', () => {
     expect(results[0].ok).toBe(false);
     expect((results[0].error as { code?: string })?.code).toBe('ERR_BULK_RESULT_MISMATCH');
   });
+
+  it('passes the 1-based attempt number to writeBatch across a transient retry (#3149)', async () => {
+    const seen: number[] = [];
+    let attempts = 0;
+    const writeBatch = vi.fn(async (batch: { n: number }[], ctx: { attempt: number }) => {
+      seen.push(ctx.attempt);
+      attempts++;
+      if (attempts === 1) throw new Error('fetch failed');
+      return batch.map((r) => ({ id: `r${r.n}` }));
+    });
+    const writeOne = vi.fn(async () => ({ id: 'x' }));
+
+    await bulkWrite([{ n: 1 }, { n: 2 }], { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(seen).toEqual([1, 2]); // attempt 1 threw, attempt 2 succeeded
+  });
+
+  it('passes an attempt counter to writeOne during degradation (#3149)', async () => {
+    const seen: number[] = [];
+    const writeBatch = vi.fn(async () => { throw new Error('logical'); }); // force degradation
+    const writeOne = vi.fn(async (row: { n: number }, ctx: { attempt: number }) => {
+      seen.push(ctx.attempt);
+      if (ctx.attempt < 2 && row.n === 1) throw new Error('fetch failed'); // retry row 1 once
+      return { id: `r${row.n}` };
+    });
+
+    await bulkWrite([{ n: 1 }, { n: 2 }], { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    // row 1: attempt 1 (transient throw) → attempt 2 (ok); row 2: attempt 1 (ok)
+    expect(seen).toEqual([1, 2, 1]);
+  });
 });
 
 describe('withTransientRetry', () => {

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

@@ -67,10 +67,18 @@ export interface BulkWriteOptions<TRow, TRecord = any> extends RetryOptions {
    * `batch[i]` positionally (this is how every `bulkCreate` implementation in
    * this repo already behaves: sql's single `INSERT ... VALUES (...), (...)
    * RETURNING *`, memory's `Promise.all`, mongodb's ordered `insertMany`).
+   *
+   * `ctx.attempt` is the 1-based attempt number. `attempt > 1` means a prior
+   * attempt's outcome is UNKNOWN (a transient blip that may have landed after
+   * commit) — an exactly-once caller should recheck by natural key and skip
+   * rows already present before re-writing (framework#3149).
    */
-  writeBatch: (batch: TRow[]) => Promise<TRecord[]>;
-  /** Write a single row — used only to degrade a failed batch. */
-  writeOne: (row: TRow) => Promise<TRecord>;
+  writeBatch: (batch: TRow[], ctx: { attempt: number }) => Promise<TRecord[]>;
+  /**
+   * Write a single row — used only to degrade a failed batch. `ctx.attempt`
+   * carries the same recheck signal as {@link writeBatch}.
+   */
+  writeOne: (row: TRow, ctx: { attempt: number }) => Promise<TRecord>;
 }
 
 const DEFAULT_BATCH_SIZE = 200;
@@ -136,11 +144,11 @@ interface ResolvedRetryOptions {
   sleep: (ms: number) => Promise<void>;
 }
 
-async function withRetry<T>(fn: () => Promise<T>, opts: ResolvedRetryOptions): Promise<T> {
+async function withRetry<T>(fn: (attempt: number) => Promise<T>, opts: ResolvedRetryOptions): Promise<T> {
   let lastError: unknown;
   for (let attempt = 1; attempt <= opts.maxRetries; attempt++) {
     try {
-      return await fn();
+      return await fn(attempt);
     } catch (err) {
       lastError = err;
       if (attempt >= opts.maxRetries || !opts.isTransientError(err)) throw err;
@@ -159,7 +167,7 @@ async function withRetry<T>(fn: () => Promise<T>, opts: ResolvedRetryOptions): P
  * transient-error backoff {@link bulkWrite} applies to batches — so a
  * network blip doesn't drop an update the way it used to drop an insert.
  */
-export async function withTransientRetry<T>(fn: () => Promise<T>, opts: RetryOptions = {}): Promise<T> {
+export async function withTransientRetry<T>(fn: (attempt: number) => Promise<T>, opts: RetryOptions = {}): Promise<T> {
   return withRetry(fn, {
     maxRetries: Math.max(1, opts.maxRetries ?? DEFAULT_MAX_RETRIES),
     backoffBaseMs: opts.backoffBaseMs ?? DEFAULT_BACKOFF_BASE_MS,
@@ -194,7 +202,7 @@ export async function bulkWrite<TRow, TRecord = any>(
   for (let start = 0; start < rows.length; start += batchSize) {
     const batch = rows.slice(start, start + batchSize);
     try {
-      const records = await withRetry(() => opts.writeBatch(batch), retryOpts);
+      const records = await withRetry((attempt) => opts.writeBatch(batch, { attempt }), 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
@@ -233,7 +241,7 @@ export async function bulkWrite<TRow, TRecord = any>(
       for (let i = 0; i < batch.length; i++) {
         const idx = start + i;
         try {
-          const record = await withRetry(() => opts.writeOne(batch[i]), retryOpts);
+          const record = await withRetry((attempt) => opts.writeOne(batch[i], { attempt }), retryOpts);
           results[idx] = { index: idx, ok: true, record };
         } catch (err) {
           results[idx] = { index: idx, ok: false, error: err };

packages/metadata-protocol/src/seed-loader-retry.test.ts

@@ -71,6 +71,14 @@ function createMetadata(): IMetadataService {
         manager: { type: 'lookup', reference: 'my_app_employee' },
       },
     },
+    // A plain object (no self-ref) → the batched flushPendingInserts path.
+    my_app_widget: {
+      name: 'my_app_widget',
+      fields: {
+        name: { type: 'text' },
+        sku: { type: 'text' },
+      },
+    },
   };
   return {
     getObject: vi.fn(async (name: string) => objects[name]),
@@ -129,3 +137,45 @@ describe('seed self-ref sequential path — transient retry (framework#3150)', (
     expect((store.my_app_employee ?? []).map((r) => r.name).sort()).toEqual(['Alice', 'Bob']);
   });
 });
+
+describe('seed batched path — idempotent retry after commit-then-lost-response (framework#3149)', () => {
+  it('does not duplicate rows when the batch insert commits but its response is lost', async () => {
+    const { engine, store } = createFaithfulEngine();
+    const metadata = createMetadata();
+
+    // First array insert writes both rows to the store, then throws a transient
+    // error (turso's commit-then-lost-response shape). The retry must recheck
+    // by externalId and NOT insert them again.
+    const realInsert = (engine.insert as any).getMockImplementation();
+    let arrayInsertCalls = 0;
+    (engine.insert as any).mockImplementation(async (obj: string, data: any, opts: any) => {
+      if (obj === 'my_app_widget' && Array.isArray(data)) {
+        arrayInsertCalls++;
+        if (arrayInsertCalls === 1) {
+          await realInsert(obj, data, opts); // commit lands
+          throw new Error('fetch failed');   // ...but the response is lost
+        }
+      }
+      return realInsert(obj, data, opts);
+    });
+
+    const result = await new SeedLoaderService(engine, metadata, createLogger()).load({
+      seeds: [{
+        object: 'my_app_widget',
+        externalId: 'sku',
+        mode: 'insert',
+        env: ['prod', 'dev', 'test'],
+        records: [{ name: 'Widget A', sku: 'W-A' }, { name: 'Widget B', sku: 'W-B' }],
+      }] as any,
+      config: CONFIG,
+    });
+
+    // The array insert ran once (attempt 1, which committed); the retry's
+    // recheck found both rows already present and did NOT re-insert.
+    expect(arrayInsertCalls).toBe(1);
+    expect(store.my_app_widget.filter((r) => r.sku === 'W-A')).toHaveLength(1);
+    expect(store.my_app_widget.filter((r) => r.sku === 'W-B')).toHaveLength(1);
+    expect(store.my_app_widget).toHaveLength(2); // no duplicates
+    expect(result.summary.totalErrored).toBe(0);
+  });
+});

packages/metadata-protocol/src/seed-loader.ts

@@ -15,7 +15,7 @@ import type {
 } from '@objectstack/spec/data';
 import { SeedLoaderConfigSchema } from '@objectstack/spec/data';
 import { resolveSeedRecord } from '@objectstack/formula';
-import { bulkWrite, withTransientRetry, type BulkWriteRowResult } from '@objectstack/core';
+import { bulkWrite, withTransientRetry, defaultIsTransientError, type BulkWriteRowResult } from '@objectstack/core';
 
 interface Logger {
   info(message: string, meta?: Record<string, any>): void;
@@ -262,20 +262,69 @@ export class SeedLoaderService implements ISeedLoaderService {
     // logical/validation failure. See framework#2678.
     const pendingInserts: Array<{ recordIndex: number; externalIdValue: string; record: Record<string, unknown> }> = [];
     const opts = SeedLoaderService.SEED_OPTIONS as any;
+    const extIdOf = (rec: Record<string, unknown>) => String(rec[externalId] ?? '');
+    // bulkWrite is at-least-once: a retry (or a mismatch-driven degradation)
+    // may re-run a write whose prior attempt already committed. Guard against
+    // duplicate seed rows by rechecking natural keys before re-inserting
+    // (framework#3149). `lastBatchUncertain` carries the "prior batch outcome
+    // unknown" signal into the per-row degradation writeOne calls.
+    let lastBatchUncertain = false;
+    const isUncertainOutcome = (e: unknown) =>
+      defaultIsTransientError(e) || (e as { code?: unknown } | null)?.code === 'ERR_BULK_RESULT_MISMATCH';
+    // Reassemble one record per input row in order: rows already present are
+    // represented by the existing record; the rest are consumed in order from
+    // the freshly-inserted list.
+    const assembleInOrder = (rows: Record<string, unknown>[], existing: Map<string, any>, freshlyInserted: any[]): any[] => {
+      let k = 0;
+      return rows.map((r) => {
+        const key = extIdOf(r);
+        if (key && existing.has(key)) return existing.get(key);
+        return freshlyInserted[k++];
+      });
+    };
     const flushPendingInserts = async (): Promise<void> => {
       if (pendingInserts.length === 0) return;
       const batch = pendingInserts.splice(0, pendingInserts.length);
       const writeResults: BulkWriteRowResult[] = await bulkWrite(
         batch.map(b => b.record),
         {
           batchSize: SeedLoaderService.BULK_BATCH_SIZE,
-          // A lone row keeps the historical bare-record insert() call shape
-          // (no array wrapping) so single-record datasets are byte-for-byte
-          // unchanged; only a real batch (>1) uses the array/bulk form.
-          writeBatch: (rows) => rows.length === 1
-            ? this.engine.insert(objectName, rows[0], opts).then((r: any) => [r])
-            : this.engine.insert(objectName, rows, opts),
-          writeOne: (row) => this.engine.insert(objectName, row, opts),
+          writeBatch: async (rows, { attempt }) => {
+            let toInsert = rows;
+            let existing = new Map<string, any>();
+            if (attempt > 1) {
+              // Prior attempt may have committed before its response was lost:
+              // insert only rows not already present so a retry can't duplicate.
+              existing = await this.loadExistingRecords(objectName, externalId, config.organizationId);
+              toInsert = rows.filter((r) => { const k = extIdOf(r); return !(k && existing.has(k)); });
+            }
+            try {
+              // A lone row keeps the historical bare-record insert() call shape
+              // (no array wrapping) so single-record datasets are byte-for-byte
+              // unchanged; only a real batch (>1) uses the array/bulk form.
+              const freshlyInserted = toInsert.length === 0
+                ? []
+                : toInsert.length === 1
+                  ? [await this.engine.insert(objectName, toInsert[0], opts)]
+                  : await this.engine.insert(objectName, toInsert, opts);
+              lastBatchUncertain = false;
+              return assembleInOrder(rows, existing, freshlyInserted as any[]);
+            } catch (e) {
+              lastBatchUncertain = isUncertainOutcome(e);
+              throw e;
+            }
+          },
+          writeOne: async (row, { attempt }) => {
+            if (attempt > 1 || lastBatchUncertain) {
+              const key = extIdOf(row);
+              if (key) {
+                const existing = await this.loadExistingRecords(objectName, externalId, config.organizationId);
+                const hit = existing.get(key);
+                if (hit) return hit; // already committed by a prior attempt
+              }
+            }
+            return this.engine.insert(objectName, row, opts);
+          },
         },
       );
       for (const res of writeResults) {

packages/rest/src/import-runner-idempotency.test.ts

@@ -0,0 +1,106 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * framework#3149: bulkWrite is at-least-once — a retry (or a mismatch-driven
+ * degradation) may re-run a create whose prior attempt already committed. When
+ * the import has natural keys (matchFields), runImport rechecks before
+ * re-creating so a retry can't duplicate the row. A pure-insert import has no
+ * natural key and stays at-least-once by contract.
+ */
+
+import { describe, it, expect, vi } from 'vitest';
+import { runImport, type ImportProtocolLike } from './import-runner';
+import type { ExportFieldMeta } from './export-format.js';
+
+const metaMap = new Map<string, ExportFieldMeta>([['name', { name: 'name', type: 'text' }]]);
+
+const baseOpts = {
+  objectName: 'task',
+  metaMap,
+  dryRun: false,
+  runAutomations: false,
+  trimWhitespace: true,
+  createMissingOptions: false,
+  skipBlankMatchKey: false,
+};
+
+/**
+ * Mock protocol backed by an in-memory store. `createManyData` optionally
+ * commits-then-throws (or short-returns) on its first call to model a lost
+ * response / mismatch; findData filters the store for the recheck.
+ */
+function makeProtocol(opts: { firstCall?: 'throw' | 'shortReturn' } = {}) {
+  const store: Array<Record<string, any>> = [];
+  let idc = 0;
+  let calls = 0;
+  const createManyData = vi.fn(async (args: { records: any[] }) => {
+    calls++;
+    const recs = args.records.map((r) => {
+      const rec = { id: `id-${++idc}`, ...r };
+      store.push(rec);
+      return rec;
+    });
+    if (calls === 1 && opts.firstCall === 'throw') throw new Error('fetch failed'); // committed, response lost
+    if (calls === 1 && opts.firstCall === 'shortReturn') return { records: [] };     // committed, bad count
+    return { records: recs };
+  });
+  const createData = vi.fn(async (args: { data: { name: string } }) => {
+    const rec = { id: `id-${++idc}`, ...args.data };
+    store.push(rec);
+    return rec;
+  });
+  const findData = vi.fn(async (args: { query?: { $filter?: Record<string, any> } }) => {
+    const filter = args.query?.$filter ?? {};
+    return store.filter((row) => Object.entries(filter).every(([k, v]) => row[k] === v));
+  });
+  const p: ImportProtocolLike = { findData, createData, updateData: vi.fn(), createManyData };
+  return { p, store, createManyData, createData };
+}
+
+describe('runImport — idempotent retry with natural keys (framework#3149)', () => {
+  it('upsert+matchFields: a transient retry after commit does not duplicate rows', async () => {
+    const { p, store, createManyData } = makeProtocol({ firstCall: 'throw' });
+
+    const summary = await runImport({
+      ...baseOpts, p, writeMode: 'upsert', matchFields: ['name'],
+      rows: [{ name: 'x' }, { name: 'y' }],
+    });
+
+    // createManyData ran once (attempt 1, which committed); the retry's recheck
+    // found both rows already present and did NOT re-create them.
+    expect(createManyData).toHaveBeenCalledTimes(1);
+    expect(store.filter((r) => r.name === 'x')).toHaveLength(1);
+    expect(store.filter((r) => r.name === 'y')).toHaveLength(1);
+    expect(store).toHaveLength(2); // no duplicates
+    expect(summary.created).toBe(2);
+    expect(summary.errors).toBe(0);
+  });
+
+  it('upsert+matchFields: a short createManyData return degrades and still does not duplicate', async () => {
+    const { p, store } = makeProtocol({ firstCall: 'shortReturn' });
+
+    const summary = await runImport({
+      ...baseOpts, p, writeMode: 'upsert', matchFields: ['name'],
+      rows: [{ name: 'x' }, { name: 'y' }],
+    });
+
+    // The empty return voids the batch → per-row degradation, which rechecks
+    // and finds both rows already committed rather than re-creating them.
+    expect(store).toHaveLength(2);
+    expect(summary.created).toBe(2);
+    expect(summary.errors).toBe(0);
+  });
+
+  it('pure insert (no matchFields): retry is at-least-once — duplicates are the documented contract', async () => {
+    const { p, store } = makeProtocol({ firstCall: 'throw' });
+
+    await runImport({
+      ...baseOpts, p, writeMode: 'insert', matchFields: [],
+      rows: [{ name: 'x' }, { name: 'y' }],
+    });
+
+    // No natural key to recheck against: the committed-then-retried batch is
+    // written twice. This pins the contract (exactly-once needs matchFields).
+    expect(store).toHaveLength(4);
+  });
+});