fix hanging stdio servers (#1200)
Co-authored-by: Johnny Santos <johnnyadsantos@gmail.com> Co-authored-by: TomasRup <tomasr@wix.com>
This commit is contained in:
parent
5aaf6ee67e
commit
ace9b62d3c
15 changed files with 220 additions and 47 deletions
42
src/__fixtures__/serverThatHangs.ts
Normal file
42
src/__fixtures__/serverThatHangs.ts
Normal file
|
|
@ -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);
|
||||
19
src/__fixtures__/testServer.ts
Normal file
19
src/__fixtures__/testServer.ts
Normal file
|
|
@ -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);
|
||||
22
src/__fixtures__/zodTestMatrix.ts
Normal file
22
src/__fixtures__/zodTestMatrix.ts
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
import * as z3 from 'zod/v3';
|
||||
import * as z4 from 'zod/v4';
|
||||
|
||||
// Shared Zod namespace type that exposes the common surface area used in tests.
|
||||
export type ZNamespace = typeof z3 & typeof z4;
|
||||
|
||||
export const zodTestMatrix = [
|
||||
{
|
||||
zodVersionLabel: 'Zod v3',
|
||||
z: z3 as ZNamespace,
|
||||
isV3: true as const,
|
||||
isV4: false as const
|
||||
},
|
||||
{
|
||||
zodVersionLabel: 'Zod v4',
|
||||
z: z4 as ZNamespace,
|
||||
isV3: false as const,
|
||||
isV4: true as const
|
||||
}
|
||||
] as const;
|
||||
|
||||
export type ZodMatrixEntry = (typeof zodTestMatrix)[number];
|
||||
|
|
@ -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();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,28 +0,0 @@
|
|||
import { Server } from '../server/index.js';
|
||||
import { StdioServerTransport } from '../server/stdio.js';
|
||||
|
||||
describe('Process cleanup', () => {
|
||||
vi.setConfig({ testTimeout: 5000 }); // 5 second timeout
|
||||
|
||||
it('should exit cleanly after closing transport', async () => {
|
||||
const server = new Server(
|
||||
{
|
||||
name: 'test-server',
|
||||
version: '1.0.0'
|
||||
},
|
||||
{
|
||||
capabilities: {}
|
||||
}
|
||||
);
|
||||
|
||||
const transport = new StdioServerTransport();
|
||||
await server.connect(transport);
|
||||
|
||||
// Close the transport
|
||||
await transport.close();
|
||||
|
||||
// If we reach here without hanging, the test passes
|
||||
// The test runner will fail if the process hangs
|
||||
expect(true).toBe(true);
|
||||
});
|
||||
});
|
||||
113
src/integration-tests/processCleanup.test.ts
Normal file
113
src/integration-tests/processCleanup.test.ts
Normal file
|
|
@ -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;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import { McpServer } from './mcp.js';
|
|||
import { createServer, type Server } from 'node:http';
|
||||
import { AddressInfo } from 'node:net';
|
||||
import { CallToolResult, JSONRPCMessage } from '../types.js';
|
||||
import { zodTestMatrix, type ZodMatrixEntry } from '../shared/zodTestMatrix.js';
|
||||
import { zodTestMatrix, type ZodMatrixEntry } from '../__fixtures__/zodTestMatrix.js';
|
||||
|
||||
const createMockResponse = () => {
|
||||
const res = {
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import { EventStore, StreamableHTTPServerTransport, EventId, StreamId } from './
|
|||
import { McpServer } from './mcp.js';
|
||||
import { CallToolResult, JSONRPCMessage } from '../types.js';
|
||||
import { AuthInfo } from './auth/types.js';
|
||||
import { zodTestMatrix, type ZodMatrixEntry } from '../shared/zodTestMatrix.js';
|
||||
import { zodTestMatrix, type ZodMatrixEntry } from '../__fixtures__/zodTestMatrix.js';
|
||||
|
||||
async function getFreePort() {
|
||||
return new Promise(res => {
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import { Server } from './index.js';
|
|||
import { Client } from '../client/index.js';
|
||||
import { InMemoryTransport } from '../inMemory.js';
|
||||
import { McpServer, ResourceTemplate } from './mcp.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;
|
||||
|
|
|
|||
|
|
@ -5,5 +5,5 @@
|
|||
"moduleResolution": "node",
|
||||
"outDir": "./dist/cjs"
|
||||
},
|
||||
"exclude": ["**/*.test.ts", "src/__mocks__/**/*"]
|
||||
"exclude": ["**/*.test.ts", "src/__mocks__/**/*", "src/__fixtures__/**/*"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,5 +3,5 @@
|
|||
"compilerOptions": {
|
||||
"outDir": "./dist/esm"
|
||||
},
|
||||
"exclude": ["**/*.test.ts", "src/__mocks__/**/*", "src/server/zodTestMatrix.ts"]
|
||||
"exclude": ["**/*.test.ts", "src/__mocks__/**/*", "src/__fixtures__/**/*"]
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue