feat(rest): exact import-retry idempotency via pre-assigned row ids; adopt partial-success create (#3173, #3172)

Pure-insert imports (no matchFields) had no natural key to recheck on a
commit-then-lost-response retry, so they stayed at-least-once — a retry
duplicated the whole batch (#3149 left this documented, #3173 tracked
closing it).

Every buffered CREATE row now gets a client-generated id at flush time
(stable across attempts; an explicit user-supplied id is respected — both
the SQL driver and bulkCreate honor caller ids). The retry path rechecks
by id ($in) and re-inserts only rows that truly did not land. This is
exact for every write mode: legitimate duplicate rows in insert mode
resolve correctly because each copy has its own id — the case a
natural-key or fingerprint recheck cannot distinguish. The old
matchFields-based recheck is replaced by the id recheck.

The import runner also prefers the protocol's new insertManyData
(framework#3172) via bulkWrite's writeBatchPartial: a row that fails
validation is a per-row verdict from one call, so a bad row no longer
forces the whole-batch degradation that re-ran beforeInsert hooks on its
siblings. Protocols without insertManyData keep the createManyData path.

Flips the previously pinned at-least-once test: pure-insert retry after
commit now expects zero duplicates.

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

Author: claude

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

@@ -53,8 +53,11 @@ describe('runImport — bulk create batching (framework#2678)', () => {
     expect(summary.created).toBe(250);
     expect(summary.results).toHaveLength(250);
     expect(summary.results.map((r) => r.row)).toEqual(Array.from({ length: 250 }, (_, i) => i + 1));
-    expect(summary.results[0]).toMatchObject({ ok: true, action: 'created', id: 'id_r0' });
-    expect(summary.results[249]).toMatchObject({ ok: true, action: 'created', id: 'id_r249' });
+    // Rows carry a pre-assigned id (framework#3173), echoed back by the mock.
+    expect(summary.results[0]).toMatchObject({ ok: true, action: 'created' });
+    expect(summary.results[0].id).toBeTruthy();
+    expect(summary.results[249]).toMatchObject({ ok: true, action: 'created' });
+    expect(summary.results[249].id).toBeTruthy();
   });
 
   it('retries a transient createManyData failure instead of dropping the batch', async () => {

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

@@ -51,7 +51,12 @@ function makeProtocol(opts: { firstCall?: 'throw' | 'shortReturn' } = {}) {
   });
   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));
+    // Supports equality and { $in: [...] } — the id recheck (framework#3173)
+    // queries by pre-assigned id $in, like the real SQL driver does.
+    return store.filter((row) => Object.entries(filter).every(([k, v]) => {
+      if (v && typeof v === 'object' && Array.isArray((v as any).$in)) return (v as any).$in.includes(row[k]);
+      return row[k] === v;
+    }));
   });
   const p: ImportProtocolLike = { findData, createData, updateData: vi.fn(), createManyData };
   return { p, store, createManyData, createData };
@@ -91,17 +96,67 @@ describe('runImport — idempotent retry with natural keys (framework#3149)', ()
     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' });
+  it('pure insert (no matchFields): pre-assigned ids make the retry exactly-once too (#3173)', async () => {
+    const { p, store, createManyData } = makeProtocol({ firstCall: 'throw' });
 
-    await runImport({
+    const summary = 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);
+    // Previously pinned as at-least-once (4 rows). With pre-assigned row ids
+    // the retry rechecks by id and re-inserts nothing — no natural key needed.
+    expect(createManyData).toHaveBeenCalledTimes(1); // committed once; retry only rechecked
+    expect(store).toHaveLength(2);
+    expect(summary.created).toBe(2);
+    expect(summary.errors).toBe(0);
+  });
+
+  it('pure insert: legitimate duplicate rows survive the retry intact (each copy has its own id) (#3173)', async () => {
+    const { p, store } = makeProtocol({ firstCall: 'throw' });
+
+    const summary = await runImport({
+      ...baseOpts, p, writeMode: 'insert', matchFields: [],
+      rows: [{ name: 'same' }, { name: 'same' }], // two intentional copies
+    });
+
+    // A natural-key recheck could not tell the copies apart; the per-row id
+    // recheck keeps exactly the two intended rows — no loss, no duplication.
+    expect(store).toHaveLength(2);
+    expect(summary.created).toBe(2);
+    expect(summary.errors).toBe(0);
+  });
+
+  it('insertManyData (partial success): a bad row is a per-row verdict — good rows never re-run (framework#3172)', async () => {
+    const store: Array<Record<string, any>> = [];
+    const insertManyData = vi.fn(async (args: { records: any[] }) => ({
+      outcomes: args.records.map((r) => {
+        if (r.name === 'bad') return { ok: false, error: new Error('validation failed: bad name') };
+        const rec = { ...r };
+        store.push(rec);
+        return { ok: true, record: rec };
+      }),
+    }));
+    const createManyData = vi.fn();
+    const createData = vi.fn();
+    const p: ImportProtocolLike = {
+      findData: vi.fn(async () => []), createData, updateData: vi.fn(), createManyData, insertManyData,
+    };
+
+    const summary = await runImport({
+      ...baseOpts, p, writeMode: 'insert', matchFields: [],
+      rows: [{ name: 'good1' }, { name: 'bad' }, { name: 'good2' }],
+    });
+
+    expect(insertManyData).toHaveBeenCalledTimes(1); // one call, per-row verdicts
+    expect(createManyData).not.toHaveBeenCalled();   // partial path preferred
+    expect(createData).not.toHaveBeenCalled();       // NO degradation re-run for good rows
+    expect(summary.created).toBe(2);
+    expect(summary.errors).toBe(1);
+    expect(summary.results[0]).toMatchObject({ ok: true, action: 'created' });
+    expect(summary.results[1]).toMatchObject({ ok: false, action: 'failed' });
+    expect(summary.results[2]).toMatchObject({ ok: true, action: 'created' });
+    expect(store).toHaveLength(2);
   });
 
   it('marks rows created-with-warning on a summary recompute failure, without failing or duplicating (framework#3147)', async () => {

packages/rest/src/import-runner.ts

@@ -1,5 +1,6 @@
 // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
 
+import { randomUUID } from 'node:crypto';
 import { coerceRow, firstMissingRequiredField, type RefResolver, type RefMatch } from './import-coerce.js';
 import type { ExportFieldMeta } from './export-format.js';
 import { bulkWrite, withTransientRetry, defaultIsTransientError, type BulkWriteRowResult } from '@objectstack/core';
@@ -76,6 +77,14 @@ export interface ImportProtocolLike {
    * record per input row, in the same order.
    */
   createManyData?(args: { object: string; records: any[]; context?: any; environmentId?: string }): Promise<{ records: any[] }>;
+  /**
+   * Optional partial-success bulk create (framework#3172). When present it is
+   * preferred over `createManyData`: one outcome per input row, in order — a
+   * row that fails validation is a per-row verdict, so a bad row never forces
+   * the whole-batch degradation that re-runs beforeInsert hooks on the good
+   * rows.
+   */
+  insertManyData?(args: { object: string; records: any[]; context?: any; environmentId?: string }): Promise<{ outcomes: Array<{ ok: boolean; record?: any; error?: unknown }> }>;
 }
 
 export interface RunImportOptions {
@@ -272,12 +281,19 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
   // without `createManyData` never buffers — `canBulkCreate` is false and
   // creates fall back to the original inline per-row `createData` call.
   const canBulkCreate = typeof p.createManyData === 'function';
+  // Partial-success flush (framework#3172): preferred when the protocol
+  // offers it — a row that fails validation is a per-row verdict from one
+  // batch call, so a bad row never forces the whole-batch degradation that
+  // re-runs beforeInsert hooks on its siblings.
+  const canPartialCreate = typeof p.insertManyData === 'function';
   const pendingCreates: Array<{ index: number; rowNo: number; data: Record<string, any> }> = [];
   // 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), recheck before re-creating so a retry can't
-  // duplicate a row (framework#3149). A pure-insert import (no matchFields) has
-  // no natural key to recheck against and stays at-least-once by contract.
+  // re-run a create whose prior attempt already committed. Every buffered
+  // CREATE row is therefore pre-assigned a client-generated id at flush time
+  // (framework#3173) — stable across attempts — so a retry can recheck by id
+  // ($in) and re-insert only the rows that truly did not land. This is exact
+  // for EVERY write mode (including pure insert with legitimate duplicate
+  // rows, where a natural-key recheck could not distinguish copies).
   let lastBatchUncertain = false;
   // Set when a flush's write succeeded but its post-write roll-up summary
   // recompute exhausted retries (framework#3147). The rows ARE written; we mark
@@ -296,15 +312,42 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
     }
     return null;
   };
-  const existingByMatch = async (data: Record<string, any>): Promise<Record<string, any> | null> => {
-    if (matchFields.length === 0) return null;
-    const found = await findExisting(data);
-    return found && typeof found === 'object' ? found : null; // ignore 'blank'/'none'/'ambiguous'
+  // Exact idempotency recheck (framework#3173): buffered CREATE rows carry a
+  // pre-assigned id, so "did the lost-response attempt actually commit?" is
+  // answered precisely by an id $in query — no natural key, no clocks, and
+  // legitimate duplicate rows (pure insert mode) resolve correctly because
+  // each copy has its own id.
+  const recheckByIds = async (chunk: Array<Record<string, any>>): Promise<Map<string, any>> => {
+    const ids = chunk.map((r) => r.id).filter((v) => v != null && v !== '');
+    if (ids.length === 0) return new Map();
+    const r = await p.findData({
+      ...findArgsBase({ $filter: { id: { $in: ids } }, $top: ids.length }),
+      object: objectName,
+    });
+    return new Map(findRows(r).map((rec: any) => [String(rec.id), rec]));
   };
   const flushPendingCreates = async (): Promise<void> => {
     if (pendingCreates.length === 0) return;
     flushSummaryStale = false;
     const batch = pendingCreates.splice(0, pendingCreates.length);
+    // Pre-assign ids once per row (framework#3173) — the closures below see
+    // the same row objects on every retry attempt, so the ids are stable. An
+    // id the user supplied explicitly is respected.
+    for (const b of batch) {
+      if (b.data.id == null || b.data.id === '') b.data.id = randomUUID();
+    }
+    // Recheck helper shared by both write paths: on attempt > 1 split the
+    // chunk into rows that already landed (by id) and rows still to create.
+    const splitByExisting = async (chunk: Array<Record<string, any>>) => {
+      const existingByIdx = new Map<number, Record<string, any>>();
+      const toCreate: Array<Record<string, any>> = [];
+      const found = await recheckByIds(chunk);
+      chunk.forEach((row, i) => {
+        const hit = row.id != null ? found.get(String(row.id)) : undefined;
+        if (hit) existingByIdx.set(i, hit); else toCreate.push(row);
+      });
+      return { existingByIdx, toCreate };
+    };
     const writeResults: BulkWriteRowResult[] = await bulkWrite(
       batch.map(b => b.data),
       {
@@ -313,17 +356,57 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
         // the issue's suggested 100-500 rows/batch must not translate into
         // one oversized multi-row INSERT statement.
         batchSize: Math.min(progressEvery, MAX_CREATE_BATCH_SIZE),
+        // Partial-success path (framework#3172): per-row verdicts from one
+        // call; a bad row never degrades the batch.
+        ...(canPartialCreate ? {
+          writeBatchPartial: async (chunk: Array<Record<string, any>>, { attempt }: { attempt: number }) => {
+            let toCreate = chunk;
+            let existingByIdx = new Map<number, Record<string, any>>();
+            if (attempt > 1) {
+              ({ existingByIdx, toCreate } = await splitByExisting(chunk));
+            }
+            try {
+              let freshOutcomes: Array<{ ok: boolean; record?: any; error?: unknown }>;
+              if (toCreate.length === 0) {
+                freshOutcomes = [];
+              } else {
+                try {
+                  freshOutcomes = (await p.insertManyData!({
+                    object: objectName, records: toCreate, context: writeCtx,
+                    ...(environmentId ? { environmentId } : {}),
+                  })).outcomes;
+                } catch (e) {
+                  // Rows written but summary recompute failed: recover the
+                  // outcome array carried on the error (framework#3147).
+                  const recovered = recoverSummaryStale(e);
+                  if (!recovered) throw e;
+                  freshOutcomes = recovered as Array<{ ok: boolean; record?: any; error?: unknown }>;
+                }
+              }
+              if (!Array.isArray(freshOutcomes) || freshOutcomes.length !== toCreate.length) {
+                throw Object.assign(
+                  new Error(`insertManyData returned ${Array.isArray(freshOutcomes) ? `${freshOutcomes.length} outcome(s)` : String(typeof freshOutcomes)} for ${toCreate.length} row(s)`),
+                  { code: 'ERR_BULK_RESULT_MISMATCH' },
+                );
+              }
+              lastBatchUncertain = false;
+              let k = 0;
+              return chunk.map((_row, i) => existingByIdx.has(i)
+                ? { ok: true, record: existingByIdx.get(i)! }
+                : freshOutcomes[k++]);
+            } catch (e) {
+              lastBatchUncertain = isUncertainOutcome(e);
+              throw e;
+            }
+          },
+        } : {}),
         writeBatch: async (chunk, { attempt }) => {
           let toCreate = chunk;
-          const existingByIdx = new Map<number, Record<string, any>>();
-          if (attempt > 1 && matchFields.length > 0) {
+          let existingByIdx = new Map<number, Record<string, any>>();
+          if (attempt > 1) {
             // A prior attempt may have committed before its response was lost:
-            // recheck each row by matchFields and only create the ones missing.
-            toCreate = [];
-            for (let i = 0; i < chunk.length; i++) {
-              const hit = await existingByMatch(chunk[i]);
-              if (hit) existingByIdx.set(i, hit); else toCreate.push(chunk[i]);
-            }
+            // recheck by pre-assigned id and only create the missing rows.
+            ({ existingByIdx, toCreate } = await splitByExisting(chunk));
           }
           try {
             let createdRecords: any[];
@@ -362,8 +445,9 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
           }
         },
         writeOne: async (row, { attempt }) => {
-          if ((attempt > 1 || lastBatchUncertain) && matchFields.length > 0) {
-            const hit = await existingByMatch(row);
+          if (attempt > 1 || lastBatchUncertain) {
+            const found = await recheckByIds([row]);
+            const hit = row.id != null ? found.get(String(row.id)) : undefined;
             if (hit) return hit; // already committed by a prior attempt
           }
           try {