From c63a4da8eaf5dfd28ab38b5169f7c5dfd1704ff0 Mon Sep 17 00:00:00 2001 From: doomsday1906 Date: Fri, 23 Oct 2026 23:31:07 +0300 Subject: [PATCH] Release excess query workers after idle bursts --- CHANGELOG.md | 1 + __tests__/query-pool.test.ts | 92 ++++++++++++++++++++++++++++++++++++ src/mcp/query-pool.ts | 36 ++++++++++++-- 3 files changed, 126 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4222048f58..20ed54a175 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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("/me")` or `await api.get(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()` or `render()`, now counts as rendered there, so an app's entry reaches its root component and a component's tests show among its callers. (#2587) diff --git a/__tests__/query-pool.test.ts b/__tests__/query-pool.test.ts index 7c28f48f04..e6310e746f 100644 --- a/__tests__/query-pool.test.ts +++ b/__tests__/query-pool.test.ts @@ -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(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(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; diff --git a/src/mcp/query-pool.ts b/src/mcp/query-pool.ts index 6b1f473991..9b6b688949 100644 --- a/src/mcp/query-pool.ts +++ b/src/mcp/query-pool.ts @@ -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 @@ -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 @@ -157,6 +161,8 @@ export class QueryPool { private queue: Job[] = []; private inflight = new Map(); private workers = new Set(); + private idleTimers = new Map(); + private retirements = new Set>(); // 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 + @@ -250,19 +256,39 @@ 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. @@ -270,6 +296,7 @@ export class QueryPool { 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); @@ -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 }); } @@ -369,6 +397,8 @@ export class QueryPool { async destroy(): Promise { 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(); @@ -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))]); } }