diff --git a/.changeset/stdio-server-stdin-eof-close.md b/.changeset/stdio-server-stdin-eof-close.md new file mode 100644 index 0000000000..e62f046293 --- /dev/null +++ b/.changeset/stdio-server-stdin-eof-close.md @@ -0,0 +1,7 @@ +--- +'@modelcontextprotocol/sdk': patch +--- + +`StdioServerTransport` now closes itself and fires `onclose` when its stdin ends or closes, so a server whose client hangs up tears down through `onclose` and can exit, instead of running on with no parent. A stdin that had already ended or been destroyed before `start()` is +treated the same way. `close()` is idempotent, so `onclose` fires once when `close()` is also called. Requests still in flight when stdin ends are aborted and not answered, and a notification sent after that rejects with `Not connected`; a client that expects responses keeps +stdin open until it has read them. An `onclose` that calls `process.exit()` can cut off output that is still being written. diff --git a/src/__fixtures__/serverWithKeepAlive.ts b/src/__fixtures__/serverWithKeepAlive.ts new file mode 100644 index 0000000000..83fd815103 --- /dev/null +++ b/src/__fixtures__/serverWithKeepAlive.ts @@ -0,0 +1,21 @@ +import { setInterval } from 'node:timers'; +import { McpServer } from '../server/mcp.js'; +import { StdioServerTransport } from '../server/stdio.js'; + +const transport = new StdioServerTransport(); + +const server = new McpServer({ + name: 'server-with-keep-alive', + version: '1.0.0' +}); + +await server.connect(transport); + +// Simulates a real server holding a keep-alive handle (connection pool, file +// watcher, heartbeat timer, ...). A well-behaved server releases its handles +// when the connection closes, and the process then exits naturally. +const keepAlive = setInterval(() => {}, 60_000); + +server.server.onclose = () => { + clearInterval(keepAlive); +}; diff --git a/src/server/stdio.ts b/src/server/stdio.ts index ced465bb81..1ebce2c6a9 100644 --- a/src/server/stdio.ts +++ b/src/server/stdio.ts @@ -8,10 +8,16 @@ import { Transport } from '../shared/transport.js'; * Server transport for stdio: this communicates with an MCP client by reading from the current process' stdin and writing to stdout. * * This transport is only available in Node.js environments. + * + * When the client closes its end of the pipe (stdin reaches end-of-file), the transport + * closes itself and fires `onclose`. A server that holds no other keep-alive handles will + * then exit naturally. Requests still in flight when stdin ends are not answered; a client + * that expects responses keeps stdin open until it has read them. */ export class StdioServerTransport implements Transport { private _readBuffer: ReadBuffer; private _started = false; + private _closed = false; constructor( private _stdin: Readable = process.stdin, @@ -46,6 +52,16 @@ export class StdioServerTransport implements Transport { _onerror = (error: Error) => { this.onerror?.(error); }; + _onstdinclose = () => { + // stdin EOF means the client hung up and nothing more can arrive. + const open = !this._closed && !this._stdin.destroyed; + this.close().catch(() => {}); + // A socket can still fail a pending write after 'end': keep reporting errors until it closes. + if (open) { + this._stdin.on('error', this._onerror); + this._stdin.once('close', () => this._stdin.off('error', this._onerror)); + } + }; /** * Starts listening for messages on stdin. @@ -58,8 +74,15 @@ export class StdioServerTransport implements Transport { } this._started = true; + // A stream that already ended or was destroyed will never emit 'end'/'close' again. + // Deferred so an onclose assigned right after connect() still sees it. + if (this._stdin.readableEnded || this._stdin.destroyed) { + setImmediate(this._onstdinclose); + } this._stdin.on('data', this._ondata); this._stdin.on('error', this._onerror); + this._stdin.on('end', this._onstdinclose); + this._stdin.on('close', this._onstdinclose); } private processReadBuffer() { @@ -78,9 +101,16 @@ export class StdioServerTransport implements Transport { } async close(): Promise { + if (this._closed) { + return; + } + this._closed = true; + // Remove our event listeners first this._stdin.off('data', this._ondata); this._stdin.off('error', this._onerror); + this._stdin.off('end', this._onstdinclose); + this._stdin.off('close', this._onstdinclose); // Check if we were the only data listener const remainingDataListeners = this._stdin.listenerCount('data'); diff --git a/test/integration-tests/processCleanup.test.ts b/test/integration-tests/processCleanup.test.ts index 11940697b9..4f0da5409c 100644 --- a/test/integration-tests/processCleanup.test.ts +++ b/test/integration-tests/processCleanup.test.ts @@ -1,3 +1,4 @@ +import { spawn } from 'node:child_process'; import path from 'node:path'; import { Readable, Writable } from 'node:stream'; import { Client } from '../../src/client/index.js'; @@ -76,6 +77,50 @@ describe('Process cleanup', () => { expect(onCloseWasCalled).toBe(1); }); + it('server process should exit on its own when the client closes stdin', async () => { + // Regression test for zombie stdio servers: when the client drops its end of + // the pipe (window closed, session restarted) without sending any signal, the + // server must notice stdin EOF, close its transport, and fire `onclose` so the + // server can release its keep-alive handles and the process exits naturally. + const child = spawn('node', ['--import', 'tsx', 'serverWithKeepAlive.ts'], { + cwd: FIXTURES_DIR, + stdio: ['pipe', 'pipe', 'inherit'] + }); + + const exited = new Promise(resolve => { + child.on('exit', code => resolve(code)); + }); + + // Confirm the server is up by completing an initialize round-trip over raw stdio. + const initialized = new Promise(resolve => { + child.stdout.on('data', () => resolve()); + }); + child.stdin.write( + JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { + protocolVersion: '2025-06-18', + capabilities: {}, + clientInfo: { name: 'stdin-eof-test', version: '1.0.0' } + } + }) + '\n' + ); + await initialized; + + // Hang up: close our end of the pipe without any signal. + child.stdin.end(); + + // The server must exit on its own — no SIGTERM/SIGKILL involved. Bound the + // wait so a regression fails the test instead of leaking the child. + const exitCode = await Promise.race([exited, new Promise<'zombie'>(resolve => setTimeout(() => resolve('zombie'), 8000))]); + if (exitCode === 'zombie') { + child.kill('SIGKILL'); + } + expect(exitCode).toBe(0); + }, 15_000); + it('should exit cleanly for a server that hangs', async () => { const client = new Client({ name: 'test-client', diff --git a/test/server/stdio.test.ts b/test/server/stdio.test.ts index 9613e0ec48..2433f6fd55 100644 --- a/test/server/stdio.test.ts +++ b/test/server/stdio.test.ts @@ -1,4 +1,4 @@ -import { Readable, Writable } from 'node:stream'; +import { Duplex, Readable, Writable } from 'node:stream'; import { ReadBuffer, serializeMessage } from '../../src/shared/stdio.js'; import { JSONRPCMessage } from '../../src/types.js'; import { StdioServerTransport } from '../../src/server/stdio.js'; @@ -148,3 +148,245 @@ test('should fire onerror and close when ReadBuffer overflows', async () => { expect(receivedError?.message).toMatch(/ReadBuffer exceeded maximum size/); expect(closeCount).toBe(1); }); + +test('should close and fire onclose when stdin ends (client hung up)', async () => { + // `autoDestroy: false, emitClose: false` so that pushing EOF emits only 'end', + // proving the 'end' listener works on its own (without relying on 'close'). + const endOnlyInput = new Readable({ read: () => {}, autoDestroy: false, emitClose: false }); + const server = new StdioServerTransport(endOnlyInput, output); + server.onerror = error => { + throw error; + }; + + let closeCount = 0; + const closed = new Promise(resolve => { + server.onclose = () => { + closeCount++; + resolve(); + }; + }); + + await server.start(); + endOnlyInput.push(null); // EOF — the client closed its end of the pipe + + await closed; + expect(closeCount).toBe(1); +}); + +test('should close and fire onclose when stdin closes', async () => { + const server = new StdioServerTransport(input, output); + server.onerror = error => { + throw error; + }; + + let closeCount = 0; + const closed = new Promise(resolve => { + server.onclose = () => { + closeCount++; + resolve(); + }; + }); + + await server.start(); + input.destroy(); // emits 'close' + + await closed; + expect(closeCount).toBe(1); +}); + +test('should fire onclose when stdin was already destroyed before start()', async () => { + // A custom stream (e.g. a net.Socket the peer reset during async setup) can + // be dead before start() runs: its one 'close' event already fired, so a + // listener registered in start() would never see it. + input.destroy(); + await new Promise(resolve => { + input.once('close', resolve); + }); + expect(input.destroyed).toBe(true); + + const server = new StdioServerTransport(input, output); + server.onerror = error => { + throw error; + }; + + let closeCount = 0; + const closed = new Promise(resolve => { + server.onclose = () => { + closeCount++; + resolve(); + }; + }); + + await server.start(); + + await closed; + expect(closeCount).toBe(1); +}); + +test('should fire onclose when stdin had already ended before start()', async () => { + // `autoDestroy: false, emitClose: false` so consuming the stream to EOF + // leaves it ended but NOT destroyed — its one 'end' event fired before + // start(), so only the `readableEnded` check can catch this shape. + const endedInput = new Readable({ read: () => {}, autoDestroy: false, emitClose: false }); + endedInput.push(null); // EOF + endedInput.resume(); // flow, so 'end' actually fires + await new Promise(resolve => { + endedInput.once('end', resolve); + }); + expect(endedInput.readableEnded).toBe(true); + expect(endedInput.destroyed).toBe(false); + + const server = new StdioServerTransport(endedInput, output); + server.onerror = error => { + throw error; + }; + + let closeCount = 0; + const closed = new Promise(resolve => { + server.onclose = () => { + closeCount++; + resolve(); + }; + }); + + await server.start(); + + await closed; + expect(closeCount).toBe(1); +}); + +test('should fire onclose assigned after start() when stdin was already dead', async () => { + // The synthetic hangup for a pre-dead stream must land after the caller's + // start()/connect() continuation — exactly like a real 'end' — otherwise an + // onclose assigned right after `await server.connect(transport)` never runs. + input.destroy(); + await new Promise(resolve => { + input.once('close', resolve); + }); + + const server = new StdioServerTransport(input, output); + server.onerror = error => { + throw error; + }; + + await server.start(); + + let closeCount = 0; + const closed = new Promise(resolve => { + server.onclose = () => { + closeCount++; + resolve(); + }; + }); + + await closed; + expect(closeCount).toBe(1); +}); + +test('should not fire onclose twice when close() is called after stdin ends', async () => { + const server = new StdioServerTransport(input, output); + server.onerror = error => { + throw error; + }; + + let closeCount = 0; + const closed = new Promise(resolve => { + server.onclose = () => { + closeCount++; + resolve(); + }; + }); + + await server.start(); + input.push(null); // EOF fires 'end', and stream teardown may fire 'close' too + + await closed; + await server.close(); + // Allow any late 'close' event from the stream teardown to be delivered + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(closeCount).toBe(1); +}); + +test('should remove its stdin end and close listeners on close', async () => { + // stdin defaults to the process-global process.stdin, so listeners left by a + // closed transport would pile up across transport lifecycles in one process. + const server = new StdioServerTransport(input, output); + await server.start(); + await server.close(); + + expect(input.listenerCount('end')).toBe(0); + expect(input.listenerCount('close')).toBe(0); +}); + +test('should still deliver messages that arrived before stdin ended', async () => { + const server = new StdioServerTransport(input, output); + server.onerror = error => { + throw error; + }; + + const messages: JSONRPCMessage[] = []; + const closed = new Promise(resolve => { + server.onclose = () => resolve(); + }); + server.onmessage = message => { + messages.push(message); + }; + + const message: JSONRPCMessage = { jsonrpc: '2.0', id: 1, method: 'ping' }; + input.push(serializeMessage(message)); + input.push(null); // EOF right behind the message + + await server.start(); + await closed; + + expect(messages).toEqual([message]); +}); + +test('should keep reporting stream errors after stdin ends, until the stream closes', async () => { + // One duplex for both directions, as with `new StdioServerTransport(socket, socket)`. + const socket = new Duplex({ + read() {}, + write(_chunk, _encoding, callback) { + callback(); + } + }); + const server = new StdioServerTransport(socket, socket); + const errors: Error[] = []; + server.onerror = error => errors.push(error); + const closed = new Promise(resolve => { + server.onclose = () => resolve(); + }); + + await server.start(); + socket.push(null); + await closed; + + // A pending write can still fail after 'end'; with no listener the process would crash. + const failure = new Error('write EPIPE'); + socket.emit('error', failure); + expect(errors).toEqual([failure]); + + socket.destroy(); + await new Promise(resolve => socket.once('close', resolve)); + expect(socket.listenerCount('error')).toBe(0); +}); + +test('should leave no listeners when close() follows start() on a stream that had already ended', async () => { + const endedInput = new Readable({ read: () => {}, autoDestroy: false, emitClose: false }); + endedInput.push(null); + endedInput.resume(); + await new Promise(resolve => { + endedInput.once('end', resolve); + }); + + const server = new StdioServerTransport(endedInput, output); + await server.start(); + await server.close(); + // start() deferred its own close by one turn; let it run. + await new Promise(resolve => setImmediate(resolve)); + + expect(endedInput.listenerCount('error')).toBe(0); + expect(endedInput.listenerCount('close')).toBe(0); + expect(endedInput.listenerCount('end')).toBe(0); +});