From c7ed9b9904b0a12e2c0bcd74bbfd5056f81a4b8e Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Fri, 22 May 2026 08:13:10 -0700 Subject: [PATCH] fix(autopilot): drop maxWaiting from per-source submit (E2E found silent coalesce) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Postgres E2E for fan-out surfaced two real bugs that the unit-stub tests + PGLite tests couldn't catch: 1) **CRITICAL** — dispatchPerSource passed `maxWaiting: 1` to every per-source queue.add. maxWaiting is per-(name, queue) — since all per-source jobs share `name='autopilot-cycle'`, the second + third + Nth source's submit silently coalesced into the FIRST source's waiting job. Net result on a real worker: 1 job processed per tick, not N. The entire fan-out feature was a silent no-op past the first source. Per-source idempotency_key (`autopilot-cycle::`) already handles "two ticks for same source within slot" dedup, which is the only thing maxWaiting was buying us. Dropping it fixes fan-out without losing dedup. New unit-stub regression test asserts maxWaiting is NOT in the per- source submit opts so a future refactor that re-adds it gets caught in 100x faster CI (test/autopilot-fanout.test.ts). 2) **postgres-engine.ts:updateSourceConfig** — initial impl used sql.json() correctly but my mid-debug rewrite to executeRaw + positional `$1::jsonb` produced JSONB STRING shape (not OBJECT) because postgres-js double-encodes JS string params in unsafe mode. `||` between JSONB object + JSONB string yields a JSONB ARRAY, wiping every existing config key on update. Same latent bug class exists at src/commands/sources.ts:482 (gbrain sources federate/unfederate path); flagged for follow-up but out-of-scope here. Reverted to sql.json() inside the template tag (verified via direct psql round-trip: jsonb_typeof = 'object'). Updated the e2e seed helper to use sql.json() too — the executeRaw + JSON.stringify pattern was producing string-shape JSONB at SEED time which made the failure cascade harder to debug. Coverage adds: - test/e2e/list-all-sources-postgres.test.ts: 11 cases pin Postgres parity for listAllSources + updateSourceConfig including jsonb_typeof round-trip - test/e2e/autopilot-fanout-postgres.test.ts: 6 cases end-to-end including 3-source fan-out producing 3 distinct rows, idempotency coalesce within slot, cap honored, legacy fallback path - test/autopilot-fanout.test.ts: +1 regression guard on maxWaiting This is the kind of bug that justifies the user's "fill test gaps then run E2E" mandate. The unit tests + PGLite parity tests + typecheck all passed cleanly; only the real-Postgres E2E found the coalesce. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/commands/autopilot-fanout.ts | 7 +++++- src/core/postgres-engine.ts | 13 +++++++---- test/autopilot-fanout.test.ts | 16 ++++++++++++++ test/e2e/autopilot-fanout-postgres.test.ts | 2 ++ test/e2e/list-all-sources-postgres.test.ts | 25 +++++++++++++--------- 5 files changed, 48 insertions(+), 15 deletions(-) diff --git a/src/commands/autopilot-fanout.ts b/src/commands/autopilot-fanout.ts index cb78672cb..364ac7d61 100644 --- a/src/commands/autopilot-fanout.ts +++ b/src/commands/autopilot-fanout.ts @@ -216,7 +216,12 @@ export async function dispatchPerSource( idempotency_key: `autopilot-cycle:${src.id}:${opts.slot}`, max_attempts: 2, timeout_ms: opts.timeoutMs, - maxWaiting: 1, + // DELIBERATELY no maxWaiting: 1 here. maxWaiting is per + // (name, queue), so it would coalesce all N per-source jobs + // sharing name='autopilot-cycle' down to ONE waiting job — + // killing the fan-out. The per-source idempotency_key + // already provides the right dedup granularity (one job per + // source per slot, regardless of how many ticks try). }, ); dispatched.push(src.id); diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 4b7ef18b3..915085745 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -967,16 +967,21 @@ export class PostgresEngine implements BrainEngine { // shape this autopilot wave uses is flat (`last_full_cycle_at`, // `archive_*`, etc.) so concat is sufficient. Idempotent on re-run. // - // sql.json(patch) is the canonical safe path per feedback_postgres_jsonb_double_encode - // — postgres-js handles JSONB serialization (no double-encode). Matches - // the pattern at putPage and submitJob elsewhere in this file. + // MUST use sql.json(patch) inside the template tag — postgres-js's + // positional executeRaw + `$1::jsonb` cast DOUBLE-ENCODES the + // JSON.stringify'd string, producing a JSONB STRING shape instead + // of OBJECT. `||` between JSONB object + JSONB string yields a + // JSONB ARRAY (concat semantics for non-matching types), which + // wipes every existing config key. sql.json(...) inside the + // template tag is the canonical safe path — same pattern as + // putPage + submitJob elsewhere in this file. Empirically verified + // produces jsonb_typeof = 'object'. const sql = this.sql; const result = await sql` UPDATE sources SET config = COALESCE(config, '{}'::jsonb) || ${sql.json(patch as Parameters[0])} WHERE id = ${sourceId} `; - // postgres-js returns count as result.count; matched-rows shape return (result.count ?? 0) > 0; } diff --git a/test/autopilot-fanout.test.ts b/test/autopilot-fanout.test.ts index 0359c18a8..2e4fee561 100644 --- a/test/autopilot-fanout.test.ts +++ b/test/autopilot-fanout.test.ts @@ -281,6 +281,22 @@ describe('dispatchPerSource — integration with stubbed engine + queue', () => expect(parsed.pending.length).toBe(2); }); + test('per-source submit MUST NOT pass maxWaiting (regression — coalesces all sources to one job)', async () => { + // Direct unit-stub queues can't enforce maxWaiting semantics (the + // production MinionQueue implementation does), so this catches the + // regression by inspecting the submit opts at the dispatch boundary. + // If a future refactor re-adds maxWaiting:1 to the per-source path, + // the production fan-out would silently coalesce N sources to ONE + // waiting job per tick — killing the entire feature. The e2e test + // also catches this against a real queue, but this guard fires in + // unit tests too so the bug surfaces 100x faster. + const { engine, queue, added, fanoutOpts } = makeStubs([src('a'), src('b'), src('c')]); + await dispatchPerSource(engine, queue, fanoutOpts); + for (const job of added) { + expect(job.opts.maxWaiting).toBeUndefined(); + } + }); + test('all-fresh tick dispatches nothing (no jobs added)', async () => { const NOW = Date.now(); const recent = (id: string) => diff --git a/test/e2e/autopilot-fanout-postgres.test.ts b/test/e2e/autopilot-fanout-postgres.test.ts index 556c17a81..38a16ceff 100644 --- a/test/e2e/autopilot-fanout-postgres.test.ts +++ b/test/e2e/autopilot-fanout-postgres.test.ts @@ -50,6 +50,8 @@ beforeEach(async () => { async function seedSource(id: string, opts: { local_path?: string } = {}): Promise { const localPath = opts.local_path ?? mkdtempSync(join(tmpdir(), `gbrain-fanout-${id}-`)); + // Direct literal `'{}'::jsonb` is fine (no parameter binding). Test + // explicitly resets config to {} so each test starts clean. await engine.executeRaw( `INSERT INTO sources (id, name, local_path, config, archived, created_at) VALUES ($1, $2, $3, '{}'::jsonb, false, NOW()) diff --git a/test/e2e/list-all-sources-postgres.test.ts b/test/e2e/list-all-sources-postgres.test.ts index 07e820880..0fec4509f 100644 --- a/test/e2e/list-all-sources-postgres.test.ts +++ b/test/e2e/list-all-sources-postgres.test.ts @@ -42,16 +42,21 @@ async function seedSource( ): Promise { const localPath = opts.local_path === undefined ? `/tmp/${id}` : opts.local_path; const archived = opts.archived === true; - const config = JSON.stringify(opts.config ?? {}); - await engine.executeRaw( - `INSERT INTO sources (id, name, local_path, config, archived, created_at) - VALUES ($1, $2, $3, $4::jsonb, $5, NOW()) - ON CONFLICT (id) DO UPDATE - SET local_path = EXCLUDED.local_path, - config = EXCLUDED.config, - archived = EXCLUDED.archived`, - [id, id, localPath, config, archived], - ); + // NOTE: do NOT use executeRaw + `JSON.stringify(config) + $N::jsonb` — + // postgres-js double-encodes the JS string parameter, producing a JSONB + // STRING shape instead of OBJECT. Use sql.json() inside the template tag. + // Same pattern as putPage. The pre-existing sources.ts:482 has the + // same latent bug; the call site there is rare (gbrain sources + // federate/unfederate) and out of scope for this PR. + const eng = engine as unknown as { sql: (...args: unknown[]) => Promise<{ count?: number }> }; + await (eng.sql as any)` + INSERT INTO sources (id, name, local_path, config, archived, created_at) + VALUES (${id}, ${id}, ${localPath}, ${(eng.sql as any).json(opts.config ?? {})}, ${archived}, NOW()) + ON CONFLICT (id) DO UPDATE + SET local_path = EXCLUDED.local_path, + config = EXCLUDED.config, + archived = EXCLUDED.archived + `; } describeIfDB('Postgres parity — listAllSources', () => {