perf(seed,import): route bulk writes through the engine's batch insert path (#2680)

* perf(seed,import): route bulk writes through the engine's batch insert path

Seed loader and data-import wrote records one at a time even though the
engine's array-form insert() already does the efficient thing (one
driver.bulkCreate round-trip + parent-deduplicated summary recompute).
Add a shared bulkWrite/withTransientRetry helper (@objectstack/core) that
batches writes, retries transient driver errors with backoff, and degrades
to per-row writes when a batch fails for a non-transient reason so one bad
row can't drop the rest. Wire it into SeedLoaderService.loadDataset() and
import-runner's runImport(), and fix Protocol.createManyData to forward
context to engine.insert() like createData already does.

Fixes #2678.

* fix(driver-sql): bulkCreate must assign a client-side id like create() does

SqlDriver.bulkCreate() never ran the id/_id normalization create() applies,
so a row inserted via the array path without a driver-native id default
landed with id: NULL. This bulk path was rarely exercised before #2678
routed real seed/import data through it at scale, surfacing the gap (the
showcase invoice-line dogfood test failed: product/invoice references
resolved to NULL because the referenced object's bulk-inserted rows had no
id to record). bulkCreate() now assigns nanoid(DEFAULT_ID_LENGTH) per row
exactly like create() does.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: os-zhuang

.changeset/seed-import-bulk-insert.md

@@ -0,0 +1,17 @@
+---
+'@objectstack/core': minor
+'@objectstack/metadata-protocol': minor
+'@objectstack/rest': minor
+'@objectstack/driver-sql': patch
+---
+
+Seed loader and data-import now route bulk writes through the engine's array-form `insert()` (one round-trip per batch, with parent-deduplicated summary recompute) instead of one `insert()`/`createData()` call per record, and both retry transient driver errors instead of silently dropping the row (#2678).
+
+A new shared helper, `bulkWrite` (`@objectstack/core`), batches rows through a caller-supplied batch-write function, retries a whole-batch transient failure (network blip / timeout) with exponential backoff, and degrades to per-row writes (each itself retried) when a batch fails for a non-transient reason — so one bad row can't drop the other N-1. `withTransientRetry` wraps a single write (e.g. an update) with the same retry behavior.
+
+- `SeedLoaderService.loadDataset()` (`@objectstack/metadata-protocol`) buffers insert-mode records and flushes them in batches of 200 via the engine's array `insert()`. Datasets with a self-referencing field (e.g. `employee.manager_id -> employee`) keep the historical per-record write path, since a later record may need an earlier one's freshly-assigned id.
+- `runImport()` (`@objectstack/rest`) buffers create-resolved rows and flushes them via `protocol.createManyData()` when the protocol supports it, falling back to the original per-row `createData()` call otherwise. `Protocol.createManyData` (`@objectstack/metadata-protocol`) now forwards `context` to `engine.insert()` like `createData` already did, so tenant-scoped bulk creates work correctly.
+
+Previously, a 1000-row seed or import into an object with a rollup summary issued 1000+ round-trips and up to 1000 summary recomputes; a single transient network error on any one row silently dropped it with no retry (the 2026-07-06 HotCRM first-boot incident). A `bulkCreate`-capable driver now sees roughly `ceil(N/batch)` writes, and a transient error is retried before a row is ever reported as failed.
+
+**Fix (`@objectstack/driver-sql`):** `SqlDriver.bulkCreate()` never generated a client-side id for a row missing one, unlike `create()` — a latent gap that this change is the first to exercise at scale (a bulk-inserted row without a driver-native id default silently landed with `id: NULL`). `bulkCreate()` now mirrors `create()`'s id/`_id` normalization per row.

packages/core/src/index.ts

@@ -26,6 +26,9 @@ export * from './utils/env.js';
 // Export timezone-aware calendar utilities (ADR-0053 Phase 2)
 export * from './utils/datetime.js';
 
+// Export the shared batched-write helper (framework#2678)
+export * from './utils/bulk-write.js';
+
 // Export in-memory fallbacks for core-criticality services
 export * from './fallbacks/index.js';
 

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

@@ -0,0 +1,154 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+import { describe, it, expect, vi } from 'vitest';
+import { bulkWrite, defaultIsTransientError, withTransientRetry } from './bulk-write';
+
+const noopSleep = async () => {};
+
+describe('bulkWrite', () => {
+  it('writes N rows in ceil(N/batchSize) batch calls, not N single-row calls', async () => {
+    const rows = Array.from({ length: 250 }, (_, i) => ({ n: i }));
+    const writeBatch = vi.fn(async (batch: { n: number }[]) => batch.map((r) => ({ id: `r${r.n}` })));
+    const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
+
+    const results = await bulkWrite(rows, { batchSize: 100, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeBatch).toHaveBeenCalledTimes(3); // ceil(250/100)
+    expect(writeOne).not.toHaveBeenCalled();
+    expect(results).toHaveLength(250);
+    expect(results.every((r) => r.ok)).toBe(true);
+    // Results stay correlated to original row order across batch boundaries.
+    expect(results[0].record).toEqual({ id: 'r0' });
+    expect(results[249].record).toEqual({ id: 'r249' });
+  });
+
+  it('retries a whole-batch transient failure and does not degrade to per-row on success', async () => {
+    let attempts = 0;
+    const writeBatch = vi.fn(async (batch: { n: number }[]) => {
+      attempts++;
+      if (attempts === 1) throw new Error('fetch failed');
+      return batch.map((r) => ({ id: `r${r.n}` }));
+    });
+    const writeOne = vi.fn(async () => ({ id: 'x' }));
+
+    const rows = [{ n: 1 }, { n: 2 }];
+    const results = await bulkWrite(rows, { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeBatch).toHaveBeenCalledTimes(2);
+    expect(writeOne).not.toHaveBeenCalled();
+    expect(results.every((r) => r.ok)).toBe(true);
+  });
+
+  it('degrades to per-row on a logical (non-transient) batch error, without failing the other rows', async () => {
+    const writeBatch = vi.fn(async () => {
+      throw new Error('CHECK constraint failed: score >= 0');
+    });
+    const writeOne = vi.fn(async (row: { n: number; bad?: boolean }) => {
+      if (row.bad) throw new Error('CHECK constraint failed: score >= 0');
+      return { id: `r${row.n}` };
+    });
+
+    const rows = [{ n: 1 }, { n: 2, bad: true }, { n: 3 }];
+    const results = await bulkWrite(rows, { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeBatch).toHaveBeenCalledTimes(1);
+    expect(writeOne).toHaveBeenCalledTimes(3);
+    expect(results[0]).toMatchObject({ index: 0, ok: true, record: { id: 'r1' } });
+    expect(results[1].ok).toBe(false);
+    expect(results[1].error).toBeInstanceOf(Error);
+    expect(results[2]).toMatchObject({ index: 2, ok: true, record: { id: 'r3' } });
+  });
+
+  it('exhausts batch retries on a persistent transient error, then still degrades to per-row', async () => {
+    const writeBatch = vi.fn(async () => {
+      throw new Error('ETIMEDOUT');
+    });
+    const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
+
+    const rows = [{ n: 1 }, { n: 2 }];
+    const results = await bulkWrite(rows, {
+      batchSize: 10, maxRetries: 2, writeBatch, writeOne, sleep: noopSleep,
+    });
+
+    expect(writeBatch).toHaveBeenCalledTimes(2); // maxRetries batch attempts
+    expect(writeOne).toHaveBeenCalledTimes(2); // then one call per row
+    expect(results.every((r) => r.ok)).toBe(true);
+  });
+
+  it('keeps per-row index correlation intact across multiple batches with a mid-stream failure', async () => {
+    const writeBatch = vi.fn(async (batch: { n: number; bad?: boolean }[]) => {
+      if (batch.some((r) => r.bad)) throw new Error('validation error');
+      return batch.map((r) => ({ id: `r${r.n}` }));
+    });
+    const writeOne = vi.fn(async (row: { n: number; bad?: boolean }) => {
+      if (row.bad) throw new Error('validation error');
+      return { id: `r${row.n}` };
+    });
+
+    const rows = [{ n: 0 }, { n: 1 }, { n: 2, bad: true }, { n: 3 }, { n: 4 }];
+    const results = await bulkWrite(rows, { batchSize: 2, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(results.map((r) => r.index)).toEqual([0, 1, 2, 3, 4]);
+    expect(results[2].ok).toBe(false);
+    expect(results.filter((r) => r.ok)).toHaveLength(4);
+  });
+
+  it('never retries a logical error at the row level either', async () => {
+    const writeBatch = vi.fn(async () => { throw new Error('logical'); });
+    const writeOne = vi.fn(async () => { throw new Error('NOT NULL constraint failed'); });
+
+    const results = await bulkWrite([{ n: 1 }, { n: 2 }], { batchSize: 10, maxRetries: 3, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeOne).toHaveBeenCalledTimes(2); // no retry per row — not transient
+    expect(results.every((r) => !r.ok)).toBe(true);
+  });
+
+  it('a single-row batch failure is the row result directly, without a redundant writeOne call', async () => {
+    const writeBatch = vi.fn(async () => { throw new Error('logical'); });
+    const writeOne = vi.fn(async () => ({ id: 'should-not-be-called' }));
+
+    const results = await bulkWrite([{ n: 1 }], { batchSize: 10, writeBatch, writeOne, sleep: noopSleep });
+
+    expect(writeOne).not.toHaveBeenCalled();
+    expect(results[0]).toMatchObject({ index: 0, ok: false });
+  });
+});
+
+describe('withTransientRetry', () => {
+  it('retries a transient failure and returns the eventual success', async () => {
+    let attempts = 0;
+    const fn = vi.fn(async () => {
+      attempts++;
+      if (attempts < 3) throw new Error('fetch failed');
+      return 'ok';
+    });
+
+    const result = await withTransientRetry(fn, { sleep: noopSleep });
+
+    expect(result).toBe('ok');
+    expect(fn).toHaveBeenCalledTimes(3);
+  });
+
+  it('does not retry a logical error', async () => {
+    const fn = vi.fn(async () => { throw new Error('UNIQUE constraint failed'); });
+
+    await expect(withTransientRetry(fn, { sleep: noopSleep })).rejects.toThrow('UNIQUE constraint failed');
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe('defaultIsTransientError', () => {
+  it('recognizes common transient network signatures', () => {
+    expect(defaultIsTransientError(new Error('fetch failed'))).toBe(true);
+    expect(defaultIsTransientError(new Error('request timed out after 30000ms'))).toBe(true);
+    expect(defaultIsTransientError(new Error('socket hang up'))).toBe(true);
+    expect(defaultIsTransientError(Object.assign(new Error('connect ECONNRESET'), { code: 'ECONNRESET' }))).toBe(true);
+    expect(defaultIsTransientError(new Error('Service returned 503'))).toBe(true);
+  });
+
+  it('does not classify validation/constraint errors as transient', () => {
+    expect(defaultIsTransientError(new Error('NOT NULL constraint failed: task.title'))).toBe(false);
+    expect(defaultIsTransientError(new Error('Validation failed: email is required'))).toBe(false);
+    expect(defaultIsTransientError(new Error('UNIQUE constraint failed: task.id'))).toBe(false);
+  });
+});

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

@@ -0,0 +1,194 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * `bulkWrite` — the shared batched-write helper used by BOTH the seed loader
+ * (`@objectstack/metadata-protocol`) and the data-import runner
+ * (`@objectstack/rest`), so neither reimplements batching, transient-error
+ * retry, or per-row degradation. See framework#2678.
+ *
+ * ObjectQL's engine already does the efficient thing when handed an ARRAY —
+ * one `driver.bulkCreate` round-trip plus parent-deduplicated summary
+ * recompute (`engine.insert(object, rows[])`) — but seed/import fed it one
+ * record at a time, so neither got the benefit. This module re-chunks rows
+ * into batches and drives them through a caller-supplied batch-write
+ * function, adding:
+ *
+ *  - transient-error retry (network blip / timeout) with exponential
+ *    backoff, so a dropped connection doesn't silently drop the row (the
+ *    2026-07-06 HotCRM incident: a turso `fetch failed` mid-seed dropped rows
+ *    silently because nothing retried);
+ *  - per-row degradation when a batch fails for a non-transient (logical /
+ *    validation) reason, so one bad row can't fail the other N-1 — needed
+ *    because `driver.bulkCreate` is a single multi-row statement/`Promise.all`
+ *    on every driver in this repo (sql, memory, mongodb): one bad row fails
+ *    the whole call;
+ *  - a stable per-row result keyed by the row's original index, so callers
+ *    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).
+ */
+
+export interface BulkWriteRowResult<TRecord = any> {
+  /** Index into the original `rows` array passed to {@link bulkWrite}. */
+  index: number;
+  ok: boolean;
+  record?: TRecord;
+  error?: unknown;
+}
+
+export interface RetryOptions {
+  /** Max attempts for one write (batch or single-row), including the first. Default 3. */
+  maxRetries?: number;
+  /** Base backoff in ms; doubled each retry, plus jitter. Default 200. */
+  backoffBaseMs?: number;
+  /** Classifies an error as transient (worth retrying) vs logical (the row/batch is just bad). */
+  isTransientError?: (err: unknown) => boolean;
+  /** Injectable sleep, for deterministic tests. */
+  sleep?: (ms: number) => Promise<void>;
+}
+
+export interface BulkWriteOptions<TRow, TRecord = any> extends RetryOptions {
+  /** Rows per batch. Default 200 (framework#2678 suggests 100-500). */
+  batchSize?: number;
+  /**
+   * Write one batch. MUST resolve to one record per input row, in the SAME
+   * order as `batch` — {@link bulkWrite} correlates `records[i]` back to
+   * `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`).
+   */
+  writeBatch: (batch: TRow[]) => Promise<TRecord[]>;
+  /** Write a single row — used only to degrade a failed batch. */
+  writeOne: (row: TRow) => Promise<TRecord>;
+}
+
+const DEFAULT_BATCH_SIZE = 200;
+const DEFAULT_MAX_RETRIES = 3;
+const DEFAULT_BACKOFF_BASE_MS = 200;
+
+/**
+ * Transient-error signatures shared by common HTTP/TCP-backed drivers
+ * (turso/libsql's fetch-based transport included). Deliberately excludes
+ * anything that looks like a validation/constraint error — those must NOT be
+ * retried, only degraded to per-row.
+ */
+const TRANSIENT_PATTERNS: RegExp[] = [
+  /fetch failed/i,
+  /network/i,
+  /timed?\s*out/i,
+  /timeout/i,
+  /socket hang ?up/i,
+  /connection.*(closed|reset|refused|terminated|aborted)/i,
+  /\b(502|503|504)\b/,
+  /server.*unavailable/i,
+  /too many connections/i,
+];
+
+const TRANSIENT_CODES = /^(ECONNRESET|ECONNREFUSED|ECONNABORTED|EPIPE|EAI_AGAIN|ETIMEDOUT|EHOSTUNREACH|ENETUNREACH|ENOTFOUND)$/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 ?? '');
+  return TRANSIENT_PATTERNS.some((re) => re.test(text));
+}
+
+const defaultSleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
+
+interface ResolvedRetryOptions {
+  maxRetries: number;
+  backoffBaseMs: number;
+  isTransientError: (err: unknown) => boolean;
+  sleep: (ms: number) => Promise<void>;
+}
+
+async function withRetry<T>(fn: () => Promise<T>, opts: ResolvedRetryOptions): Promise<T> {
+  let lastError: unknown;
+  for (let attempt = 1; attempt <= opts.maxRetries; attempt++) {
+    try {
+      return await fn();
+    } catch (err) {
+      lastError = err;
+      if (attempt >= opts.maxRetries || !opts.isTransientError(err)) throw err;
+      const jitter = Math.floor(Math.random() * 50);
+      await opts.sleep(opts.backoffBaseMs * 2 ** (attempt - 1) + jitter);
+    }
+  }
+  // Unreachable — the loop above always returns or throws — but keeps TS's
+  // control-flow analysis happy about a guaranteed return type.
+  throw lastError;
+}
+
+/**
+ * Retry a single write (e.g. an `engine.update()` call the seed loader or
+ * import runner makes outside the batched-insert path) with the same
+ * 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> {
+  return withRetry(fn, {
+    maxRetries: Math.max(1, opts.maxRetries ?? DEFAULT_MAX_RETRIES),
+    backoffBaseMs: opts.backoffBaseMs ?? DEFAULT_BACKOFF_BASE_MS,
+    isTransientError: opts.isTransientError ?? defaultIsTransientError,
+    sleep: opts.sleep ?? defaultSleep,
+  });
+}
+
+/**
+ * Write `rows` through `opts.writeBatch` in chunks of `opts.batchSize`,
+ * retrying a whole-batch transient failure with backoff, and degrading to
+ * per-row `opts.writeOne` calls (each itself retried) when a batch fails for
+ * a non-transient reason — so one bad row can't drop the rest of the batch.
+ *
+ * Returns one {@link BulkWriteRowResult} per input row, indexed to match
+ * `rows`' original order.
+ */
+export async function bulkWrite<TRow, TRecord = any>(
+  rows: TRow[],
+  opts: BulkWriteOptions<TRow, TRecord>,
+): Promise<BulkWriteRowResult<TRecord>[]> {
+  const batchSize = Math.max(1, opts.batchSize ?? DEFAULT_BATCH_SIZE);
+  const retryOpts: ResolvedRetryOptions = {
+    maxRetries: Math.max(1, opts.maxRetries ?? DEFAULT_MAX_RETRIES),
+    backoffBaseMs: opts.backoffBaseMs ?? DEFAULT_BACKOFF_BASE_MS,
+    isTransientError: opts.isTransientError ?? defaultIsTransientError,
+    sleep: opts.sleep ?? defaultSleep,
+  };
+
+  const results: BulkWriteRowResult<TRecord>[] = new Array(rows.length);
+
+  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);
+      for (let i = 0; i < batch.length; i++) {
+        results[start + i] = { index: start + i, ok: true, record: records[i] };
+      }
+    } catch (batchErr) {
+      // A single-row "batch" already IS the per-row attempt — its failure
+      // (after transient retry) is the row's final outcome; calling
+      // `writeOne` again would just repeat the identical work.
+      if (batch.length === 1) {
+        results[start] = { index: start, ok: false, error: batchErr };
+        continue;
+      }
+      // The batch failed even after transient retry, or failed for a logical
+      // reason retry wouldn't fix. Degrade to per-row so one bad row can't
+      // fail the other rows in this batch. Each row still gets its own
+      // transient retry — the batch-level failure doesn't tell us which row
+      // (if any) was actually the transient one.
+      for (let i = 0; i < batch.length; i++) {
+        const idx = start + i;
+        try {
+          const record = await withRetry(() => opts.writeOne(batch[i]), retryOpts);
+          results[idx] = { index: idx, ok: true, record };
+        } catch (err) {
+          results[idx] = { index: idx, ok: false, error: err };
+        }
+      }
+    }
+  }
+
+  return results;
+}

packages/metadata-protocol/src/protocol.ts

@@ -2918,8 +2918,12 @@ export class ObjectStackProtocolImplementation implements ObjectStackProtocol {
         } as BatchUpdateResponse;
     }
 
-    async createManyData(request: { object: string, records: any[] }): Promise<any> {
-        const records = await this.engine.insert(request.object, request.records);
+    async createManyData(request: { object: string, records: any[], context?: any }): Promise<any> {
+        const records = await this.engine.insert(
+            request.object,
+            request.records,
+            request.context !== undefined ? { context: request.context } as any : undefined,
+        );
         return {
             object: request.object,
             records,