Skip to content

Commit 9992e02

Browse files
committed
feat(objectql,core,seed): partial-success batch insert — beforeInsert fires once (#3172)
Root fix for the #3152 leftover: a batch that died in validation forced bulkWrite's whole-batch degradation, re-running beforeInsert hooks on the good rows (at-least-once side effects). - engine.insertMany(object, rows, options): new partial-success entry. Bad rows are culled per-row AFTER beforeInsert (encryption/validation/ autonumber failures each captured per row), survivors are written in one driver batch, and the return is one outcome per input row in input order ({ok,record} | {ok:false,error}). afterInsert, summary recompute and realtime fire only for written rows; dead rows consume no autonumber. Plain insert(array) semantics are untouched. - bulkWrite gains an optional writeBatchPartial: per-row failures from it are final verdicts (no writeOne degradation — that would re-fire the hooks); only a THROWN error (transient, mismatch) still degrades. - seed-loader's flushPendingInserts switches to writeBatchPartial and feature-detects engine.insertMany (legacy engines keep the old whole-array + degradation behavior). - protocol.insertManyData: partial-success sibling of createManyData (same readonly ingress strip), for the import runner. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga
1 parent 1cd4264 commit 9992e02

8 files changed

Lines changed: 435 additions & 40 deletions

File tree

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

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,60 @@ describe('bulkWrite', () => {
173173
expect(seen).toEqual([1, 2]); // attempt 1 threw, attempt 2 succeeded
174174
});
175175

176+
it('writeBatchPartial: per-row failures are final verdicts — no writeOne degradation (#3172)', async () => {
177+
const writeBatchPartial = vi.fn(async (batch: { n: number; bad?: boolean }[]) =>
178+
batch.map((r) => (r.bad
179+
? { ok: false, error: new Error('validation failed: bad row') }
180+
: { ok: true, record: { id: `r${r.n}` } })));
181+
const writeBatch = vi.fn();
182+
const writeOne = vi.fn();
183+
184+
const results = await bulkWrite(
185+
[{ n: 1 }, { n: 2, bad: true }, { n: 3 }],
186+
{ batchSize: 10, writeBatchPartial, writeBatch, writeOne, sleep: noopSleep },
187+
);
188+
189+
expect(writeBatchPartial).toHaveBeenCalledTimes(1);
190+
expect(writeBatch).not.toHaveBeenCalled(); // partial path replaces writeBatch
191+
expect(writeOne).not.toHaveBeenCalled(); // per-row failure ≠ batch failure
192+
expect(results[0]).toMatchObject({ index: 0, ok: true, record: { id: 'r1' } });
193+
expect(results[1].ok).toBe(false);
194+
expect(results[2]).toMatchObject({ index: 2, ok: true, record: { id: 'r3' } });
195+
});
196+
197+
it('writeBatchPartial: a THROWN error still degrades to writeOne, with transient retry first (#3172)', async () => {
198+
let attempts = 0;
199+
const writeBatchPartial = vi.fn(async (batch: { n: number }[]) => {
200+
attempts++;
201+
if (attempts === 1) throw new Error('fetch failed'); // transient → retried
202+
if (attempts === 2) throw new Error('CHECK constraint failed'); // logical → degrade
203+
return batch.map((r) => ({ ok: true, record: { id: `r${r.n}` } }));
204+
});
205+
const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
206+
207+
const results = await bulkWrite(
208+
[{ n: 1 }, { n: 2 }],
209+
{ batchSize: 10, writeBatchPartial, writeBatch: vi.fn(), writeOne, sleep: noopSleep },
210+
);
211+
212+
expect(writeBatchPartial).toHaveBeenCalledTimes(2); // transient retry happened
213+
expect(writeOne).toHaveBeenCalledTimes(2); // then per-row degradation
214+
expect(results.every((r) => r.ok)).toBe(true);
215+
});
216+
217+
it('writeBatchPartial: a wrong-length outcome array is rejected as a failed batch (#3172)', async () => {
218+
const writeBatchPartial = vi.fn(async () => [{ ok: true, record: { id: 'only-one' } }]);
219+
const writeOne = vi.fn(async (row: { n: number }) => ({ id: `r${row.n}` }));
220+
221+
const results = await bulkWrite(
222+
[{ n: 1 }, { n: 2 }],
223+
{ batchSize: 10, writeBatchPartial, writeBatch: vi.fn(), writeOne, sleep: noopSleep },
224+
);
225+
226+
expect(writeOne).toHaveBeenCalledTimes(2); // degraded instead of trusting the short array
227+
expect(results.every((r) => r.ok)).toBe(true);
228+
});
229+
176230
it('passes an attempt counter to writeOne during degradation (#3149)', async () => {
177231
const seen: number[] = [];
178232
const writeBatch = vi.fn(async () => { throw new Error('logical'); }); // force degradation

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

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,21 @@ export interface BulkWriteOptions<TRow, TRecord = any> extends RetryOptions {
7979
* carries the same recheck signal as {@link writeBatch}.
8080
*/
8181
writeOne: (row: TRow, ctx: { attempt: number }) => Promise<TRecord>;
82+
/**
83+
* Partial-success batch write (framework#3172). When provided it is used
84+
* INSTEAD of {@link writeBatch}: it must resolve one outcome per input row,
85+
* in input order — `{ ok: true, record }` for written rows, `{ ok: false,
86+
* error }` for rows that failed individually (e.g. validation). Per-row
87+
* failures are final verdicts: bulkWrite records them as-is and does NOT
88+
* degrade to `writeOne` for them — that is the whole point (a degradation
89+
* re-run would re-fire beforeInsert hooks on the good rows). Only a THROWN
90+
* error (a transient infra failure, a result-count mismatch) falls back to
91+
* the per-row `writeOne` degradation, exactly like `writeBatch`.
92+
*/
93+
writeBatchPartial?: (
94+
batch: TRow[],
95+
ctx: { attempt: number },
96+
) => Promise<Array<{ ok: boolean; record?: TRecord; error?: unknown }>>;
8297
}
8398

8499
const DEFAULT_BATCH_SIZE = 200;
@@ -202,6 +217,29 @@ export async function bulkWrite<TRow, TRecord = any>(
202217
for (let start = 0; start < rows.length; start += batchSize) {
203218
const batch = rows.slice(start, start + batchSize);
204219
try {
220+
// Partial-success path (framework#3172): one call yields a final per-row
221+
// verdict, so a row that fails validation never triggers the whole-batch
222+
// degradation that re-runs beforeInsert hooks on its siblings.
223+
if (opts.writeBatchPartial) {
224+
const outcomes = await withRetry((attempt) => opts.writeBatchPartial!(batch, { attempt }), retryOpts);
225+
if (!Array.isArray(outcomes) || outcomes.length !== batch.length) {
226+
throw Object.assign(
227+
new Error(
228+
`bulkWrite: writeBatchPartial returned ${
229+
Array.isArray(outcomes) ? `${outcomes.length} outcome(s)` : String(typeof outcomes)
230+
} for a ${batch.length}-row batch — treating batch as failed`,
231+
),
232+
{ code: 'ERR_BULK_RESULT_MISMATCH' },
233+
);
234+
}
235+
for (let i = 0; i < batch.length; i++) {
236+
const o = outcomes[i];
237+
results[start + i] = o.ok
238+
? { index: start + i, ok: true, record: o.record }
239+
: { index: start + i, ok: false, error: o.error };
240+
}
241+
continue;
242+
}
205243
const records = await withRetry((attempt) => opts.writeBatch(batch, { attempt }), retryOpts);
206244
// Contract guard (framework#3151): `writeBatch` must resolve one record
207245
// per input row. A short / long / non-array return breaks the positional

packages/metadata-protocol/src/protocol.ts

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3265,6 +3265,34 @@ export class ObjectStackProtocolImplementation implements ObjectStackProtocol {
32653265
count: records.length
32663266
};
32673267
}
3268+
3269+
/**
3270+
* Partial-success bulk create (framework#3172): like createManyData, but a
3271+
* row that fails validation is a per-row verdict instead of aborting the
3272+
* whole batch — the import runner uses this so a bad row never forces a
3273+
* degradation re-run of the good rows' beforeInsert hooks. Requires an
3274+
* engine with `insertMany` (ObjectQL has it); absent that, callers should
3275+
* fall back to createManyData.
3276+
*/
3277+
async insertManyData(request: { object: string, records: any[], context?: any }): Promise<{ object: string; outcomes: Array<{ ok: boolean; record?: any; error?: unknown }> }> {
3278+
const engineInsertMany = (this.engine as any)?.insertMany;
3279+
if (typeof engineInsertMany !== 'function') {
3280+
throw new Error('insertManyData requires an engine with insertMany (framework#3172)');
3281+
}
3282+
// Same ingress strip as createManyData (#3043).
3283+
const rows = stripReadonlyForInsert(
3284+
this.engine.registry?.getObject(request.object),
3285+
request.records,
3286+
request.context,
3287+
);
3288+
const outcomes = await engineInsertMany.call(
3289+
this.engine,
3290+
request.object,
3291+
rows,
3292+
request.context !== undefined ? { context: request.context } as any : undefined,
3293+
);
3294+
return { object: request.object, outcomes };
3295+
}
32683296

32693297
async updateManyData(request: UpdateManyDataRequest): Promise<BatchUpdateResponse> {
32703298
const { object, records, options } = request;

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -180,6 +180,55 @@ describe('seed batched path — idempotent retry after commit-then-lost-response
180180
});
181181
});
182182

183+
describe('seed batched path — partial-success engine (framework#3172)', () => {
184+
it('a bad row is a per-row verdict: good rows insert once, no degradation re-run', async () => {
185+
const { engine, store } = createFaithfulEngine();
186+
const metadata = createMetadata();
187+
188+
// Engine with insertMany (partial success): the bad row comes back as a
189+
// per-row error, good rows as records — one call, no thrown batch error.
190+
let insertManyCalls = 0;
191+
(engine as any).insertMany = vi.fn(async (obj: string, rows: any[], opts: any) => {
192+
insertManyCalls++;
193+
const outcomes = [];
194+
for (const r of rows) {
195+
if (r.sku === 'BAD') {
196+
outcomes.push({ ok: false, error: new Error('validation failed: bad widget') });
197+
} else {
198+
outcomes.push({ ok: true, record: await (engine.insert as any)(obj, r, opts) });
199+
}
200+
}
201+
return outcomes;
202+
});
203+
204+
const result = await new SeedLoaderService(engine, metadata, createLogger()).load({
205+
seeds: [{
206+
object: 'my_app_widget',
207+
externalId: 'sku',
208+
mode: 'insert',
209+
env: ['prod', 'dev', 'test'],
210+
records: [
211+
{ name: 'Good A', sku: 'W-A' },
212+
{ name: 'Bad', sku: 'BAD' },
213+
{ name: 'Good B', sku: 'W-B' },
214+
],
215+
}] as any,
216+
config: CONFIG,
217+
});
218+
219+
expect(insertManyCalls).toBe(1);
220+
expect(result.summary.totalInserted).toBe(2);
221+
expect(result.summary.totalErrored).toBe(1);
222+
expect(store.my_app_widget.map((r: any) => r.sku).sort()).toEqual(['W-A', 'W-B']);
223+
// The good rows were written exactly once — no whole-batch degradation
224+
// re-ran them through engine.insert (single-row form).
225+
const singleInsertCalls = (engine.insert as any).mock.calls.filter(
226+
([obj, data]: [string, any]) => obj === 'my_app_widget' && !Array.isArray(data),
227+
);
228+
expect(singleInsertCalls).toHaveLength(2); // only the two calls made INSIDE insertMany's mock
229+
});
230+
});
231+
183232
describe('seed batched path — summary recompute failure is a warning, not an error (framework#3147)', () => {
184233
it('records the rows as inserted (not errored) and does not re-insert on ERR_SUMMARY_RECOMPUTE', async () => {
185234
const { engine, store } = createFaithfulEngine();

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

Lines changed: 38 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -271,25 +271,23 @@ export class SeedLoaderService implements ISeedLoaderService {
271271
let lastBatchUncertain = false;
272272
const isUncertainOutcome = (e: unknown) =>
273273
defaultIsTransientError(e) || (e as { code?: unknown } | null)?.code === 'ERR_BULK_RESULT_MISMATCH';
274-
// Reassemble one record per input row in order: rows already present are
275-
// represented by the existing record; the rest are consumed in order from
276-
// the freshly-inserted list.
277-
const assembleInOrder = (rows: Record<string, unknown>[], existing: Map<string, any>, freshlyInserted: any[]): any[] => {
278-
let k = 0;
279-
return rows.map((r) => {
280-
const key = extIdOf(r);
281-
if (key && existing.has(key)) return existing.get(key);
282-
return freshlyInserted[k++];
283-
});
284-
};
274+
// Partial-success entry (framework#3172): when the engine offers
275+
// insertMany, a row that fails validation is a final per-row verdict from
276+
// ONE batch call — bulkWrite never degrades, so beforeInsert hooks run
277+
// exactly once per row. Feature-detected so a legacy engine (tests, other
278+
// IDataEngine impls) falls back to the whole-array insert + degradation.
279+
const engineInsertMany: ((o: string, rows: any[], op: any) => Promise<Array<{ ok: boolean; record?: any; error?: unknown }>>) | undefined =
280+
typeof (this.engine as any).insertMany === 'function'
281+
? (o, rows, op) => (this.engine as any).insertMany(o, rows, op)
282+
: undefined;
285283
const flushPendingInserts = async (): Promise<void> => {
286284
if (pendingInserts.length === 0) return;
287285
const batch = pendingInserts.splice(0, pendingInserts.length);
288286
const writeResults: BulkWriteRowResult[] = await bulkWrite(
289287
batch.map(b => b.record),
290288
{
291289
batchSize: SeedLoaderService.BULK_BATCH_SIZE,
292-
writeBatch: async (rows, { attempt }) => {
290+
writeBatchPartial: async (rows, { attempt }) => {
293291
let toInsert = rows;
294292
let existing = new Map<string, any>();
295293
if (attempt > 1) {
@@ -299,21 +297,42 @@ export class SeedLoaderService implements ISeedLoaderService {
299297
toInsert = rows.filter((r) => { const k = extIdOf(r); return !(k && existing.has(k)); });
300298
}
301299
try {
302-
// A lone row keeps the historical bare-record insert() call shape
303-
// (no array wrapping) so single-record datasets are byte-for-byte
304-
// unchanged; only a real batch (>1) uses the array/bulk form.
305-
const freshlyInserted = toInsert.length === 0
306-
? []
307-
: toInsert.length === 1
300+
let freshOutcomes: Array<{ ok: boolean; record?: any; error?: unknown }>;
301+
if (toInsert.length === 0) {
302+
freshOutcomes = [];
303+
} else if (engineInsertMany) {
304+
// Partial-success batch: per-row verdicts, hooks fire once.
305+
// On ERR_SUMMARY_RECOMPUTE, writeRecoveringSummary hands back
306+
// e.written — which for insertMany IS the outcome array.
307+
freshOutcomes = await this.writeRecoveringSummary(() => engineInsertMany(objectName, toInsert, opts));
308+
} else {
309+
// Legacy whole-array insert: any bad row throws the batch, and
310+
// bulkWrite's per-row degradation (writeOne) sorts it out. A
311+
// lone row keeps the historical bare-record insert() shape.
312+
const recs = toInsert.length === 1
308313
? [await this.writeRecoveringSummary(() => this.engine.insert(objectName, toInsert[0], opts))]
309314
: await this.writeRecoveringSummary(() => this.engine.insert(objectName, toInsert, opts));
315+
freshOutcomes = (recs as any[]).map((r) => ({ ok: true, record: r }));
316+
}
310317
lastBatchUncertain = false;
311-
return assembleInOrder(rows, existing, freshlyInserted as any[]);
318+
// Reassemble one outcome per input row in order: recheck hits use
319+
// the existing record; the rest consume freshOutcomes in order.
320+
let k = 0;
321+
return rows.map((r) => {
322+
const key = extIdOf(r);
323+
if (key && existing.has(key)) return { ok: true, record: existing.get(key) };
324+
return freshOutcomes[k++];
325+
});
312326
} catch (e) {
313327
lastBatchUncertain = isUncertainOutcome(e);
314328
throw e;
315329
}
316330
},
331+
writeBatch: async () => {
332+
// Unreachable — writeBatchPartial above takes precedence — but the
333+
// BulkWriteOptions contract requires it.
334+
throw new Error('seed-loader uses writeBatchPartial');
335+
},
317336
writeOne: async (row, { attempt }) => {
318337
if (attempt > 1 || lastBatchUncertain) {
319338
const key = extIdOf(row);

0 commit comments

Comments
 (0)