diff --git a/CHANGELOG.md b/CHANGELOG.md index cb651f556..7c20b7208 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -119,7 +119,7 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - An MCP tool call whose `projectPath` names a file inside a project, such as the file an agent is looking at, now answers from that project. Before, it failed with "Path is not a directory", even though a path that doesn't exist at all worked. Sensitive folders such as `/etc` are still refused when the path names a file inside them. Thanks @inth3shadows for the report and @maxmilian for the fix. (#2982) - An index left incomplete by an indexing run that was stopped partway, by a crash, an out-of-memory kill or the server's own hang check, no longer gets the background server killed with "Main thread unresponsive" every time it starts: on a large project, restoring the database indexes the run had set aside took longer than the hang check allows, so each restart died at the same point and the index was never repaired. They are now restored in the background, and the project is served once they are back. Thanks @cbeaulieu-gt for the report and the first fix. (#1887) - A new agent session no longer waits up to about 12 minutes for its first answer when the background server is stuck: a server whose process still holds its connection open but never responds made the session wait out a long series of retries. The session now gives up within seconds and answers on its own, and switches back to the background server once that has been restarted. -- The background server now recycles its query workers after five minutes idle, releasing the memory a burst of large queries left behind while keeping one fresh worker ready for the next call. Thanks @danusha2345 and @doomsday1906. +- The background server now gives back the memory a burst of large queries left behind after five minutes idle: the extra query workers exit and the last one stays up with its caches trimmed, so the next call after a pause is as quick as any other. Thanks @danusha2345 and @doomsday1906. - An MCP session whose agent stops reading its answers, while keeping the session open, now shuts down the normal way the next time it answers: it stops its work, frees the project for the next session and exits cleanly, and other sessions sharing the background server carry on. Before, that answer crashed it with "Uncaught exception: Error: EPIPE". Thanks @iiijiashu for the report and the first fix. (#2991) - A tool call whose arguments aren't an object, like `"arguments": "alpha"` from a hand-written MCP client or script, now gets the standard invalid-params error saying the arguments must be an object, and no tool runs. Before, a string or a number made most tools fail and come back as an internal CodeGraph error that told the agent to retry and then to continue without CodeGraph, and an array was run as if it were the arguments. Thanks @inth3shadows for the report and @kjgggggg for the first fix. (#2981) - The bundled Node runtime moves to 24.22.1, Node's October security release, and every release now checks the Node it bundles against the checksums nodejs.org publishes. The new runtime also fixes a crash on Windows as a command like `codegraph sync` exits, after its work is done, printing `Assertion failed: !(handle->flags & UV_HANDLE_CLOSING)`, which made a script or CI run that had succeeded fail. Thanks @Ancienttwo for the report and the first fix, and @IdealBoy709 for the report on the security release. (#2617, #2780) diff --git a/__tests__/dominant-file-cache.test.ts b/__tests__/dominant-file-cache.test.ts index f9ba57eef..c93205f3b 100644 --- a/__tests__/dominant-file-cache.test.ts +++ b/__tests__/dominant-file-cache.test.ts @@ -56,6 +56,26 @@ describe('dominant file — computed once per index state (#1864)', () => { return cg; } + it('keeps the answer when the connection lets go of its memory, and reads on', async () => { + const cg = await setup(); + const q = queriesOf(cg); + const compute = spyCompute(cg); + const answer = q.getDominantFile(); + const hits = cg.searchNodes('engineStep0'); + const db = (cg as unknown as { db: DatabaseConnection }).db.getDb(); + const mapped = db.pragma('mmap_size', { simple: true }); + expect(mapped).toBeGreaterThan(0); + const exec = vi.spyOn(db, 'exec'); + + cg.releaseMemory(); + + expect(exec.mock.calls.map(([sql]) => sql)).toEqual(expect.arrayContaining(['PRAGMA shrink_memory', 'PRAGMA mmap_size = 0'])); + expect(q.getDominantFile()).toEqual(answer); + expect(compute).toHaveBeenCalledTimes(1); + expect(db.pragma('mmap_size', { simple: true })).toBe(mapped); + expect(cg.searchNodes('engineStep0')).toEqual(hits); + }); + it('reuses the answer across explores while the database is unchanged', async () => { const cg = await setup(); const compute = spyCompute(cg); diff --git a/__tests__/query-pool.test.ts b/__tests__/query-pool.test.ts index 3fcf73f2e..932d7c86f 100644 --- a/__tests__/query-pool.test.ts +++ b/__tests__/query-pool.test.ts @@ -17,7 +17,7 @@ import { CodeGraph } from '../src'; import { ExploreSessionState, EXPLORE_EMISSION_KEY } from '../src/mcp/explore-session-state'; import type { MCPEngine } from '../src/mcp/engine'; import { QueryPool, resolvePoolSize, type PoolWorker } from '../src/mcp/query-pool'; -import type { ToolResult } from '../src/mcp/tools'; +import type { ToolHandler, ToolResult } from '../src/mcp/tools'; const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); @@ -40,6 +40,11 @@ class FakeWorker implements PoolWorker { exitCode: number | undefined; /** Ignore the pool's 'close', to exercise its terminate fallback. */ ignoresClose = false; + /** What a trim reports. Too much to keep unless a test says otherwise, so a used worker is replaced. */ + heldBytes = Number.MAX_SAFE_INTEGER; + /** Hold the 'trimmed' reply until the test sends it. */ + holdsTrimReport = false; + trims = 0; constructor(private behavior: (m: CallMsg) => Action, readyOk: boolean | null = true) { setTimeout(() => { if (this.alive && readyOk !== null) this.emitMessage({ type: 'ready', ok: readyOk }); }, 0); } @@ -55,7 +60,12 @@ class FakeWorker implements PoolWorker { if (this.alive) this.msgCb?.({ type: 'result', id, result }); } postMessage(msg: unknown): void { - const m = msg as CallMsg | { type: 'close' }; + const m = msg as CallMsg | { type: 'close' } | { type: 'trim' }; + if (m?.type === 'trim') { + this.trims++; + if (!this.holdsTrimReport && this.alive) this.emitMessage({ type: 'trimmed', heldBytes: this.heldBytes }); + return; + } if (m?.type === 'close') { // Like the real worker: close up and exit by itself, with code 0. if (this.ignoresClose) return; @@ -258,6 +268,170 @@ describe('QueryPool', () => { expect(workers[0]!.terminated).toBe(true); // the timeout ended it, inside destroy() }); + describe('going idle keeps one worker, trimmed in place', () => { + const light = 1024; + + /** A pool whose workers are light enough to keep, each answering with its own number. */ + function keepingPool(size: number, ready: (index: number) => boolean | null = () => true) { + const workers: FakeWorker[] = []; + const pool = new QueryPool({ + root: '/x', size, idleShrinkMs: 5, idleKeepBytes: light, + createWorker: () => { + const index = workers.length; + const worker = new FakeWorker(() => ({ result: ok(`worker ${index}`) }), ready(index)); + worker.heldBytes = light; + workers.push(worker); + return worker; + }, + }); + return { pool, workers }; + } + + it('answers the first call after idle on the worker already running, starting none', async () => { + const { pool, workers } = keepingPool(2); + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(1)); + expect(workers).toHaveLength(1); + expect(workers[0]!.alive).toBe(true); + expect((await pool.run('codegraph_search', {})).content[0].text).toBe('worker 0'); + expect(workers).toHaveLength(1); + await pool.destroy(); + }); + + it('stays ready through the shrink, so a call then never falls back to in-process', async () => { + // Any worker started here would never report ready, as if still loading. + const { pool, workers } = keepingPool(2, (index) => (index === 0 ? true : null)); + await pool.run('codegraph_search', {}); + await sleep(30); // well past the deadline + expect(pool.ready).toBe(true); + expect(workers).toHaveLength(1); + expect((await pool.run('codegraph_search', {})).content[0].text).toBe('worker 0'); + await pool.destroy(); + }); + + it('retires the workers a burst added, each by itself, and keeps the last to serve', async () => { + const workers: FakeWorker[] = []; + const releases: Array<() => void> = []; + const pool = new QueryPool({ + root: '/x', size: 3, idleShrinkMs: 5, idleKeepBytes: light, + createWorker: () => { + const index = workers.length; + const worker = new FakeWorker(() => ({ wait: new Promise((resolve) => { releases[index] = () => resolve(ok('r')); }) })); + worker.heldBytes = light; + workers.push(worker); + return worker; + }, + }); + const calls = Promise.all(Array.from({ length: 3 }, (_, i) => pool.run('codegraph_search', { i }))); + await vi.waitFor(() => expect(releases.filter(Boolean)).toHaveLength(3)); + for (const index of [2, 0, 1]) { + releases[index]!(); + await sleep(0); // its result lands before the next one's + } + await calls; + await vi.waitFor(() => expect(pool.liveWorkers).toBe(1)); + await vi.waitFor(() => expect(workers[1]!.trims).toBe(1)); + expect(workers).toHaveLength(3); + for (const retired of [workers[0]!, workers[2]!]) { + await vi.waitFor(() => expect(retired.exitCode).toBe(0)); + expect(retired.terminated).toBe(false); + expect(retired.trims).toBe(0); + } + expect(workers[1]!.alive).toBe(true); + await pool.destroy(); + }); + + it('keeps a worker that has served over a spare that only just started', async () => { + // The second worker reports ready after the last result, so it is the newest idle one. + const { pool, workers } = keepingPool(2, (index) => (index === 0 ? true : null)); + await vi.waitFor(() => expect(pool.ready).toBe(true)); + await Promise.all([pool.run('codegraph_search', { i: 0 }), pool.run('codegraph_search', { i: 1 })]); + expect(workers).toHaveLength(2); + workers[1]!.emitMessage({ type: 'ready', ok: true }); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(1)); + await vi.waitFor(() => expect(workers[1]!.exitCode).toBe(0)); + expect(pool.liveWorkers).toBe(1); + expect((await pool.run('codegraph_search', {})).content[0].text).toBe('worker 0'); + await pool.destroy(); + }); + + it('trims once per quiet spell, and again only after serving', async () => { + const { pool, workers } = keepingPool(1); + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(1)); + await sleep(30); + expect(workers[0]!.trims).toBe(1); + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(2)); + await pool.destroy(); + }); + + it('replaces a worker that holds more than the limit after its trim, not one at it', async () => { + const { pool, workers } = keepingPool(1); + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(1)); + await sleep(10); + expect(workers).toHaveLength(1); // exactly at the limit: kept + + workers[0]!.heldBytes = light + 1; + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers).toHaveLength(2)); + await vi.waitFor(() => expect(workers[0]!.exitCode).toBe(0)); + expect(workers[0]!.terminated).toBe(false); + expect((await pool.run('codegraph_search', {})).content[0].text).toBe('worker 1'); + await pool.destroy(); + }); + + it('leaves a heavy worker alone when a call has reached it since the trim', async () => { + const { pool, workers } = keepingPool(1); + const worker = workers[0]!; + worker.holdsTrimReport = true; + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(worker.trims).toBe(1)); + let finish!: (r: ToolResult) => void; + const held = new Promise((resolve) => { finish = resolve; }); + (worker as unknown as { behavior: (m: CallMsg) => Action }).behavior = () => ({ wait: held }); + const call = pool.run('codegraph_search', {}); + worker.emitMessage({ type: 'trimmed', heldBytes: Number.MAX_SAFE_INTEGER }); + await sleep(10); + expect(workers).toHaveLength(1); // no replacement, and the call is still its own + expect(worker.alive).toBe(true); + finish(ok('still here')); + expect((await call).content[0].text).toBe('still here'); + await pool.destroy(); + }); + + it('serves a call that arrives in the middle of the trim', async () => { + const { pool, workers } = keepingPool(1); + workers[0]!.holdsTrimReport = true; + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(1)); + expect((await pool.run('codegraph_search', {})).content[0].text).toBe('worker 0'); + expect(workers).toHaveLength(1); + await pool.destroy(); + }); + + it('recovers when the kept worker dies while idle', async () => { + const { pool, workers } = keepingPool(1); + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(1)); + workers[0]!.alive = false; + workers[0]!.emitExit(); + expect((await pool.run('codegraph_search', {})).content[0].text).toBe('worker 1'); + expect(pool.healthy).toBe(true); + await pool.destroy(); + }); + + it('ignores a late report from a worker it has let go of', async () => { + const { pool, workers } = keepingPool(2); + await pool.run('codegraph_search', {}); + await vi.waitFor(() => expect(workers[0]!.trims).toBe(1)); + await pool.destroy(); + workers[0]!.emitMessage({ type: 'trimmed', heldBytes: Number.MAX_SAFE_INTEGER }); + expect(workers).toHaveLength(1); + }); + }); + it('recovers from a worker crash: retries the in-flight call and respawns', async () => { let calls = 0; const pool = new QueryPool({ @@ -484,6 +658,8 @@ describe('QueryPool', () => { // Use the built engine so its pool loads the real compiled worker sibling. const BuiltEngine: typeof MCPEngine = require('../dist/mcp/engine').MCPEngine; +const BuiltHandler: typeof ToolHandler = require('../dist/mcp/tools').ToolHandler; +const BuiltGraph: typeof CodeGraph = require('../dist/index').default; describe('MCP query pool with real projects (#1465)', () => { let tempDir: string; @@ -602,7 +778,7 @@ describe('MCP query pool with real projects (#1465)', () => { const terminated: boolean[] = []; const exits: Promise[] = []; pool = new QueryPool({ - root: alpha, size: 1, idleShrinkMs: 50, + root: alpha, size: 1, idleShrinkMs: 50, idleKeepBytes: 0, createWorker: () => { const worker = new Worker(path.resolve(__dirname, '../dist/mcp/query-worker.js'), { workerData: { root: alpha } }); const i = terminated.push(false) - 1; @@ -622,6 +798,74 @@ describe('MCP query pool with real projects (#1465)', () => { expect(exits).toHaveLength(2); // a fresh worker replaced it }, 30000); + it('keeps a real worker through idle: it trims, stays up, and answers the next call', async () => { + const alpha = await indexProject('alpha', 'alphaSymbol'); + const started: Worker[] = []; + const held: number[] = []; + let released = 0; + // The real worker script, with its handler's release reported on the way through. + const script = ` + const { parentPort } = require('worker_threads'); + const { ToolHandler } = require(${JSON.stringify(path.resolve(__dirname, '../dist/mcp/tools.js'))}); + const release = ToolHandler.prototype.releaseIdleMemory; + ToolHandler.prototype.releaseIdleMemory = function () { + parentPort.postMessage({ type: 'released' }); + return release.call(this); + }; + require(${JSON.stringify(path.resolve(__dirname, '../dist/mcp/query-worker.js'))}); + `; + pool = new QueryPool({ + root: alpha, size: 2, idleShrinkMs: 50, + createWorker: () => { + const worker = new Worker(script, { eval: true, workerData: { root: alpha } }); + worker.on('message', (m: { type?: string; heldBytes?: number }) => { + if (m?.type === 'trimmed') held.push(m.heldBytes!); + if (m?.type === 'released') released++; + }); + started.push(worker); + return worker; + }, + }); + const first = await pool.run('codegraph_explore', { query: 'alphaSymbol' }); + expect(first.content[0].text).toContain('alphaSymbol'); + await vi.waitFor(() => expect(held).toHaveLength(1), { timeout: 15000 }); + expect(released).toBe(1); + expect(held[0]).toBeGreaterThan(0); + expect(held[0]).toBeLessThan(256 * 1024 * 1024); + + const after = await pool.run('codegraph_explore', { query: 'alphaSymbol' }); + expect(after).toEqual(first); + expect(started).toHaveLength(1); + expect(started[0]!.threadId).toBeGreaterThan(0); // -1 once a worker has exited + }, 30000); + + it('a kept worker closes all but the last project it used, and goes on answering', async () => { + const alpha = await indexProject('alpha', 'alphaSymbol'); + const beta = await indexProject('beta', 'betaSymbol'); + const gamma = await indexProject('gamma', 'gammaSymbol'); + const cg = BuiltGraph.openSync(alpha); + const handler = new BuiltHandler(cg); + try { + await handler.executeReadTool('codegraph_search', { projectPath: beta, query: 'betaSymbol' }); + await handler.executeReadTool('codegraph_search', { projectPath: gamma, query: 'gammaSymbol' }); + const open = (handler as unknown as { projectCache: Map }).projectCache; + const betaGraph = [...open.values()][0]!; + expect(open.size).toBe(2); + + handler.releaseIdleMemory(); + + expect([...open.keys()]).toEqual([gamma]); + expect(() => betaGraph.searchNodes('betaSymbol')).toThrow(); + for (const [projectPath, symbol] of [[gamma, 'gammaSymbol'], [alpha, 'alphaSymbol'], [beta, 'betaSymbol']]) { + const result = await handler.executeReadTool('codegraph_search', { projectPath, query: symbol }); + expect(result.content[0].text).toContain(symbol); + } + } finally { + await handler.closeAll(); + cg.close(); + } + }, 30000); + it('never terminates a real worker before it has started', async () => { const alpha = await indexProject('alpha', 'alphaSymbol'); const startedWhenEnded: boolean[] = []; diff --git a/src/db/index.ts b/src/db/index.ts index c165d7e13..3e3b1a68a 100644 --- a/src/db/index.ts +++ b/src/db/index.ts @@ -16,6 +16,9 @@ export { SqliteDatabase, SqliteBackend } from './sqlite-adapter'; export { DatabaseOpenError } from './open-error'; export { NodeSqliteUnavailableError } from './node-sqlite'; +/** SQLite's memory-mapped window on the database file: 256 MB. */ +const MMAP_SIZE_BYTES = 268435456; + /** * Apply connection-level PRAGMAs. Shared by `initialize` and `open` so the two * paths can't drift. @@ -37,7 +40,7 @@ function configureConnection(db: SqliteDatabase, readOnly = false): void { db.pragma('synchronous = NORMAL'); // safe with WAL mode db.pragma('cache_size = -64000'); // 64 MB page cache db.pragma('temp_store = MEMORY'); // temp tables in memory - db.pragma('mmap_size = 268435456'); // 256 MB memory-mapped I/O + db.pragma(`mmap_size = ${MMAP_SIZE_BYTES}`); // memory-mapped I/O // Without a journal_size_limit the -wal file never shrinks below its // high-water mark while a connection lives: checkpoints fold frames back but // leave the file at full size, so one giant deferred-sync WAL stays giant @@ -956,6 +959,19 @@ export class DatabaseConnection { } } + /** + * Give back what this connection holds only to read faster: its page cache, + * and the file mapping, whose touched pages count toward the process until + * they are unmapped. The next reads fill both again. It stays the same + * connection, so its prepared statements stay, and so do the answers + * remembered against its change stamp. + */ + releaseMemory(): void { + this.db.exec('PRAGMA shrink_memory'); + this.db.pragma('mmap_size = 0'); + this.db.pragma(`mmap_size = ${MMAP_SIZE_BYTES}`); + } + /** * Close the database connection */ diff --git a/src/index.ts b/src/index.ts index d14ab409b..5b0fa212c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -606,6 +606,15 @@ export class CodeGraph { return isInitialized(path.resolve(projectRoot)); } + /** + * Give back the memory the database connection holds only to read faster, + * keeping it open and keeping every answer it has worked out (see + * {@link DatabaseConnection.releaseMemory}). + */ + releaseMemory(): void { + this.db.releaseMemory(); + } + /** * Close the CodeGraph instance and release resources */ diff --git a/src/mcp/query-pool.ts b/src/mcp/query-pool.ts index 653d50039..7f43cc8e1 100644 --- a/src/mcp/query-pool.ts +++ b/src/mcp/query-pool.ts @@ -17,6 +17,9 @@ * once; a poison call that keeps crashing fails gracefully (never wedges the * pool). A crash budget trips a circuit breaker (`healthy` → false) so the * caller falls back to in-process dispatch instead of thrashing respawns. + * - idle trim: a pool that goes quiet gives its memory back. The workers a + * burst added exit, and the last one to serve stays up and shrinks in + * place, so the next call finds a warm worker rather than a cold start. * - graceful backstop: a call that can't be served within `softTimeoutMs` * resolves with SUCCESS-shaped "busy, retry" guidance — never `isError`, so * a momentary overload can't teach the agent to abandon codegraph — instead @@ -48,15 +51,22 @@ export interface PoolWorker { /** Default linger before a queued call is answered with busy-guidance. */ const DEFAULT_BUSY_TIMEOUT_MS = 45_000; // < the ~60s MCP client request timeout -/** Recycle a quiescent pool after a burst so worker-local graph/SQLite caches +/** Shrink a quiescent pool after a burst so worker-local graph/SQLite caches * do not remain resident for the daemon's whole lifetime. Long enough that an - * agent pausing to edit or build keeps its warm workers: the first explore on - * a fresh one takes about 200 ms more. */ + * agent pausing to edit or build keeps every worker it grew. */ const DEFAULT_IDLE_SHRINK_MS = 300_000; /** How long a retired idle worker gets to exit on 'close' before it is terminated. */ const RETIRE_TIMEOUT_MS = 5_000; +/** + * Most a worker may still hold after its idle trim and stay. A trimmed worker + * holds about 80 MB, mostly heap and tree-sitter's wasm, whatever the size of + * the project; one above this holds something that won't come back in place, + * so it costs less to replace it than to keep it for the rest of the session. + */ +const DEFAULT_IDLE_KEEP_BYTES = 256 * 1024 * 1024; + /** Hard ceiling on pool size regardless of core count / env. */ const MAX_POOL_SIZE = 16; @@ -95,6 +105,8 @@ interface WorkerMessage { ok?: boolean; id?: number; result?: ToolResult; + /** On 'trimmed': the heap and buffers the worker still holds. */ + heldBytes?: number; } interface Job { @@ -117,8 +129,10 @@ export interface QueryPoolOptions { softTimeoutMs?: number; /** Retries for an in-flight call whose worker crashed. Default 1. */ maxRetries?: number; - /** Idle delay before recycling back to one fresh worker. Default 5 minutes; 0 disables. */ + /** Idle delay before shrinking to one trimmed worker. Default 5 minutes; 0 disables. */ idleShrinkMs?: number; + /** A worker holding more than this after its trim is replaced. Default 256 MB; 0 replaces any. */ + idleKeepBytes?: number; /** How long a retired idle worker gets to exit by itself (tests shorten it). Default 5s. */ retireTimeoutMs?: number; /** Worker factory (tests inject a fake). Defaults to a real `worker_threads` Worker. */ @@ -188,6 +202,7 @@ export class QueryPool { private readonly softTimeoutMs: number; private readonly maxRetries: number; private readonly idleShrinkMs: number; + private readonly idleKeepBytes: number; private readonly retireTimeoutMs: number; private readonly createWorker: () => PoolWorker; private readonly startSettleMs: number; @@ -195,9 +210,12 @@ export class QueryPool { // Retired idle workers asked to exit: the timer that terminates one that // hasn't by retireTimeoutMs, and what destroy() waits on. private retiring = new Map; settle: () => void }>(); - // Set by a served call and cleared by a recycle, so a clean warm worker that - // has answered nothing is never recycled for its own sake. - private servedSinceRecycle = false; + // Set by a served call and cleared by a shrink, so a worker that has answered + // nothing since is left alone rather than trimmed again for its own sake. + private servedSinceShrink = false; + // The worker that answered last, which is the one worth keeping: its caches + // and its compiled code are what the next call would otherwise rebuild. + private lastServer?: PoolWorker; constructor(opts: QueryPoolOptions) { this.root = opts.root; @@ -205,6 +223,7 @@ export class QueryPool { this.softTimeoutMs = opts.softTimeoutMs ?? resolveBusyTimeoutMs(); this.maxRetries = opts.maxRetries ?? 1; this.idleShrinkMs = Math.max(0, opts.idleShrinkMs ?? DEFAULT_IDLE_SHRINK_MS); + this.idleKeepBytes = opts.idleKeepBytes ?? DEFAULT_IDLE_KEEP_BYTES; this.retireTimeoutMs = opts.retireTimeoutMs ?? RETIRE_TIMEOUT_MS; this.createWorker = opts.createWorker ?? (() => new Worker(WORKER_FILE, { workerData: { root: this.root } })); this.startSettleMs = opts.startSettleMs ?? WORKER_START_SETTLE_MS; @@ -235,9 +254,10 @@ export class QueryPool { * hitting that window was the recurring #662 test flake (and a real * first-call stall for agents). The pool exists for CONCURRENT load, which * by definition arrives after warm-up; the pre-pool in-process path is - * strictly better while nothing is warm, which also holds while an idle - * recycle's fresh worker starts. Stays true across crash-respawn gaps, which - * are covered by retry + backstop. + * strictly better while nothing is warm, which also holds while the + * replacement for a worker too heavy to keep starts. Stays true across an + * idle shrink, whose kept worker never stops serving, and across crash-respawn + * gaps, which are covered by retry + backstop. */ get ready(): boolean { return this.everReady && !this.destroyed; @@ -284,11 +304,20 @@ export class QueryPool { const job = this.inflight.get(w); this.inflight.delete(w); this.idle.push(w); - this.servedSinceRecycle = true; + this.servedSinceShrink = true; + this.lastServer = w; if (job) this.settle(job, m.result ?? busyGuidance(0)); this.drain(); this.armIdleShrink(); + return; } + if (m.type === 'trimmed') this.replaceIfHeavy(w, m.heldBytes ?? 0); + } + + /** Nothing queued or running, nothing starting, and every worker free. */ + private get quiescent(): boolean { + return !this.destroyed && this.queue.length === 0 && this.inflight.size === 0 && + this.pendingWorkers.size === 0 && this.idle.length === this.workers.size; } private clearIdleShrink(): void { @@ -298,31 +327,50 @@ export class QueryPool { private armIdleShrink(): void { this.clearIdleShrink(); - if ( - this.idleShrinkMs === 0 || this.destroyed || !this.servedSinceRecycle || - this.queue.length > 0 || this.inflight.size > 0 || this.pendingWorkers.size > 0 || - this.idle.length !== this.workers.size - ) return; + if (this.idleShrinkMs === 0 || !this.servedSinceShrink || !this.quiescent) return; this.idleShrinkTimer = setTimeout(() => this.shrinkIdle(), this.idleShrinkMs); this.idleShrinkTimer.unref?.(); } private shrinkIdle(): void { this.idleShrinkTimer = undefined; - if ( - this.destroyed || this.queue.length > 0 || this.inflight.size > 0 || - this.pendingWorkers.size > 0 || this.idle.length !== this.workers.size - ) return; - - // Drop every used isolate, including the last one: a single worker can hold - // hundreds of MB in graph and SQLite caches after a large query. Replace it - // with one clean warm worker so the next burst still avoids a cold queue. - const stale = [...this.workers]; + if (!this.quiescent) return; + + // Keep one worker, trimmed in place, rather than ending them all and + // starting a clean one. A clean worker has to work out again everything a + // connection remembers (the dominant file, the name indexes): half a second + // on the next call on a mid-sized index, seconds on a large one. The trim + // leaves this one about 70 MB above a clean worker, on any index. Staying + // up also means no moment here when a call finds no worker. + const keep = this.lastServer && this.idle.includes(this.lastServer) + ? this.lastServer + : this.idle[this.idle.length - 1]; + if (!keep) return; + for (const w of this.idle) { + if (w === keep) continue; + this.workers.delete(w); + this.retire(w); + } + this.idle = [keep]; + this.servedSinceShrink = false; + try { + keep.postMessage({ type: 'trim' }); + } catch { /* already gone: its 'exit' recovers the pool */ } + } + + /** + * The last check on a kept worker: one that still holds a lot after its trim + * is ended like the rest, with one clean worker in its place, so that keeping + * a worker can't keep hold of what a large query left behind. Calls that come + * while the clean one starts are served in process. Back at work by now, it + * is left alone, and judged at the next shrink. + */ + private replaceIfHeavy(w: PoolWorker, heldBytes: number): void { + if (heldBytes <= this.idleKeepBytes || !this.quiescent || this.workers.size !== 1) return; this.workers.clear(); this.idle = []; this.everReady = false; - this.servedSinceRecycle = false; - for (const w of stale) this.retire(w); + this.retire(w); this.spawnOne(); } diff --git a/src/mcp/query-worker.ts b/src/mcp/query-worker.ts index f03af5eee..ca755199b 100644 --- a/src/mcp/query-worker.ts +++ b/src/mcp/query-worker.ts @@ -41,6 +41,11 @@ interface CloseMessage { type: 'close'; } +/** The pool keeping this idle worker: give back what it can, and say what is left. */ +interface TrimMessage { + type: 'trim'; +} + // Mirror the engine's lazy-require of the heavy CodeGraph + tools chain. This // module is only ever loaded as a Worker, so the require runs once on spawn. const loadCodeGraph = (): typeof import('../index').default => @@ -70,7 +75,7 @@ if (parentPort) { // against its crash budget (→ fall back to in-process) without hanging. port.postMessage({ type: 'ready', ok: initError === null, error: initError }); - port.on('message', (msg: CallMessage | CloseMessage) => { + port.on('message', (msg: CallMessage | CloseMessage | TrimMessage) => { if (msg?.type === 'close') { try { void handler?.closeAll(); // projects opened through projectPath @@ -82,6 +87,15 @@ if (parentPort) { collectBeforeExit(); process.exit(0); } + if (msg?.type === 'trim') { + try { + handler?.releaseIdleMemory(); + } catch { /* a worker that can't let go still serves */ } + // This thread's own heap and buffers: `rss` is the whole process's. + const { heapUsed, external } = process.memoryUsage(); + port.postMessage({ type: 'trimmed', heldBytes: heapUsed + external }); + return; + } if (!msg || msg.type !== 'call') return; void serve(msg); }); diff --git a/src/mcp/tools.ts b/src/mcp/tools.ts index 07dfdc2c1..bfe967e7c 100644 --- a/src/mcp/tools.ts +++ b/src/mcp/tools.ts @@ -2623,13 +2623,13 @@ export class ToolHandler { * Evicts over the LRU bound, on close, and once idle past the timeout * (#2087). The cache is in last-use order, so idle entries lead it. */ - private trimProjects(): void { + private trimProjects(maxOpen = MAX_CACHED_PROJECTS): void { if (this.activeCalls > 0) return; const idleMs = resolveProjectIdleTimeoutMs(); const now = Date.now(); for (const [root, cg] of this.projectCache) { const idle = idleMs > 0 && now - (this.projectUsedAt.get(root) ?? now) >= idleMs; - if (!this.closing && this.projectCache.size <= MAX_CACHED_PROJECTS && !idle) break; + if (!this.closing && this.projectCache.size <= maxOpen && !idle) break; if (this.projectGates.has(cg)) continue; this.projectCache.delete(root); this.projectUsedAt.delete(root); @@ -2696,6 +2696,22 @@ export class ToolHandler { return cg; } + /** + * A query worker kept through a quiet spell gives back what it can without + * going cold: every project but the last one used closes, and what stays + * open lets go of the memory it holds only to read faster. The heap needs + * nothing from here: V8 shrinks an idle thread's by itself, well before the + * pool's idle delay is up. + */ + releaseIdleMemory(): void { + this.trimProjects(1); + for (const cg of [this.cg, ...this.projectCache.values()]) { + try { + cg?.releaseMemory(); + } catch { /* one that can't let go still answers, and the others still can */ } + } + } + /** * Close all cached project connections */