feat: partial-success batch insert + exact import-retry idempotency (#3191)

Fixes #3172, #3173.

- #3172 engine.insertMany: partial-success batch insert (bad rows culled per-row, beforeInsert fires once) + bulkWrite.writeBatchPartial; seed/import adopt via feature detection.
- #3173 pre-assigned import row ids + id $in recheck → exactly-once retry for every write mode (incl. legitimate duplicate rows).
- Real-SqlDriver (better-sqlite3) regression test exercising the turso failure shapes the mock couldn't.

Author: os-zhuang

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

@@ -173,6 +173,60 @@ describe('bulkWrite', () => {
     expect(seen).toEqual([1, 2]); // attempt 1 threw, attempt 2 succeeded
   });
 
+  it('writeBatchPartial: per-row failures are final verdicts — no writeOne degradation (#3172)', async () => {
+    const writeBatchPartial = vi.fn(async (batch: { n: number; bad?: boolean }[]) =>
+      batch.map((r) => (r.bad
+        ? { ok: false, error: new Error('validation failed: bad row') }
+        : { ok: true, record: { id: `r${r.n}` } })));
+    const writeBatch = vi.fn();
+    const writeOne = vi.fn();
+
+    const results = await bulkWrite(
+      [{ n: 1 }, { n: 2, bad: true }, { n: 3 }],
+      { batchSize: 10, writeBatchPartial, writeBatch, writeOne, sleep: noopSleep },
+    );
+
+    expect(writeBatchPartial).toHaveBeenCalledTimes(1);
+    expect(writeBatch).not.toHaveBeenCalled();  // partial path replaces writeBatch
+    expect(writeOne).not.toHaveBeenCalled();    // per-row failure ≠ batch failure
+    expect(results[0]).toMatchObject({ index: 0, ok: true, record: { id: 'r1' } });
+    expect(results[1].ok).toBe(false);
+    expect(results[2]).toMatchObject({ index: 2, ok: true, record: { id: 'r3' } });
+  });
+
+  it('writeBatchPartial: a THROWN error still degrades to writeOne, with transient retry first (#3172)', async () => {
+    let attempts = 0;
+    const writeBatchPartial = vi.fn(async (batch: { n: number }[]) => {
+      attempts++;
+      if (attempts === 1) throw new Error('fetch failed'); // transient → retried
+      if (attempts === 2) throw new Error('CHECK constraint failed'); // logical → degrade
+      return batch.map((r) => ({ ok: true, record: { 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, writeBatchPartial, writeBatch: vi.fn(), writeOne, sleep: noopSleep },
+    );
+
+    expect(writeBatchPartial).toHaveBeenCalledTimes(2); // transient retry happened
+    expect(writeOne).toHaveBeenCalledTimes(2);          // then per-row degradation
+    expect(results.every((r) => r.ok)).toBe(true);
+  });
+
+  it('writeBatchPartial: a wrong-length outcome array is rejected as a failed batch (#3172)', async () => {
+    const writeBatchPartial = vi.fn(async () => [{ ok: true, record: { id: 'only-one' } }]);
+    const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
+
+    const results = await bulkWrite(
+      [{ n: 1 }, { n: 2 }],
+      { batchSize: 10, writeBatchPartial, writeBatch: vi.fn(), writeOne, sleep: noopSleep },
+    );
+
+    expect(writeOne).toHaveBeenCalledTimes(2); // degraded instead of trusting the short array
+    expect(results.every((r) => r.ok)).toBe(true);
+  });
+
   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

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

@@ -79,6 +79,21 @@ export interface BulkWriteOptions<TRow, TRecord = any> extends RetryOptions {
    * carries the same recheck signal as {@link writeBatch}.
    */
   writeOne: (row: TRow, ctx: { attempt: number }) => Promise<TRecord>;
+  /**
+   * Partial-success batch write (framework#3172). When provided it is used
+   * INSTEAD of {@link writeBatch}: it must resolve one outcome per input row,
+   * in input order — `{ ok: true, record }` for written rows, `{ ok: false,
+   * error }` for rows that failed individually (e.g. validation). Per-row
+   * failures are final verdicts: bulkWrite records them as-is and does NOT
+   * degrade to `writeOne` for them — that is the whole point (a degradation
+   * re-run would re-fire beforeInsert hooks on the good rows). Only a THROWN
+   * error (a transient infra failure, a result-count mismatch) falls back to
+   * the per-row `writeOne` degradation, exactly like `writeBatch`.
+   */
+  writeBatchPartial?: (
+    batch: TRow[],
+    ctx: { attempt: number },
+  ) => Promise<Array<{ ok: boolean; record?: TRecord; error?: unknown }>>;
 }
 
 const DEFAULT_BATCH_SIZE = 200;
@@ -202,6 +217,29 @@ export async function bulkWrite<TRow, TRecord = any>(
   for (let start = 0; start < rows.length; start += batchSize) {
     const batch = rows.slice(start, start + batchSize);
     try {
+      // Partial-success path (framework#3172): one call yields a final per-row
+      // verdict, so a row that fails validation never triggers the whole-batch
+      // degradation that re-runs beforeInsert hooks on its siblings.
+      if (opts.writeBatchPartial) {
+        const outcomes = await withRetry((attempt) => opts.writeBatchPartial!(batch, { attempt }), retryOpts);
+        if (!Array.isArray(outcomes) || outcomes.length !== batch.length) {
+          throw Object.assign(
+            new Error(
+              `bulkWrite: writeBatchPartial returned ${
+                Array.isArray(outcomes) ? `${outcomes.length} outcome(s)` : String(typeof outcomes)
+              } for a ${batch.length}-row batch — treating batch as failed`,
+            ),
+            { code: 'ERR_BULK_RESULT_MISMATCH' },
+          );
+        }
+        for (let i = 0; i < batch.length; i++) {
+          const o = outcomes[i];
+          results[start + i] = o.ok
+            ? { index: start + i, ok: true, record: o.record }
+            : { index: start + i, ok: false, error: o.error };
+        }
+        continue;
+      }
       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

packages/metadata-protocol/src/protocol.ts

@@ -3262,6 +3262,34 @@ export class ObjectStackProtocolImplementation implements ObjectStackProtocol {
             count: records.length
         };
     }
+
+    /**
+     * Partial-success bulk create (framework#3172): like createManyData, but a
+     * row that fails validation is a per-row verdict instead of aborting the
+     * whole batch — the import runner uses this so a bad row never forces a
+     * degradation re-run of the good rows' beforeInsert hooks. Requires an
+     * engine with `insertMany` (ObjectQL has it); absent that, callers should
+     * fall back to createManyData.
+     */
+    async insertManyData(request: { object: string, records: any[], context?: any }): Promise<{ object: string; outcomes: Array<{ ok: boolean; record?: any; error?: unknown }> }> {
+        const engineInsertMany = (this.engine as any)?.insertMany;
+        if (typeof engineInsertMany !== 'function') {
+            throw new Error('insertManyData requires an engine with insertMany (framework#3172)');
+        }
+        // Same ingress strip as createManyData (#3043).
+        const rows = stripReadonlyForInsert(
+            this.engine.registry?.getObject(request.object),
+            request.records,
+            request.context,
+        );
+        const outcomes = await engineInsertMany.call(
+            this.engine,
+            request.object,
+            rows,
+            request.context !== undefined ? { context: request.context } as any : undefined,
+        );
+        return { object: request.object, outcomes };
+    }
 
     async updateManyData(request: UpdateManyDataRequest): Promise<BatchUpdateResponse> {
         const { object, records, options } = request;

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

@@ -180,6 +180,55 @@ describe('seed batched path — idempotent retry after commit-then-lost-response
   });
 });
 
+describe('seed batched path — partial-success engine (framework#3172)', () => {
+  it('a bad row is a per-row verdict: good rows insert once, no degradation re-run', async () => {
+    const { engine, store } = createFaithfulEngine();
+    const metadata = createMetadata();
+
+    // Engine with insertMany (partial success): the bad row comes back as a
+    // per-row error, good rows as records — one call, no thrown batch error.
+    let insertManyCalls = 0;
+    (engine as any).insertMany = vi.fn(async (obj: string, rows: any[], opts: any) => {
+      insertManyCalls++;
+      const outcomes = [];
+      for (const r of rows) {
+        if (r.sku === 'BAD') {
+          outcomes.push({ ok: false, error: new Error('validation failed: bad widget') });
+        } else {
+          outcomes.push({ ok: true, record: await (engine.insert as any)(obj, r, opts) });
+        }
+      }
+      return outcomes;
+    });
+
+    const result = await new SeedLoaderService(engine, metadata, createLogger()).load({
+      seeds: [{
+        object: 'my_app_widget',
+        externalId: 'sku',
+        mode: 'insert',
+        env: ['prod', 'dev', 'test'],
+        records: [
+          { name: 'Good A', sku: 'W-A' },
+          { name: 'Bad', sku: 'BAD' },
+          { name: 'Good B', sku: 'W-B' },
+        ],
+      }] as any,
+      config: CONFIG,
+    });
+
+    expect(insertManyCalls).toBe(1);
+    expect(result.summary.totalInserted).toBe(2);
+    expect(result.summary.totalErrored).toBe(1);
+    expect(store.my_app_widget.map((r: any) => r.sku).sort()).toEqual(['W-A', 'W-B']);
+    // The good rows were written exactly once — no whole-batch degradation
+    // re-ran them through engine.insert (single-row form).
+    const singleInsertCalls = (engine.insert as any).mock.calls.filter(
+      ([obj, data]: [string, any]) => obj === 'my_app_widget' && !Array.isArray(data),
+    );
+    expect(singleInsertCalls).toHaveLength(2); // only the two calls made INSIDE insertMany's mock
+  });
+});
+
 describe('seed batched path — summary recompute failure is a warning, not an error (framework#3147)', () => {
   it('records the rows as inserted (not errored) and does not re-insert on ERR_SUMMARY_RECOMPUTE', async () => {
     const { engine, store } = createFaithfulEngine();

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

@@ -271,25 +271,23 @@ export class SeedLoaderService implements ISeedLoaderService {
     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++];
-      });
-    };
+    // Partial-success entry (framework#3172): when the engine offers
+    // insertMany, a row that fails validation is a final per-row verdict from
+    // ONE batch call — bulkWrite never degrades, so beforeInsert hooks run
+    // exactly once per row. Feature-detected so a legacy engine (tests, other
+    // IDataEngine impls) falls back to the whole-array insert + degradation.
+    const engineInsertMany: ((o: string, rows: any[], op: any) => Promise<Array<{ ok: boolean; record?: any; error?: unknown }>>) | undefined =
+      typeof (this.engine as any).insertMany === 'function'
+        ? (o, rows, op) => (this.engine as any).insertMany(o, rows, op)
+        : undefined;
     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,
-          writeBatch: async (rows, { attempt }) => {
+          writeBatchPartial: async (rows, { attempt }) => {
             let toInsert = rows;
             let existing = new Map<string, any>();
             if (attempt > 1) {
@@ -299,21 +297,42 @@ export class SeedLoaderService implements ISeedLoaderService {
               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
+              let freshOutcomes: Array<{ ok: boolean; record?: any; error?: unknown }>;
+              if (toInsert.length === 0) {
+                freshOutcomes = [];
+              } else if (engineInsertMany) {
+                // Partial-success batch: per-row verdicts, hooks fire once.
+                // On ERR_SUMMARY_RECOMPUTE, writeRecoveringSummary hands back
+                // e.written — which for insertMany IS the outcome array.
+                freshOutcomes = await this.writeRecoveringSummary(() => engineInsertMany(objectName, toInsert, opts));
+              } else {
+                // Legacy whole-array insert: any bad row throws the batch, and
+                // bulkWrite's per-row degradation (writeOne) sorts it out. A
+                // lone row keeps the historical bare-record insert() shape.
+                const recs = toInsert.length === 1
                   ? [await this.writeRecoveringSummary(() => this.engine.insert(objectName, toInsert[0], opts))]
                   : await this.writeRecoveringSummary(() => this.engine.insert(objectName, toInsert, opts));
+                freshOutcomes = (recs as any[]).map((r) => ({ ok: true, record: r }));
+              }
               lastBatchUncertain = false;
-              return assembleInOrder(rows, existing, freshlyInserted as any[]);
+              // Reassemble one outcome per input row in order: recheck hits use
+              // the existing record; the rest consume freshOutcomes in order.
+              let k = 0;
+              return rows.map((r) => {
+                const key = extIdOf(r);
+                if (key && existing.has(key)) return { ok: true, record: existing.get(key) };
+                return freshOutcomes[k++];
+              });
             } catch (e) {
               lastBatchUncertain = isUncertainOutcome(e);
               throw e;
             }
           },
+          writeBatch: async () => {
+            // Unreachable — writeBatchPartial above takes precedence — but the
+            // BulkWriteOptions contract requires it.
+            throw new Error('seed-loader uses writeBatchPartial');
+          },
           writeOne: async (row, { attempt }) => {
             if (attempt > 1 || lastBatchUncertain) {
               const key = extIdOf(row);