From 5522a2a758ce6863e2e2c8eab1f43a0ae9de1222 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 03:17:41 +0000 Subject: [PATCH 1/8] fix(core): validate writeBatch result contract in bulkWrite (#3151) bulkWrite correlated writeBatch's return to input rows purely by position, with no check that it returned one record per row. A driver that returned fewer rows silently backfilled `record: undefined` and reported phantom successes; a longer return dropped records; a non-array return marked the whole batch ok with no records. Guard the return right after the batch write succeeds: a short / long / non-array result now throws ERR_BULK_RESULT_MISMATCH, which falls into the existing per-row degradation path (each row re-attempted via writeOne) instead of being trusted. The error is thrown outside withRetry and its message avoids any transient signature, so the batch is never retried on it. Also document bulkWrite's at-least-once delivery semantics in the module header, ahead of the idempotency work in #3149. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- packages/core/src/utils/bulk-write.test.ts | 44 ++++++++++++++++++++++ packages/core/src/utils/bulk-write.ts | 30 +++++++++++++++ 2 files changed, 74 insertions(+) diff --git a/packages/core/src/utils/bulk-write.test.ts b/packages/core/src/utils/bulk-write.test.ts index a372472e94..8fef57434e 100644 --- a/packages/core/src/utils/bulk-write.test.ts +++ b/packages/core/src/utils/bulk-write.test.ts @@ -112,6 +112,50 @@ 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'); + }); }); describe('withTransientRetry', () => { diff --git a/packages/core/src/utils/bulk-write.ts b/packages/core/src/utils/bulk-write.ts index 35dc6db11a..a295746693 100644 --- a/packages/core/src/utils/bulk-write.ts +++ b/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 { @@ -162,6 +173,25 @@ export async function bulkWrite( const batch = rows.slice(start, start + batchSize); try { const records = await withRetry(() => opts.writeBatch(batch), 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] }; } From ac781b78c0e6d4fad1afefc8b8dbdee45fe0f25b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 03:24:58 +0000 Subject: [PATCH 2/8] fix(seed,rest,core): wrap bare writes in transient retry; short-circuit logical errors (#3150) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two write points were left out of the bulk-write transient-retry rework, so a single network blip dropped the row with no retry — while adjacent paths retried: - seed-loader `writeRecord` (the self-referencing sequential path) and `resolveDeferredUpdates` called engine.insert/update bare; now wrapped in withTransientRetry, matching the batched update path. - import-runner's no-createManyData fallback called p.createData bare; now wrapped, matching the update path (L352) and the bulkWrite batch path. Also harden defaultIsTransientError: a message carrying a definitive logical signature (constraint / validation / required / unique / not null / out of range / not allowed) is now classified non-transient even when a transient keyword also appears (e.g. `CHECK constraint failed: network_zone`), so we don't burn retries on a row that fails identically every time. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- packages/core/src/utils/bulk-write.test.ts | 10 ++ packages/core/src/utils/bulk-write.ts | 26 +++- .../src/seed-loader-retry.test.ts | 131 ++++++++++++++++++ packages/metadata-protocol/src/seed-loader.ts | 18 +-- packages/rest/src/import-runner-bulk.test.ts | 21 +++ packages/rest/src/import-runner.ts | 6 +- 6 files changed, 200 insertions(+), 12 deletions(-) create mode 100644 packages/metadata-protocol/src/seed-loader-retry.test.ts diff --git a/packages/core/src/utils/bulk-write.test.ts b/packages/core/src/utils/bulk-write.test.ts index 8fef57434e..b73f072e1a 100644 --- a/packages/core/src/utils/bulk-write.test.ts +++ b/packages/core/src/utils/bulk-write.test.ts @@ -195,4 +195,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); + }); }); diff --git a/packages/core/src/utils/bulk-write.ts b/packages/core/src/utils/bulk-write.ts index a295746693..de99b5a2a1 100644 --- a/packages/core/src/utils/bulk-write.ts +++ b/packages/core/src/utils/bulk-write.ts @@ -97,11 +97,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)); } diff --git a/packages/metadata-protocol/src/seed-loader-retry.test.ts b/packages/metadata-protocol/src/seed-loader-retry.test.ts new file mode 100644 index 0000000000..30c4a7d801 --- /dev/null +++ b/packages/metadata-protocol/src/seed-loader-retry.test.ts @@ -0,0 +1,131 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect, vi } from 'vitest'; +import { SeedLoaderService } from './seed-loader'; +import type { IDataEngine, IMetadataService } from '@objectstack/spec/contracts'; + +/** + * framework#3150: the self-referencing seed path (`hasSelfRef`) writes records + * sequentially via `writeRecord`, which — unlike the batched path (bulkWrite's + * internal retry) and the update path — used to call `engine.insert` bare. A + * single transient blip (`fetch failed`) therefore dropped the row with no + * retry. These tests pin the now-wrapped behaviour. + */ + +function createLogger() { + return { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }; +} + +function createFaithfulEngine(): { engine: IDataEngine; store: Record } { + const store: Record = {}; + let idCounter = 0; + + const engine = { + find: vi.fn(async (objectName: string, query?: any) => { + let records = store[objectName] || []; + if (query?.where) { + records = records.filter((r) => + Object.entries(query.where).every(([k, v]) => r[k] === v), + ); + } + if (typeof query?.limit === 'number') records = records.slice(0, query.limit); + return records; + }), + findOne: vi.fn(async (objectName: string, query?: any) => { + const rows = await (engine.find as any)(objectName, { ...query, limit: 1 }); + return rows[0] ?? null; + }), + insert: vi.fn(async (objectName: string, data: any) => { + if (!store[objectName]) store[objectName] = []; + 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; + }), + update: vi.fn(async (objectName: string, data: any) => { + const records = store[objectName] || []; + const idx = records.findIndex((r) => r.id === data.id); + if (idx >= 0) { records[idx] = { ...records[idx], ...data }; return records[idx]; } + return data; + }), + delete: vi.fn(async () => ({ deleted: 1 })), + count: vi.fn(async (objectName: string) => (store[objectName] || []).length), + aggregate: vi.fn(async () => []), + } as unknown as IDataEngine; + + return { engine, store }; +} + +// A self-referencing object: `manager` is a lookup back to the same object, so +// the loader takes the historical sequential `writeRecord` path (`hasSelfRef`). +function createMetadata(): IMetadataService { + const objects: Record = { + my_app_employee: { + name: 'my_app_employee', + fields: { + name: { type: 'text' }, + manager: { type: 'lookup', reference: 'my_app_employee' }, + }, + }, + }; + return { + getObject: vi.fn(async (name: string) => objects[name]), + listObjects: vi.fn(async () => Object.values(objects)), + register: vi.fn(async () => {}), + get: vi.fn(async (_t: string, name: string) => objects[name]), + list: vi.fn(async () => []), + unregister: vi.fn(async () => {}), + exists: vi.fn(async () => false), + listNames: vi.fn(async () => []), + } as unknown as IMetadataService; +} + +const CONFIG = { + dryRun: false, + haltOnError: false, + multiPass: true, + defaultMode: 'insert', + batchSize: 1000, + transaction: false, +} as any; + +const SEEDS = [ + { + object: 'my_app_employee', + externalId: 'name', + mode: 'insert', + env: ['prod', 'dev', 'test'], + records: [{ name: 'Alice' }, { name: 'Bob' }], + }, +] as any[]; + +describe('seed self-ref sequential path — transient retry (framework#3150)', () => { + it('retries a transient insert on the self-referencing path instead of dropping the row', async () => { + const { engine, store } = createFaithfulEngine(); + const metadata = createMetadata(); + + // Bob's first insert hits a network blip, then succeeds on retry. + const realInsert = (engine.insert as any).getMockImplementation(); + let bobAttempts = 0; + (engine.insert as any).mockImplementation(async (obj: string, data: any, opts: any) => { + if (obj === 'my_app_employee' && !Array.isArray(data) && data.name === 'Bob') { + bobAttempts++; + if (bobAttempts === 1) throw new Error('fetch failed'); + } + return realInsert(obj, data, opts); + }); + + const result = await new SeedLoaderService(engine, metadata, createLogger()).load({ + seeds: SEEDS, + config: CONFIG, + }); + + expect(bobAttempts).toBe(2); // first threw, retried, succeeded + expect(result.summary.totalErrored).toBe(0); + expect((store.my_app_employee ?? []).map((r) => r.name).sort()).toEqual(['Alice', 'Bob']); + }); +}); diff --git a/packages/metadata-protocol/src/seed-loader.ts b/packages/metadata-protocol/src/seed-loader.ts index 0daf777292..6db70f9e6e 100644 --- a/packages/metadata-protocol/src/seed-loader.ts +++ b/packages/metadata-protocol/src/seed-loader.ts @@ -690,10 +690,10 @@ export class SeedLoaderService implements ISeedLoaderService { if (recordId) { try { - await this.engine.update(deferred.objectName, { + await withTransientRetry(() => this.engine.update(deferred.objectName, { id: recordId, [deferred.field]: resolvedId, - }, { context: { isSystem: true } } as any); + }, { context: { isSystem: true } } as any)); // Update result stats const resultEntry = allResults.find(r => r.object === deferred.objectName); @@ -764,7 +764,7 @@ export class SeedLoaderService implements ISeedLoaderService { switch (mode) { case 'insert': { - const result = await this.engine.insert(objectName, record, opts); + const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); return { action: 'inserted', id: this.extractId(result) }; } @@ -776,7 +776,7 @@ export class SeedLoaderService implements ISeedLoaderService { if (this.isNoOpReplay(record, existing)) { return { action: 'skipped', id }; } - await this.engine.update(objectName, { ...record, id }, opts); + await withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts)); return { action: 'updated', id }; } @@ -786,10 +786,10 @@ export class SeedLoaderService implements ISeedLoaderService { if (this.isNoOpReplay(record, existing)) { return { action: 'skipped', id }; } - await this.engine.update(objectName, { ...record, id }, opts); + await withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts)); return { action: 'updated', id }; } else { - const result = await this.engine.insert(objectName, record, opts); + const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); return { action: 'inserted', id: this.extractId(result) }; } } @@ -798,18 +798,18 @@ export class SeedLoaderService implements ISeedLoaderService { if (existing) { return { action: 'skipped', id: this.extractId(existing) }; } - const result = await this.engine.insert(objectName, record, opts); + const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); return { action: 'inserted', id: this.extractId(result) }; } case 'replace': { // Replace mode: just insert (caller should have cleared the table) - const result = await this.engine.insert(objectName, record, opts); + const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); return { action: 'inserted', id: this.extractId(result) }; } default: { - const result = await this.engine.insert(objectName, record, opts); + const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); return { action: 'inserted', id: this.extractId(result) }; } } diff --git a/packages/rest/src/import-runner-bulk.test.ts b/packages/rest/src/import-runner-bulk.test.ts index 185b86cde0..ef818924a2 100644 --- a/packages/rest/src/import-runner-bulk.test.ts +++ b/packages/rest/src/import-runner-bulk.test.ts @@ -118,6 +118,27 @@ describe('runImport — bulk create batching (framework#2678)', () => { expect(summary.created).toBe(5); }); + it('retries a transient createData failure on the no-createManyData fallback path (#3150)', async () => { + let attempts = 0; + const createData = vi.fn(async (args: { data: { name: string } }) => { + attempts++; + if (attempts === 1) throw new Error('fetch failed'); // one transient blip, then succeeds + return { id: `id_${args.data.name}` }; + }); + const p: ImportProtocolLike = { + findData: vi.fn(async () => []), + createData, + updateData: vi.fn(), + // no createManyData → inline per-row fallback path (previously un-retried) + }; + + const summary = await runImport({ ...baseOpts, p, rows: rowsOf(1) }); + + expect(createData).toHaveBeenCalledTimes(2); // first throws, retried, succeeds + expect(summary.created).toBe(1); + expect(summary.errors).toBe(0); + }); + 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 })), diff --git a/packages/rest/src/import-runner.ts b/packages/rest/src/import-runner.ts index 22f32366a6..f5bc711ba3 100644 --- a/packages/rest/src/import-runner.ts +++ b/packages/rest/src/import-runner.ts @@ -361,7 +361,11 @@ export function runImport(opts: RunImportOptions): Promise { 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 } : {}) }); + // Wrap in transient retry to match the update path above (L352) + // and the batched create path (bulkWrite's internal retry) — a + // single `fetch failed` blip must not silently drop the row + // (framework#3150). + const res2 = await withTransientRetry(() => p.createData({ object: objectName, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) })); const id = extractRecordId(res2); okCount++; created++; if (collectUndo && id != null) undoLog.created.push(id); From 323421612974bea8d00217a1cd897ebc0ac4e2b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 03:27:32 +0000 Subject: [PATCH 3/8] fix(rest): import resolveRef flushes pending creates and drops negative cache (#3148) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Batched CREATE rows buffer into pendingCreates and flush later, but resolveRef only queried the DB — so a later row referencing an earlier same-file CREATE always missed with reference_not_found and was dropped. refCache compounded it by caching the empty miss, so even after a progress-checkpoint flush landed the row, the name stayed pinned to the cached miss. resolveRef now, on a same-object miss with a non-empty buffer, flushes pending creates and retries the lookup once (the buffered rows are all earlier than the current one, so the flush is safe and idempotent). A bare miss is no longer cached — only a definitive id/ambiguous verdict is — so a name that misses early can still resolve once a later flush creates it. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- .../rest/src/import-runner-selfref.test.ts | 116 ++++++++++++++++++ packages/rest/src/import-runner.ts | 44 +++++-- 2 files changed, 147 insertions(+), 13 deletions(-) create mode 100644 packages/rest/src/import-runner-selfref.test.ts diff --git a/packages/rest/src/import-runner-selfref.test.ts b/packages/rest/src/import-runner-selfref.test.ts new file mode 100644 index 0000000000..710f6249d1 --- /dev/null +++ b/packages/rest/src/import-runner-selfref.test.ts @@ -0,0 +1,116 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * framework#3148: within a single import file, a later row must be able to + * reference a record an earlier row created — even though CREATE rows are + * buffered into a batched flush rather than written immediately. resolveRef + * flushes the pending-create buffer on a same-object miss and retries, and a + * bare miss is never negatively cached (which would pin it forever once a + * later flush actually creates the row). + */ + +import { describe, it, expect, vi } from 'vitest'; +import { runImport, type ImportProtocolLike } from './import-runner'; +import type { ExportFieldMeta } from './export-format.js'; + +// `parent` is a lookup back to this same object (a category tree). +const metaMap = new Map([ + ['name', { name: 'name', type: 'text' }], + ['parent', { name: 'parent', type: 'lookup', reference: 'showcase_category' }], +]); + +const baseOpts = { + objectName: 'showcase_category', + metaMap, + writeMode: 'insert' as const, + matchFields: [] as string[], + dryRun: false, + runAutomations: false, + trimWhitespace: true, + createMissingOptions: false, + skipBlankMatchKey: false, +}; + +/** Mock protocol backed by an in-memory store: createManyData writes, findData filters. */ +function makeProtocol(seed: Array> = []) { + const store: Array> = [...seed]; + let idc = 0; + const createManyData = vi.fn(async (args: { records: any[] }) => ({ + records: args.records.map((r) => { + const rec = { id: `new-${++idc}`, ...r }; + store.push(rec); + return rec; + }), + })); + const findData = vi.fn(async (args: { query?: { $filter?: Record } }) => { + const filter = args.query?.$filter ?? {}; + return store.filter((row) => Object.entries(filter).every(([k, v]) => row[k] === v)); + }); + const p: ImportProtocolLike = { findData, createData: vi.fn(), updateData: vi.fn(), createManyData }; + return { p, store, createManyData }; +} + +describe('runImport — same-file forward references (framework#3148)', () => { + it('resolves a row that references a record an earlier buffered row created', async () => { + // Existing-Parent is already in the DB; New-Parent is created by row 2 and + // referenced by row 3 — while still sitting in the create buffer. + const { p, store, createManyData } = makeProtocol([{ id: 'existing-1', name: 'Existing-Parent' }]); + + const summary = await runImport({ + ...baseOpts, p, + rows: [ + { name: 'Control-Child', parent: 'Existing-Parent' }, // references a pre-existing row + { name: 'New-Parent' }, + { name: 'New-Child', parent: 'New-Parent' }, // references row 2 (still buffered) + ], + }); + + expect(summary.errors).toBe(0); + expect(summary.created).toBe(3); + + const newParent = store.find((r) => r.name === 'New-Parent')!; + const newChild = store.find((r) => r.name === 'New-Child')!; + const controlChild = store.find((r) => r.name === 'Control-Child')!; + expect(newChild.parent).toBe(newParent.id); // resolved to the real created id + expect(controlChild.parent).toBe('existing-1'); // existing-reference behaviour unchanged + + // The miss on 'New-Parent' forced an early flush mid-loop, so there are two + // createManyData calls (the forced flush + the end-of-loop flush). + expect(createManyData).toHaveBeenCalledTimes(2); + }); + + it('does not negatively cache a miss: a name that misses early resolves once created (progressEvery=1)', async () => { + const { p, store } = makeProtocol(); + + const summary = await runImport({ + ...baseOpts, p, progressEvery: 1, + rows: [ + { name: 'Decoy', parent: 'New-Parent' }, // miss — New-Parent doesn't exist yet → fails + { name: 'New-Parent' }, // created; flushed after this row (progressEvery=1) + { name: 'Child', parent: 'New-Parent' }, // must resolve — the earlier miss must not be cached + ], + }); + + // Row 1 legitimately fails (the target truly did not exist at that point). + expect(summary.results[0]).toMatchObject({ ok: false, code: 'reference_not_found' }); + // Rows 2 and 3 succeed; the child links to the real parent. + const newParent = store.find((r) => r.name === 'New-Parent')!; + const child = store.find((r) => r.name === 'Child')!; + expect(child.parent).toBe(newParent.id); + expect(summary.created).toBe(2); + }); + + it('a genuine miss (no such record, no pending create) still reports reference_not_found', async () => { + const { p, createManyData } = makeProtocol(); + + const summary = await runImport({ + ...baseOpts, p, + rows: [{ name: 'Orphan', parent: 'Nobody' }], + }); + + expect(summary.results[0]).toMatchObject({ ok: false, code: 'reference_not_found' }); + expect(summary.created).toBe(0); + // Nothing to create → no batch flush ever ran. + expect(createManyData).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/rest/src/import-runner.ts b/packages/rest/src/import-runner.ts index f5bc711ba3..a25829ec8c 100644 --- a/packages/rest/src/import-runner.ts +++ b/packages/rest/src/import-runner.ts @@ -199,20 +199,38 @@ export function runImport(opts: RunImportOptions): Promise { ...(meta.displayField ? [meta.displayField] : []), 'name', 'title', 'label', 'full_name', 'email', 'username', ])]; - let match: RefMatch = {}; - for (const f of candidates) { - try { - const r = await p.findData({ - ...findArgsBase({ $filter: { [f]: display }, $top: 2 }), - object: referenceObject, - }); - const recs = findRows(r); - if (recs.length === 0) continue; - if (recs.length > 1) { match = { ambiguous: true, matchedField: f }; break; } - if (recs[0]?.id != null) { match = { id: String(recs[0].id), matchedField: f }; break; } - } catch { /* field absent on target object — try the next candidate */ } + const lookup = async (): Promise => { + let match: RefMatch = {}; + for (const f of candidates) { + try { + const r = await p.findData({ + ...findArgsBase({ $filter: { [f]: display }, $top: 2 }), + object: referenceObject, + }); + const recs = findRows(r); + if (recs.length === 0) continue; + if (recs.length > 1) { match = { ambiguous: true, matchedField: f }; break; } + if (recs[0]?.id != null) { match = { id: String(recs[0].id), matchedField: f }; break; } + } catch { /* field absent on target object — try the next candidate */ } + } + return match; + }; + let match = await lookup(); + // A miss may just mean the referenced row is still buffered as a pending + // create — the same-file "later row references an earlier CREATE" case that + // the batched-create rework regressed. Flush the buffer and retry the + // lookup once: the buffered rows are all EARLIER than this one (resolveRef + // runs mid row-loop), so the flush is safe and, once drained, a no-op. + // Only a reference to THIS object can be satisfied from the buffer, so we + // don't flush for a miss on some other object (framework#3148). + if (!match.id && !match.ambiguous && referenceObject === objectName && pendingCreates.length > 0) { + await flushPendingCreates(); + match = await lookup(); } - refCache.set(cacheKey, match); + // Cache only a definitive verdict. A bare miss ({}) is deliberately NOT + // cached: the referenced row may be created by a later flush, and a + // negative-cache entry would pin the miss forever (the pre-fix regression). + if (match.id != null || match.ambiguous) refCache.set(cacheKey, match); return match; }; From d9e0f5168da89d48fe4512973693e3fda9d94a50 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 03:29:19 +0000 Subject: [PATCH 4/8] fix(objectql): reject short/non-array bulkCreate driver results (#3151) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit engine.insert(array) built afterInsert contexts by index over rowHookContexts, so a driver bulkCreate that returned fewer records than input rows padded the tail with `undefined` — afterInsert hooks ran with `ctx.result === undefined` and the engine returned undefined entries, corrupting seed externalId→id maps and import undo logs downstream. Guard the driver result before the afterInsert loop: a batch whose result is not an array of exactly rows.length records now throws ERR_BULK_RESULT_MISMATCH, so the caller sees a real failure. Every in-repo driver already returns one-per-row in order; this defends against third-party drivers. The single-record path is untouched. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- .../objectql/src/engine-bulk-contract.test.ts | 94 +++++++++++++++++++ packages/objectql/src/engine.ts | 19 ++++ 2 files changed, 113 insertions(+) create mode 100644 packages/objectql/src/engine-bulk-contract.test.ts diff --git a/packages/objectql/src/engine-bulk-contract.test.ts b/packages/objectql/src/engine-bulk-contract.test.ts new file mode 100644 index 0000000000..3e6257fa49 --- /dev/null +++ b/packages/objectql/src/engine-bulk-contract.test.ts @@ -0,0 +1,94 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. +// +// framework#3151 (engine layer): engine.insert(array) must not fabricate +// afterInsert contexts (or caller results) when the driver's bulkCreate +// returns the wrong number of records. A short / non-array return is refused +// with ERR_BULK_RESULT_MISMATCH rather than padded with undefined. + +import { describe, it, expect } from 'vitest'; +import { ObjectQL } from './engine.js'; + +function makeDriver(opts: { bulkCreate?: (object: string, rows: any[]) => Promise } = {}) { + const stores = new Map>(); + const storeFor = (o: string) => { + let s = stores.get(o); + if (!s) { s = new Map(); stores.set(o, s); } + return s; + }; + let n = 0; + const driver: any = { + name: 'memory', version: '0.0.0', supports: {}, + async connect() {}, async disconnect() {}, async checkHealth() { return true; }, async execute() { return null; }, + async find(object: string) { return Array.from(storeFor(object).values()); }, + findStream() { throw new Error('ni'); }, + async findOne() { return null; }, + async create(object: string, data: Record) { + n += 1; + const id = (data.id as string) ?? `r_${n}`; + const row = { ...data, id }; + storeFor(object).set(id, row); + return row; + }, + async update(object: string, id: string, data: Record) { + const s = storeFor(object); + const row = { ...s.get(id), ...data, id }; + s.set(id, row); + return row; + }, + async delete(object: string, id: string) { return storeFor(object).delete(id); }, + async count() { return 0; }, + async bulkCreate(object: string, rows: Record[]) { + if (opts.bulkCreate) return opts.bulkCreate(object, rows); + const out: any[] = []; + for (const r of rows) out.push(await this.create(object, r)); + return out; + }, + async bulkUpdate() { return []; }, async bulkDelete() {}, + async beginTransaction() { return { __trx: true, commit: async () => {}, rollback: async () => {} }; }, + async commit() {}, async rollback() {}, + }; + return { driver, storeFor }; +} + +async function makeEngine(driverOpts?: { bulkCreate?: (object: string, rows: any[]) => Promise }) { + const engine = new ObjectQL(); + const d = makeDriver(driverOpts); + engine.registerDriver(d.driver, true); + await engine.init(); + engine.registry.registerObject({ name: 'task', fields: { title: { type: 'text' } } } as any); + return { engine, storeFor: d.storeFor }; +} + +describe('engine.insert(array) — driver result contract (framework#3151)', () => { + it('rejects a short bulkCreate return and runs no afterInsert for the phantom rows', async () => { + // Driver drops one row from its return set. + const { engine } = await makeEngine({ + bulkCreate: async (_o, rows) => rows.slice(1).map((r, i) => ({ id: `id-${i}`, ...r })), + }); + const afterCalls: any[] = []; + engine.registerHook('afterInsert', async (ctx: any) => { afterCalls.push(ctx.result); }, { object: 'task' }); + + await expect(engine.insert('task', [{ title: 'a' }, { title: 'b' }])) + .rejects.toMatchObject({ code: 'ERR_BULK_RESULT_MISMATCH' }); + expect(afterCalls).toHaveLength(0); // never fed undefined + }); + + it('rejects a non-array bulkCreate return', async () => { + const { engine } = await makeEngine({ bulkCreate: async () => undefined }); + await expect(engine.insert('task', [{ title: 'a' }, { title: 'b' }])) + .rejects.toMatchObject({ code: 'ERR_BULK_RESULT_MISMATCH' }); + }); + + it('accepts a correct one-per-row bulkCreate return (regression)', async () => { + const { engine } = await makeEngine(); + const res = await engine.insert('task', [{ title: 'a' }, { title: 'b' }]); + expect(res).toHaveLength(2); + expect((res as any[]).every((r) => r?.id)).toBe(true); + }); + + it('leaves the single-record insert path unaffected by the batch guard', async () => { + const { engine } = await makeEngine(); + const res = await engine.insert('task', { title: 'solo' }); + expect((res as any).id).toBeTruthy(); + }); +}); diff --git a/packages/objectql/src/engine.ts b/packages/objectql/src/engine.ts index 06f55b6475..dfc8cf2840 100644 --- a/packages/objectql/src/engine.ts +++ b/packages/objectql/src/engine.ts @@ -2274,6 +2274,25 @@ export class ObjectQL implements IDataEngine { result = await driver.create(object, rows[0], driverOptions); } + // Driver-result contract guard (framework#3151): a batch write must + // return one record per input row. A short / non-array return would + // otherwise be padded with `undefined` below and fed to afterInsert + // hooks (`ctx.result === undefined`) and back to the caller as phantom + // records — corrupting seed externalId→id maps and import undo logs. + // Refuse it: throw so the caller sees a real failure rather than + // silent data loss. (Every driver in this repo already returns + // one-per-row in order; this defends against third-party drivers.) + if (isBatch && (!Array.isArray(result) || result.length !== rows.length)) { + throw Object.assign( + new Error( + `bulkCreate for '${object}' returned ${ + Array.isArray(result) ? `${result.length} record(s)` : String(typeof result) + } for ${rows.length} input row(s) — refusing to fabricate afterInsert contexts`, + ), + { code: 'ERR_BULK_RESULT_MISMATCH' }, + ); + } + // Coerce `boolean` fields (SQLite/libsql return 0/1) to real booleans on // the after-hook view so flow trigger conditions (`record.is_escalated // != true`) and `{record.}` interpolation see JS booleans, not From 4434f300287e2ea40cc3511671e4d9ad4d1b510e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 03:37:35 +0000 Subject: [PATCH 5/8] fix(core,seed,rest): idempotent batch retry via attempt-aware natural-key recheck (#3149) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bulkWrite's transient retry is at-least-once: when a driver commits a batch but its response is lost (turso's fetch-based transport fails after commit), the retry re-inserted the whole batch — 2 input rows became 4 stored rows, all reported success, each firing record-change and summary recompute twice. Rather than a generic dedup inside bulkWrite, thread a 1-based `attempt` counter into writeBatch/writeOne and let each caller enforce batch-level idempotency with its own natural key: - core: withRetry passes the attempt number; writeBatch/writeOne gain a `{ attempt }` context (backward compatible — callers may ignore it). - seed: on attempt > 1, recheck existing rows by externalId (reusing loadExistingRecords) and insert only the missing ones; reassemble the per-row result in order. - import: on attempt > 1 with matchFields, recheck by matchFields and create only the missing rows. A short createManyData return is now surfaced as a failed batch (so degradation, which also rechecks, runs) instead of being padded. A pure-insert import (no matchFields) has no natural key and stays at-least-once by documented contract. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- packages/core/src/utils/bulk-write.test.ts | 31 +++++ packages/core/src/utils/bulk-write.ts | 24 ++-- .../src/seed-loader-retry.test.ts | 50 +++++++++ packages/metadata-protocol/src/seed-loader.ts | 65 +++++++++-- .../src/import-runner-idempotency.test.ts | 106 ++++++++++++++++++ packages/rest/src/import-runner.ts | 71 ++++++++++-- 6 files changed, 322 insertions(+), 25 deletions(-) create mode 100644 packages/rest/src/import-runner-idempotency.test.ts diff --git a/packages/core/src/utils/bulk-write.test.ts b/packages/core/src/utils/bulk-write.test.ts index b73f072e1a..25c7cdd098 100644 --- a/packages/core/src/utils/bulk-write.test.ts +++ b/packages/core/src/utils/bulk-write.test.ts @@ -156,6 +156,37 @@ describe('bulkWrite', () => { 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', () => { diff --git a/packages/core/src/utils/bulk-write.ts b/packages/core/src/utils/bulk-write.ts index de99b5a2a1..9c803ef7c7 100644 --- a/packages/core/src/utils/bulk-write.ts +++ b/packages/core/src/utils/bulk-write.ts @@ -67,10 +67,18 @@ export interface BulkWriteOptions 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[]) => Promise; - /** Write a single row — used only to degrade a failed batch. */ - writeOne: (row: TRow) => Promise; + writeBatch: (batch: TRow[], ctx: { attempt: number }) => Promise; + /** + * Write a single row — used only to degrade a failed batch. `ctx.attempt` + * carries the same recheck signal as {@link writeBatch}. + */ + writeOne: (row: TRow, ctx: { attempt: number }) => Promise; } const DEFAULT_BATCH_SIZE = 200; @@ -136,11 +144,11 @@ interface ResolvedRetryOptions { sleep: (ms: number) => Promise; } -async function withRetry(fn: () => Promise, opts: ResolvedRetryOptions): Promise { +async function withRetry(fn: (attempt: number) => Promise, opts: ResolvedRetryOptions): Promise { 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; @@ -159,7 +167,7 @@ async function withRetry(fn: () => Promise, 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(fn: () => Promise, opts: RetryOptions = {}): Promise { +export async function withTransientRetry(fn: (attempt: number) => Promise, opts: RetryOptions = {}): Promise { return withRetry(fn, { maxRetries: Math.max(1, opts.maxRetries ?? DEFAULT_MAX_RETRIES), backoffBaseMs: opts.backoffBaseMs ?? DEFAULT_BACKOFF_BASE_MS, @@ -194,7 +202,7 @@ export async function bulkWrite( 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 @@ -233,7 +241,7 @@ export async function bulkWrite( 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 }; diff --git a/packages/metadata-protocol/src/seed-loader-retry.test.ts b/packages/metadata-protocol/src/seed-loader-retry.test.ts index 30c4a7d801..d91845676d 100644 --- a/packages/metadata-protocol/src/seed-loader-retry.test.ts +++ b/packages/metadata-protocol/src/seed-loader-retry.test.ts @@ -71,6 +71,14 @@ function createMetadata(): IMetadataService { manager: { type: 'lookup', reference: 'my_app_employee' }, }, }, + // A plain object (no self-ref) → the batched flushPendingInserts path. + my_app_widget: { + name: 'my_app_widget', + fields: { + name: { type: 'text' }, + sku: { type: 'text' }, + }, + }, }; return { getObject: vi.fn(async (name: string) => objects[name]), @@ -129,3 +137,45 @@ describe('seed self-ref sequential path — transient retry (framework#3150)', ( expect((store.my_app_employee ?? []).map((r) => r.name).sort()).toEqual(['Alice', 'Bob']); }); }); + +describe('seed batched path — idempotent retry after commit-then-lost-response (framework#3149)', () => { + it('does not duplicate rows when the batch insert commits but its response is lost', async () => { + const { engine, store } = createFaithfulEngine(); + const metadata = createMetadata(); + + // First array insert writes both rows to the store, then throws a transient + // error (turso's commit-then-lost-response shape). The retry must recheck + // by externalId and NOT insert them again. + const realInsert = (engine.insert as any).getMockImplementation(); + let arrayInsertCalls = 0; + (engine.insert as any).mockImplementation(async (obj: string, data: any, opts: any) => { + if (obj === 'my_app_widget' && Array.isArray(data)) { + arrayInsertCalls++; + if (arrayInsertCalls === 1) { + await realInsert(obj, data, opts); // commit lands + throw new Error('fetch failed'); // ...but the response is lost + } + } + return realInsert(obj, data, opts); + }); + + const result = await new SeedLoaderService(engine, metadata, createLogger()).load({ + seeds: [{ + object: 'my_app_widget', + externalId: 'sku', + mode: 'insert', + env: ['prod', 'dev', 'test'], + records: [{ name: 'Widget A', sku: 'W-A' }, { name: 'Widget B', sku: 'W-B' }], + }] as any, + config: CONFIG, + }); + + // The array insert ran once (attempt 1, which committed); the retry's + // recheck found both rows already present and did NOT re-insert. + expect(arrayInsertCalls).toBe(1); + expect(store.my_app_widget.filter((r) => r.sku === 'W-A')).toHaveLength(1); + expect(store.my_app_widget.filter((r) => r.sku === 'W-B')).toHaveLength(1); + expect(store.my_app_widget).toHaveLength(2); // no duplicates + expect(result.summary.totalErrored).toBe(0); + }); +}); diff --git a/packages/metadata-protocol/src/seed-loader.ts b/packages/metadata-protocol/src/seed-loader.ts index 6db70f9e6e..b631be8e5a 100644 --- a/packages/metadata-protocol/src/seed-loader.ts +++ b/packages/metadata-protocol/src/seed-loader.ts @@ -15,7 +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'; +import { bulkWrite, withTransientRetry, defaultIsTransientError, type BulkWriteRowResult } from '@objectstack/core'; interface Logger { info(message: string, meta?: Record): void; @@ -262,6 +262,26 @@ export class SeedLoaderService implements ISeedLoaderService { // logical/validation failure. See framework#2678. const pendingInserts: Array<{ recordIndex: number; externalIdValue: string; record: Record }> = []; const opts = SeedLoaderService.SEED_OPTIONS as any; + const extIdOf = (rec: Record) => String(rec[externalId] ?? ''); + // bulkWrite is at-least-once: a retry (or a mismatch-driven degradation) + // may re-run a write whose prior attempt already committed. Guard against + // duplicate seed rows by rechecking natural keys before re-inserting + // (framework#3149). `lastBatchUncertain` carries the "prior batch outcome + // unknown" signal into the per-row degradation writeOne calls. + 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[], existing: Map, 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++]; + }); + }; const flushPendingInserts = async (): Promise => { if (pendingInserts.length === 0) return; const batch = pendingInserts.splice(0, pendingInserts.length); @@ -269,13 +289,42 @@ export class SeedLoaderService implements ISeedLoaderService { 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), + writeBatch: async (rows, { attempt }) => { + let toInsert = rows; + let existing = new Map(); + if (attempt > 1) { + // Prior attempt may have committed before its response was lost: + // insert only rows not already present so a retry can't duplicate. + existing = await this.loadExistingRecords(objectName, externalId, config.organizationId); + 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 + ? [await this.engine.insert(objectName, toInsert[0], opts)] + : await this.engine.insert(objectName, toInsert, opts); + lastBatchUncertain = false; + return assembleInOrder(rows, existing, freshlyInserted as any[]); + } catch (e) { + lastBatchUncertain = isUncertainOutcome(e); + throw e; + } + }, + writeOne: async (row, { attempt }) => { + if (attempt > 1 || lastBatchUncertain) { + const key = extIdOf(row); + if (key) { + const existing = await this.loadExistingRecords(objectName, externalId, config.organizationId); + const hit = existing.get(key); + if (hit) return hit; // already committed by a prior attempt + } + } + return this.engine.insert(objectName, row, opts); + }, }, ); for (const res of writeResults) { diff --git a/packages/rest/src/import-runner-idempotency.test.ts b/packages/rest/src/import-runner-idempotency.test.ts new file mode 100644 index 0000000000..e01ffa4062 --- /dev/null +++ b/packages/rest/src/import-runner-idempotency.test.ts @@ -0,0 +1,106 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * framework#3149: bulkWrite is at-least-once — a retry (or a mismatch-driven + * degradation) may re-run a create whose prior attempt already committed. When + * the import has natural keys (matchFields), runImport rechecks before + * re-creating so a retry can't duplicate the row. A pure-insert import has no + * natural key and stays at-least-once by contract. + */ + +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, + dryRun: false, + runAutomations: false, + trimWhitespace: true, + createMissingOptions: false, + skipBlankMatchKey: false, +}; + +/** + * Mock protocol backed by an in-memory store. `createManyData` optionally + * commits-then-throws (or short-returns) on its first call to model a lost + * response / mismatch; findData filters the store for the recheck. + */ +function makeProtocol(opts: { firstCall?: 'throw' | 'shortReturn' } = {}) { + const store: Array> = []; + let idc = 0; + let calls = 0; + const createManyData = vi.fn(async (args: { records: any[] }) => { + calls++; + const recs = args.records.map((r) => { + const rec = { id: `id-${++idc}`, ...r }; + store.push(rec); + return rec; + }); + if (calls === 1 && opts.firstCall === 'throw') throw new Error('fetch failed'); // committed, response lost + if (calls === 1 && opts.firstCall === 'shortReturn') return { records: [] }; // committed, bad count + return { records: recs }; + }); + const createData = vi.fn(async (args: { data: { name: string } }) => { + const rec = { id: `id-${++idc}`, ...args.data }; + store.push(rec); + return rec; + }); + const findData = vi.fn(async (args: { query?: { $filter?: Record } }) => { + const filter = args.query?.$filter ?? {}; + return store.filter((row) => Object.entries(filter).every(([k, v]) => row[k] === v)); + }); + const p: ImportProtocolLike = { findData, createData, updateData: vi.fn(), createManyData }; + return { p, store, createManyData, createData }; +} + +describe('runImport — idempotent retry with natural keys (framework#3149)', () => { + it('upsert+matchFields: a transient retry after commit does not duplicate rows', async () => { + const { p, store, createManyData } = makeProtocol({ firstCall: 'throw' }); + + const summary = await runImport({ + ...baseOpts, p, writeMode: 'upsert', matchFields: ['name'], + rows: [{ name: 'x' }, { name: 'y' }], + }); + + // createManyData ran once (attempt 1, which committed); the retry's recheck + // found both rows already present and did NOT re-create them. + expect(createManyData).toHaveBeenCalledTimes(1); + expect(store.filter((r) => r.name === 'x')).toHaveLength(1); + expect(store.filter((r) => r.name === 'y')).toHaveLength(1); + expect(store).toHaveLength(2); // no duplicates + expect(summary.created).toBe(2); + expect(summary.errors).toBe(0); + }); + + it('upsert+matchFields: a short createManyData return degrades and still does not duplicate', async () => { + const { p, store } = makeProtocol({ firstCall: 'shortReturn' }); + + const summary = await runImport({ + ...baseOpts, p, writeMode: 'upsert', matchFields: ['name'], + rows: [{ name: 'x' }, { name: 'y' }], + }); + + // The empty return voids the batch → per-row degradation, which rechecks + // and finds both rows already committed rather than re-creating them. + expect(store).toHaveLength(2); + expect(summary.created).toBe(2); + expect(summary.errors).toBe(0); + }); + + it('pure insert (no matchFields): retry is at-least-once — duplicates are the documented contract', async () => { + const { p, store } = makeProtocol({ firstCall: 'throw' }); + + await runImport({ + ...baseOpts, p, writeMode: 'insert', matchFields: [], + rows: [{ name: 'x' }, { name: 'y' }], + }); + + // No natural key to recheck against: the committed-then-retried batch is + // written twice. This pins the contract (exactly-once needs matchFields). + expect(store).toHaveLength(4); + }); +}); diff --git a/packages/rest/src/import-runner.ts b/packages/rest/src/import-runner.ts index a25829ec8c..65030e3adb 100644 --- a/packages/rest/src/import-runner.ts +++ b/packages/rest/src/import-runner.ts @@ -2,7 +2,7 @@ import { coerceRow, firstMissingRequiredField, type RefResolver, type RefMatch } from './import-coerce.js'; import type { ExportFieldMeta } from './export-format.js'; -import { bulkWrite, withTransientRetry, type BulkWriteRowResult } from '@objectstack/core'; +import { bulkWrite, withTransientRetry, defaultIsTransientError, type BulkWriteRowResult } from '@objectstack/core'; /** * import-runner — the shared row-processing core for bulk import. @@ -273,6 +273,19 @@ export function runImport(opts: RunImportOptions): Promise { // 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 }> = []; + // bulkWrite is at-least-once: a retry (or a mismatch-driven degradation) may + // re-run a create whose prior attempt already committed. When the import has + // natural keys (matchFields), recheck before re-creating so a retry can't + // duplicate a row (framework#3149). A pure-insert import (no matchFields) has + // no natural key to recheck against and stays at-least-once by contract. + let lastBatchUncertain = false; + const isUncertainOutcome = (e: unknown) => + defaultIsTransientError(e) || (e as { code?: unknown } | null)?.code === 'ERR_BULK_RESULT_MISMATCH'; + const existingByMatch = async (data: Record): Promise | null> => { + if (matchFields.length === 0) return null; + const found = await findExisting(data); + return found && typeof found === 'object' ? found : null; // ignore 'blank'/'none'/'ambiguous' + }; const flushPendingCreates = async (): Promise => { if (pendingCreates.length === 0) return; const batch = pendingCreates.splice(0, pendingCreates.length); @@ -284,14 +297,54 @@ export function runImport(opts: RunImportOptions): Promise { // 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 } : {}), - }), + writeBatch: async (chunk, { attempt }) => { + let toCreate = chunk; + const existingByIdx = new Map>(); + if (attempt > 1 && matchFields.length > 0) { + // A prior attempt may have committed before its response was lost: + // recheck each row by matchFields and only create the ones missing. + toCreate = []; + for (let i = 0; i < chunk.length; i++) { + const hit = await existingByMatch(chunk[i]); + if (hit) existingByIdx.set(i, hit); else toCreate.push(chunk[i]); + } + } + try { + const createdRecords = toCreate.length === 0 + ? [] + : (await p.createManyData!({ + object: objectName, records: toCreate, context: writeCtx, + ...(environmentId ? { environmentId } : {}), + })).records; + // Surface a short/non-array createManyData return as a failed batch + // (framework#3151) rather than padding the reassembly with undefined + // — this drops into per-row degradation, which rechecks first. + if (!Array.isArray(createdRecords) || createdRecords.length !== toCreate.length) { + throw Object.assign( + new Error(`createManyData returned ${Array.isArray(createdRecords) ? `${createdRecords.length} record(s)` : String(typeof createdRecords)} for ${toCreate.length} row(s)`), + { code: 'ERR_BULK_RESULT_MISMATCH' }, + ); + } + lastBatchUncertain = false; + // Reassemble one record per input row: rechecked-existing rows use + // the found record, the rest are consumed in order from created. + let k = 0; + return chunk.map((_row, i) => existingByIdx.has(i) ? existingByIdx.get(i)! : createdRecords[k++]); + } catch (e) { + lastBatchUncertain = isUncertainOutcome(e); + throw e; + } + }, + writeOne: async (row, { attempt }) => { + if ((attempt > 1 || lastBatchUncertain) && matchFields.length > 0) { + const hit = await existingByMatch(row); + if (hit) return hit; // already committed by a prior attempt + } + return p.createData({ + object: objectName, data: row, context: writeCtx, + ...(environmentId ? { environmentId } : {}), + }); + }, }, ); for (const res of writeResults) { From 45c5df6498be8c7c5f1a45c7ddeaf114c8b7e8b4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 03:48:12 +0000 Subject: [PATCH 6/8] fix(objectql,seed,rest): retry roll-up recompute and surface failures (#3147) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit recomputeSummaries wrapped each parent's aggregate+update in a try/catch that only warned — a transient turso blip on the parent summary write left the value silently stale, and a child batch insert still reported full success. Engine: retry each parent recompute with transient backoff (withTransientRetry, injectable via engine.summaryRetryOptions); collect per-parent failures so one bad parent doesn't abort the rest. When any remain after retries, insert/update/delete throw SummaryRecomputeError (code ERR_SUMMARY_RECOMPUTE) AFTER the realtime publish — the records ARE written, so the error carries them on `written`. Callers (identified by `code`, no objectql import — that would cycle): seed and import recover ERR_SUMMARY_RECOMPUTE to success using `written` rather than re-writing (which would duplicate), logging a warning; import marks the affected rows `SUMMARY_RECOMPUTE_FAILED` while keeping them created/updated. A direct API caller simply sees the thrown error. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- .../src/seed-loader-retry.test.ts | 37 +++++ packages/metadata-protocol/src/seed-loader.ts | 45 ++++-- .../objectql/src/engine-summary-retry.test.ts | 150 ++++++++++++++++++ packages/objectql/src/engine.ts | 71 ++++++--- packages/objectql/src/index.ts | 2 + packages/objectql/src/summary-errors.ts | 35 ++++ .../src/import-runner-idempotency.test.ts | 23 +++ packages/rest/src/import-runner.ts | 64 ++++++-- 8 files changed, 387 insertions(+), 40 deletions(-) create mode 100644 packages/objectql/src/engine-summary-retry.test.ts create mode 100644 packages/objectql/src/summary-errors.ts diff --git a/packages/metadata-protocol/src/seed-loader-retry.test.ts b/packages/metadata-protocol/src/seed-loader-retry.test.ts index d91845676d..94304bf18f 100644 --- a/packages/metadata-protocol/src/seed-loader-retry.test.ts +++ b/packages/metadata-protocol/src/seed-loader-retry.test.ts @@ -179,3 +179,40 @@ describe('seed batched path — idempotent retry after commit-then-lost-response expect(result.summary.totalErrored).toBe(0); }); }); + +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(); + const metadata = createMetadata(); + + // The array insert writes the rows, then reports a post-write summary + // recompute failure — the records ARE written (carried on `written`). + const realInsert = (engine.insert as any).getMockImplementation(); + let arrayCalls = 0; + (engine.insert as any).mockImplementation(async (obj: string, data: any, opts: any) => { + if (obj === 'my_app_widget' && Array.isArray(data)) { + arrayCalls++; + const written = await realInsert(obj, data, opts); + throw Object.assign(new Error('summary recompute failed'), { + code: 'ERR_SUMMARY_RECOMPUTE', written, failures: [{ parentObject: 'x' }], + }); + } + return realInsert(obj, data, opts); + }); + + const result = await new SeedLoaderService(engine, metadata, createLogger()).load({ + seeds: [{ + object: 'my_app_widget', + externalId: 'sku', + mode: 'insert', + env: ['prod', 'dev', 'test'], + records: [{ name: 'A', sku: 'W-A' }, { name: 'B', sku: 'W-B' }], + }] as any, + config: CONFIG, + }); + + expect(arrayCalls).toBe(1); // recovered, NOT re-inserted + expect(store.my_app_widget).toHaveLength(2); // no duplicates + expect(result.summary.totalErrored).toBe(0); // a stale summary is not a write error + }); +}); diff --git a/packages/metadata-protocol/src/seed-loader.ts b/packages/metadata-protocol/src/seed-loader.ts index b631be8e5a..2f6fc691a5 100644 --- a/packages/metadata-protocol/src/seed-loader.ts +++ b/packages/metadata-protocol/src/seed-loader.ts @@ -305,8 +305,8 @@ export class SeedLoaderService implements ISeedLoaderService { const freshlyInserted = toInsert.length === 0 ? [] : toInsert.length === 1 - ? [await this.engine.insert(objectName, toInsert[0], opts)] - : await this.engine.insert(objectName, toInsert, opts); + ? [await this.writeRecoveringSummary(() => this.engine.insert(objectName, toInsert[0], opts))] + : await this.writeRecoveringSummary(() => this.engine.insert(objectName, toInsert, opts)); lastBatchUncertain = false; return assembleInOrder(rows, existing, freshlyInserted as any[]); } catch (e) { @@ -323,7 +323,7 @@ export class SeedLoaderService implements ISeedLoaderService { if (hit) return hit; // already committed by a prior attempt } } - return this.engine.insert(objectName, row, opts); + return this.writeRecoveringSummary(() => this.engine.insert(objectName, row, opts)); }, }, ); @@ -596,7 +596,7 @@ export class SeedLoaderService implements ISeedLoaderService { insertedRecords.get(objectName)!.set(externalIdValue, decision.id); } try { - await withTransientRetry(() => this.engine.update(objectName, { ...record, id: decision.id }, opts)); + await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.update(objectName, { ...record, id: decision.id }, opts))); updated++; } catch (err: any) { errored++; @@ -800,6 +800,29 @@ export class SeedLoaderService implements ISeedLoaderService { */ private static readonly SEED_OPTIONS = { context: { isSystem: true, skipTriggers: true } } as const; + /** + * Run an engine write; if it fails ONLY because a post-write roll-up summary + * recompute exhausted its retries (framework#3147, `code` + * 'ERR_SUMMARY_RECOMPUTE'), the record WAS written — treat it as a warning + * and return the written value rather than re-writing (which would + * duplicate). Matched by `code` so we needn't import objectql (which depends + * on this package — importing back would cycle). Any other error propagates. + */ + private async writeRecoveringSummary(fn: () => Promise): Promise { + try { + return await fn(); + } catch (e: any) { + if (e?.code === 'ERR_SUMMARY_RECOMPUTE') { + this.logger.warn( + '[SeedLoader] roll-up summary recompute failed after retries; records were written (summary values may be stale)', + { failures: Array.isArray(e.failures) ? e.failures.length : undefined }, + ); + return e.written as T; + } + throw e; + } + } + private async writeRecord( objectName: string, record: Record, @@ -813,7 +836,7 @@ export class SeedLoaderService implements ISeedLoaderService { switch (mode) { case 'insert': { - const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); + const result = await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.insert(objectName, record, opts))); return { action: 'inserted', id: this.extractId(result) }; } @@ -825,7 +848,7 @@ export class SeedLoaderService implements ISeedLoaderService { if (this.isNoOpReplay(record, existing)) { return { action: 'skipped', id }; } - await withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts)); + await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts))); return { action: 'updated', id }; } @@ -835,10 +858,10 @@ export class SeedLoaderService implements ISeedLoaderService { if (this.isNoOpReplay(record, existing)) { return { action: 'skipped', id }; } - await withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts)); + await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts))); return { action: 'updated', id }; } else { - const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); + const result = await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.insert(objectName, record, opts))); return { action: 'inserted', id: this.extractId(result) }; } } @@ -847,18 +870,18 @@ export class SeedLoaderService implements ISeedLoaderService { if (existing) { return { action: 'skipped', id: this.extractId(existing) }; } - const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); + const result = await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.insert(objectName, record, opts))); return { action: 'inserted', id: this.extractId(result) }; } case 'replace': { // Replace mode: just insert (caller should have cleared the table) - const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); + const result = await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.insert(objectName, record, opts))); return { action: 'inserted', id: this.extractId(result) }; } default: { - const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts)); + const result = await this.writeRecoveringSummary(() => withTransientRetry(() => this.engine.insert(objectName, record, opts))); return { action: 'inserted', id: this.extractId(result) }; } } diff --git a/packages/objectql/src/engine-summary-retry.test.ts b/packages/objectql/src/engine-summary-retry.test.ts new file mode 100644 index 0000000000..9bebc2590a --- /dev/null +++ b/packages/objectql/src/engine-summary-retry.test.ts @@ -0,0 +1,150 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. +// +// framework#3147: a roll-up summary recompute that fails on the parent +// aggregate/update must be retried (transient), and — if it still fails — must +// be surfaced to the caller (SummaryRecomputeError / ERR_SUMMARY_RECOMPUTE) +// rather than swallowed with a warn. The triggering child records ARE written. + +import { describe, it, expect } from 'vitest'; +import { ObjectQL } from './engine.js'; + +function makeDriver(opts: { onParentUpdate?: (parentId: string, callN: number) => void } = {}) { + const stores = new Map>(); + const storeFor = (o: string) => { + let s = stores.get(o); + if (!s) { s = new Map(); stores.set(o, s); } + return s; + }; + const matches = (row: any, where: any): boolean => { + if (!where || typeof where !== 'object') return true; + return Object.entries(where).every(([k, v]) => row?.[k] === v); + }; + const parentCalls = new Map(); + let n = 0; + const driver: any = { + name: 'memory', version: '0.0.0', supports: {}, + async connect() {}, async disconnect() {}, async checkHealth() { return true; }, async execute() { return null; }, + async find(object: string, ast: any) { + return Array.from(storeFor(object).values()).filter((r) => matches(r, ast?.where)); + }, + findStream() { throw new Error('ni'); }, + async findOne(object: string, ast: any) { + for (const r of storeFor(object).values()) if (matches(r, ast?.where)) return r; + return null; + }, + async create(object: string, data: Record) { + n += 1; + const id = (data.id as string) ?? `r_${n}`; + const row = { ...data, id }; + storeFor(object).set(id, row); + return row; + }, + async update(object: string, id: string, data: Record) { + if (object === 'inv' && opts.onParentUpdate) { + const callN = (parentCalls.get(id) ?? 0) + 1; + parentCalls.set(id, callN); + opts.onParentUpdate(id, callN); // may throw + } + const s = storeFor(object); + const row = { ...s.get(id), ...data, id }; + s.set(id, row); + return row; + }, + async delete(object: string, id: string) { return storeFor(object).delete(id); }, + async count() { return 0; }, + async bulkCreate(object: string, rows: Record[]) { + return Promise.all(rows.map((r) => this.create(object, r))); + }, + async bulkUpdate() { return []; }, async bulkDelete() {}, + async beginTransaction() { return { __trx: true, commit: async () => {}, rollback: async () => {} }; }, + async commit() {}, async rollback() {}, + }; + return { driver, storeFor, parentCalls }; +} + +async function makeEngine(driverOpts?: { onParentUpdate?: (parentId: string, callN: number) => void }) { + const engine = new ObjectQL(); + const d = makeDriver(driverOpts); + engine.registerDriver(d.driver, true); + await engine.init(); + // Deterministic, fast backoff for the retry tests. + engine.summaryRetryOptions = { sleep: async () => {}, backoffBaseMs: 0 }; + engine.registry.registerObject({ + name: 'inv', + fields: { + name: { type: 'text' }, + line_total: { type: 'summary', summaryOperations: { object: 'inv_line', field: 'amount', function: 'sum' } }, + }, + } as any); + engine.registry.registerObject({ + name: 'inv_line', + fields: { amount: { type: 'number' }, inv: { type: 'master_detail', reference: 'inv' } }, + } as any); + return { engine, storeFor: d.storeFor, parentCalls: d.parentCalls }; +} + +describe('roll-up summary recompute — retry + surface failures (framework#3147)', () => { + it('retries a transient parent-update failure and lands the correct summary', async () => { + const { engine, storeFor } = await makeEngine({ + onParentUpdate: (_id, callN) => { if (callN === 1) throw new Error('fetch failed'); }, + }); + const p = await engine.insert('inv', { name: 'INV-1' }); + // Child insert succeeds; the parent summary update throws once then retries. + await engine.insert('inv_line', { inv: p.id, amount: 42 }); + + expect(storeFor('inv').get(p.id).line_total).toBe(42); + }); + + it('surfaces ERR_SUMMARY_RECOMPUTE when retries are exhausted, and the child is still written', async () => { + const { engine, storeFor } = await makeEngine({ + onParentUpdate: () => { throw new Error('ETIMEDOUT'); }, // persistent transient + }); + const p = await engine.insert('inv', { name: 'INV-2' }); + + let caught: any; + await engine.insert('inv_line', { inv: p.id, amount: 10 }).catch((e) => { caught = e; }); + + expect(caught?.code).toBe('ERR_SUMMARY_RECOMPUTE'); + expect(Array.isArray(caught?.failures)).toBe(true); + expect(caught.failures[0]).toMatchObject({ parentObject: 'inv', field: 'line_total' }); + // The child record WAS written (only the summary recompute failed). + expect(caught?.written).toBeTruthy(); + expect(Array.from(storeFor('inv_line').values())).toHaveLength(1); + }); + + it('does not retry a non-transient parent-update failure (single attempt), still surfaces it', async () => { + const { engine, parentCalls } = await makeEngine({ + onParentUpdate: () => { throw new Error('validation failed: bad summary'); }, + }); + const p = await engine.insert('inv', { name: 'INV-3' }); + + let caught: any; + await engine.insert('inv_line', { inv: p.id, amount: 5 }).catch((e) => { caught = e; }); + + expect(caught?.code).toBe('ERR_SUMMARY_RECOMPUTE'); + expect(parentCalls.get(p.id)).toBe(1); // no retry on a logical error + }); + + it('one failing parent does not block another parent recompute; failures list only the failed one', async () => { + let failId: string | null = null; + const { engine, storeFor } = await makeEngine({ + onParentUpdate: (id) => { if (id === failId) throw new Error('ETIMEDOUT'); }, + }); + const a = await engine.insert('inv', { name: 'A' }); + const b = await engine.insert('inv', { name: 'B' }); + failId = a.id; // only A's summary update fails + + // Insert one child per parent as a single batch so both recompute in one call. + let caught: any; + await engine.insert('inv_line', [ + { inv: a.id, amount: 7 }, + { inv: b.id, amount: 3 }, + ]).catch((e) => { caught = e; }); + + expect(caught?.code).toBe('ERR_SUMMARY_RECOMPUTE'); + expect(caught.failures).toHaveLength(1); + expect(caught.failures[0].parentId).toBe(a.id); + // B's summary recomputed correctly despite A failing. + expect(storeFor('inv').get(b.id).line_total).toBe(3); + }); +}); diff --git a/packages/objectql/src/engine.ts b/packages/objectql/src/engine.ts index dfc8cf2840..846d7bd535 100644 --- a/packages/objectql/src/engine.ts +++ b/packages/objectql/src/engine.ts @@ -12,7 +12,8 @@ import { } from '@objectstack/spec/data'; import { parseAutonumberFormat, renderAutonumber, missingFieldValues } from '@objectstack/spec/data'; import { ExecutionContext, ExecutionContextSchema } from '@objectstack/spec/kernel'; -import { IDataDriver, IDataEngine, Logger, createLogger } from '@objectstack/core'; +import { IDataDriver, IDataEngine, Logger, createLogger, withTransientRetry, type RetryOptions } from '@objectstack/core'; +import { SummaryRecomputeError, type SummaryRecomputeFailure } from './summary-errors.js'; import { CoreServiceName, StorageNameMapping } from '@objectstack/spec/system'; import { IRealtimeService, RealtimeEventPayload } from '@objectstack/spec/contracts'; import type { ICryptoProvider, CryptoHandle } from '@objectstack/spec/contracts'; @@ -1742,6 +1743,13 @@ export class ObjectQL implements IDataEngine { * parent objects that aggregate it. Invalidated when packages register. */ private summaryIndex: Map | null = null; + /** + * Retry options for roll-up summary recompute (framework#3147). Public so a + * test can inject a no-op sleep for deterministic backoff; production uses + * the transient-retry defaults. + */ + summaryRetryOptions: RetryOptions = {}; + /** Invalidate the cached roll-up summary index (call when metadata changes). */ private invalidateSummaryIndex(): void { this.summaryIndex = null; @@ -1802,37 +1810,48 @@ export class ObjectQL implements IDataEngine { records: any, previous: any, execCtx?: ExecutionContext, - ): Promise { + ): Promise { const descriptors = this.getSummaryDescriptors(childObject); - if (descriptors.length === 0) return; + if (descriptors.length === 0) return []; const recs = Array.isArray(records) ? records : records ? [records] : []; const prevs = Array.isArray(previous) ? previous : previous ? [previous] : []; + const failures: SummaryRecomputeFailure[] = []; for (const desc of descriptors) { const ids = new Set(); for (const r of recs) { const v = r?.[desc.fkField]; if (v != null && v !== '') ids.add(String(v)); } for (const p of prevs) { const v = p?.[desc.fkField]; if (v != null && v !== '') ids.add(String(v)); } for (const parentId of ids) { try { - const rows = await this.aggregate(childObject, { - where: { [desc.fkField]: parentId }, - aggregations: [{ - function: desc.fn, - ...(desc.fn === 'count' ? {} : { field: desc.sourceField }), - alias: 'value', - }], - context: execCtx, - } as any); - let value = rows?.[0]?.value; - if (value == null) value = (desc.fn === 'count' || desc.fn === 'sum') ? 0 : null; - await this.update(desc.parentObject, { id: parentId, [desc.summaryField]: value }, { context: execCtx } as any); + // Retry a transient failure (a turso `fetch failed` on the parent + // aggregate/update) with backoff — a network blip here used to leave + // the parent summary silently stale (framework#3147). + await withTransientRetry(async () => { + const rows = await this.aggregate(childObject, { + where: { [desc.fkField]: parentId }, + aggregations: [{ + function: desc.fn, + ...(desc.fn === 'count' ? {} : { field: desc.sourceField }), + alias: 'value', + }], + context: execCtx, + } as any); + let value = rows?.[0]?.value; + if (value == null) value = (desc.fn === 'count' || desc.fn === 'sum') ? 0 : null; + await this.update(desc.parentObject, { id: parentId, [desc.summaryField]: value }, { context: execCtx } as any); + }, this.summaryRetryOptions); } catch (err) { + // Retries exhausted (or a non-transient failure). Record it so the + // caller can surface it — one parent's failure must not abort the + // remaining parents' recompute. this.logger.warn('Roll-up summary recompute failed', { childObject, parentObject: desc.parentObject, parentId, field: desc.summaryField, error: (err as any)?.message, }); + failures.push({ childObject, parentObject: desc.parentObject, parentId, field: desc.summaryField, error: err }); } } } + return failures; } /** @@ -2306,7 +2325,7 @@ export class ObjectQL implements IDataEngine { } // Roll-up: recompute parent summary fields that aggregate this object. - await this.recomputeSummaries(object, result, null, opCtx.context); + const summaryFailures = await this.recomputeSummaries(object, result, null, opCtx.context); // Publish data.record.created event to realtime service if (this.realtimeService) { @@ -2346,7 +2365,13 @@ export class ObjectQL implements IDataEngine { // Return the (possibly hook-mutated) after-view: the array of per-row // results for batch, the single record otherwise. - return isBatch ? rowHookContexts.map((rowCtx) => rowCtx.result) : rowHookContexts[0].result; + const written = isBatch ? rowHookContexts.map((rowCtx) => rowCtx.result) : rowHookContexts[0].result; + // Records ARE written; a summary that could not be recomputed after + // retries must not be silent (framework#3147). Thrown after realtime + // publish, carrying the written records so a bulk caller (seed/import) + // can treat it as a warning rather than re-writing. + if (summaryFailures.length > 0) throw new SummaryRecomputeError(summaryFailures, written); + return written; } catch (e) { this.logger.error('Insert operation failed', e as Error, { object }); throw e; @@ -2520,7 +2545,7 @@ export class ObjectQL implements IDataEngine { // Roll-up: recompute parent summaries; pass priorRecord too so a child // that moved to a different parent updates BOTH old and new parent. - await this.recomputeSummaries(object, result, priorRecord, opCtx.context); + const summaryFailures = await this.recomputeSummaries(object, result, priorRecord, opCtx.context); // Publish data.record.updated event to realtime service if (this.realtimeService) { @@ -2544,6 +2569,9 @@ export class ObjectQL implements IDataEngine { } } + // The record IS updated; a summary that could not recompute after + // retries must surface, not stay silent (framework#3147). + if (summaryFailures.length > 0) throw new SummaryRecomputeError(summaryFailures, hookContext.result); return hookContext.result; } catch (e) { this.logger.error('Update operation failed', e as Error, { object }); @@ -2754,7 +2782,9 @@ export class ObjectQL implements IDataEngine { await this.triggerHooks('afterDelete', hookContext); // Roll-up: recompute the parent summary now that the child is gone. - if (summaryPrev) await this.recomputeSummaries(object, null, summaryPrev, opCtx.context); + const summaryFailures = summaryPrev + ? await this.recomputeSummaries(object, null, summaryPrev, opCtx.context) + : []; // Publish data.record.deleted event to realtime service if (this.realtimeService) { @@ -2776,6 +2806,9 @@ export class ObjectQL implements IDataEngine { } } + // The record IS deleted; a summary that could not recompute after + // retries must surface, not stay silent (framework#3147). + if (summaryFailures.length > 0) throw new SummaryRecomputeError(summaryFailures, hookContext.result); return hookContext.result; } catch (e) { this.logger.error('Delete operation failed', e as Error, { object }); diff --git a/packages/objectql/src/index.ts b/packages/objectql/src/index.ts index c8467c97c0..3b75923ea0 100644 --- a/packages/objectql/src/index.ts +++ b/packages/objectql/src/index.ts @@ -36,6 +36,8 @@ export type { SysMetadataEngine, SysMetadataRepositoryOptions } from '@objectsta // Export Engine export { ObjectQL, ObjectRepository, ScopedContext } from './engine.js'; export type { ObjectQLHostContext, HookHandler, HookEntry, OperationContext, EngineMiddleware } from './engine.js'; +export { SummaryRecomputeError } from './summary-errors.js'; +export type { SummaryRecomputeFailure } from './summary-errors.js'; // Export in-memory aggregation fallback (used by engine.aggregate when the // driver lacks native groupBy/aggregations support; also useful for tests). diff --git a/packages/objectql/src/summary-errors.ts b/packages/objectql/src/summary-errors.ts new file mode 100644 index 0000000000..2424711017 --- /dev/null +++ b/packages/objectql/src/summary-errors.ts @@ -0,0 +1,35 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** One parent whose roll-up summary could not be recomputed after retries. */ +export interface SummaryRecomputeFailure { + childObject: string; + parentObject: string; + parentId: string; + field: string; + error: unknown; +} + +/** + * Thrown by engine.insert/update/delete when one or more parent roll-up + * summaries fail to recompute after transient retries (framework#3147). + * + * The triggering records WERE written — this signals a stale/incorrect + * summary, not a failed write. `written` carries the write's result (the array + * for a batch, the single record otherwise) so a caller that can tolerate a + * stale summary (e.g. a bulk seed/import, which treats it as a warning) can + * recover the records instead of re-running the write. Identified by `code` + * rather than `instanceof` so it survives crossing package boundaries. + */ +export class SummaryRecomputeError extends Error { + readonly code = 'ERR_SUMMARY_RECOMPUTE' as const; + constructor( + public readonly failures: SummaryRecomputeFailure[], + public readonly written: unknown, + ) { + super( + `Roll-up summary recompute failed after retries for ${failures.length} parent record(s); ` + + `the triggering records WERE written (summary values may be stale).`, + ); + this.name = 'SummaryRecomputeError'; + } +} diff --git a/packages/rest/src/import-runner-idempotency.test.ts b/packages/rest/src/import-runner-idempotency.test.ts index e01ffa4062..4ef839db5a 100644 --- a/packages/rest/src/import-runner-idempotency.test.ts +++ b/packages/rest/src/import-runner-idempotency.test.ts @@ -103,4 +103,27 @@ describe('runImport — idempotent retry with natural keys (framework#3149)', () // written twice. This pins the contract (exactly-once needs matchFields). expect(store).toHaveLength(4); }); + + it('marks rows created-with-warning on a summary recompute failure, without failing or duplicating (framework#3147)', async () => { + const store: Array> = []; + let idc = 0; + const createManyData = vi.fn(async (args: { records: any[] }) => { + const recs = args.records.map((r) => { const rec = { id: `id-${++idc}`, ...r }; store.push(rec); return rec; }); + // Records written, but the post-write summary recompute failed. + throw Object.assign(new Error('summary recompute failed'), { code: 'ERR_SUMMARY_RECOMPUTE', written: recs }); + }); + const createData = vi.fn(); + const p: ImportProtocolLike = { findData: vi.fn(async () => []), createData, updateData: vi.fn(), createManyData }; + + const summary = await runImport({ + ...baseOpts, p, writeMode: 'insert', matchFields: [], + rows: [{ name: 'x' }, { name: 'y' }], + }); + + expect(createData).not.toHaveBeenCalled(); // not degraded / re-created + expect(store).toHaveLength(2); // no duplicate + expect(summary.created).toBe(2); + expect(summary.errors).toBe(0); + expect(summary.results.every((r) => r.ok && r.code === 'SUMMARY_RECOMPUTE_FAILED')).toBe(true); + }); }); diff --git a/packages/rest/src/import-runner.ts b/packages/rest/src/import-runner.ts index 65030e3adb..112d436283 100644 --- a/packages/rest/src/import-runner.ts +++ b/packages/rest/src/import-runner.ts @@ -279,8 +279,23 @@ export function runImport(opts: RunImportOptions): Promise { // duplicate a row (framework#3149). A pure-insert import (no matchFields) has // no natural key to recheck against and stays at-least-once by contract. let lastBatchUncertain = false; + // Set when a flush's write succeeded but its post-write roll-up summary + // recompute exhausted retries (framework#3147). The rows ARE written; we mark + // them created-with-a-warning code rather than failing (or re-writing) them. + let flushSummaryStale = false; const isUncertainOutcome = (e: unknown) => defaultIsTransientError(e) || (e as { code?: unknown } | null)?.code === 'ERR_BULK_RESULT_MISMATCH'; + // A post-write summary recompute failure (ERR_SUMMARY_RECOMPUTE) means the + // records were written; recover the written records from the error rather + // than letting the write look failed (which would re-create → duplicate). + const recoverSummaryStale = (e: unknown): unknown[] | null => { + const err = e as { code?: unknown; written?: unknown } | null; + if (err?.code === 'ERR_SUMMARY_RECOMPUTE') { + flushSummaryStale = true; + return Array.isArray(err.written) ? err.written : (err.written != null ? [err.written] : []); + } + return null; + }; const existingByMatch = async (data: Record): Promise | null> => { if (matchFields.length === 0) return null; const found = await findExisting(data); @@ -288,6 +303,7 @@ export function runImport(opts: RunImportOptions): Promise { }; const flushPendingCreates = async (): Promise => { if (pendingCreates.length === 0) return; + flushSummaryStale = false; const batch = pendingCreates.splice(0, pendingCreates.length); const writeResults: BulkWriteRowResult[] = await bulkWrite( batch.map(b => b.data), @@ -310,12 +326,22 @@ export function runImport(opts: RunImportOptions): Promise { } } try { - const createdRecords = toCreate.length === 0 - ? [] - : (await p.createManyData!({ + let createdRecords: any[]; + if (toCreate.length === 0) { + createdRecords = []; + } else { + try { + createdRecords = (await p.createManyData!({ object: objectName, records: toCreate, context: writeCtx, ...(environmentId ? { environmentId } : {}), })).records; + } catch (e) { + // Records written but summary recompute failed: recover them. + const recovered = recoverSummaryStale(e); + if (!recovered) throw e; + createdRecords = recovered; + } + } // Surface a short/non-array createManyData return as a failed batch // (framework#3151) rather than padding the reassembly with undefined // — this drops into per-row degradation, which rechecks first. @@ -340,10 +366,16 @@ export function runImport(opts: RunImportOptions): Promise { const hit = await existingByMatch(row); if (hit) return hit; // already committed by a prior attempt } - return p.createData({ - object: objectName, data: row, context: writeCtx, - ...(environmentId ? { environmentId } : {}), - }); + try { + return await p.createData({ + object: objectName, data: row, context: writeCtx, + ...(environmentId ? { environmentId } : {}), + }); + } catch (e) { + const recovered = recoverSummaryStale(e); + if (recovered) return recovered[0]; // record written; summary stale + throw e; + } }, }, ); @@ -353,7 +385,8 @@ export function runImport(opts: RunImportOptions): Promise { const id = extractRecordId(res.record); okCount++; created++; if (collectUndo && id != null) undoLog.created.push(id); - results[index] = { row: rowNo, ok: true, action: 'created', id }; + results[index] = { row: rowNo, ok: true, action: 'created', id, + ...(flushSummaryStale ? { code: 'SUMMARY_RECOMPUTE_FAILED' } : {}) }; } else { errCount++; results[index] = toFailedResult(rowNo, res.error); @@ -420,13 +453,24 @@ export function runImport(opts: RunImportOptions): Promise { else { created++; results[i] = { row: rowNo, ok: true, action: 'created' }; } } else if (willUpdate) { const target = existing as Record; - const res2 = await withTransientRetry(() => p.updateData({ object: objectName, id: target.id, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) })); + let res2: unknown; + let updateSummaryStale = false; + try { + res2 = await withTransientRetry(() => p.updateData({ object: objectName, id: target.id, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) })); + } catch (e) { + // Record updated but summary recompute failed (framework#3147): + // the update landed, so recover rather than fail the row. + const recovered = recoverSummaryStale(e); + if (!recovered) throw e; + res2 = recovered[0]; updateSummaryStale = true; + } 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[i] = { row: rowNo, ok: true, action: 'updated', id }; + results[i] = { row: rowNo, ok: true, action: 'updated', id, + ...(updateSummaryStale ? { code: 'SUMMARY_RECOMPUTE_FAILED' } : {}) }; } else if (canBulkCreate) { // Buffer — the actual write happens in a batched flush below. pendingCreates.push({ index: i, rowNo, data }); From ba7e2377893652d3a208a74444e058fb6796525a Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 03:51:37 +0000 Subject: [PATCH 7/8] fix(objectql): assign autonumbers after validation; document at-least-once hooks (#3152) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In the batch insert path, beforeInsert hooks and autonumber assignment ran before per-row validation. When bulkWrite degrades a batch to per-row (or retries it), the whole engine.insert re-runs, so a batch that dies in validation had already consumed an autonumber sequence value for every good row — leaving number-range gaps — and re-fired side-effecting beforeInsert hooks on the retry. Move autonumber assignment to after validation: a doomed attempt now consumes no sequence value (required-validation already exempts autonumber fields, and a driver-owned autonumber assigns nothing here, so no validation rule can depend on the value — the reorder is safe). The remaining beforeInsert double-execution on a degraded batch is documented as at-least-once hook semantics on engine.insert (hooks must be idempotent); root-causing it needs a partial-success engine API, out of scope here. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- .../src/engine-autonumber-batch.test.ts | 90 +++++++++++++++++++ packages/objectql/src/engine.ts | 28 +++++- 2 files changed, 115 insertions(+), 3 deletions(-) create mode 100644 packages/objectql/src/engine-autonumber-batch.test.ts diff --git a/packages/objectql/src/engine-autonumber-batch.test.ts b/packages/objectql/src/engine-autonumber-batch.test.ts new file mode 100644 index 0000000000..fc42e1c7dc --- /dev/null +++ b/packages/objectql/src/engine-autonumber-batch.test.ts @@ -0,0 +1,90 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. +// +// framework#3152: a batch insert that dies in validation must not have already +// consumed autonumber sequence values for its good rows. Autonumbers are now +// assigned AFTER validation, so a doomed attempt (which bulkWrite then degrades +// to per-row) leaves no gaps in the number range. + +import { describe, it, expect } from 'vitest'; +import { ObjectQL } from './engine.js'; + +function makeDriver() { + const stores = new Map>(); + const storeFor = (o: string) => { + let s = stores.get(o); + if (!s) { s = new Map(); stores.set(o, s); } + return s; + }; + let n = 0; + const driver: any = { + name: 'memory', version: '0.0.0', supports: {}, + async connect() {}, async disconnect() {}, async checkHealth() { return true; }, async execute() { return null; }, + async find(object: string) { return Array.from(storeFor(object).values()); }, + findStream() { throw new Error('ni'); }, + async findOne() { return null; }, + async create(object: string, data: Record) { + n += 1; + const id = (data.id as string) ?? `r_${n}`; + const row = { ...data, id }; + storeFor(object).set(id, row); + return row; + }, + async update(object: string, id: string, data: Record) { + const s = storeFor(object); + const row = { ...s.get(id), ...data, id }; + s.set(id, row); + return row; + }, + async delete(object: string, id: string) { return storeFor(object).delete(id); }, + async count() { return 0; }, + async bulkCreate(object: string, rows: Record[]) { + const out: any[] = []; + for (const r of rows) out.push(await this.create(object, r)); + return out; + }, + async bulkUpdate() { return []; }, async bulkDelete() {}, + async beginTransaction() { return { __trx: true, commit: async () => {}, rollback: async () => {} }; }, + async commit() {}, async rollback() {}, + }; + return { driver, storeFor }; +} + +async function makeEngine() { + const engine = new ObjectQL(); + const d = makeDriver(); + engine.registerDriver(d.driver, true); + await engine.init(); + engine.registry.registerObject({ + name: 'task', + fields: { + name: { type: 'text', required: true }, + code: { type: 'autonumber', format: 'T-{000}' }, + }, + } as any); + return { engine, storeFor: d.storeFor }; +} + +describe('batch insert — autonumber assigned after validation (framework#3152)', () => { + it('a validation-failed batch consumes no autonumber, leaving no gap for the degraded per-row retry', async () => { + const { engine } = await makeEngine(); + + // A batch with a bad (name-less) middle row fails in validation before any + // driver write. Previously the good rows had already been numbered. + await expect(engine.insert('task', [{ name: 'a' }, { slug: 'no-name' }, { name: 'b' }])) + .rejects.toThrow(); + + // bulkWrite would now degrade to per-row; simulate that by inserting the + // good rows individually. Their numbers must be contiguous from 1. + const a = await engine.insert('task', { name: 'a' }); + const b = await engine.insert('task', { name: 'b' }); + + expect(a.code).toBe('T-001'); + expect(b.code).toBe('T-002'); // no gap — the failed batch consumed nothing + }); + + it('a fully-valid batch still numbers every row contiguously', async () => { + const { engine } = await makeEngine(); + const rows = await engine.insert('task', [{ name: 'a' }, { name: 'b' }, { name: 'c' }]); + expect((rows as any[]).map((r) => r.code)).toEqual(['T-001', 'T-002', 'T-003']); + }); +}); diff --git a/packages/objectql/src/engine.ts b/packages/objectql/src/engine.ts index 846d7bd535..ec224fbe54 100644 --- a/packages/objectql/src/engine.ts +++ b/packages/objectql/src/engine.ts @@ -2200,6 +2200,20 @@ export class ObjectQL implements IDataEngine { return opCtx.result; } + /** + * Insert one record or an array of records. + * + * At-least-once hook semantics (framework#3152): when this call is driven by + * `bulkWrite` (seed / import), a transient batch retry or a per-row + * degradation re-runs the whole insert, so a `beforeInsert` hook may fire + * MORE THAN ONCE for the same input row (a first attempt that failed in + * validation, then the degraded retry). Side-effecting `beforeInsert` hooks + * (notifications, external calls, counters) must therefore be idempotent. + * `afterInsert` hooks fire only on a successful write, so they are not + * re-run by a validation failure. Autonumbers are assigned only after + * validation passes, so a doomed attempt no longer consumes a sequence value + * (no number-range gaps from a rejected batch). + */ async insert(object: string, data: any | any[], options?: DataEngineInsertOptions): Promise { object = this.resolveObjectName(object); this.logger.debug('Insert operation starting', { object, isBatch: Array.isArray(data) }); @@ -2271,9 +2285,6 @@ export class ObjectQL implements IDataEngine { // Defaults are already resolved above (pre-hook, #2703); a hook may // have overridden fields or replaced `input.data` — take its data as-is. const rows = rowHookContexts.map((rowCtx) => rowCtx.input.data as Record); - for (const r of rows) { - await this.applyAutonumbers(object, r, opCtx.context, driverOwnsAutonumber); - } for (const r of rows) { await this.encryptSecretFields(object, r, opCtx.context, driverOptions); } @@ -2282,6 +2293,17 @@ export class ObjectQL implements IDataEngine { validateRecord(schemaForValidation, r, 'insert'); evaluateValidationRules(schemaForValidation as any, r, 'insert', { logger: this.logger, currentUser: this.buildEvalUser(opCtx.context) }); } + // Autonumbers are assigned AFTER validation (framework#3152): in the + // batch path a bulkWrite retry / per-row degradation re-runs the whole + // engine.insert, and a batch that dies in validation would otherwise + // have already consumed a sequence value for every good row on the + // failed attempt, leaving gaps. Required-validation exempts autonumber + // fields either way (they are engine-assigned), and a driver that owns + // autonumber assigns nothing here — so no validation rule can depend on + // the value, making this reorder safe. + for (const r of rows) { + await this.applyAutonumbers(object, r, opCtx.context, driverOwnsAutonumber); + } if (isBatch) { if (driver.bulkCreate) { result = await driver.bulkCreate(object, rows, driverOptions); From 2ae283aa8ca8a31b065aebeaed04a90e2cb78a85 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 04:01:38 +0000 Subject: [PATCH 8/8] =?UTF-8?q?chore:=20add=20changeset=20for=20bulk-write?= =?UTF-8?q?=20hardening=20(#3147=E2=80=93#3152)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga --- .changeset/bulk-write-hardening.md | 31 ++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) create mode 100644 .changeset/bulk-write-hardening.md diff --git a/.changeset/bulk-write-hardening.md b/.changeset/bulk-write-hardening.md new file mode 100644 index 0000000000..e32a7e46ba --- /dev/null +++ b/.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).