From bac3354b078f4df2a1c469ec42c3eddd31fff1d1 Mon Sep 17 00:00:00 2001 From: Gursimran Singh Date: Mon, 24 Aug 2026 06:27:42 +0530 Subject: [PATCH 1/2] fix(stdio): share a single drain listener under backpressure --- .changeset/shared-stdio-drain-wait.md | 6 ++ packages/client/src/client/stdio.ts | 12 ++- .../test/client/stdioBackpressure.test.ts | 80 +++++++++++++++++++ packages/core-internal/src/shared/stdio.ts | 44 ++++++++++ packages/server/src/server/stdio.ts | 30 ++----- packages/server/test/server/stdio.test.ts | 32 ++++++++ 6 files changed, 176 insertions(+), 28 deletions(-) create mode 100644 .changeset/shared-stdio-drain-wait.md create mode 100644 packages/client/test/client/stdioBackpressure.test.ts diff --git a/.changeset/shared-stdio-drain-wait.md b/.changeset/shared-stdio-drain-wait.md new file mode 100644 index 0000000000..f69c5a1eae --- /dev/null +++ b/.changeset/shared-stdio-drain-wait.md @@ -0,0 +1,6 @@ +--- +'@modelcontextprotocol/client': patch +'@modelcontextprotocol/server': patch +--- + +Share a single 'drain' listener across concurrent backpressured stdio writes, fixing MaxListenersExceededWarning when many messages are written while the pipe is backed up (e.g. bulk `sendToolListChanged` notifications, or a slow-starting child process that isn't reading stdin yet). `StdioClientTransport.send()` now also rejects on stdin errors instead of waiting forever. diff --git a/packages/client/src/client/stdio.ts b/packages/client/src/client/stdio.ts index a4664e1c93..aafc3c0dd4 100644 --- a/packages/client/src/client/stdio.ts +++ b/packages/client/src/client/stdio.ts @@ -4,7 +4,7 @@ import type { Stream } from 'node:stream'; import { PassThrough } from 'node:stream'; import type { JSONRPCMessage, Transport } from '@modelcontextprotocol/core-internal'; -import { ReadBuffer, SdkError, SdkErrorCode, serializeMessage } from '@modelcontextprotocol/core-internal'; +import { DrainWait, ReadBuffer, SdkError, SdkErrorCode, serializeMessage } from '@modelcontextprotocol/core-internal'; import spawn from 'cross-spawn'; export type StdioServerParameters = { @@ -103,6 +103,7 @@ export class StdioClientTransport implements Transport { private _readBuffer: ReadBuffer; private _serverParams: StdioServerParameters; private _stderrStream: PassThrough | null = null; + private _drainWait = new DrainWait(); onclose?: () => void; onerror?: (error: Error) => void; @@ -313,16 +314,19 @@ export class StdioClientTransport implements Transport { } send(message: JSONRPCMessage): Promise { - return new Promise(resolve => { + return new Promise((resolve, reject) => { if (!this._process?.stdin) { - throw new SdkError(SdkErrorCode.NotConnected, 'Not connected'); + reject(new SdkError(SdkErrorCode.NotConnected, 'Not connected')); + return; } const json = serializeMessage(message); if (this._process.stdin.write(json)) { resolve(); } else { - this._process.stdin.once('drain', resolve); + // Backpressure: wait on a shared drain so concurrent writes + // don't stack up 'drain' listeners on the pipe. + this._drainWait.wait(this._process.stdin).then(resolve, reject); } }); } diff --git a/packages/client/test/client/stdioBackpressure.test.ts b/packages/client/test/client/stdioBackpressure.test.ts new file mode 100644 index 0000000000..e0880a1521 --- /dev/null +++ b/packages/client/test/client/stdioBackpressure.test.ts @@ -0,0 +1,80 @@ +import { EventEmitter } from 'node:events'; + +import type { ChildProcess } from 'node:child_process'; +import spawn from 'cross-spawn'; +import type { Mock, MockedFunction } from 'vitest'; + +import { StdioClientTransport } from '../../src/client/stdio'; + +// mock cross-spawn +vi.mock('cross-spawn'); +const mockSpawn = spawn as unknown as MockedFunction; + +describe('StdioClientTransport backpressure', () => { + let stdin: EventEmitter & { write: Mock }; + + beforeEach(() => { + stdin = new EventEmitter() as EventEmitter & { write: Mock }; + stdin.write = vi.fn().mockReturnValue(false); + + mockSpawn.mockImplementation(() => { + const mockProcess = { + on: vi.fn((event: string, callback: () => void) => { + if (event === 'spawn') { + callback(); + } + return mockProcess; + }), + stdin, + stdout: { on: vi.fn() }, + stderr: null + }; + return mockProcess as unknown as ChildProcess; + }); + }); + + afterEach(() => { + vi.clearAllMocks(); + }); + + test('shares a single drain listener across concurrent backpressured sends', async () => { + const transport = new StdioClientTransport({ command: 'test-command' }); + await transport.start(); + + const messages = Array.from({ length: 15 }, (_, i) => ({ + jsonrpc: '2.0' as const, + id: i, + method: 'ping' + })); + + const sends = Promise.allSettled(messages.map(m => transport.send(m))); + await new Promise(resolve => setImmediate(resolve)); + + // every message is written immediately, but the backed-up pipe means + // all sends wait for drain - and they must share ONE listener + expect(stdin.write).toHaveBeenCalledTimes(15); + expect(stdin.listenerCount('drain')).toBe(1); + + stdin.emit('drain'); + + const results = await sends; + expect(results.every(r => r.status === 'fulfilled')).toBe(true); + expect(stdin.listenerCount('drain')).toBe(0); + }); + + test('rejects pending sends when stdin errors instead of hanging', async () => { + const transport = new StdioClientTransport({ command: 'test-command' }); + await transport.start(); + + const sends = Promise.allSettled([ + transport.send({ jsonrpc: '2.0', id: 1, method: 'ping' }), + transport.send({ jsonrpc: '2.0', id: 2, method: 'ping' }) + ]); + await new Promise(resolve => setImmediate(resolve)); + + stdin.emit('error', new Error('EPIPE')); + + const results = await sends; + expect(results.every(r => r.status === 'rejected')).toBe(true); + }); +}); diff --git a/packages/core-internal/src/shared/stdio.ts b/packages/core-internal/src/shared/stdio.ts index 8bd794b87b..3ee531bbc5 100644 --- a/packages/core-internal/src/shared/stdio.ts +++ b/packages/core-internal/src/shared/stdio.ts @@ -1,3 +1,5 @@ +import type { Writable } from 'node:stream'; + import type { JSONRPCMessage } from '../types/index'; import { JSONRPCMessageSchema } from '../types/index'; @@ -60,3 +62,45 @@ export function deserializeMessage(line: string): JSONRPCMessage { export function serializeMessage(message: JSONRPCMessage): string { return JSON.stringify(message) + '\n'; } + +/** + * Shared backpressure wait for a writable stream. + * + * Registers at most one 'drain' listener at a time no matter how many writes + * are waiting for the stream to drain. Node and Bun emit + * MaxListenersExceededWarning once more than 10 listeners pile up on a single + * event, which previously happened whenever several messages were written + * while the pipe was backed up (e.g. a slow-starting child process, or bulk + * notifications like sendToolListChanged). + */ +export class DrainWait { + private _pending: Promise | null = null; + + /** + * Returns a promise that resolves when the stream emits 'drain'. All + * callers that overlap share one listener and one promise. Rejects if the + * stream emits 'error' before draining. + */ + wait(stream: Writable): Promise { + if (!this._pending) { + this._pending = new Promise((resolve, reject) => { + const onDrain = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const cleanup = () => { + stream.off('drain', onDrain); + stream.off('error', onError); + this._pending = null; + }; + stream.once('drain', onDrain); + stream.once('error', onError); + }); + } + return this._pending; + } +} diff --git a/packages/server/src/server/stdio.ts b/packages/server/src/server/stdio.ts index e8fda03257..3e52f06ef7 100644 --- a/packages/server/src/server/stdio.ts +++ b/packages/server/src/server/stdio.ts @@ -1,7 +1,7 @@ import type { Readable, Writable } from 'node:stream'; import type { JSONRPCMessage, Transport } from '@modelcontextprotocol/core-internal'; -import { ReadBuffer, serializeMessage } from '@modelcontextprotocol/core-internal'; +import { DrainWait, ReadBuffer, serializeMessage } from '@modelcontextprotocol/core-internal'; import { process } from '@modelcontextprotocol/server/_shims'; /** @@ -20,6 +20,7 @@ export class StdioServerTransport implements Transport { private _readBuffer: ReadBuffer; private _started = false; private _closed = false; + private _drainWait = new DrainWait(); constructor( private _stdin: Readable = process.stdin, @@ -123,31 +124,12 @@ export class StdioServerTransport implements Transport { return new Promise((resolve, reject) => { const json = serializeMessage(message); - let settled = false; - const onError = (error: Error) => { - if (settled) return; - settled = true; - this._stdout.off('error', onError); - this._stdout.off('drain', onDrain); - reject(error); - }; - const onDrain = () => { - if (settled) return; - settled = true; - this._stdout.off('error', onError); - this._stdout.off('drain', onDrain); - resolve(); - }; - - this._stdout.once('error', onError); - if (this._stdout.write(json)) { - if (settled) return; - settled = true; - this._stdout.off('error', onError); resolve(); - } else if (!settled) { - this._stdout.once('drain', onDrain); + } else { + // Backpressure: wait on a shared drain so concurrent writes + // don't stack up 'drain' listeners on stdout. + this._drainWait.wait(this._stdout).then(resolve, reject); } }); } diff --git a/packages/server/test/server/stdio.test.ts b/packages/server/test/server/stdio.test.ts index fe79e3679c..f01fda86b1 100644 --- a/packages/server/test/server/stdio.test.ts +++ b/packages/server/test/server/stdio.test.ts @@ -227,3 +227,35 @@ test('should fire onerror and close when ReadBuffer overflows', async () => { expect(receivedError?.message).toMatch(/ReadBuffer exceeded maximum size/); expect(closeCount).toBe(1); }); + +test('shares a single drain listener across concurrent backpressured sends', async () => { + const backedUpOutput = new Writable({ + highWaterMark: 1, + write(_chunk, _encoding, _callback) { + // never invoke the callback so the stream stays backed up + } + }); + const server = new StdioServerTransport(input, backedUpOutput); + await server.start(); + + const messages = Array.from({ length: 15 }, (_, i) => ({ + jsonrpc: '2.0' as const, + id: i, + method: 'ping' + })); + + const sends = Promise.allSettled(messages.map(m => server.send(m))); + await new Promise(resolve => setImmediate(resolve)); + + // all writes buffered, but the sends must share ONE drain listener + expect(backedUpOutput.listenerCount('drain')).toBe(1); + + backedUpOutput.emit('drain'); + + const results = await sends; + expect(results.every(r => r.status === 'fulfilled')).toBe(true); + expect(backedUpOutput.listenerCount('drain')).toBe(0); + + backedUpOutput.destroy(); + await server.close(); +}); From 58ca732777b4820388938c8ab157086d806f737c Mon Sep 17 00:00:00 2001 From: Gursimran Singh Date: Tue, 25 Aug 2026 06:40:26 +0530 Subject: [PATCH 2/2] fix(stdio): scope drain waits to their stream and handle close --- .../test/client/stdioBackpressure.test.ts | 49 ++++++++++++++ packages/core-internal/src/shared/stdio.ts | 65 +++++++++++++------ 2 files changed, 94 insertions(+), 20 deletions(-) diff --git a/packages/client/test/client/stdioBackpressure.test.ts b/packages/client/test/client/stdioBackpressure.test.ts index e0880a1521..ca46d8b31f 100644 --- a/packages/client/test/client/stdioBackpressure.test.ts +++ b/packages/client/test/client/stdioBackpressure.test.ts @@ -1,6 +1,8 @@ import { EventEmitter } from 'node:events'; +import { PassThrough } from 'node:stream'; import type { ChildProcess } from 'node:child_process'; +import { DrainWait } from '@modelcontextprotocol/core-internal'; import spawn from 'cross-spawn'; import type { Mock, MockedFunction } from 'vitest'; @@ -78,3 +80,50 @@ describe('StdioClientTransport backpressure', () => { expect(results.every(r => r.status === 'rejected')).toBe(true); }); }); + +describe('DrainWait stream lifecycle', () => { + test('rejects pending waits when the stream closes instead of draining', async () => { + const stream = new PassThrough({ highWaterMark: 1 }); + expect(stream.write('payload')).toBe(false); + + const wait = new DrainWait(); + const pending = wait.wait(stream); + + stream.destroy(); // no error argument -> emits 'close', never 'error' + + await expect(pending).rejects.toThrow('closed before it drained'); + expect(stream.listenerCount('drain')).toBe(0); + }); + + test('does not reuse a wait bound to a different stream', async () => { + const a = new PassThrough({ highWaterMark: 1 }); + const b = new PassThrough({ highWaterMark: 1 }); + expect(a.write('payload')).toBe(false); + expect(b.write('payload')).toBe(false); + + const wait = new DrainWait(); + const onA = wait.wait(a); + const onB = wait.wait(b); + + expect(a.listenerCount('drain')).toBe(1); + expect(b.listenerCount('drain')).toBe(1); + + // draining stream a must not resolve the wait on stream b + a.read(); + await expect(onA).resolves.toBeUndefined(); + await expect(Promise.race([onB, sleep(50).then(() => 'pending')])).resolves.toBe('pending'); + + b.read(); + await expect(onB).resolves.toBeUndefined(); + }); + + test('rejects immediately when the stream is already destroyed', async () => { + const stream = new PassThrough({ highWaterMark: 1 }); + stream.destroy(); + + const wait = new DrainWait(); + await expect(wait.wait(stream)).rejects.toThrow('already destroyed'); + }); +}); + +const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); diff --git a/packages/core-internal/src/shared/stdio.ts b/packages/core-internal/src/shared/stdio.ts index 3ee531bbc5..fc83dd0148 100644 --- a/packages/core-internal/src/shared/stdio.ts +++ b/packages/core-internal/src/shared/stdio.ts @@ -75,32 +75,57 @@ export function serializeMessage(message: JSONRPCMessage): string { */ export class DrainWait { private _pending: Promise | null = null; + private _stream: Writable | null = null; /** * Returns a promise that resolves when the stream emits 'drain'. All - * callers that overlap share one listener and one promise. Rejects if the - * stream emits 'error' before draining. + * callers that overlap on the same stream share one listener and one + * promise. Rejects if the stream emits 'error' or 'close' before + * draining. */ wait(stream: Writable): Promise { - if (!this._pending) { - this._pending = new Promise((resolve, reject) => { - const onDrain = () => { - cleanup(); - resolve(); - }; - const onError = (error: Error) => { - cleanup(); - reject(error); - }; - const cleanup = () => { - stream.off('drain', onDrain); - stream.off('error', onError); - this._pending = null; - }; - stream.once('drain', onDrain); - stream.once('error', onError); - }); + // The cached wait is only valid for the stream it was created for. + // A transport can wrap a new stream after a close/start cycle, and a + // promise bound to a dead stream must not resolve a later send. + if (this._pending && this._stream === stream) { + return this._pending; + } + if (stream.destroyed) { + return Promise.reject(new Error('Cannot wait for drain: stream is already destroyed')); } + + this._stream = stream; + this._pending = new Promise((resolve, reject) => { + const onDrain = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + // destroy() without an error emits 'close', never 'error' or + // 'drain', so waiters would otherwise hang forever. + const onClose = () => { + cleanup(); + reject(new Error('Stream closed before it drained')); + }; + const cleanup = () => { + stream.off('drain', onDrain); + stream.off('error', onError); + stream.off('close', onClose); + // clear before waiters resume in their microtasks, so a caller + // that writes again in the same turn gets a fresh listener + // rather than a settled promise + if (this._stream === stream) { + this._pending = null; + this._stream = null; + } + }; + stream.once('drain', onDrain); + stream.once('error', onError); + stream.once('close', onClose); + }); return this._pending; } }