diff --git a/.changeset/fix-stdio-close-listener-cleanup.md b/.changeset/fix-stdio-close-listener-cleanup.md new file mode 100644 index 0000000000..8a9d35a404 --- /dev/null +++ b/.changeset/fix-stdio-close-listener-cleanup.md @@ -0,0 +1,19 @@ +--- +"@modelcontextprotocol/client": patch +--- + +fix(stdio): stop EventEmitter listener accumulation in the stdio transport + +Two sources of unbounded listener growth are fixed in `StdioClientTransport`: + +- `send()` used to add a dedicated `drain` listener to `stdin` for every write + that hit a full buffer, so sustained back-pressure accumulated listeners and + triggered `MaxListenersExceededWarning`. Sends now share a single `drain` + wait, and any still-pending sends are settled when the process is closed or + disposed so their promises cannot hang. +- `close()` now detaches the listeners installed in `start()` (the `data` + listener on `stdout` and the `error` listeners) and explicitly unpipes and + ends the `stderr` pass-through. Ending the pass-through tears down any + stream the caller piped it into (e.g. a shared log) deterministically, even + when a grandchild keeps the child's `stderr` write-end open so the natural + end cascade never fires. diff --git a/packages/client/src/client/stdio.ts b/packages/client/src/client/stdio.ts index 6dbddb667c..801d093d19 100644 --- a/packages/client/src/client/stdio.ts +++ b/packages/client/src/client/stdio.ts @@ -109,6 +109,12 @@ export class StdioClientTransport implements Transport { private _readBuffer: ReadBuffer; private _serverParams: StdioServerParameters; private _stderrStream: PassThrough | null = null; + private _stdoutDataHandler?: (chunk: Buffer) => void; + private _stdoutErrorHandler?: (error: Error) => void; + private _stdinErrorHandler?: (error: Error) => void; + private _pendingDrainWaiters: Array<() => void> = []; + private _drainHandler?: () => void; + private _drainStream?: Stream; onclose?: () => void; onerror?: (error: Error) => void; @@ -154,16 +160,23 @@ export class StdioClientTransport implements Transport { resolve(); }); + this._process.on('exit', () => { + // Settle at exit, not just close: 'close' can lag behind if a grandchild inherits the child's stdio. + this.settlePendingSends(); + }); + this._process.on('close', _code => { this._process = undefined; this.onclose?.(); + this.settlePendingSends(); }); - this._process.stdin?.on('error', error => { + this._stdinErrorHandler = error => { this.onerror?.(error); - }); + }; + this._process.stdin?.on('error', this._stdinErrorHandler); - this._process.stdout?.on('data', chunk => { + this._stdoutDataHandler = chunk => { try { this._readBuffer.append(chunk); this.processReadBuffer(); @@ -171,11 +184,13 @@ export class StdioClientTransport implements Transport { this.onerror?.(error as Error); this.close().catch(() => {}); } - }); + }; + this._process.stdout?.on('data', this._stdoutDataHandler); - this._process.stdout?.on('error', error => { + this._stdoutErrorHandler = error => { this.onerror?.(error); - }); + }; + this._process.stdout?.on('error', this._stdoutErrorHandler); if (this._stderrStream && this._process.stderr) { this._process.stderr.pipe(this._stderrStream); @@ -233,6 +248,7 @@ export class StdioClientTransport implements Transport { private async _dispose(): Promise { const proc = this._process; this._process = undefined; + this.settlePendingSends(); if (proc && proc.exitCode === null && proc.signalCode === null) { const exited = new Promise(resolve => proc.once('exit', () => resolve())); try { @@ -282,6 +298,26 @@ export class StdioClientTransport implements Transport { const processToClose = this._process; this._process = undefined; + // Detach the listeners installed in start() so late stdout cannot reach the JSON-RPC + // parser. The stderr pass-through is deliberately left attached until the child has + // fully shut down below, so stderr the child emits while dying still reaches consumers. + if (this._stdoutDataHandler) { + processToClose.stdout?.removeListener('data', this._stdoutDataHandler); + this._stdoutDataHandler = undefined; + } + if (this._stdoutErrorHandler) { + processToClose.stdout?.removeListener('error', this._stdoutErrorHandler); + this._stdoutErrorHandler = undefined; + } + if (this._stdinErrorHandler) { + processToClose.stdin?.removeListener('error', this._stdinErrorHandler); + this._stdinErrorHandler = undefined; + } + // Keep no-op error handlers so late errors cannot throw uncaught. + processToClose.stdin?.on('error', () => {}); + processToClose.stdout?.on('error', () => {}); + processToClose.stderr?.on('error', () => {}); + const closePromise = new Promise(resolve => { processToClose.once('close', () => { resolve(); @@ -313,6 +349,21 @@ export class StdioClientTransport implements Transport { // ignore } } + + // The child is gone: tear down the stderr pass-through so a shared stream the caller piped + // it into deterministically drops its pipe/drain listeners, even when the child never + // exited cleanly (e.g. a grandchild kept the child's stderr write-end open). + if (this._stderrStream) { + try { + processToClose.stderr?.unpipe(this._stderrStream); + } catch { + // ignore + } + this._stderrStream.end(); + } + + // A pending send may still wait on a 'drain' the dead process will never emit. + this.settlePendingSends(); } this._readBuffer.clear(); @@ -327,9 +378,45 @@ export class StdioClientTransport implements Transport { const json = serializeMessage(message); if (this._process.stdin.write(json)) { resolve(); - } else { - this._process.stdin.once('drain', resolve); + return; + } else if (this._pendingDrainWaiters.length === 0) { + // Share one 'drain' wait so back-pressure cannot accumulate listeners on stdin. + this._drainStream = this._process.stdin; + this._drainHandler = () => { + this._drainHandler = undefined; + this._drainStream = undefined; + const waiters = this._pendingDrainWaiters; + this._pendingDrainWaiters = []; + for (const wait of waiters) { + wait(); + } + }; + this._drainStream.once('drain', this._drainHandler); } + this._pendingDrainWaiters.push(resolve); }); } + + /** + * Resolve the send() promises still waiting on a stdin 'drain' that the process + * will no longer emit once it is gone. They resolve rather than reject: the bytes + * were already handed to stdin.write(), so delivery past process exit is out of scope + * — use onclose to learn the connection is gone. + * + * @internal + */ + private settlePendingSends() { + const waiters = this._pendingDrainWaiters; + this._pendingDrainWaiters = []; + for (const wait of waiters) { + wait(); + } + // The shared 'drain' wait may still be attached to a stdin the process will never drain; + // drop it so it does not linger on a dead stream. + if (this._drainHandler && this._drainStream) { + this._drainStream.removeListener('drain', this._drainHandler); + this._drainHandler = undefined; + this._drainStream = undefined; + } + } } diff --git a/packages/client/test/client/stdio.test.ts b/packages/client/test/client/stdio.test.ts index fe6d9258ce..3363e0c29c 100644 --- a/packages/client/test/client/stdio.test.ts +++ b/packages/client/test/client/stdio.test.ts @@ -2,6 +2,7 @@ import { existsSync } from 'node:fs'; import { tmpdir } from 'node:os'; import type { JSONRPCMessage } from '@modelcontextprotocol/core-internal'; +import { vi } from 'vitest'; import type { StdioServerParameters } from '../../src/client/stdio'; import { DEFAULT_INHERITED_ENV_VARS, StdioClientTransport } from '../../src/client/stdio'; @@ -151,6 +152,183 @@ test('_dispose releases the parent-side pipe handles even when a helper process expect(proc.stdin?.destroyed).toBe(true); }, 10_000); +test('close() removes the data listener from stdout to prevent stale reads after close', async () => { + // Regression test for #780: the 'data' listener on stdout must be removed + // in close(), otherwise late non-JSON output from the child triggers + // processReadBuffer() → JSON.parse() → SyntaxError after close. + const client = new StdioClientTransport({ + command: 'node', + args: ['-e', 'process.stdin.resume(); process.stdin.on("end", () => process.exit(0));'] + }); + + await client.start(); + + const proc = (client as unknown as { _process: import('node:child_process').ChildProcess })._process; + expect(proc).toBeDefined(); + expect(proc.stdout).toBeDefined(); + + // Before close, the data listener should be present + expect(proc.stdout!.listenerCount('data')).toBeGreaterThan(0); + + await client.close(); + + // After close, the data listener should be removed + expect(proc.stdout!.listenerCount('data')).toBe(0); +}); + +test('send() under back-pressure does not accumulate drain listeners on stdin', async () => { + // Regression test for #3070: each send() that hits a full stdin buffer used + // to add its own 'drain' listener, so sustained back-pressure piled them up + // (MaxListenersExceededWarning). Sends now share a single drain wait. + const warnings: string[] = []; + const onWarning = (warning: Error) => { + if (warning.name === 'MaxListenersExceededWarning') warnings.push(warning.message); + }; + process.on('warning', onWarning); + + try { + // A child that never reads stdin keeps the pipe permanently full. + const client = new StdioClientTransport({ + command: process.execPath, + args: ['-e', 'setTimeout(() => {}, 10_000)'] + }); + await client.start(); + + const pending: Promise[] = []; + for (let i = 0; i < 30; i++) { + pending.push(client.send({ jsonrpc: '2.0', id: i, method: 'x', params: { blob: 'a'.repeat(4096) } })); + } + await new Promise(resolve => setTimeout(resolve, 100)); + + const proc = (client as unknown as { _process: import('node:child_process').ChildProcess })._process; + expect(proc!.stdin!.listenerCount('drain')).toBeLessThanOrEqual(1); + expect(warnings).toHaveLength(0); + + await client.close(); + // The back-pressured sends settle once the process is gone (no dangling promises). + await Promise.all(pending); + } finally { + process.off('warning', onWarning); + } +}, 15_000); + +test('close() removes the shared drain listener from stdin when the child never drains', async () => { + // A back-pressured send() registers the shared 'drain' wait on the child's stdin. If the child + // is closed before it ever drains, close() (via settlePendingSends()) must remove that listener + // rather than leave it dangling on a dead stream. A child that never reads stdin keeps the pipe + // full, so 'drain' never fires on its own — only the explicit removal in close() can clear it. + const client = new StdioClientTransport({ + command: process.execPath, + args: ['-e', 'setTimeout(() => {}, 10_000)'] + }); + await client.start(); + + const proc = (client as unknown as { _process: import('node:child_process').ChildProcess })._process; + + // One large back-pressured send registers the shared 'drain' wait on stdin. + const pending = client.send({ jsonrpc: '2.0', id: 0, method: 'x', params: { blob: 'a'.repeat(256 * 1024) } }); + await vi.waitFor(() => expect(proc!.stdin!.listenerCount('drain')).toBe(1), { timeout: 3000, interval: 25 }); + + await client.close(); + await pending; + + // close() settled the pending send and removed the drain listener the dead stdin will never emit. + expect(proc!.stdin!.listenerCount('drain')).toBe(0); +}, 15_000); + +test('a back-pressured send() resolves and onclose fires exactly once when the child dies', async () => { + // Pins the settle-on-death contract: a send() still waiting on a stdin 'drain' when the child + // dies resolves (rather than hanging or rejecting), and the connection-gone signal (onclose) + // fires exactly once. A child that never reads stdin keeps the pipe full, so 'drain' never + // fires on its own — only the settle-on-death path can resolve the pending send. + let oncloseCount = 0; + const client = new StdioClientTransport({ + command: process.execPath, + args: ['-e', 'setTimeout(() => {}, 10_000)'] + }); + client.onclose = () => { + oncloseCount += 1; + }; + + await client.start(); + + const proc = (client as unknown as { _process: import('node:child_process').ChildProcess })._process; + + // One large back-pressured send leaves the shared 'drain' wait pending on the full pipe. + const pending = client.send({ jsonrpc: '2.0', id: 0, method: 'x', params: { blob: 'a'.repeat(256 * 1024) } }); + await vi.waitFor(() => expect(proc!.stdin!.listenerCount('drain')).toBe(1), { timeout: 3000, interval: 25 }); + + await client.close(); + + // The still-pending send resolves once the child is gone... + await pending; + // ...and the connection-gone signal fired exactly once. + expect(oncloseCount).toBe(1); +}, 15_000); + +test('a back-pressured send() still pending when the child exits on its own resolves and onclose fires once', async () => { + // A child that never reads stdin keeps the pipe full, so a large send() stays pending on the + // shared 'drain' wait. Here the child exits on its own (not via close()), so the pending send + // is settled by settle-on-exit (the 'exit' handler) rather than by an explicit close() call — + // settling at the earliest point the process is gone, so a send can never wait on a 'close' + // that lags behind exit (e.g. a grandchild inheriting the child's stdio on platforms where + // 'close' waits for the stdio to fully close). Pins: the send resolves, onclose fires once, + // and the drain listener the dead stdin will never emit is removed. + let oncloseCount = 0; + const client = new StdioClientTransport({ + command: process.execPath, + args: ['-e', 'setTimeout(() => process.exit(0), 2000)'] + }); + client.onclose = () => { + oncloseCount += 1; + }; + + await client.start(); + + const proc = (client as unknown as { _process: import('node:child_process').ChildProcess })._process; + + // One large back-pressured send leaves the shared 'drain' wait pending on the full pipe. + const pending = client.send({ jsonrpc: '2.0', id: 0, method: 'x', params: { blob: 'a'.repeat(256 * 1024) } }); + await vi.waitFor(() => expect(proc!.stdin!.listenerCount('drain')).toBe(1), { timeout: 3000, interval: 25 }); + + // The child exits on its own at ~2s; settle-on-exit resolves the send. + await pending; + expect(oncloseCount).toBe(1); + expect(proc!.stdin!.listenerCount('drain')).toBe(0); +}, 15_000); + +test('close() removes the drain listener the stderr pipe added on the pass-through', async () => { + // Regression test for #3070. child.stderr.pipe(_stderrStream) attaches a 'drain' listener to + // _stderrStream (the pipe's destination) once it backs up. A grandchild holding the child's + // stderr write-end means child.stderr never 'end's, so pipe() never auto-cleans that listener + // up — close() must unpipe it. Counting 'drain' on the pass-through distinguishes fixed from + // unfixed; counting it on the source (stderr) would always read 0. + const HOLDING_SCRIPT = [ + String.raw`const { spawn } = require('child_process');`, + String.raw`spawn(process.execPath, ['-e', 'setTimeout(() => {}, 3000)'], { stdio: 'inherit' });`, + String.raw`process.stderr.write(Buffer.alloc(256 * 1024, 0x41));` + ].join('\n'); + const client = new StdioClientTransport({ + command: process.execPath, + args: ['-e', HOLDING_SCRIPT], + stderr: 'pipe' + }); + + await client.start(); + + const stderrStream = (client as unknown as { _stderrStream: import('node:stream').PassThrough })._stderrStream; + if (stderrStream === null) throw new Error('expected a stderr pass-through'); + + // Wait until the child's stderr back-pressure has made pipe() attach the 'drain' listener. + await vi.waitFor(() => expect(stderrStream.listenerCount('drain')).toBeGreaterThan(0), { timeout: 3000, interval: 25 }); + + await client.close(); + + // The unpipe in close() removed the drain listener even though the grandchild keeps the + // child's stderr open, so the pass-through cannot accumulate drain listeners. + expect(stderrStream.listenerCount('drain')).toBe(0); +}, 20_000); + test('DEFAULT_INHERITED_ENV_VARS matches the host platform', () => { if (process.platform === 'win32') { // Variables Windows tooling needs to resolve executables and shells.