From f05ff04d98460d880b8ef0bae441ccfd8ad094e7 Mon Sep 17 00:00:00 2001 From: Berkant Acun Date: Sat, 12 Sep 2026 01:46:19 +0300 Subject: [PATCH] [v1.x] fix(client): drain piped stderr so unread pipe cannot deadlock v1.x backport of #2788. `StdioClientTransport` with `stderr: 'pipe'` (or `'overlapped'`) pipes child stderr into a PassThrough that stays paused until a reader attaches. A server that logs a normal amount fills that buffer and the OS pipe, blocks on write(2), stops reading stdin, and the session hangs with no error, no rejection and no timeout. After piping, the PassThrough is put in flowing mode so unread chunks drain; listeners attached before start() still receive every chunk, a late listener sees only post-attach data. Fixes #2776 on v1.x. --- src/client/stdio.ts | 16 ++++++ test/client/stdio.test.ts | 117 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+) diff --git a/src/client/stdio.ts b/src/client/stdio.ts index 816a686fce..bff8724eb7 100644 --- a/src/client/stdio.ts +++ b/src/client/stdio.ts @@ -28,6 +28,11 @@ export type StdioServerParameters = { * How to handle stderr of the child process. This matches the semantics of Node's `child_process.spawn`. * * The default is "inherit", meaning messages to stderr will be printed to the parent process's stderr. + * + * When set to "pipe" or "overlapped", stderr is exposed on `StdioClientTransport.stderr`. + * The SDK drains that stream so an unread pipe cannot fill and deadlock the session. + * Attach a `data` listener (or `.pipe()` it) before `start()` / `Client.connect` to + * receive every chunk; without a listener the bytes are discarded. */ stderr?: IOType | Stream | number; @@ -172,6 +177,10 @@ export class StdioClientTransport implements Transport { if (this._stderrStream && this._process.stderr) { this._process.stderr.pipe(this._stderrStream); + // Flowing mode discards unread chunks so a chatty child cannot + // fill the PassThrough (16 KiB highWaterMark) and block on + // write(2). Listeners attached before start() still receive data. + this._stderrStream.resume(); } }); } @@ -182,6 +191,13 @@ export class StdioClientTransport implements Transport { * If stderr piping was requested, a PassThrough stream is returned _immediately_, allowing callers to * attach listeners before the start method is invoked. This prevents loss of any early * error output emitted by the child process. + * + * After `start()`, the SDK puts this stream in flowing mode so piping is safe without a + * consumer — unread stderr is drained and cannot deadlock the session. A listener (or + * `.pipe()` destination) attached before `start()` still receives every chunk. A late + * listener sees only data written after it attached; bytes already drained are not + * replayed. The paused-mode API (`read()` in a loop) is not supported once `start()` + * has put the stream in flowing mode — use `data` events or `.pipe()`. */ get stderr(): Stream | null { if (this._stderrStream) { diff --git a/test/client/stdio.test.ts b/test/client/stdio.test.ts index 577e0c946f..0e2461e42d 100644 --- a/test/client/stdio.test.ts +++ b/test/client/stdio.test.ts @@ -1,4 +1,5 @@ import { JSONRPCMessage } from '../../src/types.js'; +import { Client } from '../../src/client/index.js'; import { StdioClientTransport, StdioServerParameters } from '../../src/client/stdio.js'; // Configure default server parameters based on OS @@ -116,3 +117,119 @@ test('should fire onerror and close when ReadBuffer overflows', async () => { expect(error.message).toMatch(/ReadBuffer exceeded maximum size/); await closed; }); + +/** Unique markers so late-attach tests can tell drained startup bytes from post-request bytes. */ +const STDERR_STARTUP_MARKER = 'STDERR_STARTUP_MARKER'; +const STDERR_POST_REQUEST_MARKER = 'STDERR_POST_REQUEST_MARKER'; + +/** + * Minimal MCP server that floods stderr on every post-handshake request. + * Enough writes to fill a paused PassThrough (16 KiB) and the OS pipe (~64 KiB) + * so an undrained `stderr: 'pipe'` would block the child on write(2). + */ +function chattyStderrServerScript(): string { + return String.raw` + const { createInterface } = require('readline'); + const send = (o) => process.stdout.write(JSON.stringify(o) + '\n'); + process.stderr.write(${JSON.stringify(`${STDERR_STARTUP_MARKER}\n`)}); + createInterface({ input: process.stdin }).on('line', (line) => { + const m = JSON.parse(line); + if (m.method === 'initialize') { + return send({ + jsonrpc: '2.0', + id: m.id, + result: { + protocolVersion: '2025-06-18', + capabilities: { tools: {} }, + serverInfo: { name: 'chatty', version: '1.0.0' } + } + }); + } + if (m.method === 'notifications/initialized') return; + for (let i = 0; i < 256; i++) process.stderr.write('x'.repeat(512) + '\n'); + process.stderr.write(${JSON.stringify(`${STDERR_POST_REQUEST_MARKER}\n`)}); + send({ + jsonrpc: '2.0', + id: m.id, + result: { tools: [{ name: 'x', description: 'd', inputSchema: { type: 'object' } }] } + }); + }); + `; +} + +function chattyStderrTransport(): StdioClientTransport { + return new StdioClientTransport({ + command: process.execPath, + args: ['-e', chattyStderrServerScript()], + stderr: 'pipe' + }); +} + +/** stdout can settle before every flowing-mode stderr chunk is delivered. */ +async function waitForStderrMarker(captured: { text: string }, marker: string): Promise { + await vi.waitFor( + () => { + if (!captured.text.includes(marker)) { + throw new Error(`stderr has not yet included ${marker}`); + } + }, + { timeout: 3000, interval: 10 } + ); +} + +test('piped stderr without a reader does not deadlock listTools', async () => { + const transport = chattyStderrTransport(); + const client = new Client({ name: 'demo', version: '1.0.0' }); + try { + await client.connect(transport); + const result = await client.listTools(undefined, { timeout: 5000 }); + expect(result.tools).toEqual([{ name: 'x', description: 'd', inputSchema: { type: 'object' } }]); + } finally { + await client.close(); + } +}, 8000); + +test('piped stderr listener attached before start still receives chunks', async () => { + const transport = chattyStderrTransport(); + const stderr = transport.stderr; + expect(stderr).not.toBeNull(); + const captured = { text: '' }; + stderr!.on('data', (chunk: Buffer) => { + captured.text += chunk.toString(); + }); + + const client = new Client({ name: 'demo', version: '1.0.0' }); + try { + await client.connect(transport); + await waitForStderrMarker(captured, STDERR_STARTUP_MARKER); + const result = await client.listTools(undefined, { timeout: 5000 }); + expect(result.tools).toHaveLength(1); + await waitForStderrMarker(captured, STDERR_POST_REQUEST_MARKER); + } finally { + await client.close(); + } +}, 8000); + +test('late stderr listener does not deadlock and sees only post-attach chunks', async () => { + const transport = chattyStderrTransport(); + const client = new Client({ name: 'demo', version: '1.0.0' }); + try { + await client.connect(transport); + // Flowing-mode drain of startup stderr needs a turn to settle before we attach. + await new Promise(resolve => setImmediate(resolve)); + + const stderr = transport.stderr; + expect(stderr).not.toBeNull(); + const captured = { text: '' }; + stderr!.on('data', (chunk: Buffer) => { + captured.text += chunk.toString(); + }); + + const result = await client.listTools(undefined, { timeout: 5000 }); + expect(result.tools).toHaveLength(1); + await waitForStderrMarker(captured, STDERR_POST_REQUEST_MARKER); + expect(captured.text.includes(STDERR_STARTUP_MARKER)).toBe(false); + } finally { + await client.close(); + } +}, 8000);