Skip to content

Commit ac781b7

Browse files
committed
fix(seed,rest,core): wrap bare writes in transient retry; short-circuit logical errors (#3150)
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga
1 parent 5522a2a commit ac781b7

6 files changed

Lines changed: 200 additions & 12 deletions

File tree

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,4 +195,14 @@ describe('defaultIsTransientError', () => {
195195
expect(defaultIsTransientError(new Error('Validation failed: email is required'))).toBe(false);
196196
expect(defaultIsTransientError(new Error('UNIQUE constraint failed: task.id'))).toBe(false);
197197
});
198+
199+
it('classifies a mixed constraint+network message as logical, not transient (#3150)', () => {
200+
// A logical signature must win even when a transient keyword also appears,
201+
// so we do not burn retries on a row that will fail identically each time.
202+
expect(defaultIsTransientError(new Error('CHECK constraint failed: network_zone'))).toBe(false);
203+
expect(defaultIsTransientError(new Error('column network_id is not allowed'))).toBe(false);
204+
expect(defaultIsTransientError(new Error('value out of range at row 503'))).toBe(false);
205+
// Pure transient signatures are unaffected.
206+
expect(defaultIsTransientError(new Error('fetch failed'))).toBe(true);
207+
});
198208
});

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

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -97,11 +97,33 @@ const TRANSIENT_PATTERNS: RegExp[] = [
9797

9898
const TRANSIENT_CODES = /^(ECONNRESET|ECONNREFUSED|ECONNABORTED|EPIPE|EAI_AGAIN|ETIMEDOUT|EHOSTUNREACH|ENETUNREACH|ENOTFOUND)$/i;
9999

100+
/**
101+
* Validation / constraint / schema signatures that are DEFINITIVELY logical,
102+
* never worth retrying. Checked before {@link TRANSIENT_PATTERNS} so a message
103+
* that happens to mention both (e.g. `CHECK constraint failed: network_zone`,
104+
* `column network_id is not allowed`) is classified as logical rather than
105+
* burning retries on a row that will fail identically every time (framework
106+
* #3150).
107+
*/
108+
const NON_TRANSIENT_PATTERNS: RegExp[] = [
109+
/validation/i,
110+
/constraint/i,
111+
/\brequired\b/i,
112+
/\bunique\b/i,
113+
/duplicate/i,
114+
/not[\s_-]*null/i,
115+
/invalid/i,
116+
/not allowed/i,
117+
/out of range/i,
118+
];
119+
100120
export function defaultIsTransientError(err: unknown): boolean {
101-
const code = (err as { code?: unknown } | null)?.code;
102-
if (typeof code === 'string' && TRANSIENT_CODES.test(code)) return true;
103121
const message = (err as { message?: unknown } | null)?.message;
104122
const text = typeof message === 'string' ? message : String(err ?? '');
123+
// A definitive logical signature wins even if a transient word also appears.
124+
if (NON_TRANSIENT_PATTERNS.some((re) => re.test(text))) return false;
125+
const code = (err as { code?: unknown } | null)?.code;
126+
if (typeof code === 'string' && TRANSIENT_CODES.test(code)) return true;
105127
return TRANSIENT_PATTERNS.some((re) => re.test(text));
106128
}
107129

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
import { describe, it, expect, vi } from 'vitest';
4+
import { SeedLoaderService } from './seed-loader';
5+
import type { IDataEngine, IMetadataService } from '@objectstack/spec/contracts';
6+
7+
/**
8+
* framework#3150: the self-referencing seed path (`hasSelfRef`) writes records
9+
* sequentially via `writeRecord`, which — unlike the batched path (bulkWrite's
10+
* internal retry) and the update path — used to call `engine.insert` bare. A
11+
* single transient blip (`fetch failed`) therefore dropped the row with no
12+
* retry. These tests pin the now-wrapped behaviour.
13+
*/
14+
15+
function createLogger() {
16+
return { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() };
17+
}
18+
19+
function createFaithfulEngine(): { engine: IDataEngine; store: Record<string, any[]> } {
20+
const store: Record<string, any[]> = {};
21+
let idCounter = 0;
22+
23+
const engine = {
24+
find: vi.fn(async (objectName: string, query?: any) => {
25+
let records = store[objectName] || [];
26+
if (query?.where) {
27+
records = records.filter((r) =>
28+
Object.entries(query.where).every(([k, v]) => r[k] === v),
29+
);
30+
}
31+
if (typeof query?.limit === 'number') records = records.slice(0, query.limit);
32+
return records;
33+
}),
34+
findOne: vi.fn(async (objectName: string, query?: any) => {
35+
const rows = await (engine.find as any)(objectName, { ...query, limit: 1 });
36+
return rows[0] ?? null;
37+
}),
38+
insert: vi.fn(async (objectName: string, data: any) => {
39+
if (!store[objectName]) store[objectName] = [];
40+
if (Array.isArray(data)) {
41+
const records = data.map((d) => ({ id: `gen-${++idCounter}`, ...d }));
42+
store[objectName].push(...records);
43+
return records;
44+
}
45+
const record = { id: `gen-${++idCounter}`, ...data };
46+
store[objectName].push(record);
47+
return record;
48+
}),
49+
update: vi.fn(async (objectName: string, data: any) => {
50+
const records = store[objectName] || [];
51+
const idx = records.findIndex((r) => r.id === data.id);
52+
if (idx >= 0) { records[idx] = { ...records[idx], ...data }; return records[idx]; }
53+
return data;
54+
}),
55+
delete: vi.fn(async () => ({ deleted: 1 })),
56+
count: vi.fn(async (objectName: string) => (store[objectName] || []).length),
57+
aggregate: vi.fn(async () => []),
58+
} as unknown as IDataEngine;
59+
60+
return { engine, store };
61+
}
62+
63+
// A self-referencing object: `manager` is a lookup back to the same object, so
64+
// the loader takes the historical sequential `writeRecord` path (`hasSelfRef`).
65+
function createMetadata(): IMetadataService {
66+
const objects: Record<string, any> = {
67+
my_app_employee: {
68+
name: 'my_app_employee',
69+
fields: {
70+
name: { type: 'text' },
71+
manager: { type: 'lookup', reference: 'my_app_employee' },
72+
},
73+
},
74+
};
75+
return {
76+
getObject: vi.fn(async (name: string) => objects[name]),
77+
listObjects: vi.fn(async () => Object.values(objects)),
78+
register: vi.fn(async () => {}),
79+
get: vi.fn(async (_t: string, name: string) => objects[name]),
80+
list: vi.fn(async () => []),
81+
unregister: vi.fn(async () => {}),
82+
exists: vi.fn(async () => false),
83+
listNames: vi.fn(async () => []),
84+
} as unknown as IMetadataService;
85+
}
86+
87+
const CONFIG = {
88+
dryRun: false,
89+
haltOnError: false,
90+
multiPass: true,
91+
defaultMode: 'insert',
92+
batchSize: 1000,
93+
transaction: false,
94+
} as any;
95+
96+
const SEEDS = [
97+
{
98+
object: 'my_app_employee',
99+
externalId: 'name',
100+
mode: 'insert',
101+
env: ['prod', 'dev', 'test'],
102+
records: [{ name: 'Alice' }, { name: 'Bob' }],
103+
},
104+
] as any[];
105+
106+
describe('seed self-ref sequential path — transient retry (framework#3150)', () => {
107+
it('retries a transient insert on the self-referencing path instead of dropping the row', async () => {
108+
const { engine, store } = createFaithfulEngine();
109+
const metadata = createMetadata();
110+
111+
// Bob's first insert hits a network blip, then succeeds on retry.
112+
const realInsert = (engine.insert as any).getMockImplementation();
113+
let bobAttempts = 0;
114+
(engine.insert as any).mockImplementation(async (obj: string, data: any, opts: any) => {
115+
if (obj === 'my_app_employee' && !Array.isArray(data) && data.name === 'Bob') {
116+
bobAttempts++;
117+
if (bobAttempts === 1) throw new Error('fetch failed');
118+
}
119+
return realInsert(obj, data, opts);
120+
});
121+
122+
const result = await new SeedLoaderService(engine, metadata, createLogger()).load({
123+
seeds: SEEDS,
124+
config: CONFIG,
125+
});
126+
127+
expect(bobAttempts).toBe(2); // first threw, retried, succeeded
128+
expect(result.summary.totalErrored).toBe(0);
129+
expect((store.my_app_employee ?? []).map((r) => r.name).sort()).toEqual(['Alice', 'Bob']);
130+
});
131+
});

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

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -690,10 +690,10 @@ export class SeedLoaderService implements ISeedLoaderService {
690690

691691
if (recordId) {
692692
try {
693-
await this.engine.update(deferred.objectName, {
693+
await withTransientRetry(() => this.engine.update(deferred.objectName, {
694694
id: recordId,
695695
[deferred.field]: resolvedId,
696-
}, { context: { isSystem: true } } as any);
696+
}, { context: { isSystem: true } } as any));
697697

698698
// Update result stats
699699
const resultEntry = allResults.find(r => r.object === deferred.objectName);
@@ -764,7 +764,7 @@ export class SeedLoaderService implements ISeedLoaderService {
764764

765765
switch (mode) {
766766
case 'insert': {
767-
const result = await this.engine.insert(objectName, record, opts);
767+
const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts));
768768
return { action: 'inserted', id: this.extractId(result) };
769769
}
770770

@@ -776,7 +776,7 @@ export class SeedLoaderService implements ISeedLoaderService {
776776
if (this.isNoOpReplay(record, existing)) {
777777
return { action: 'skipped', id };
778778
}
779-
await this.engine.update(objectName, { ...record, id }, opts);
779+
await withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts));
780780
return { action: 'updated', id };
781781
}
782782

@@ -786,10 +786,10 @@ export class SeedLoaderService implements ISeedLoaderService {
786786
if (this.isNoOpReplay(record, existing)) {
787787
return { action: 'skipped', id };
788788
}
789-
await this.engine.update(objectName, { ...record, id }, opts);
789+
await withTransientRetry(() => this.engine.update(objectName, { ...record, id }, opts));
790790
return { action: 'updated', id };
791791
} else {
792-
const result = await this.engine.insert(objectName, record, opts);
792+
const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts));
793793
return { action: 'inserted', id: this.extractId(result) };
794794
}
795795
}
@@ -798,18 +798,18 @@ export class SeedLoaderService implements ISeedLoaderService {
798798
if (existing) {
799799
return { action: 'skipped', id: this.extractId(existing) };
800800
}
801-
const result = await this.engine.insert(objectName, record, opts);
801+
const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts));
802802
return { action: 'inserted', id: this.extractId(result) };
803803
}
804804

805805
case 'replace': {
806806
// Replace mode: just insert (caller should have cleared the table)
807-
const result = await this.engine.insert(objectName, record, opts);
807+
const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts));
808808
return { action: 'inserted', id: this.extractId(result) };
809809
}
810810

811811
default: {
812-
const result = await this.engine.insert(objectName, record, opts);
812+
const result = await withTransientRetry(() => this.engine.insert(objectName, record, opts));
813813
return { action: 'inserted', id: this.extractId(result) };
814814
}
815815
}

packages/rest/src/import-runner-bulk.test.ts

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,27 @@ describe('runImport — bulk create batching (framework#2678)', () => {
118118
expect(summary.created).toBe(5);
119119
});
120120

121+
it('retries a transient createData failure on the no-createManyData fallback path (#3150)', async () => {
122+
let attempts = 0;
123+
const createData = vi.fn(async (args: { data: { name: string } }) => {
124+
attempts++;
125+
if (attempts === 1) throw new Error('fetch failed'); // one transient blip, then succeeds
126+
return { id: `id_${args.data.name}` };
127+
});
128+
const p: ImportProtocolLike = {
129+
findData: vi.fn(async () => []),
130+
createData,
131+
updateData: vi.fn(),
132+
// no createManyData → inline per-row fallback path (previously un-retried)
133+
};
134+
135+
const summary = await runImport({ ...baseOpts, p, rows: rowsOf(1) });
136+
137+
expect(createData).toHaveBeenCalledTimes(2); // first throws, retried, succeeds
138+
expect(summary.created).toBe(1);
139+
expect(summary.errors).toBe(0);
140+
});
141+
121142
it('preserves row order in results even with update/skip rows interleaved between buffered creates', async () => {
122143
const createManyData = vi.fn(async (args: { records: any[] }) => ({
123144
records: args.records.map((r) => ({ id: `id_${r.name}`, ...r })),

packages/rest/src/import-runner.ts

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -361,7 +361,11 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
361361
pendingCreates.push({ index: i, rowNo, data });
362362
} else {
363363
// No bulk-create primitive on this protocol: original inline path.
364-
const res2 = await p.createData({ object: objectName, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) });
364+
// Wrap in transient retry to match the update path above (L352)
365+
// and the batched create path (bulkWrite's internal retry) — a
366+
// single `fetch failed` blip must not silently drop the row
367+
// (framework#3150).
368+
const res2 = await withTransientRetry(() => p.createData({ object: objectName, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) }));
365369
const id = extractRecordId(res2);
366370
okCount++; created++;
367371
if (collectUndo && id != null) undoLog.created.push(id);

0 commit comments

Comments
 (0)