Skip to content

fix(stdio): stop EventEmitter listener accumulation in the stdio transport #3073

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
amd-clb wants to merge 1 commit into
base: main
Choose a base branch
from
+292 −8
Open

fix(stdio): stop EventEmitter listener accumulation in the stdio transport #3073

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
19 changes: 19 additions & 0 deletions .changeset/fix-stdio-close-listener-cleanup.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
103 changes: 95 additions & 8 deletions packages/client/src/client/stdio.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -154,28 +160,37 @@ 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();
} catch (error) {
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);
Expand Down Expand Up @@ -233,6 +248,7 @@ export class StdioClientTransport implements Transport {
private async _dispose(): Promise<void> {
const proc = this._process;
this._process = undefined;
this.settlePendingSends();
if (proc && proc.exitCode === null && proc.signalCode === null) {
const exited = new Promise<void>(resolve => proc.once('exit', () => resolve()));
try {
Expand Down Expand Up @@ -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<void>(resolve => {
processToClose.once('close', () => {
resolve();
Expand Down Expand Up @@ -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();
Expand All @@ -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;
}
}
}
178 changes: 178 additions & 0 deletions packages/client/test/client/stdio.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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<void>[] = [];
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.
Expand Down
Loading