From 3dbb3a472d5687139d9cfbde099cb9f4335f9fa9 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Jul 2026 17:46:40 +0000 Subject: [PATCH 1/2] 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. --- .changeset/seed-import-bulk-insert.md | 14 ++ packages/core/src/index.ts | 3 + packages/core/src/utils/bulk-write.test.ts | 154 ++++++++++++++ packages/core/src/utils/bulk-write.ts | 194 +++++++++++++++++ packages/metadata-protocol/src/protocol.ts | 8 +- packages/metadata-protocol/src/seed-loader.ts | 195 +++++++++++++++--- .../protocol-publish-package-drafts.test.ts | 12 +- packages/rest/src/import-runner-bulk.test.ts | 140 +++++++++++++ packages/rest/src/import-runner.ts | 127 ++++++++++-- packages/runtime/src/http-dispatcher.test.ts | 17 +- packages/runtime/src/seed-loader.test.ts | 17 +- 11 files changed, 821 insertions(+), 60 deletions(-) create mode 100644 .changeset/seed-import-bulk-insert.md create mode 100644 packages/core/src/utils/bulk-write.test.ts create mode 100644 packages/core/src/utils/bulk-write.ts create mode 100644 packages/rest/src/import-runner-bulk.test.ts diff --git a/.changeset/seed-import-bulk-insert.md b/.changeset/seed-import-bulk-insert.md new file mode 100644 index 0000000000..c15d842d6b --- /dev/null +++ b/.changeset/seed-import-bulk-insert.md @@ -0,0 +1,14 @@ +--- +'@objectstack/core': minor +'@objectstack/metadata-protocol': minor +'@objectstack/rest': minor +--- + +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. diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index ff3ce5392a..bc6b78035b 100644 --- a/packages/core/src/index.ts +++ b/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'; diff --git a/packages/core/src/utils/bulk-write.test.ts b/packages/core/src/utils/bulk-write.test.ts new file mode 100644 index 0000000000..a372472e94 --- /dev/null +++ b/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); + }); +}); diff --git a/packages/core/src/utils/bulk-write.ts b/packages/core/src/utils/bulk-write.ts new file mode 100644 index 0000000000..35dc6db11a --- /dev/null +++ b/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 { + /** 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; +} + +export interface BulkWriteOptions 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; + /** Write a single row — used only to degrade a failed batch. */ + writeOne: (row: TRow) => Promise; +} + +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 => new Promise((resolve) => setTimeout(resolve, ms)); + +interface ResolvedRetryOptions { + maxRetries: number; + backoffBaseMs: number; + isTransientError: (err: unknown) => boolean; + sleep: (ms: number) => Promise; +} + +async function withRetry(fn: () => Promise, opts: ResolvedRetryOptions): Promise { + 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(fn: () => Promise, opts: RetryOptions = {}): Promise { + 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( + rows: TRow[], + opts: BulkWriteOptions, +): Promise[]> { + 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[] = 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; +} diff --git a/packages/metadata-protocol/src/protocol.ts b/packages/metadata-protocol/src/protocol.ts index 2e5c3124ff..afc5f59457 100644 --- a/packages/metadata-protocol/src/protocol.ts +++ b/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 { - const records = await this.engine.insert(request.object, request.records); + async createManyData(request: { object: string, records: any[], context?: any }): Promise { + const records = await this.engine.insert( + request.object, + request.records, + request.context !== undefined ? { context: request.context } as any : undefined, + ); return { object: request.object, records, diff --git a/packages/metadata-protocol/src/seed-loader.ts b/packages/metadata-protocol/src/seed-loader.ts index 20a5f22cd6..6b30ab5fa0 100644 --- a/packages/metadata-protocol/src/seed-loader.ts +++ b/packages/metadata-protocol/src/seed-loader.ts @@ -15,6 +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'; interface Logger { info(message: string, meta?: Record): void; @@ -220,6 +221,60 @@ export class SeedLoaderService implements ISeedLoaderService { // Get reference resolutions for this object const objectRefs = refMap.get(objectName) || []; + // Self-referencing objects (e.g. `employee.manager_id -> employee`) can + // have record i reference record j ref.targetObject === objectName); + + // Records resolved as inserts (mode 'insert'/'replace', or an unmatched + // upsert/ignore) are buffered here and flushed in batches through the + // engine's array-form insert() — one round-trip per batch instead of one + // per record, with transient-error retry and per-row degradation on a + // logical/validation failure. See framework#2678. + const pendingInserts: Array<{ recordIndex: number; externalIdValue: string; record: Record }> = []; + const opts = SeedLoaderService.SEED_OPTIONS as any; + const flushPendingInserts = async (): Promise => { + 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), + }, + ); + for (const res of writeResults) { + const { recordIndex, externalIdValue, record } = batch[res.index]; + if (res.ok) { + inserted++; + const internalId = this.extractId(res.record); + if (externalIdValue && internalId) { + insertedRecords.get(objectName)!.set(externalIdValue, internalId); + } + } else { + errored++; + const error = this.buildWriteError(objectName, record, externalId, recordIndex, res.error); + errors.push(error); + allErrors.push(error); + this.logger.warn(`[SeedLoader] ${error.message}`, { recordIndex }); + } + } + }; + // Pin a single `now()` snapshot for the entire dataset so multi-pass // loads see one logical clock — the M9 determinism guarantee for seeds. const seedNow = new Date(); @@ -395,39 +450,61 @@ export class SeedLoaderService implements ISeedLoaderService { // Insert/upsert the record if (!config.dryRun) { - try { - const result = await this.writeRecord( - objectName, record, mode, externalId, existingRecords - ); - - if (result.action === 'inserted') inserted++; - else if (result.action === 'updated') updated++; - else if (result.action === 'skipped') skipped++; - - // Track the inserted/updated record's ID for reference resolution + if (hasSelfRef) { + // Self-referencing dataset: keep the historical sequential + // per-record write so a later record can resolve its self-ref + // against an earlier one via `insertedRecords` — see `hasSelfRef`. + try { + const result = await this.writeRecord( + objectName, record, mode, externalId, existingRecords + ); + + if (result.action === 'inserted') inserted++; + else if (result.action === 'updated') updated++; + else if (result.action === 'skipped') skipped++; + + const externalIdValue = String(record[externalId] ?? ''); + const internalId = result.id; + if (externalIdValue && internalId) { + insertedRecords.get(objectName)!.set(externalIdValue, String(internalId)); + } + } catch (err: any) { + errored++; + const error = this.buildWriteError(objectName, record, externalId, i, err); + errors.push(error); + allErrors.push(error); + this.logger.warn(`[SeedLoader] ${error.message}`, { recordIndex: i }); + } + } else { + const decision = this.decideWriteAction(record, mode, externalId, existingRecords); const externalIdValue = String(record[externalId] ?? ''); - const internalId = result.id; - if (externalIdValue && internalId) { - insertedRecords.get(objectName)!.set(externalIdValue, String(internalId)); + + if (decision.action === 'skip') { + skipped++; + if (decision.id && externalIdValue) { + insertedRecords.get(objectName)!.set(externalIdValue, decision.id); + } + } else if (decision.action === 'update') { + try { + await withTransientRetry(() => this.engine.update(objectName, { ...record, id: decision.id }, opts)); + updated++; + if (externalIdValue) { + insertedRecords.get(objectName)!.set(externalIdValue, decision.id); + } + } catch (err: any) { + errored++; + const error = this.buildWriteError(objectName, record, externalId, i, err); + errors.push(error); + allErrors.push(error); + this.logger.warn(`[SeedLoader] ${error.message}`, { recordIndex: i }); + } + } else { + // Insert: buffer for the batched flush rather than writing now. + pendingInserts.push({ recordIndex: i, externalIdValue, record }); + if (pendingInserts.length >= SeedLoaderService.BULK_BATCH_SIZE) { + await flushPendingInserts(); + } } - } catch (err: any) { - // LOUD FAILURE: write errors were previously only counted + - // warn-logged, so dropped rows were invisible in result.errors and - // the boot summary. Surface them as actionable errors too, so the - // overall load is marked unsuccessful and the reason is reported. - errored++; - const error: ReferenceResolutionError = { - sourceObject: objectName, - field: '(write)', - targetObject: objectName, - targetField: externalId, - attemptedValue: record[externalId] ?? null, - recordIndex: i, - message: `Failed to write ${objectName} record #${i} (${externalId}=${String(record[externalId] ?? '')}): ${err.message}`, - }; - errors.push(error); - allErrors.push(error); - this.logger.warn(`[SeedLoader] ${error.message}`, { recordIndex: i }); } } else { // Dry-run: simulate insert tracking @@ -439,6 +516,10 @@ export class SeedLoaderService implements ISeedLoaderService { } } + if (!config.dryRun) { + await flushPendingInserts(); + } + return { object: objectName, mode, @@ -675,6 +756,58 @@ export class SeedLoaderService implements ISeedLoaderService { } } + /** Rows per batch for the buffered-insert flush. See framework#2678. */ + private static readonly BULK_BATCH_SIZE = 200; + + /** + * Decide what {@link loadDataset}'s non-self-referencing (batched) path + * should do with a record — mirrors {@link writeRecord}'s mode/existing + * logic exactly, but WITHOUT performing the write, so insert decisions can + * be buffered and flushed as a batch instead of one call per record. + */ + private decideWriteAction( + record: Record, + mode: string, + externalId: string, + existingRecords?: Map, + ): { action: 'insert' } | { action: 'update'; id: string } | { action: 'skip'; id?: string } { + const externalIdValue = record[externalId]; + const existing = existingRecords?.get(String(externalIdValue ?? '')); + + switch (mode) { + case 'update': + return existing ? { action: 'update', id: this.extractId(existing)! } : { action: 'skip' }; + case 'upsert': + return existing ? { action: 'update', id: this.extractId(existing)! } : { action: 'insert' }; + case 'ignore': + return existing ? { action: 'skip', id: this.extractId(existing) } : { action: 'insert' }; + case 'insert': + case 'replace': + default: + return { action: 'insert' }; + } + } + + /** Builds the same `ReferenceResolutionError` shape a failed write has always reported. */ + private buildWriteError( + objectName: string, + record: Record, + externalId: string, + recordIndex: number, + err: unknown, + ): ReferenceResolutionError { + const message = (err as { message?: unknown } | null)?.message ?? String(err); + return { + sourceObject: objectName, + field: '(write)', + targetObject: objectName, + targetField: externalId, + attemptedValue: record[externalId] ?? null, + recordIndex, + message: `Failed to write ${objectName} record #${recordIndex} (${externalId}=${String(record[externalId] ?? '')}): ${message}`, + }; + } + // ========================================================================== // Internal: Dependency Graph // ========================================================================== diff --git a/packages/objectql/src/protocol-publish-package-drafts.test.ts b/packages/objectql/src/protocol-publish-package-drafts.test.ts index 87314c553c..dde5cc2456 100644 --- a/packages/objectql/src/protocol-publish-package-drafts.test.ts +++ b/packages/objectql/src/protocol-publish-package-drafts.test.ts @@ -175,8 +175,16 @@ describe('protocol.applySeedBodies — real loader smoke test', () => { const inserted: Array<{ object: string; record: any }> = []; (protocol as any).engine = { find: async () => [], - insert: async (object: string, record: any) => { - inserted.push({ object, record }); + insert: async (object: string, data: any) => { + // Mirror the real engine's array-form insert (bulk path): an array in + // → an array of created records out, same order — see framework#2678. + if (Array.isArray(data)) { + return data.map((record) => { + inserted.push({ object, record }); + return { id: `${object}_${inserted.length}` }; + }); + } + inserted.push({ object, record: data }); return { id: `${object}_${inserted.length}` }; }, update: async () => ({}), diff --git a/packages/rest/src/import-runner-bulk.test.ts b/packages/rest/src/import-runner-bulk.test.ts new file mode 100644 index 0000000000..185b86cde0 --- /dev/null +++ b/packages/rest/src/import-runner-bulk.test.ts @@ -0,0 +1,140 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * runImport() batches CREATE-resolved rows through `p.createManyData` instead + * of one `createData` call per row (framework#2678). These tests exercise + * that integration directly against a mock `ImportProtocolLike`, independent + * of the generic `bulkWrite` unit tests in `@objectstack/core`. + */ + +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([ + ['name', { name: 'name', type: 'text' }], +]); + +const baseOpts = { + objectName: 'task', + metaMap, + writeMode: 'insert' as const, + matchFields: [] as string[], + dryRun: false, + runAutomations: false, + trimWhitespace: true, + createMissingOptions: false, + skipBlankMatchKey: false, +}; + +function rowsOf(n: number): Array> { + return Array.from({ length: n }, (_, i) => ({ name: `r${i}` })); +} + +describe('runImport — bulk create batching (framework#2678)', () => { + it('routes N insert-mode rows through ceil(N/batch) createManyData calls, not N createData calls', async () => { + const createManyData = vi.fn(async (args: { records: any[] }) => ({ + records: args.records.map((r) => ({ id: `id_${r.name}`, ...r })), + })); + const createData = vi.fn(); + const p: ImportProtocolLike = { + findData: vi.fn(async () => []), + createData, + updateData: vi.fn(), + createManyData, + }; + + const summary = await runImport({ + ...baseOpts, p, rows: rowsOf(250), progressEvery: 100, + }); + + expect(createData).not.toHaveBeenCalled(); + expect(createManyData).toHaveBeenCalledTimes(3); // ceil(250/100) + 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' }); + }); + + it('retries a transient createManyData failure instead of dropping the batch', async () => { + let attempts = 0; + const createManyData = vi.fn(async (args: { records: any[] }) => { + attempts++; + if (attempts === 1) throw new Error('fetch failed'); + return { records: args.records.map((r) => ({ id: `id_${r.name}`, ...r })) }; + }); + const p: ImportProtocolLike = { + findData: vi.fn(async () => []), + createData: vi.fn(), + updateData: vi.fn(), + createManyData, + }; + + const summary = await runImport({ ...baseOpts, p, rows: rowsOf(3) }); + + expect(createManyData).toHaveBeenCalledTimes(2); + expect(summary.errors).toBe(0); + expect(summary.created).toBe(3); + }); + + it('degrades to per-row createData on a logical batch failure, without failing the whole batch', async () => { + const createManyData = vi.fn(async () => { + throw new Error('CHECK constraint failed'); + }); + const createData = vi.fn(async (args: { data: { name: string } }) => { + if (args.data.name === 'r1') throw new Error('CHECK constraint failed: name'); + return { id: `id_${args.data.name}`, record: { id: `id_${args.data.name}` } }; + }); + const p: ImportProtocolLike = { + findData: vi.fn(async () => []), + createData, + updateData: vi.fn(), + createManyData, + }; + + const summary = await runImport({ ...baseOpts, p, rows: rowsOf(3) }); + + expect(createData).toHaveBeenCalledTimes(3); + expect(summary.errors).toBe(1); + expect(summary.created).toBe(2); + 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' }); + }); + + it('falls back to one createData call per row when the protocol has no createManyData', async () => { + const createData = vi.fn(async (args: { data: { name: string } }) => ({ id: `id_${args.data.name}` })); + const p: ImportProtocolLike = { + findData: vi.fn(async () => []), + createData, + updateData: vi.fn(), + // no createManyData + }; + + const summary = await runImport({ ...baseOpts, p, rows: rowsOf(5) }); + + expect(createData).toHaveBeenCalledTimes(5); + expect(summary.created).toBe(5); + }); + + it('preserves row order in results even with update/skip rows interleaved between buffered creates', async () => { + const createManyData = vi.fn(async (args: { records: any[] }) => ({ + records: args.records.map((r) => ({ id: `id_${r.name}`, ...r })), + })); + const updateData = vi.fn(async (args: { id: string }) => ({ id: args.id })); + // Row 1 ('existing') matches an existing record → update; the rest are creates. + const findData = vi.fn(async (args: { query?: { $filter?: { name?: string } } }) => + (args.query?.$filter?.name === 'existing' ? [{ id: 'existing_id', name: 'existing' }] : [])); + const p: ImportProtocolLike = { findData, createData: vi.fn(), updateData, createManyData }; + + const summary = await runImport({ + ...baseOpts, p, writeMode: 'upsert', matchFields: ['name'], + rows: [{ name: 'a' }, { name: 'existing' }, { name: 'b' }], + }); + + expect(summary.results.map((r) => r.action)).toEqual(['created', 'updated', 'created']); + expect(summary.created).toBe(2); + expect(summary.updated).toBe(1); + }); +}); diff --git a/packages/rest/src/import-runner.ts b/packages/rest/src/import-runner.ts index f6b343fcef..fa2f5e8e70 100644 --- a/packages/rest/src/import-runner.ts +++ b/packages/rest/src/import-runner.ts @@ -2,6 +2,7 @@ import { coerceRow, type RefResolver, type RefMatch } from './import-coerce.js'; import type { ExportFieldMeta } from './export-format.js'; +import { bulkWrite, withTransientRetry, type BulkWriteRowResult } from '@objectstack/core'; /** * import-runner — the shared row-processing core for bulk import. @@ -11,6 +12,13 @@ import type { ExportFieldMeta } from './export-format.js'; * keeps the two paths byte-for-byte identical in coercion, upsert matching, and * per-row reporting — the async worker only adds progress persistence and * cancellation on top. + * + * Rows resolved to a CREATE are batched through `p.createManyData` (the + * engine's array-form `insert()` — one round-trip per batch, with transient + * retry and per-row degradation on a logical/validation failure) instead of + * one `p.createData` call per row — see framework#2678. A protocol that + * doesn't implement `createManyData` falls back to the original per-row + * `createData` path unchanged. */ export type ImportAction = 'created' | 'updated' | 'skipped' | 'failed'; @@ -61,6 +69,13 @@ export interface ImportProtocolLike { findData(args: any): Promise; createData(args: any): Promise; updateData(args: any): Promise; + /** + * Optional bulk-create primitive. When present, `runImport` batches + * CREATE-resolved rows through it instead of one `createData` call per + * row — see framework#2678. Must resolve to `{ records: any[] }` with one + * record per input row, in the same order. + */ + createManyData?(args: { object: string; records: any[]; context?: any; environmentId?: string }): Promise<{ records: any[] }>; } export interface RunImportOptions { @@ -88,7 +103,11 @@ export interface RunImportOptions { * DB write of progress completes before the next chunk. */ onProgress?: (p: ImportProgress) => void | Promise; - /** Rows between onProgress calls (default 200). */ + /** + * Rows between onProgress calls (default 200). Also the flush boundary for + * buffered creates — a batch never grows past this before being written, + * so progress numbers stay accurate at every reported checkpoint. + */ progressEvery?: number; /** * Cooperative cancellation. Checked at each progress boundary; when it returns @@ -103,6 +122,21 @@ export interface RunImportOptions { captureUndo?: boolean; } +/** Extracts a created/updated record's id regardless of which response shape the protocol returned. */ +function extractRecordId(rec: any): string | undefined { + const id = rec?.id ?? rec?.record?.id; + return id != null ? String(id) : undefined; +} + +function toFailedResult(rowNo: number, err: unknown): ImportRowResult { + const code = (err as any)?.code ?? 'IMPORT_ROW_FAILED'; + const message = typeof (err as any)?.message === 'string' ? (err as any).message.slice(0, 300) : 'Row failed'; + return { row: rowNo, ok: false, action: 'failed', error: message, code }; +} + +/** Upper bound on rows in one createManyData batch (framework#2678 suggests 100-500). */ +const MAX_CREATE_BATCH_SIZE = 200; + export function runImport(opts: RunImportOptions): Promise { const { p, objectName, environmentId, context, rows, metaMap, @@ -189,7 +223,11 @@ export function runImport(opts: RunImportOptions): Promise { const writeCtx = { ...(context ?? {}), skipAutomations: !runAutomations }; - const results: ImportRowResult[] = []; + // Sparse-indexed by row position `i` (not push-only): CREATE rows are + // resolved immediately but their write is deferred to a later batch flush, + // so their result would otherwise land out of order relative to + // immediately-written update/skip rows interleaved between them. + const results: ImportRowResult[] = new Array(rows.length); let okCount = 0, errCount = 0, created = 0, updated = 0, skipped = 0; let cancelled = false; @@ -197,6 +235,47 @@ export function runImport(opts: RunImportOptions): Promise { processed, total: rows.length, created, updated, skipped, errors: errCount, }); + // CREATE rows are buffered here and flushed through `p.createManyData` + // (one round-trip per batch) when the protocol supports it. A protocol + // 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'; + const pendingCreates: Array<{ index: number; rowNo: number; data: Record }> = []; + const flushPendingCreates = async (): Promise => { + if (pendingCreates.length === 0) return; + const batch = pendingCreates.splice(0, pendingCreates.length); + const writeResults: BulkWriteRowResult[] = await bulkWrite( + batch.map(b => b.data), + { + // Flush cadence follows progressEvery, but the write batch itself is + // capped independently — a caller-supplied progressEvery far above + // 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), + writeBatch: (chunk) => p.createManyData!({ + object: objectName, records: chunk, context: writeCtx, + ...(environmentId ? { environmentId } : {}), + }).then(r => r.records), + writeOne: (row) => p.createData({ + object: objectName, data: row, context: writeCtx, + ...(environmentId ? { environmentId } : {}), + }), + }, + ); + for (const res of writeResults) { + const { index, rowNo } = batch[res.index]; + if (res.ok) { + const id = extractRecordId(res.record); + okCount++; created++; + if (collectUndo && id != null) undoLog.created.push(id); + results[index] = { row: rowNo, ok: true, action: 'created', id }; + } else { + errCount++; + results[index] = toFailedResult(rowNo, res.error); + } + } + }; + return (async () => { for (let i = 0; i < rows.length; i++) { const rowNo = i + 1; @@ -208,7 +287,7 @@ export function runImport(opts: RunImportOptions): Promise { if (errors.length > 0) { const first = errors[0]; errCount++; - results.push({ row: rowNo, ok: false, action: 'failed', field: first.field, code: first.code, error: first.message }); + results[i] = { row: rowNo, ok: false, action: 'failed', field: first.field, code: first.code, error: first.message }; } else { // 2. Decide create vs update vs skip. let existing: Record | 'blank' | 'none' | 'ambiguous' = 'none'; @@ -217,12 +296,12 @@ export function runImport(opts: RunImportOptions): Promise { existing = await findExisting(data); if (existing === 'ambiguous') { errCount++; - results.push({ row: rowNo, ok: false, action: 'failed', code: 'AMBIGUOUS_MATCH', error: `matchFields matched more than one ${objectName} record` }); + results[i] = { row: rowNo, ok: false, action: 'failed', code: 'AMBIGUOUS_MATCH', error: `matchFields matched more than one ${objectName} record` }; handled = true; } else if (existing === 'blank' && (skipBlankMatchKey || writeMode === 'update')) { // Blank match key: skip when asked, else fall through to create. skipped++; - results.push({ row: rowNo, ok: true, action: 'skipped', code: 'BLANK_MATCH_KEY' }); + results[i] = { row: rowNo, ok: true, action: 'skipped', code: 'BLANK_MATCH_KEY' }; handled = true; } } @@ -234,47 +313,55 @@ export function runImport(opts: RunImportOptions): Promise { if (!willUpdate && !willCreate) { // update mode, no match → skip. skipped++; - results.push({ row: rowNo, ok: true, action: 'skipped', code: 'NO_MATCH' }); + results[i] = { row: rowNo, ok: true, action: 'skipped', code: 'NO_MATCH' }; } else if (dryRun) { okCount++; - if (willUpdate) { updated++; results.push({ row: rowNo, ok: true, action: 'updated', id: String((existing as any).id ?? '') || undefined }); } - else { created++; results.push({ row: rowNo, ok: true, action: 'created' }); } + if (willUpdate) { updated++; results[i] = { row: rowNo, ok: true, action: 'updated', id: String((existing as any).id ?? '') || undefined }; } + else { created++; results[i] = { row: rowNo, ok: true, action: 'created' }; } } else if (willUpdate) { const target = existing as Record; - const res2 = await p.updateData({ object: objectName, id: target.id, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) }); - const id = (res2 as any)?.id ?? (res2 as any)?.record?.id ?? target.id; + const res2 = await withTransientRetry(() => p.updateData({ object: objectName, id: target.id, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) })); + const id = extractRecordId(res2) ?? String(target.id); okCount++; updated++; if (collectUndo && target.id != null) { undoLog.updated.push({ id: String(target.id), before: captureBefore(target, data) }); } - results.push({ row: rowNo, ok: true, action: 'updated', id: id != null ? String(id) : undefined }); + results[i] = { row: rowNo, ok: true, action: 'updated', id }; + } else if (canBulkCreate) { + // Buffer — the actual write happens in a batched flush below. + pendingCreates.push({ index: i, rowNo, data }); } else { + // No bulk-create primitive on this protocol: original inline path. const res2 = await p.createData({ object: objectName, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) }); - const id = (res2 as any)?.id ?? (res2 as any)?.record?.id; + const id = extractRecordId(res2); okCount++; created++; - if (collectUndo && id != null) undoLog.created.push(String(id)); - results.push({ row: rowNo, ok: true, action: 'created', id: id != null ? String(id) : undefined }); + if (collectUndo && id != null) undoLog.created.push(id); + results[i] = { row: rowNo, ok: true, action: 'created', id }; } } } } catch (err: any) { errCount++; - const code = err?.code ?? 'IMPORT_ROW_FAILED'; - const message = typeof err?.message === 'string' ? err.message.slice(0, 300) : 'Row failed'; - results.push({ row: rowNo, ok: false, action: 'failed', error: message, code }); + results[i] = toFailedResult(rowNo, err); } const processed = i + 1; - if (onProgress && (processed % progressEvery === 0 || processed === rows.length)) { - await onProgress(snapshot(processed)); + if (processed % progressEvery === 0 || processed === rows.length) { + // Flush before reporting/cancelling so counts and `processed` reflect + // every row up to this checkpoint, not just decided-but-unwritten ones. + await flushPendingCreates(); + if (onProgress) await onProgress(snapshot(processed)); } if (shouldCancel && processed < rows.length && (processed % progressEvery === 0)) { if (await shouldCancel()) { cancelled = true; break; } } } + await flushPendingCreates(); + const compacted = results.filter((r): r is ImportRowResult => r !== undefined); + return { - ...snapshot(results.length), ok: okCount, results, cancelled, + ...snapshot(compacted.length), ok: okCount, results: compacted, cancelled, ...(collectUndo ? { undoLog } : {}), }; })(); diff --git a/packages/runtime/src/http-dispatcher.test.ts b/packages/runtime/src/http-dispatcher.test.ts index a60fdce3f9..db37b534ff 100644 --- a/packages/runtime/src/http-dispatcher.test.ts +++ b/packages/runtime/src/http-dispatcher.test.ts @@ -1227,7 +1227,11 @@ describe('HttpDispatcher', () => { type: 'seed', name: 'project_seed', lock: null, editable: true, item: { object: 'project', externalId: 'name', mode: 'upsert', records }, }); - const insert = vi.fn().mockImplementation(async (_obj: string, rec: any) => ({ id: `id_${rec.name}` })); + // Mirror the real engine's array-form insert (bulk path): an + // array in → an array of created records out — framework#2678. + const insert = vi.fn().mockImplementation(async (_obj: string, rec: any) => ( + Array.isArray(rec) ? rec.map((r) => ({ id: `id_${r.name}` })) : { id: `id_${rec.name}` } + )); const find = vi.fn().mockResolvedValue([]); // no existing rows → all insert (kernel as any).getService = vi.fn().mockImplementation((name: string) => { if (name === 'protocol') return Promise.resolve({ publishPackageDrafts, getMetaItem }); @@ -1242,9 +1246,14 @@ describe('HttpDispatcher', () => { const seedApplied = (result.response as any)?.body?.data?.seedApplied; expect(seedApplied?.success).toBe(true); expect(seedApplied?.inserted).toBe(2); - // rows actually went to the engine - expect(insert).toHaveBeenCalledTimes(2); - expect(insert).toHaveBeenCalledWith('project', expect.objectContaining({ name: 'Apollo' }), expect.anything()); + // rows actually went to the engine — batched into one bulk + // insert() call rather than one per record (framework#2678). + expect(insert).toHaveBeenCalledTimes(1); + expect(insert).toHaveBeenCalledWith( + 'project', + [expect.objectContaining({ name: 'Apollo' }), expect.objectContaining({ name: 'Gemini' })], + expect.anything(), + ); }); // ADR-0045: "Publish" = live AND visible. A materialized (additive) diff --git a/packages/runtime/src/seed-loader.test.ts b/packages/runtime/src/seed-loader.test.ts index b39e8d63e5..cc81fc0d94 100644 --- a/packages/runtime/src/seed-loader.test.ts +++ b/packages/runtime/src/seed-loader.test.ts @@ -44,6 +44,13 @@ function createMockEngine(data: Record = {}): IDataEngine { }), insert: vi.fn(async (objectName: string, data: any) => { if (!store[objectName]) store[objectName] = []; + // Mirror the real engine's array-form insert (bulk path): an array in + // → an array of created records out, same order — see framework#2678. + if (Array.isArray(data)) { + const records = data.map((d) => ({ id: `gen-${++idCounter}`, ...d })); + store[objectName].push(...records); + return records; + } const record = { id: `gen-${++idCounter}`, ...data }; store[objectName].push(record); return record; @@ -289,7 +296,15 @@ describe('SeedLoaderService', () => { expect(result.success).toBe(true); expect(result.summary.totalInserted).toBe(2); expect(result.summary.objectsProcessed).toBe(1); - expect(engine.insert).toHaveBeenCalledTimes(2); + // framework#2678: non-self-referencing insert-mode records are batched + // through the engine's array-form insert() — one round-trip, not one + // per record. + expect(engine.insert).toHaveBeenCalledTimes(1); + expect(engine.insert).toHaveBeenCalledWith( + 'account', + [expect.objectContaining({ name: 'Acme Corp' }), expect.objectContaining({ name: 'Globex' })], + expect.anything(), + ); }); it('should return empty result for no datasets', async () => { From 6e3f21deb26eded386c5c5f6de82ea5c98657ad3 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Jul 2026 18:12:57 +0000 Subject: [PATCH 2/2] 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. --- .changeset/seed-import-bulk-insert.md | 3 +++ packages/plugins/driver-sql/src/sql-driver.ts | 20 +++++++++++++++++-- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/.changeset/seed-import-bulk-insert.md b/.changeset/seed-import-bulk-insert.md index c15d842d6b..e194972e1f 100644 --- a/.changeset/seed-import-bulk-insert.md +++ b/.changeset/seed-import-bulk-insert.md @@ -2,6 +2,7 @@ '@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). @@ -12,3 +13,5 @@ A new shared helper, `bulkWrite` (`@objectstack/core`), batches rows through a c - `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. diff --git a/packages/plugins/driver-sql/src/sql-driver.ts b/packages/plugins/driver-sql/src/sql-driver.ts index f6995f4d71..0a036e1a4d 100644 --- a/packages/plugins/driver-sql/src/sql-driver.ts +++ b/packages/plugins/driver-sql/src/sql-driver.ts @@ -1180,7 +1180,23 @@ export class SqlDriver implements IDataDriver { async bulkCreate(object: string, data: any[], options?: DriverOptions): Promise { this.auditMissingTenant(object, 'bulkCreate', options); - for (const row of data) { + // Same client-side id assignment as create() (id/_id normalization, + // nanoid fallback when neither is supplied) — a row missing an id must + // not be silently inserted with a NULL primary key just because the + // engine batched it into an array. This path used to be rarely + // exercised; framework#2678 made it the common case for seed/import. + const rows = data.map((row) => { + if (!row || typeof row !== 'object') return row; + const { _id, ...rest } = row; + const toInsert: Record = { ...rest }; + if (_id !== undefined && toInsert.id === undefined) { + toInsert.id = _id; + } else if (toInsert.id === undefined) { + toInsert.id = nanoid(DEFAULT_ID_LENGTH); + } + return toInsert; + }); + for (const row of rows) { if (row && typeof row === 'object') { this.injectTenantOnInsert(object, row, options); // Reserve a persistent sequence value for each row's autonumber @@ -1192,7 +1208,7 @@ export class SqlDriver implements IDataDriver { } } const builder = this.getBuilder(object, options); - return await builder.insert(data).returning('*'); + return await builder.insert(rows).returning('*'); } /**