fix(objectql,seed,rest): retry roll-up recompute and surface failures (#3147)

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01APnSDJneEDRh3Z51KGCLga

Author: claude

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
+  });
+});

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<T>(fn: () => Promise<T>): Promise<T> {
+    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<string, unknown>,
@@ -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) };
       }
     }

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<string, Map<string, any>>();
+  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<string, number>();
+  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<string, unknown>) {
+      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<string, unknown>) {
+      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<string, unknown>[]) {
+      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);
+  });
+});