fix hanging stdio servers (#1200)

Co-authored-by: Johnny Santos <johnnyadsantos@gmail.com>
Co-authored-by: TomasRup <tomasr@wix.com>

Author: 3 people

src/__fixtures__/serverThatHangs.ts

@@ -0,0 +1,42 @@
+import { setInterval } from 'node:timers';
+import process from 'node:process';
+import { McpServer } from '../server/mcp.js';
+import { StdioServerTransport } from '../server/stdio.js';
+
+const transport = new StdioServerTransport();
+
+const server = new McpServer(
+    {
+        name: 'server-that-hangs',
+        title: 'Test Server that hangs',
+        version: '1.0.0'
+    },
+    {
+        capabilities: {
+            logging: {}
+        }
+    }
+);
+
+await server.connect(transport);
+
+// Keep process alive even after stdin closes
+const keepAlive = setInterval(() => {}, 60_000);
+
+// Prevent transport close from exiting
+transport.onclose = () => {
+    // Intentionally ignore - we want to test the signal handling
+};
+
+const doNotExitImmediately = async (signal: NodeJS.Signals) => {
+    await server.sendLoggingMessage({
+        level: 'debug',
+        data: `received signal ${signal}`
+    });
+    // Clear keepalive but delay exit to simulate slow shutdown
+    clearInterval(keepAlive);
+    setInterval(() => {}, 30_000);
+};
+
+process.on('SIGINT', doNotExitImmediately);
+process.on('SIGTERM', doNotExitImmediately);

src/__fixtures__/testServer.ts

@@ -0,0 +1,19 @@
+import { McpServer } from '../server/mcp.js';
+import { StdioServerTransport } from '../server/stdio.js';
+
+const transport = new StdioServerTransport();
+
+const server = new McpServer({
+    name: 'test-server',
+    version: '1.0.0'
+});
+
+await server.connect(transport);
+
+const exit = async () => {
+    await server.close();
+    process.exit(0);
+};
+
+process.on('SIGINT', exit);
+process.on('SIGTERM', exit);

src/shared/zodTestMatrix.ts renamed to src/__fixtures__/zodTestMatrix.ts

@@ -91,7 +91,6 @@ export function getDefaultEnvironment(): Record<string, string> {
  */
 export class StdioClientTransport implements Transport {
     private _process?: ChildProcess;
-    private _abortController: AbortController = new AbortController();
     private _readBuffer: ReadBuffer = new ReadBuffer();
     private _serverParams: StdioServerParameters;
     private _stderrStream: PassThrough | null = null;
@@ -126,18 +125,11 @@ export class StdioClientTransport implements Transport {
                 },
                 stdio: ['pipe', 'pipe', this._serverParams.stderr ?? 'inherit'],
                 shell: false,
-                signal: this._abortController.signal,
                 windowsHide: process.platform === 'win32' && isElectron(),
                 cwd: this._serverParams.cwd
             });
 
             this._process.on('error', error => {
-                if (error.name === 'AbortError') {
-                    // Expected when close() is called.
-                    this.onclose?.();
-                    return;
-                }
-
                 reject(error);
                 this.onerror?.(error);
             });
@@ -210,8 +202,43 @@ export class StdioClientTransport implements Transport {
     }
 
     async close(): Promise<void> {
-        this._abortController.abort();
-        this._process = undefined;
+        if (this._process) {
+            const processToClose = this._process;
+            this._process = undefined;
+
+            const closePromise = new Promise<void>(resolve => {
+                processToClose.once('close', () => {
+                    resolve();
+                });
+            });
+
+            try {
+                processToClose.stdin?.end();
+            } catch {
+                // ignore
+            }
+
+            await Promise.race([closePromise, new Promise(resolve => setTimeout(resolve, 2_000).unref())]);
+
+            if (processToClose.exitCode === null) {
+                try {
+                    processToClose.kill('SIGTERM');
+                } catch {
+                    // ignore
+                }
+
+                await Promise.race([closePromise, new Promise(resolve => setTimeout(resolve, 2_000).unref())]);
+            }
+
+            if (processToClose.exitCode === null) {
+                try {
+                    processToClose.kill('SIGKILL');
+                } catch {
+                    // ignore
+                }
+            }
+        }
+
         this._readBuffer.clear();
     }
 

src/client/stdio.ts

@@ -0,0 +1,113 @@
+import path from 'node:path';
+import { Readable, Writable } from 'node:stream';
+import { Client } from '../client/index.js';
+import { StdioClientTransport } from '../client/stdio.js';
+import { Server } from '../server/index.js';
+import { StdioServerTransport } from '../server/stdio.js';
+import { LoggingMessageNotificationSchema } from '../types.js';
+
+const FIXTURES_DIR = path.resolve(__dirname, '../__fixtures__');
+
+describe('Process cleanup', () => {
+    vi.setConfig({ testTimeout: 5000 }); // 5 second timeout
+
+    it('server should exit cleanly after closing transport', async () => {
+        const server = new Server(
+            {
+                name: 'test-server',
+                version: '1.0.0'
+            },
+            {
+                capabilities: {}
+            }
+        );
+
+        const mockReadable = new Readable({
+                read() {
+                    this.push(null); // signal EOF
+                }
+            }),
+            mockWritable = new Writable({
+                write(chunk, encoding, callback) {
+                    callback();
+                }
+            });
+
+        // Attach mock streams to process for the server transport
+        const transport = new StdioServerTransport(mockReadable, mockWritable);
+        await server.connect(transport);
+
+        // Close the transport
+        await transport.close();
+
+        // ensure a proper disposal mock streams
+        mockReadable.destroy();
+        mockWritable.destroy();
+
+        // If we reach here without hanging, the test passes
+        // The test runner will fail if the process hangs
+        expect(true).toBe(true);
+    });
+
+    it('onclose should be called exactly once', async () => {
+        const client = new Client({
+            name: 'test-client',
+            version: '1.0.0'
+        });
+
+        const transport = new StdioClientTransport({
+            command: 'node',
+            args: ['--import', 'tsx', 'testServer.ts'],
+            cwd: FIXTURES_DIR
+        });
+
+        await client.connect(transport);
+
+        let onCloseWasCalled = 0;
+        client.onclose = () => {
+            onCloseWasCalled++;
+        };
+
+        await client.close();
+
+        // A short delay to allow the close event to propagate
+        await new Promise(resolve => setTimeout(resolve, 50));
+
+        expect(onCloseWasCalled).toBe(1);
+    });
+
+    it('should exit cleanly for a server that hangs', async () => {
+        const client = new Client({
+            name: 'test-client',
+            version: '1.0.0'
+        });
+
+        const transport = new StdioClientTransport({
+            command: 'node',
+            args: ['--import', 'tsx', 'serverThatHangs.ts'],
+            cwd: FIXTURES_DIR
+        });
+
+        await client.connect(transport);
+        await client.setLoggingLevel('debug');
+        client.setNotificationHandler(LoggingMessageNotificationSchema, notification => {
+            console.debug('server log: ' + notification.params.data);
+        });
+        const serverPid = transport.pid!;
+
+        await client.close();
+
+        // A short delay to allow the close event to propagate
+        await new Promise(resolve => setTimeout(resolve, 50));
+
+        try {
+            process.kill(serverPid, 9);
+            throw new Error('Expected server to be dead but it is alive');
+        } catch (err: unknown) {
+            // 'ESRCH' the process doesn't exist
+            if (err && typeof err === 'object' && 'code' in err && err.code === 'ESRCH') {
+                // success
+            } else throw err;
+        }
+    });
+});

src/integration-tests/process-cleanup.test.ts

@@ -12,7 +12,7 @@ import {
     ListPromptsResultSchema,
     LATEST_PROTOCOL_VERSION
 } from '../types.js';
-import { zodTestMatrix, type ZodMatrixEntry } from '../shared/zodTestMatrix.js';
+import { zodTestMatrix, type ZodMatrixEntry } from '../__fixtures__/zodTestMatrix.js';
 
 describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
     const { z } = entry;

src/integration-tests/processCleanup.test.ts

@@ -7,7 +7,7 @@ import { McpServer } from '../server/mcp.js';
 import { StreamableHTTPServerTransport } from '../server/streamableHttp.js';
 import { CallToolResultSchema, LoggingMessageNotificationSchema } from '../types.js';
 import { InMemoryEventStore } from '../examples/shared/inMemoryEventStore.js';
-import { zodTestMatrix, type ZodMatrixEntry } from '../shared/zodTestMatrix.js';
+import { zodTestMatrix, type ZodMatrixEntry } from '../__fixtures__/zodTestMatrix.js';
 
 describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
     const { z } = entry;

src/integration-tests/stateManagementStreamableHttp.test.ts

@@ -1,5 +1,5 @@
 import { completable, getCompleter } from './completable.js';
-import { zodTestMatrix, type ZodMatrixEntry } from '../shared/zodTestMatrix.js';
+import { zodTestMatrix, type ZodMatrixEntry } from '../__fixtures__/zodTestMatrix.js';
 
 describe.each(zodTestMatrix)('completable with $zodVersionLabel', (entry: ZodMatrixEntry) => {
     const { z } = entry;

src/integration-tests/taskResumability.test.ts

@@ -22,7 +22,7 @@ import {
 import { completable } from './completable.js';
 import { McpServer, ResourceTemplate } from './mcp.js';
 import { InMemoryTaskStore } from '../experimental/tasks/stores/in-memory.js';
-import { zodTestMatrix, type ZodMatrixEntry } from '../shared/zodTestMatrix.js';
+import { zodTestMatrix, type ZodMatrixEntry } from '../__fixtures__/zodTestMatrix.js';
 
 function createLatch() {
     let latch = false;