Compare commits

..
Author SHA1 Message Date
Garry TanandClaude Fable 5 d98e6b507f test(config): use withEnv() for TTL env mutation to satisfy check-test-isolation R1
CI verify failed: test/postgres-engine-config-cache.test.ts mutated
process.env.GBRAIN_CONFIG_CACHE_TTL_MS directly in beforeEach/afterEach.
Wrap each test body in withEnv() (test/helpers/with-env.ts) instead.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-22 10:58:53 -07:00
d4bbf6eeae perf(config): batch + cache engine.getConfig to kill ~85 round-trips per query
Takeover of #1694: a single search fires ~85 serial getConfig() reads
(loadConfigWithEngine x2 plus the mode/cache/intent/rerank/graph-signals
resolvers); on a remote pooler each read is a round-trip, dominating query
latency and risking cli.ts's 10s disconnect force-exit truncating stdout.

The first read now batch-loads the whole config table into a process-
lifetime Map (single-flight under concurrency); setConfig/unsetConfig
write through; a 30s TTL bounds multi-writer staleness and
GBRAIN_CONFIG_CACHE_TTL_MS=0 restores per-key reads.

Rebased onto current master: unlike the original diff, both the batch
load and the TTL=0 per-key fallback stay inside connRetry() so the
#1603/#1891 retry+reconnect posture (pooler-drop self-heal) is preserved.
PGLite stays uncached (in-process, zero round-trips; raw-SQL test
fixtures rely on fresh reads). E2E helpers pin the cache off since those
suites seed config via raw SQL.

Co-authored-by: Omerbahari <Omerbahari@users.noreply.github.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-21 14:26:28 -07:00
10 changed files with 197 additions and 125 deletions
-12
View File
@@ -186,18 +186,6 @@ export function isSourceStale(src: SourceRow, now = Date.now(), floorMin = FULL_
return ageMin >= floorMin;
}
/**
* #2060: count sources past the per-source cycle freshness floor. Consumed
* by autopilot's dispatch decision — a stale source forces the fanout path
* even when the doctor plan is small (score 7094, plan ≤ 3, est < 300s),
* so targeted mode can't leave cycle_freshness stale indefinitely.
* dispatchPerSource's own throttles (skipped_fresh / fanoutMax / failure
* cooldown) bound the resulting work.
*/
export function countStaleSources(sources: SourceRow[], now = Date.now(), floorMin = FULL_CYCLE_FLOOR_MIN): number {
return sources.filter((s) => isSourceStale(s, now, floorMin)).length;
}
/**
* Most recent SUCCESSFUL cycle for a source. Prefers `last_source_cycle_at`
* (per-source phases, written by the split cycle) and falls back to the legacy
+2 -16
View File
@@ -901,27 +901,13 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
const FULL_CYCLE_FLOOR_MIN = 60;
const minutesSinceLastFull = (Date.now() - lastFullCycleAt) / 60000;
// #2060: stale per-source cycle freshness is a dispatch input. Without
// it, a brain sitting at score 7094 with a small targeted plan (≤3
// steps, <300s) stays in targeted mode indefinitely and no per-source
// cycle is ever dispatched — cycle_freshness never advances. A stale
// source forces the fanout path; dispatchPerSource's throttles
// (skipped_fresh / fanoutMax / failure cooldown) bound the work.
// Fail-open to 0: a read failure must not block dispatch.
let staleCycleSources = 0;
try {
const { countStaleSources } = await import('./autopilot-fanout.ts');
staleCycleSources = countStaleSources(await engine.listAllSources({ localPathOnly: true }));
} catch { /* fail-open: freshness is a dispatch hint, not a gate */ }
const shouldFullCycle =
(score >= 95 && plan.length === 0 && minutesSinceLastFull >= FULL_CYCLE_FLOOR_MIN) ||
plan.length > 3 ||
estTotal >= 300 ||
score < 70 ||
staleCycleSources > 0;
score < 70;
const shouldSleep = score >= 95 && plan.length === 0 && minutesSinceLastFull < FULL_CYCLE_FLOOR_MIN && staleCycleSources === 0;
const shouldSleep = score >= 95 && plan.length === 0 && minutesSinceLastFull < FULL_CYCLE_FLOOR_MIN;
if (shouldSleep) {
if (jsonMode) {
+10 -19
View File
@@ -854,10 +854,7 @@ interface SyncPhaseResult extends PhaseResult {
/**
* Resolve the source id for a brain directory by looking up the sources
* table. Returns undefined when no registered source matches (falls back
* to pre-v0.18 global config.sync.* keys) OR when MORE than one source
* claims the path — an ambiguous match must not scope phases or stamp
* last_full_cycle_at for an arbitrarily-picked source (the "freshness
* stamp that lies" this resolution exists to prevent).
* to pre-v0.18 global config.sync.* keys).
*/
async function resolveSourceForDir(
engine: BrainEngine,
@@ -868,10 +865,10 @@ async function resolveSourceForDir(
if (brainDir === null) return undefined;
try {
const rows = await engine.executeRaw<{ id: string }>(
`SELECT id FROM sources WHERE local_path = $1 LIMIT 2`,
`SELECT id FROM sources WHERE local_path = $1 LIMIT 1`,
[brainDir],
);
return rows.length === 1 ? rows[0]!.id : undefined;
return rows[0]?.id;
} catch {
// sources table might not exist on very old brains — fall through.
return undefined;
@@ -2368,23 +2365,17 @@ export async function runCycle(
}
// v0.38 (codex r1 P0-5): persist per-source cycle completion timestamp
// when the cycle ran successfully against a resolvable source. Read by
// autopilot's per-source freshness gate next tick.
//
// #1993: keyed off `cycleSourceId` (opts.sourceId ?? the source resolved
// from brainDir) — the SAME id the cycle locked + scoped its phases to —
// NOT raw opts.sourceId. The autopilot's inline cycle sets brainDir but
// passes no explicit sourceId, so keying off opts.sourceId alone never
// advanced last_full_cycle_at and cycle_freshness stayed stale even while
// the autopilot cycled every interval. Skipped when:
// - no source resolves (engine null, or no checkout AND no opts.sourceId)
// when the cycle ran successfully against an explicit source. Read by
// autopilot's per-source freshness gate next tick. Skipped when:
// - opts.sourceId is unset (legacy callers — autopilot still here)
// - engine is null (no-DB path)
// - status is 'failed' or 'skipped' (don't mark a non-run as fresh)
// - dryRun (writes are out of scope)
//
// Best-effort: a write failure does NOT change the CycleReport status.
// The cost of writing the wrong timestamp post-failure is higher than
// the cost of missing a successful write (next cycle will redo work).
if (cycleSourceId && engine && !dryRun && !aborted && (status === 'ok' || status === 'clean' || status === 'partial')) {
if (opts.sourceId && engine && !dryRun && !aborted && (status === 'ok' || status === 'clean' || status === 'partial')) {
try {
const nowIso = new Date().toISOString();
// #2194 fix #3 (the cycle split): `last_source_cycle_at` is the NEW gate
@@ -2394,13 +2385,13 @@ export async function runCycle(
// phases (those gate on autopilot.last_global_at), so writing it on a
// source-only cycle does not re-introduce the freshness poisoning codex
// flagged in the rejected skip-based design.
await engine.updateSourceConfig(cycleSourceId, {
await engine.updateSourceConfig(opts.sourceId, {
last_source_cycle_at: nowIso,
last_full_cycle_at: nowIso,
});
} catch (e) {
// Best-effort; cycle already succeeded by the time we get here.
console.warn(`[cycle] failed to write last_source_cycle_at for source ${cycleSourceId}: ${e instanceof Error ? e.message : String(e)}`);
console.warn(`[cycle] failed to write last_source_cycle_at for source ${opts.sourceId}: ${e instanceof Error ? e.message : String(e)}`);
}
}
+55 -7
View File
@@ -5565,30 +5565,78 @@ export class PostgresEngine implements BrainEngine {
});
}
/**
* perf (#1694 by @Omerbahari): process-lifetime config cache. A single
* search fires ~85 getConfig() reads (loadConfigWithEngine x2, plus
* mode/cache/intent/rerank/graph-signals resolvers). On a remote pooler
* each read is a round-trip; serial they dominate query latency and can
* push the op handler past cli.ts's 10s disconnect force-exit, truncating
* stdout. First read batch-loads the whole `config` table into this Map
* (inside the same connRetry posture as the per-key read #1603/#1891);
* setConfig/unsetConfig write through. TTL bounds staleness for
* multi-writer processes; GBRAIN_CONFIG_CACHE_TTL_MS=0 disables.
* Only present keys are stored Map.has() distinguishes known-absent.
*/
private _configCache: Map<string, string> | null = null;
private _configCacheLoadedAt = 0;
private _configCacheLoad: Promise<void> | null = null;
private get _configCacheTtlMs(): number {
const raw = process.env.GBRAIN_CONFIG_CACHE_TTL_MS;
if (raw !== undefined) {
const n = parseInt(raw, 10);
if (Number.isFinite(n) && n >= 0) return n;
}
return 30_000;
}
async getConfig(key: string): Promise<string | null> {
// #1603: a transient pooler drop on this read used to throw / fall through
// to defaults silently — which on remote Postgres surfaces as the wrong
// search mode/knobs and empty-stdout queries.
return this.connRetry(async () => {
const rows = await this.sql`SELECT value FROM config WHERE key = ${key}`;
return rows.length > 0 ? (rows[0].value as string) : null;
});
// search mode/knobs and empty-stdout queries. Both the batch load and the
// cache-off per-key read keep the connRetry reconnect posture.
const ttl = this._configCacheTtlMs;
if (ttl === 0) {
return this.connRetry(async () => {
const rows = await this.sql`SELECT value FROM config WHERE key = ${key}`;
return rows.length > 0 ? (rows[0].value as string) : null;
});
}
if (this._configCache === null || Date.now() - this._configCacheLoadedAt >= ttl) {
// Single-flight: concurrent cold reads share one batch load.
this._configCacheLoad ??= this.connRetry(async () => {
const rows = await this.sql`SELECT key, value FROM config` as unknown as
Array<{ key: string; value: string | null }>;
const map = new Map<string, string>();
for (const r of rows) if (r.value != null) map.set(r.key, r.value);
this._configCache = map;
this._configCacheLoadedAt = Date.now();
}).finally(() => {
this._configCacheLoad = null;
});
await this._configCacheLoad;
}
return this._configCache!.has(key) ? this._configCache!.get(key)! : null;
}
async setConfig(key: string, value: string): Promise<void> {
return this.connRetry(async () => {
await this.connRetry(async () => {
await this.sql`
INSERT INTO config (key, value) VALUES (${key}, ${value})
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value
`;
});
// Write-through so a long-lived process never serves stale config.
this._configCache?.set(key, value);
}
async unsetConfig(key: string): Promise<number> {
return this.connRetry(async () => {
const count = await this.connRetry(async () => {
const result = await this.sql`DELETE FROM config WHERE key = ${key}` as unknown as { count: number };
return result.count ?? 0;
});
// Write-through: known-absent, so the cache doesn't serve a stale value.
this._configCache?.delete(key);
return count;
}
async listConfigKeys(prefix: string): Promise<string[]> {
-14
View File
@@ -54,20 +54,6 @@ describe('autopilot.ts ↔ dispatchPerSource wiring', () => {
expect(AUTOPILOT_SRC).toMatch(/lastFullCycleAt\s*=\s*Date\.now\(\)/);
});
test('stale per-source cycle freshness is a shouldFullCycle input (#2060)', () => {
// Targeted mode (score 7094, plan ≤3, est <300s) must not be able to
// starve per-source cycle dispatch: a stale source (per countStaleSources
// over listAllSources) forces the fanout path, and the sleep gate must
// not fire while stale sources exist. Without these terms, cycle
// freshness never advances for a brain that always lands in targeted mode.
expect(AUTOPILOT_SRC).toMatch(/countStaleSources/);
const fullCycleDeclIdx = AUTOPILOT_SRC.indexOf('const shouldFullCycle');
expect(fullCycleDeclIdx).toBeGreaterThan(-1);
const decl = AUTOPILOT_SRC.slice(fullCycleDeclIdx, fullCycleDeclIdx + 700);
expect(decl).toMatch(/staleCycleSources\s*>\s*0/);
expect(decl).toMatch(/const shouldSleep[^;]*staleCycleSources\s*===\s*0/);
});
test('does NOT regress to the single-job dispatch on the full-cycle path', () => {
// Pre-PR: the shouldFullCycle branch did:
// const job = await queue.add('autopilot-cycle', { repoPath }, {
-18
View File
@@ -14,7 +14,6 @@ import { describe, test, expect } from 'bun:test';
import {
readLastFullCycleAt,
isSourceStale,
countStaleSources,
selectSourcesForDispatch,
resolveFanoutMax,
dispatchPerSource,
@@ -75,23 +74,6 @@ describe('isSourceStale', () => {
});
});
describe('countStaleSources (#2060 dispatch-decision input)', () => {
const NOW = Date.parse('2026-05-22T12:00:00.000Z');
test('counts never-cycled + past-floor sources, ignores fresh', () => {
const sources = [
src('never-cycled'), // stale (null)
src('old', new Date(NOW - 2 * 60 * 60_000).toISOString()), // stale (2h)
src('fresh', new Date(NOW - 30 * 60_000).toISOString()), // fresh (30min)
];
expect(countStaleSources(sources, NOW)).toBe(2);
});
test('returns 0 for all-fresh and for empty list', () => {
const fresh = src('a', new Date(NOW - 10 * 60_000).toISOString());
expect(countStaleSources([fresh], NOW)).toBe(0);
expect(countStaleSources([], NOW)).toBe(0);
});
});
describe('selectSourcesForDispatch', () => {
const NOW = Date.parse('2026-05-22T12:00:00.000Z');
const fresh = (id: string, agoMin: number) =>
+8 -38
View File
@@ -3,10 +3,8 @@
* cycles. Closes codex round-1 P0-5 (write site for last_full_cycle_at
* was unspecified pre-PR).
*
* Conditions for write (keyed off `cycleSourceId` = opts.sourceId ?? the
* source resolved from brainDir, so the autopilot's inline cycle — brainDir
* set, no explicit sourceId — also advances the timestamp, #1993):
* - a source resolves (explicit sourceId, or brainDir matches a source)
* Conditions for write:
* - opts.sourceId is set (legacy callers without sourceId skip the write)
* - engine is non-null (no-DB path skips)
* - status is 'ok' | 'clean' | 'partial' (failed/skipped don't mark fresh)
* - dryRun is false
@@ -92,45 +90,17 @@ describe('runCycle last_full_cycle_at exit hook', () => {
});
});
test('no explicit sourceId but brainDir resolves a source → writes the resolved source timestamp', async () => {
test('legacy caller (no sourceId) does NOT write any source timestamp', async () => {
await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
// The autopilot's inline cycle sets brainDir but passes no sourceId.
// runCycle resolves the source from brainDir (local_path match) into
// cycleSourceId and stamps last_full_cycle_at for it — otherwise
// cycle_freshness reports the brain stale even while the autopilot
// cycles every interval (#1993).
await seedSource('resolved-from-dir'); // local_path = brainDir
expect(await readLastFullCycleAt('resolved-from-dir')).toBeNull();
const t0 = Date.now();
const report = await runCycle(engine, {
brainDir,
phases: ['lint'],
});
expect(['ok', 'clean']).toContain(report.status);
const after = await readLastFullCycleAt('resolved-from-dir');
expect(after).not.toBeNull();
expect(new Date(after!).getTime()).toBeGreaterThanOrEqual(t0);
});
});
test('no sourceId and brainDir matches no source → does not write', async () => {
await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
// A source exists but its local_path does NOT match brainDir, so
// resolveSourceForDir returns undefined, cycleSourceId is undefined,
// and no per-source timestamp is written.
await engine.executeRaw(
`INSERT INTO sources (id, name, local_path, config, archived, created_at)
VALUES ('unmatched', 'unmatched', '/no/such/repo', '{}'::jsonb, false, NOW())
ON CONFLICT (id) DO UPDATE SET local_path = EXCLUDED.local_path`,
[],
);
await seedSource('default-like');
// No sourceId passed; should remain untouched.
await runCycle(engine, {
brainDir,
phases: ['lint'],
});
expect(await readLastFullCycleAt('unmatched')).toBeNull();
// No per-source write happens; default source's config stays empty.
const after = await readLastFullCycleAt('default-like');
expect(after).toBeNull();
});
});
+7
View File
@@ -29,6 +29,13 @@ if (existsSync(envPath)) {
}
}
// E2E suites seed/rewrite the config table via raw SQL and expect engine
// reads to see it immediately; disable the process-lifetime config cache
// (#1694) so read semantics match pre-cache behavior. Spawned CLI
// subprocesses inherit this. Cache semantics are pinned by
// test/postgres-engine-config-cache.test.ts.
process.env.GBRAIN_CONFIG_CACHE_TTL_MS ??= '0';
const DATABASE_URL = process.env.DATABASE_URL;
const FIXTURES_DIR = resolve(import.meta.dir, 'fixtures');
+112
View File
@@ -0,0 +1,112 @@
/**
* Process-lifetime config cache (#1694 takeover, by @Omerbahari).
*
* A single search fires ~85 getConfig() reads; on a remote pooler each is a
* round-trip. The first read now batch-loads the whole `config` table into a
* Map; setConfig/unsetConfig write through; TTL bounds multi-writer
* staleness; GBRAIN_CONFIG_CACHE_TTL_MS=0 restores per-key reads.
*
* Pure: stubs `_sql` with a call-counting fake; no real DB.
*/
import { describe, it, expect } from 'bun:test';
import { PostgresEngine } from '../src/core/postgres-engine.ts';
import { withEnv } from './helpers/with-env.ts';
const FAST_RETRY = { maxRetries: 3, delayMs: 1, delayMaxMs: 1, jitter: 'none' as const };
/** Engine whose `sql` records every query's template strings and returns `rows`. */
function makeEngine(rows: unknown[]) {
const e = new PostgresEngine();
const calls: string[] = [];
(e as unknown as { _connectionStyle: string })._connectionStyle = 'instance';
(e as unknown as { _bulkRetryOptsCache: unknown })._bulkRetryOptsCache = FAST_RETRY;
(e as unknown as { _sql: unknown })._sql = (strings: TemplateStringsArray) => {
calls.push(strings.join('?'));
return Promise.resolve(rows);
};
return { engine: e, calls };
}
/** Run `fn` with GBRAIN_CONFIG_CACHE_TTL_MS set (or cleared when undefined). */
const withTtl = (ttl: string | undefined, fn: () => Promise<void>) =>
withEnv({ GBRAIN_CONFIG_CACHE_TTL_MS: ttl }, fn);
describe('PostgresEngine config cache (#1694)', () => {
it('batch-loads once and serves repeat reads from the cache', () => withTtl(undefined, async () => {
const { engine, calls } = makeEngine([
{ key: 'search.mode', value: 'balanced' },
{ key: 'embedding_multimodal', value: 'true' },
]);
expect(await engine.getConfig('search.mode')).toBe('balanced');
expect(await engine.getConfig('embedding_multimodal')).toBe('true');
expect(await engine.getConfig('search.mode')).toBe('balanced');
// One SELECT total — this is the whole point of the fix.
expect(calls.length).toBe(1);
expect(calls[0]).toContain('SELECT key, value FROM config');
}));
it('returns null for a known-absent key without an extra round-trip', () => withTtl(undefined, async () => {
const { engine, calls } = makeEngine([{ key: 'a', value: '1' }]);
expect(await engine.getConfig('missing.key')).toBeNull();
expect(await engine.getConfig('missing.key')).toBeNull();
expect(calls.length).toBe(1);
}));
it('setConfig writes through so subsequent reads see the new value', () => withTtl(undefined, async () => {
const { engine, calls } = makeEngine([{ key: 'k', value: 'old' }]);
expect(await engine.getConfig('k')).toBe('old');
await engine.setConfig('k', 'new');
expect(await engine.getConfig('k')).toBe('new');
expect(calls.length).toBe(2); // batch load + upsert; no re-read
}));
it('unsetConfig writes through so subsequent reads see absence', () => withTtl(undefined, async () => {
const { engine } = makeEngine([{ key: 'k', value: 'v' }]);
expect(await engine.getConfig('k')).toBe('v');
await engine.unsetConfig('k');
expect(await engine.getConfig('k')).toBeNull();
}));
it('concurrent cold reads share a single batch load (single-flight)', () => withTtl(undefined, async () => {
const { engine, calls } = makeEngine([{ key: 'k', value: 'v' }]);
const [a, b, c] = await Promise.all([
engine.getConfig('k'),
engine.getConfig('k'),
engine.getConfig('other'),
]);
expect([a, b, c]).toEqual(['v', 'v', null]);
expect(calls.length).toBe(1);
}));
it('GBRAIN_CONFIG_CACHE_TTL_MS=0 disables the cache (per-key reads)', () => withTtl('0', async () => {
const { engine, calls } = makeEngine([{ value: 'v' }]);
expect(await engine.getConfig('k')).toBe('v');
expect(await engine.getConfig('k')).toBe('v');
expect(calls.length).toBe(2);
expect(calls[0]).toContain('SELECT value FROM config WHERE key =');
}));
it('an expired TTL reloads from the database', () => withTtl('1', async () => {
const { engine, calls } = makeEngine([{ key: 'k', value: 'v' }]);
expect(await engine.getConfig('k')).toBe('v');
await new Promise((r) => setTimeout(r, 5));
expect(await engine.getConfig('k')).toBe('v');
expect(calls.length).toBe(2); // two batch loads
}));
it('the batch load keeps the connRetry reconnect posture (#1603/#1891)', () => withTtl(undefined, async () => {
const e = new PostgresEngine();
(e as unknown as { _connectionStyle: string })._connectionStyle = 'instance';
(e as unknown as { _sql: unknown })._sql = null; // torn-down pool → retryable
(e as unknown as { _bulkRetryOptsCache: unknown })._bulkRetryOptsCache = FAST_RETRY;
let reconnects = 0;
(e as unknown as { reconnect: () => Promise<void> }).reconnect = async () => {
reconnects++;
(e as unknown as { _sql: unknown })._sql = () =>
Promise.resolve([{ key: 'k', value: 'v' }]);
};
expect(await e.getConfig('k')).toBe('v');
expect(reconnects).toBe(1);
}));
});
@@ -43,7 +43,9 @@ function makeTornDownEngine(poolResult: unknown): { engine: PostgresEngine; reco
describe('PostgresEngine non-batch config accessors self-heal (PR #1891 takeover)', () => {
it('getConfig reconnects + retries a null instance pool, then returns the value', async () => {
const { engine, reconnects } = makeTornDownEngine([{ value: 'live-value' }]);
// Rows carry `key` too: getConfig's default cached path batch-loads
// `SELECT key, value FROM config` (#1694) through the same connRetry.
const { engine, reconnects } = makeTornDownEngine([{ key: 'some.key', value: 'live-value' }]);
expect(await engine.getConfig('some.key')).toBe('live-value');
expect(reconnects()).toBe(1); // exactly one reconnect closed the gap
});