Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

### Fixes

- MCP sessions release excess query workers after a minute idle, reducing retained memory after concurrent requests while keeping one worker ready.
- In TypeScript, an awaited call with type arguments like `await fetchApi<User>("/me")` or `await api.get<T>(url)` now links to the function it calls. Before, the call was dropped, so typed fetch wrappers looked like they had no callers. (#2586)
- In JavaScript and TypeScript, a call or `new` in a parameter's default value, like `function renderWithNote(note = buildNote())`, a destructured `{ size = defaultSize() }` or `constructor(private svc = new Service())`, now counts as the function's own call. Before, it was dropped, so `buildNote` was missing that caller and impact on it came back incomplete. Re-index JavaScript and TypeScript projects after upgrading. Thanks @inth3shadows for the report and the pull request. (#2588)
- In React, a component rendered at module scope, in a top-level constant or inside a test's `it(…)` callback, like `createRoot(el).render(<App />)` or `render(<Card />)`, now counts as rendered there, so an app's entry reaches its root component and a component's tests show among its callers. (#2587)
Expand Down
92 changes: 92 additions & 0 deletions __tests__/query-pool.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -350,6 +350,98 @@ describe('QueryPool', () => {

});

describe('QueryPool idle retirement', () => {
beforeEach(() => vi.useFakeTimers());
afterEach(() => vi.useRealTimers());

it('releases excess idle workers and repeatedly regrows without spending the crash budget', async () => {
const workers: FakeWorker[] = [];
const pool = new QueryPool({ root: '/x', size: 3, createWorker: () => {
const w = new FakeWorker(() => ({ result: ok('served') }));
w.terminate = async () => { w.alive = false; w.emitExit(); return 1; };
workers.push(w);
return w;
} });
try {
for (let wave = 0; wave < 7; wave++) {
const calls = Array.from({ length: 3 }, () => pool.run('codegraph_explore', {}));
await vi.advanceTimersByTimeAsync(10);
expect(await Promise.all(calls)).toEqual(Array.from({ length: 3 }, () => ok('served')));
expect(pool.liveWorkers).toBe(3);
await vi.advanceTimersByTimeAsync(60_000);
expect(pool.liveWorkers).toBe(1);
expect(workers.filter(w => w.alive)).toHaveLength(1);
expect(pool.healthy).toBe(true);
expect(pool.ready).toBe(true);
}
const call = pool.run('codegraph_explore', {});
await vi.advanceTimersByTimeAsync(10);
expect(await call).toEqual(ok('served'));
} finally { await pool.destroy(); }
expect(workers.every(w => !w.alive)).toBe(true);
expect(vi.getTimerCount()).toBe(0);
});

it('never retires active or starting workers', async () => {
const workers: FakeWorker[] = [];
let release!: (result: ToolResult) => void;
const gate = new Promise<ToolResult>(resolve => { release = resolve; });
const pool = new QueryPool({ root: '/x', size: 3, softTimeoutMs: 120_000, createWorker: () => {
const w = new FakeWorker(() => ({ wait: gate }), null);
workers.push(w);
return w;
} });
try {
workers[0].emitMessage({ type: 'ready', ok: true });
const calls = Array.from({ length: 3 }, () => pool.run('codegraph_explore', {}));
workers[1].emitMessage({ type: 'ready', ok: true });
await vi.advanceTimersByTimeAsync(60_010);
expect(pool.liveWorkers).toBe(3);
expect(workers.every(w => w.alive)).toBe(true);
workers[2].emitMessage({ type: 'ready', ok: true });
release(ok('finished'));
expect(await Promise.all(calls)).toEqual(Array.from({ length: 3 }, () => ok('finished')));
await vi.advanceTimersByTimeAsync(60_000);
expect(pool.liveWorkers).toBe(1);
} finally { await pool.destroy(); }
});

it('destroy waits for idle retirements already in progress', async () => {
const finishes: Array<() => void> = [];
const pool = new QueryPool({ root: '/x', size: 3, createWorker: () => {
const w = new FakeWorker(() => ({ result: ok('served') }));
w.terminate = () => new Promise<number>(resolve => finishes.push(() => {
w.alive = false;
w.emitExit();
resolve(1);
}));
return w;
} });
try {
const calls = Array.from({ length: 3 }, () => pool.run('codegraph_explore', {}));
await vi.advanceTimersByTimeAsync(10);
await Promise.all(calls);
await vi.advanceTimersByTimeAsync(60_000);
expect(finishes).toHaveLength(2);
let destroyed = false;
const down = pool.destroy().then(() => { destroyed = true; });
expect(finishes).toHaveLength(3);
finishes[2](); // The remaining warm worker exits before the retirees.
await vi.advanceTimersByTimeAsync(0);
expect(destroyed).toBe(false);
finishes[0]();
finishes[1]();
await down;
expect(destroyed).toBe(true);
expect(vi.getTimerCount()).toBe(0);
} finally {
const down = pool.destroy();
finishes.forEach(finish => finish());
await down;
}
});
});

// Use the built engine so its pool loads the real compiled worker sibling.
const BuiltEngine: typeof MCPEngine = require('../dist/mcp/engine').MCPEngine;

Expand Down
36 changes: 33 additions & 3 deletions src/mcp/query-pool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
* - lazy growth: one warm worker on construct, grows to `size` on demand, so a
* single-agent session pays for one connection and a 10-subagent burst grows
* to the core budget.
* - idle shrink: excess workers release their connections and heaps after a
* minute idle; one warm worker remains for the next call.
* - crash recovery: a dead worker is respawned and its in-flight call retried
* once; a poison call that keeps crashing fails gracefully (never wedges the
* pool). A crash budget trips a circuit breaker (`healthy` → false) so the
Expand Down Expand Up @@ -51,6 +53,8 @@ const DEFAULT_BUSY_TIMEOUT_MS = 45_000; // < the ~60s MCP client request timeout
/** Hard ceiling on pool size regardless of core count / env. */
const MAX_POOL_SIZE = 16;

const IDLE_WORKER_TIMEOUT_MS = 60_000;

/**
* Total worker deaths before the pool declares itself unhealthy and the caller
* reverts to in-process dispatch. High enough to ride out a few transient
Expand Down Expand Up @@ -157,6 +161,8 @@ export class QueryPool {
private queue: Job[] = [];
private inflight = new Map<PoolWorker, Job>();
private workers = new Set<PoolWorker>();
private idleTimers = new Map<PoolWorker, NodeJS.Timeout>();
private retirements = new Set<Promise<void>>();
// Workers spawned but not yet 'ready'. Growth must count these so a single
// first call (with the eager worker still starting) doesn't spawn the WHOLE
// pool at once — N simultaneous cold worker starts (each a full module load +
Expand Down Expand Up @@ -250,26 +256,47 @@ export class QueryPool {
return;
}
this.everReady = true;
this.idle.push(w);
this.park(w);
this.drain();
return;
}
if (m.type === 'result') {
const job = this.inflight.get(w);
this.inflight.delete(w);
this.idle.push(w);
this.park(w);
if (job) this.settle(job, m.result ?? busyGuidance(0));
this.drain();
}
}

private park(w: PoolWorker): void {
this.idle.push(w);
const timer = setTimeout(() => {
this.idleTimers.delete(w);
if (this.idle.length <= 1) return;
this.idle = this.idle.filter(x => x !== w);
this.workers.delete(w); // Intentional termination must not trigger crash recovery.
const retirement = this.terminateStarted(w);
this.retirements.add(retirement);
void retirement.then(() => this.retirements.delete(retirement));
}, IDLE_WORKER_TIMEOUT_MS);
timer.unref?.();
this.idleTimers.set(w, timer);
}

private clearIdleTimer(w: PoolWorker): void {
clearTimeout(this.idleTimers.get(w));
this.idleTimers.delete(w);
}

// A worker died (crash hook, OOM, segfault, exit≠0). Respawn a replacement and
// retry its in-flight job once; a job that keeps crashing workers fails
// gracefully so it can't loop the pool forever.
private onWorkerGone(w: PoolWorker): void {
if (!this.workers.has(w)) return; // already handled (error+exit both fire)
this.workers.delete(w);
this.pendingWorkers.delete(w);
this.clearIdleTimer(w);
this.idle = this.idle.filter((x) => x !== w);
this.totalCrashes++;
const job = this.inflight.get(w);
Expand Down Expand Up @@ -305,6 +332,7 @@ export class QueryPool {
while (this.queue.length && (job = this.queue.shift()) && job.settled) job = undefined;
if (!job || job.settled) break;
const w = this.idle.pop()!;
this.clearIdleTimer(w);
this.inflight.set(w, job);
w.postMessage({ type: 'call', id: job.id, toolName: job.toolName, args: job.args });
}
Expand Down Expand Up @@ -369,6 +397,8 @@ export class QueryPool {
async destroy(): Promise<void> {
if (this.destroyed) return;
this.destroyed = true;
for (const timer of this.idleTimers.values()) clearTimeout(timer);
this.idleTimers.clear();
const ws = [...this.workers];
this.workers.clear();
this.pendingWorkers.clear();
Expand All @@ -378,6 +408,6 @@ export class QueryPool {
}
this.inflight.clear();
this.queue = [];
await Promise.all(ws.map((w) => this.terminateStarted(w)));
await Promise.all([...this.retirements, ...ws.map((w) => this.terminateStarted(w))]);
}
}