fix: harden the bulk-write path — retries, idempotency, contracts, summary visibility (#3161)

Fixes #3147, #3148, #3149, #3150, #3151, #3152.

- #3151 bulkWrite + engine.insert(array) validate write-result contracts
- #3150 wrap remaining bare writes in withTransientRetry; short-circuit logical errors
- #3148 import resolveRef flushes pending creates; no negative caching
- #3149 attempt-aware natural-key recheck makes batch retry idempotent
- #3147 recomputeSummaries retries and surfaces SummaryRecomputeError
- #3152 assign autonumbers after validation; document at-least-once hooks

Author: os-zhuang

.changeset/bulk-write-hardening.md

@@ -0,0 +1,31 @@
+---
+"@objectstack/core": patch
+"@objectstack/objectql": patch
+"@objectstack/rest": patch
+"@objectstack/metadata-protocol": patch
+---
+
+fix: harden the bulk-write path — retries, idempotency, contracts, and summary visibility (#3147–#3152)
+
+Six reliability fixes to the batched seed/import + `engine.insert(array)` path
+introduced by the #2678 bulk-write rework:
+
+- **#3151** `bulkWrite` validates that `writeBatch` returns one record per input
+  row (a short/long/non-array return is degraded per-row, not backfilled as
+  phantom success); `engine.insert(array)` likewise rejects a short driver
+  `bulkCreate` return instead of padding afterInsert with `undefined`.
+- **#3150** wraps the two remaining un-retried write points (seed
+  `writeRecord`/`resolveDeferredUpdates`, import's no-`createManyData`
+  fallback) in `withTransientRetry`; `defaultIsTransientError` short-circuits
+  definitive logical errors to non-transient.
+- **#3148** import `resolveRef` flushes pending creates on a same-object miss so
+  a later row can reference an earlier same-file CREATE, and no longer
+  negatively caches a miss.
+- **#3149** threads an `attempt` counter through `bulkWrite`; seed rechecks by
+  `externalId` and import by `matchFields` before re-writing, so a
+  commit-then-lost-response retry cannot duplicate a batch.
+- **#3147** `recomputeSummaries` retries transient failures and, on exhaustion,
+  surfaces `SummaryRecomputeError` (`ERR_SUMMARY_RECOMPUTE`) instead of a
+  silent warn; seed/import recover it to a warning without re-writing.
+- **#3152** autonumbers are assigned after validation, so a batch that dies in
+  validation consumes no sequence value (no number-range gaps).

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

@@ -112,6 +112,81 @@ 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');
+  });
+
+  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', () => {
@@ -151,4 +226,14 @@ describe('defaultIsTransientError', () => {
     expect(defaultIsTransientError(new Error('Validation failed: email is required'))).toBe(false);
     expect(defaultIsTransientError(new Error('UNIQUE constraint failed: task.id'))).toBe(false);
   });
+
+  it('classifies a mixed constraint+network message as logical, not transient (#3150)', () => {
+    // A logical signature must win even when a transient keyword also appears,
+    // so we do not burn retries on a row that will fail identically each time.
+    expect(defaultIsTransientError(new Error('CHECK constraint failed: network_zone'))).toBe(false);
+    expect(defaultIsTransientError(new Error('column network_id is not allowed'))).toBe(false);
+    expect(defaultIsTransientError(new Error('value out of range at row 503'))).toBe(false);
+    // Pure transient signatures are unaffected.
+    expect(defaultIsTransientError(new Error('fetch failed'))).toBe(true);
+  });
 });

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> {
@@ -56,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[], 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}.
    */
-  writeBatch: (batch: TRow[]) => Promise<TRecord[]>;
-  /** Write a single row — used only to degrade a failed batch. */
-  writeOne: (row: TRow) => Promise<TRecord>;
+  writeOne: (row: TRow, ctx: { attempt: number }) => Promise<TRecord>;
 }
 
 const DEFAULT_BATCH_SIZE = 200;
@@ -86,11 +105,33 @@ const TRANSIENT_PATTERNS: RegExp[] = [
 
 const TRANSIENT_CODES = /^(ECONNRESET|ECONNREFUSED|ECONNABORTED|EPIPE|EAI_AGAIN|ETIMEDOUT|EHOSTUNREACH|ENETUNREACH|ENOTFOUND)$/i;
 
+/**
+ * Validation / constraint / schema signatures that are DEFINITIVELY logical,
+ * never worth retrying. Checked before {@link TRANSIENT_PATTERNS} so a message
+ * that happens to mention both (e.g. `CHECK constraint failed: network_zone`,
+ * `column network_id is not allowed`) is classified as logical rather than
+ * burning retries on a row that will fail identically every time (framework
+ * #3150).
+ */
+const NON_TRANSIENT_PATTERNS: RegExp[] = [
+  /validation/i,
+  /constraint/i,
+  /\brequired\b/i,
+  /\bunique\b/i,
+  /duplicate/i,
+  /not[\s_-]*null/i,
+  /invalid/i,
+  /not allowed/i,
+  /out of range/i,
+];
+
 export function defaultIsTransientError(err: unknown): boolean {
-  const code = (err as { code?: unknown } | null)?.code;
-  if (typeof code === 'string' && TRANSIENT_CODES.test(code)) return true;
   const message = (err as { message?: unknown } | null)?.message;
   const text = typeof message === 'string' ? message : String(err ?? '');
+  // A definitive logical signature wins even if a transient word also appears.
+  if (NON_TRANSIENT_PATTERNS.some((re) => re.test(text))) return false;
+  const code = (err as { code?: unknown } | null)?.code;
+  if (typeof code === 'string' && TRANSIENT_CODES.test(code)) return true;
   return TRANSIENT_PATTERNS.some((re) => re.test(text));
 }
 
@@ -103,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;
@@ -126,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,
@@ -161,7 +202,26 @@ 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
+      // (`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] };
       }
@@ -181,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 };