diff --git a/changelog.d/mcp-pipe-failed-while-connecting.fixed.md b/changelog.d/mcp-pipe-failed-while-connecting.fixed.md new file mode 100644 index 000000000..cc0eab3bb --- /dev/null +++ b/changelog.d/mcp-pipe-failed-while-connecting.fixed.md @@ -0,0 +1,21 @@ +- **A stdio MCP server whose pipe fails while it's connecting no longer reads + as connected, in both engines.** Such a connect now fails: one whose + transport found the child's stdin or stdout failed before the connection + was taken on, as when the modern engine's connect-time list-change + subscribe, or the legacy engine's answer to a ping the server sent during + the handshake, meets EPIPE. Before, unless the child had already exited, + the modern connect was taken on once that subscribe gave up, and the legacy + handshake took an initialize answer that came afterwards. The connection + then read as connected while refusing every call as not sent, and said the + host was reconnecting to it when reconnect was on, until the child exited; + for a child that outlived SIGKILL that never came, and nothing held or + reported the launch. The connect's error is now `MCP server "" stdio + pipe failed with while connecting` (modern) or `MCPL server "" + stdio pipe failed with during the handshake` (legacy). A child that + can't be reaped is held as for any failed connect: `connect()` throws + `McplUnreapedLaunchError`, and a reconnect's failure is a permanent + `reconnect-failed`. A modern connect that the SDK resolved after such a + failure had already ended the child, which read `MCP server "" closed + its connection while connecting`, now names the pipe too. A connect that + fails some other way after its pipe failed, such as one the SDK rejects, + keeps its words. diff --git a/src/mcpl/modern-connection.ts b/src/mcpl/modern-connection.ts index ceaa4c756..1be973377 100644 --- a/src/mcpl/modern-connection.ts +++ b/src/mcpl/modern-connection.ts @@ -357,11 +357,17 @@ class SpawnerStdioTransport implements Transport { line.on('unreaped', (failure: Error) => this.onUnreaped(`its stdio pipe failed with ${line.pipeFailure?.message ?? 'an error'}`, failure)); } + /** The first error on the child's stdin or stdout, once there is one: the + * line is ending the child then, and writes nothing more to it. */ + get pipeFailure(): Error | undefined { + return this.line?.pipeFailure; + } + async send(message: JSONRPCMessage): Promise { if (!this.line || this.line.closed) throw new NotWrittenError(`MCP server "${this.config.id}" stdio transport is closed`); // Once the child's stdin or stdout has failed, the line is ending it and // writes nothing more to it, so a message is refused here as not sent. - const failed = this.line.pipeFailure; + const failed = this.pipeFailure; if (failed) throw new NotWrittenError(`MCP server "${this.config.id}" stdio pipe failed with ${failed.message}`, true); this.line.writeLine(JSON.stringify(message)); } @@ -415,6 +421,8 @@ interface Session { readonly generation: number; readonly client: Client; readonly wire: ObservedTransport; + /** Its stdio transport, under `wire`; null over HTTP. */ + readonly stdio: SpawnerStdioTransport | null; /** Its transport has closed, whether or not the session was adopted yet. */ lost: boolean; } @@ -640,17 +648,19 @@ export class ModernMcpConnection extends EventEmitter { // reopen backs off from the base interval, not from the previous // generation's failures. this.relistenAttempts = 0; - const inner: Transport = this.transportKind === 'stdio' + const stdio = this.transportKind === 'stdio' ? new SpawnerStdioTransport( this.config, (line) => { if (this.isCurrent(generation)) this.deliverStderr(line); }, // A child whose pipe failed and that outlived SIGKILL: a loss whose // launch is already known unreaped, so it's held as the loss is // handled (handleLost). During a connect there is no session to lose - // yet: the connect's own cleanup holds it, once the connect fails. + // yet: a connect whose pipe has failed fails, whether or not the SDK + // resolves it (establish), and its cleanup holds the launch. (reason, failure) => { if (this.isCurrent(generation)) this.handleLost(generation, reason, failure); }, ) - : new StreamableHTTPClientTransport(new URL(this.config.url!), { authProvider: bearerAuth(this.config) }); + : null; + const inner: Transport = stdio ?? new StreamableHTTPClientTransport(new URL(this.config.url!), { authProvider: bearerAuth(this.config) }); const wire = new ObservedTransport(inner); const client = new Client( @@ -669,7 +679,7 @@ export class ModernMcpConnection extends EventEmitter { }, }, ); - const session: Session = { generation, client, wire, lost: false }; + const session: Session = { generation, client, wire, stdio, lost: false }; client.onerror = (error: Error) => { // The SDK reports late responses to deadline-abandoned calls here // ("unknown message ID"), and a list-change subscription it could not @@ -714,16 +724,23 @@ export class ModernMcpConnection extends EventEmitter { if (failure) this.holdUnreaped(session, failure); return; } - if (session.lost) { - // The transport closed while the client connected, such as a child - // that exited at the connect-time list-change subscribe: the SDK - // reports that subscribe's failure only as an error, and resolves the - // connect anyway (R1-2). Adopted, the session would read as connected - // with nothing behind it, and no later close would come to reconnect - // it. So the connect failed. + // The transport closed while the client connected, such as a child that + // exited at the connect-time list-change subscribe, or the child's stdio + // pipe failed, such as that subscribe's write meeting EPIPE: the SDK + // reports the subscribe's failure only as an error, and resolves the + // connect anyway (R1-2). Adopted, the session would read as connected + // with nothing it can write to behind it, and from a child that outlived + // SIGKILL, whose 'unreaped' came while there was no session to lose, no + // later close would come to reconnect it. So the connect failed: its + // close reaps the child, and a child it can't reap is held. A failed + // pipe is named whether or not the child has exited by now. + const pipe = session.stdio?.pipeFailure; + if (session.lost || pipe) { const failure = await ModernMcpConnection.closeSession(session); if (failure) this.holdUnreaped(session, failure); - throw new Error(`MCP server "${this.id}" closed its connection while connecting`); + throw new Error(pipe + ? `MCP server "${this.id}" stdio pipe failed with ${pipe.message} while connecting` + : `MCP server "${this.id}" closed its connection while connecting`); } this.session = session; this.protocolVersion = MODERN_MCP_PROTOCOL_VERSION; diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 0e96ac352..059f10779 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -1194,21 +1194,22 @@ export class McplServerConnection extends EventEmitter { }, }; + // When the transport fails, the end of the server's stderr often says + // why, such as a missing token or a crash at startup. It goes on the + // error as `stderrTail`, never into the message: an error's message + // travels to traces and tool results, and a server's stderr is its own + // text, which can hold a credential. + const failed = (failure: string): Error & { stderrTail?: string } => { + // It goes to the host too, so it's redacted, from a window of the + // stream longer than the tail, so a credential the tail's cut + // would split is whole when it's matched (stderrTailRedacted). + const tail = transport.stderrTailRedacted(serverTextRedactor(config)).trimEnd(); + return Object.assign(new Error(failure), tail ? { stderrTail: tail } : {}); + }; + // Await the initialize response, racing an early transport close/error, // a timeout and the owner's close. Every path cleans up its listeners. const { mcpl: rawCapabilities, mcpToolsAdvertised, protocolVersion } = await new Promise<{ mcpl: McplCapabilities | null; mcpToolsAdvertised: boolean; protocolVersion: string }>((resolve, reject) => { - // When the transport fails, the end of the server's stderr often says - // why, such as a missing token or a crash at startup. It goes on the - // error as `stderrTail`, never into the message: an error's message - // travels to traces and tool results, and a server's stderr is its - // own text, which can hold a credential. - const failed = (failure: string): Error & { stderrTail?: string } => { - // It goes to the host too, so it's redacted, from a window of the - // stream longer than the tail, so a credential the tail's cut - // would split is whole when it's matched (stderrTailRedacted). - const tail = transport.stderrTailRedacted(serverTextRedactor(config)).trimEnd(); - return Object.assign(new Error(failure), tail ? { stderrTail: tail } : {}); - }; const cleanup = () => { transport.off('line', onLine); transport.off('close', onClose); @@ -1313,6 +1314,22 @@ export class McplServerConnection extends EventEmitter { transport.writeLine(JSON.stringify(initRequest)); }); + // The transport has found the child's stdin or stdout failed before the + // handshake could hand it on, such as when this host's answer to a ping + // the server sent met EPIPE. Nothing more is written to the child then, + // so the connection couldn't serve. It's read here, once the answer has + // passed its own checks, rather than as the answer arrives: a write's + // EPIPE is reported only once the callback that wrote it has returned, + // so a ping and the answer read in one callback would pass a check made + // there. A child that outlived SIGKILL may already have said so with + // nothing listening (the transport's 'unreaped' is wired only once the + // handshake hands it on). So the handshake fails, and its cleanup holds + // the launch if it can't reap the child. A failure of the write below + // is reported only once the caller has wired the transport, so it + // reaches the connection as a loss, like any later one. + const pipeFailure = transport.pipeFailure; + if (pipeFailure) throw failed(`MCPL server "${config.id}" stdio pipe failed with ${pipeFailure.message} during the handshake`); + // Send `initialized` notification (no id) transport.writeLine(JSON.stringify({ jsonrpc: '2.0', method: 'notifications/initialized' })); @@ -2606,8 +2623,10 @@ export class McplServerConnection extends EventEmitter { // reconnecting, and is lost as on a close, with the permanent failure // that raises the alert and takes its tools off the list. Its id stays // held until close() confirms the exit. A close the host has begun - // reports the failure itself. During a handshake nothing listens yet: - // the handshake's own cleanup holds the launch, once the handshake fails. + // reports the failure itself. During a handshake nothing listens yet: a + // handshake whose pipe has failed fails, when its answer comes (it's + // refused then), when the child's exit closes the transport, or at its + // timeout, and its own cleanup holds the launch. transport.on('unreaped', (failure: Error) => { if (this.closed || this.transport !== transport) return; this.holdUnreaped(transport, failure); diff --git a/test/fixtures/legacy-version-server.mjs b/test/fixtures/legacy-version-server.mjs index 7a12ee592..62ff80c18 100644 --- a/test/fixtures/legacy-version-server.mjs +++ b/test/fixtures/legacy-version-server.mjs @@ -11,6 +11,12 @@ // error a JSON-RPC error (-32603), not a version verdict // hang never answers // slow `ok`, after SLOW_INIT_MS (default 600) +// deaf-ping at initialize, closes its stdin and says so on stderr, then +// pings the client, whose answer fails with EPIPE; answers +// `ok` once it gets SIGUSR2, and stays up +// deaf-ping-answered as deaf-ping, but sends nothing until SIGUSR2, then +// writes the ping and the `ok` answer together, so the client +// reads both in one callback // ok-then- `ok` on the first start recorded in startLog, after // // IGNORE_SIGTERM=1 makes the process survive SIGTERM (only SIGKILL ends it). @@ -21,7 +27,7 @@ // nullcursor (nextCursor null), endless (a new cursor on every page, so the // list never ends), badlater (a well-formed first page, then a page with no // tools array). Tool `die` exits the process. -import { appendFileSync, existsSync, readFileSync } from 'node:fs'; +import { appendFileSync, closeSync, existsSync, openSync, readFileSync } from 'node:fs'; let mode = process.argv[2] ?? 'ok'; const startLog = process.argv[3]; @@ -59,6 +65,31 @@ process.stdin.on('data', (chunk) => { else if (mode === 'error') error(m.id, -32603, 'initialize failed on purpose'); else if (mode === 'hang') { /* never answers */ } else if (mode === 'slow') setTimeout(() => reply(m.id, { protocolVersion: '2024-11-05', capabilities: { tools: {} } }), Number(process.env.SLOW_INIT_MS ?? 600)); + else if (mode === 'deaf-ping' || mode === 'deaf-ping-answered') { + const ok = { jsonrpc: '2.0', id: m.id, result: { protocolVersion: '2024-11-05', capabilities: { tools: {} } } }; + const ping = { jsonrpc: '2.0', id: 'server-ping', method: 'ping' }; + process.stdin.removeAllListeners('data'); + process.stdin.removeAllListeners('end'); + // Node keeps a stdio descriptor open when its stream is destroyed, so + // the read end is closed by hand once the stream is done with it, and + // /dev/null takes fd 0 at once, as libuv assumes fds 0 to 2 stay taken + // (see deafen() in modern-raw-server.mjs). + process.stdin.once('close', () => { + try { closeSync(0); } catch { /* already closed */ } + openSync('/dev/null', 'r'); + // SIGUSR2's handler is in place before the stderr line, which a test + // may take as its cue to send it: until then the signal's default + // action would end the process. + process.once('SIGUSR2', mode === 'deaf-ping-answered' + ? () => process.stdout.write(`${JSON.stringify(ping)}\n${JSON.stringify(ok)}\n`) + : () => send(ok)); + process.stderr.write('deaf-ping: stdin closed\n'); + if (mode === 'deaf-ping') send(ping); + }); + process.stdin.destroy(); + setInterval(() => {}, 1 << 30); + return; + } else reply(m.id, { protocolVersion: '2024-11-05', capabilities: { tools: {} } }); } else if (m.method === 'tools/list') { const cursor = m.params?.cursor; diff --git a/test/fixtures/modern-raw-server.mjs b/test/fixtures/modern-raw-server.mjs index 9724a4fbf..e710d830c 100644 --- a/test/fixtures/modern-raw-server.mjs +++ b/test/fixtures/modern-raw-server.mjs @@ -12,6 +12,9 @@ // reject-discover-later launches after the first answer server/discover with an error // reject-discover every launch answers server/discover with an error // ignore-sigterm the process survives SIGTERM (only SIGKILL ends it) +// deaf-at-discover as server/discover arrives, closes its stdin and only then +// answers, so the client's next write (the connect-time +// subscriptions/listen) fails with EPIPE; it stays up // other-revision every launch's server/discover lists only a revision the client doesn't speak // other-revision-later launches after the first do // Tools: op (outputSchema {value: integer}, answers with a string: invalid @@ -26,7 +29,7 @@ // vanish (exits without answering), deafen (answers, then closes its stdin and // stays up, so the next write to it fails with EPIPE). import { createInterface } from 'node:readline'; -import { appendFileSync, closeSync, existsSync, readFileSync } from 'node:fs'; +import { appendFileSync, closeSync, existsSync, openSync, readFileSync } from 'node:fs'; const [flagList = '', log] = process.argv.slice(2); const flags = new Set(flagList.split(',').filter(Boolean)); @@ -44,6 +47,34 @@ const SUB = 'io.modelcontextprotocol/subscriptionId'; let listens = 0; let subscription = null; +const discovered = (id) => result(id, { + supportedVersions: ['2026-07-28'], + capabilities: { tools: { listChanged: true } }, + ttlMs: 60000, + cacheScope: 'private', + _meta: { 'io.modelcontextprotocol/serverInfo': { name: 'raw-fixture', version: '1' } }, +}); + +/** Close stdin and stay up, so the next write to this process fails with + * EPIPE; `then` runs once the read end is closed. */ +function deafen(then = () => {}) { + // Node keeps a stdio descriptor open when its stream is destroyed, so + // the read end is closed by hand once the stream is done with it, and + // /dev/null takes fd 0 at once: libuv assumes fds 0 to 2 stay taken, and + // on macOS this sequence, with fd 0 left free, was aborted at its first + // stdout write by libuv's assertion `fd > STDERR_FILENO`. + process.stdin.once('close', () => { + try { closeSync(0); } catch { /* already closed */ } + openSync('/dev/null', 'r'); + note({ event: 'deaf' }); + then(); + }); + lines.close(); + process.stdin.destroy(); + // Nothing else keeps the process up once its stdin is gone. + setInterval(() => {}, 1 << 30); +} + const lines = createInterface({ input: process.stdin }); lines.on('line', (line) => { let m; @@ -64,13 +95,8 @@ lines.on('line', (line) => { if (flags.has('hang-discover')) return; if (flags.has('reject-discover-later') && launch > 1) return error(m.id, -32603, 'not today'); if (flags.has('reject-discover')) return error(m.id, -32603, 'not today'); - return result(m.id, { - supportedVersions: ['2026-07-28'], - capabilities: { tools: { listChanged: true } }, - ttlMs: 60000, - cacheScope: 'private', - _meta: { 'io.modelcontextprotocol/serverInfo': { name: 'raw-fixture', version: '1' } }, - }); + if (flags.has('deaf-at-discover')) return deafen(() => discovered(m.id)); + return discovered(m.id); case 'subscriptions/listen': if (flags.has('exit-on-listen-first') && launch === 1) process.exit(0); listens++; @@ -122,16 +148,7 @@ lines.on('line', (line) => { if (name === 'vanish') process.exit(6); if (name === 'deafen') { result(m.id, { content: [{ type: 'text', text: 'deaf now' }] }); - // Node keeps a stdio descriptor open when its stream is destroyed, so - // the read end is closed by hand once the stream is done with it. - process.stdin.once('close', () => { - try { closeSync(0); } catch { /* already closed */ } - note({ event: 'deaf' }); - }); - lines.close(); - process.stdin.destroy(); - // Nothing else keeps the process up once its stdin is gone. - setInterval(() => {}, 1 << 30); + deafen(); return; } if (name === 'task') return send({ jsonrpc: '2.0', id: m.id, result: { resultType: 'task', taskId: 'task-1' } }); diff --git a/test/mcp-modern-outcomes.test.ts b/test/mcp-modern-outcomes.test.ts index 918f2bcc9..236bfb962 100644 --- a/test/mcp-modern-outcomes.test.ts +++ b/test/mcp-modern-outcomes.test.ts @@ -23,7 +23,7 @@ import { tmpdir } from 'node:os'; import { join, dirname } from 'node:path'; import { fileURLToPath } from 'node:url'; -import { SdkError, SdkErrorCode, SdkHttpError, UnsupportedProtocolVersionError } from '@modelcontextprotocol/client'; +import { Client, SdkError, SdkErrorCode, SdkHttpError, UnsupportedProtocolVersionError } from '@modelcontextprotocol/client'; import { ModernMcpConnection } from '../src/mcpl/modern-connection.js'; import { StdioTransport } from '../src/mcpl/transport.js'; import { McplProtocolVersionError, McplRequestError } from '../src/mcpl/server-connection.js'; @@ -389,6 +389,115 @@ test('with reconnect, a child that exits during the connect-time subscribe is a assert.equal(wire(log).find((e) => e.method === 'tools/call')!.launch, 2); }); +/** + * The first launch from here on, as the transport its spawn yields: whether + * its pipe has failed, and when it reports its child unreaped. With `stuck`, + * no signal reaches the child, so it outlives SIGTERM and SIGKILL, and + * cleanup really ends it. + */ +function firstLaunch(stuck: boolean): { readonly pid: number; pipeFailed: () => Promise; unreaped: Promise } { + const realSpawn = StdioTransport.spawn; + let first: StdioTransport | null = null; + let markUnreaped!: () => void; + const unreaped = new Promise((resolve) => { markUnreaped = resolve; }); + (StdioTransport as unknown as { spawn: typeof StdioTransport.spawn }).spawn = async (config) => { + const transport = await realSpawn.call(StdioTransport, config); + if (first) return transport; + first = transport; + transport.once('unreaped', () => markUnreaped()); + if (stuck) { + const child = (transport as unknown as { child: { pid: number; kill: (signal?: string) => boolean } }).child; + const realKill = child.kill.bind(child); + child.kill = () => true; // signals never arrive: the child outlives both bounds + cleanups.push(async () => { + realKill('SIGKILL'); + await until(() => !alive(child.pid), 'the stuck child killed'); + }); + } + return transport; + }; + cleanups.push(() => { (StdioTransport as unknown as { spawn: typeof StdioTransport.spawn }).spawn = realSpawn; }); + return { + get pid() { + assert.ok(first, 'a launch'); + return (first as unknown as { child: { pid: number } }).child.pid; + }, + pipeFailed: () => until(() => first?.pipeFailure !== undefined, "the launch's pipe to fail"), + unreaped, + }; +} + +/** + * The connect-time list-change subscribe, failed once `when` settles, as a + * subscribe whose acknowledgement never comes fails at its timeout: the SDK + * reports it and resolves the connect anyway. It's still written; only its + * end moves, to the event a test is about. + */ +function failSubscribeWhen(when: Promise): void { + const real = Client.prototype.listen; + Client.prototype.listen = function (this: Client, ...args: Parameters) { + const listening = real.apply(this, args); + listening.catch(() => { /* it ends later, at its own timeout or the client's close */ }); + return Promise.race([listening, when.then(() => { throw new Error('no acknowledgement came'); })]); + }; + cleanups.push(() => { Client.prototype.listen = real; }); +} + +test("a connect whose stdio pipe failed, and whose child was found unreaped before the connect resolved, fails holding the launch, rather than reading as connected", async () => { + const log = scratchLog(); + const launch = firstLaunch(true); + // The subscribe went to a child that had closed its stdin, so it's never + // acknowledged. It ends once the transport has found the child unreaped, + // when there was no session yet to lose. + failSubscribeWhen(launch.unreaped); + const connection = ModernMcpConnection.create({ id: 'raw', command: process.execPath, args: [SERVER, 'deaf-at-discover', log], protocol: 'modern' }); + cleanups.push(() => connection.close()); + const errors: string[] = []; + connection.on('error', (error: Error) => errors.push(error.message)); + const closes: unknown[] = []; + connection.on('close', (info: unknown) => closes.push(info)); + await assert.rejects(connection.start(), /^Error: MCP server "raw" stdio pipe failed with write EPIPE while connecting$/); + assert.equal(connection.isConnected, false); + assert.equal(connection.holdsUnreapedChild, true); + // The hold is reported once, by the connect's own cleanup. + const held = errors.filter((m) => /Not starting another launch while it may still be running/.test(m)); + assert.equal(held.length, 1, errors.join('\n')); + assert.match(held[0]!, /^MCP server "raw": stdio child \d+ of MCP server "raw" did not exit within 2000ms of SIGKILL; it could not be reaped\. /); + assert.deepEqual(closes, [], 'it never connected, so nothing closed'); + assert.equal(wire(log).filter((e) => e.event === 'start').length, 1, 'one launch'); + // close() retries the reap: refused while the child runs, then done. + await assert.rejects(connection.close(), /could not be reaped/); + process.kill(launch.pid, 'SIGKILL'); + await until(() => !alive(launch.pid), 'the child to be killed'); + await connection.close(); + assert.equal(connection.holdsUnreapedChild, false); +}); + +test("a connect whose stdio pipe failed fails even when it resolves while its child is still running and unreported: the check reads the pipe", async () => { + const log = scratchLog(); + const launch = firstLaunch(false); + // The subscribe ends as soon as the pipe has failed. The child ignores + // SIGTERM, so it runs until SIGKILL, 2 s on, and nothing has reported it. + let runningAtEnd: boolean | undefined; + failSubscribeWhen(launch.pipeFailed().then(() => { runningAtEnd = alive(launch.pid); })); + const connection = ModernMcpConnection.create({ id: 'raw', command: process.execPath, args: [SERVER, 'deaf-at-discover,ignore-sigterm', log], protocol: 'modern' }); + cleanups.push(() => connection.close()); + connection.on('error', () => { /* the subscribe's failure, reported by the SDK */ }); + await assert.rejects(connection.start(), /^Error: MCP server "raw" stdio pipe failed with write EPIPE while connecting$/); + assert.equal(runningAtEnd, true, 'the child was still running when the subscribe ended'); + assert.equal(connection.isConnected, false); + assert.equal(connection.holdsUnreapedChild, false); + assert.equal(alive(launch.pid), false, "gone once the connect failed: the connect's close reaped it"); +}); + +test("a connect whose stdio pipe failed reads that way when the failure has already ended the child", async () => { + const log = scratchLog(); + // The subscribe meets EPIPE, the transport ends the child with SIGTERM, + // and the close ends the subscribe. + await assert.rejects(connect('deaf-at-discover', log), /^Error: MCP server "raw" stdio pipe failed with write EPIPE while connecting$/); + assert.equal(wire(log).filter((e) => e.event === 'start').length, 1, 'one launch'); +}); + /** * The stdio line under a connection's live session, reached from inside, to * stand for a loss read from a failed pipe while the child still runs: the diff --git a/test/mcpl-legacy-lifetime.test.ts b/test/mcpl-legacy-lifetime.test.ts index 678abb461..feb52a666 100644 --- a/test/mcpl-legacy-lifetime.test.ts +++ b/test/mcpl-legacy-lifetime.test.ts @@ -7,6 +7,8 @@ * - a failed launch that can't be reaped is held, never orphaned: connect() * hands its owner to the caller, connectWithReconnect() and the reconnect * loop halt, and close() retries the reap; + * - a handshake whose stdio pipe failed fails, even when it's answered, so + * no connection is adopted that can't write to its child; * - close() during a reconnect handshake ends that launch before resolving; * - no reconnect launches while the previous child may still be running. */ @@ -19,7 +21,7 @@ import { join, dirname } from 'node:path'; import { fileURLToPath } from 'node:url'; import { WebSocketServer } from 'ws'; -import { McplServerConnection, McplUnreapedLaunchError } from '../src/mcpl/server-connection.js'; +import { McplServerConnection, McplUnreapedLaunchError, stderrTailOf } from '../src/mcpl/server-connection.js'; import { McplServerRegistry } from '../src/mcpl/server-registry.js'; import { StdioTransport, openTransport } from '../src/mcpl/transport.js'; import { createServer, type Socket } from 'node:net'; @@ -61,11 +63,12 @@ const RECONNECT = { reconnect: true, reconnectIntervalMs: 20, reconnectMaxInterv /** * Launches from the `from`-th on can't be signalled, so their cleanup can't * reap them (Olive-1804's pattern in mcp-modern-outcomes). `kill` really ends - * one; cleanup restores spawning and ends any left. + * one, and `unreaped` settles when its transport reports it unreaped; cleanup + * restores spawning and ends any left. */ -function unreapableFrom(from: number): Array<{ pid: number; kill: () => void }> { +function unreapableFrom(from: number): Array<{ pid: number; kill: () => void; unreaped: Promise }> { const realSpawn = StdioTransport.spawn; - const stuck: Array<{ pid: number; kill: () => void }> = []; + const stuck: Array<{ pid: number; kill: () => void; unreaped: Promise }> = []; let launches = 0; (StdioTransport as unknown as { spawn: typeof StdioTransport.spawn }).spawn = async (c) => { const transport = await realSpawn.call(StdioTransport, c); @@ -73,7 +76,8 @@ function unreapableFrom(from: number): Array<{ pid: number; kill: () => void }> const child = (transport as unknown as { child: { pid: number; kill: (signal?: string) => boolean } }).child; const realKill = child.kill.bind(child); child.kill = () => true; - stuck.push({ pid: child.pid, kill: () => { realKill('SIGKILL'); } }); + const unreaped = new Promise((resolve) => { transport.once('unreaped', () => resolve()); }); + stuck.push({ pid: child.pid, kill: () => { realKill('SIGKILL'); }, unreaped }); } return transport; }; @@ -84,6 +88,35 @@ function unreapableFrom(from: number): Array<{ pid: number; kill: () => void }> return stuck; } +/** + * The first launch from here on, as the transport its spawn yields, and for + * each line that transport hands over, whether it came in the same callback + * as the line before it. + */ +function firstLaunch(): { transport: () => StdioTransport | undefined; sameCallback: boolean[] } { + const realSpawn = StdioTransport.spawn; + let first: StdioTransport | undefined; + const sameCallback: boolean[] = []; + (StdioTransport as unknown as { spawn: typeof StdioTransport.spawn }).spawn = async (c) => { + const transport = await realSpawn.call(StdioTransport, c); + if (first) return transport; + first = transport; + // A callback's lines are handed over with no microtask between them. + let inCallback = false; + transport.on('line', () => { + sameCallback.push(inCallback); + inCallback = true; + queueMicrotask(() => { inCallback = false; }); + }); + return transport; + }; + cleanups.push(() => { (StdioTransport as unknown as { spawn: typeof StdioTransport.spawn }).spawn = realSpawn; }); + // The launch is ended once its test is over too, so a test that fails + // before its handshake ends doesn't leave the child running. + cleanups.push(() => first?.close().catch(() => {})); + return { transport: () => first, sameCallback }; +} + async function killAndWait(stuck: { pid: number; kill: () => void }): Promise { stuck.kill(); await until(() => !alive(stuck.pid), `child ${stuck.pid} gone`); @@ -160,6 +193,46 @@ test('a failed connect that cannot be reaped hands its owner to the caller, who await owner.close(); }); +test("a handshake answered after its stdio pipe failed fails, holding a child that outlived SIGKILL, rather than adopting a connection that can't write to it", async () => { + const log = scratchLog(); + const stuck = unreapableFrom(1); + const connecting = McplServerConnection.connect(config('deaf-ping', log), HOST_CAPS).then(() => null, (e: unknown) => e); + // The server closed its stdin and pinged this host, whose answer met EPIPE. + // The reap gives up on the child while the handshake still waits, with + // nothing listening for that yet; only then does the server answer. + await until(() => stuck.length === 1, 'the launch'); + await stuck[0]!.unreaped; + process.kill(stuck[0]!.pid, 'SIGUSR2'); + const error = await connecting; + assert.ok(error instanceof McplUnreapedLaunchError, String(error)); + assert.match(error.message, /^MCPL server "legacy" stdio pipe failed with write EPIPE during the handshake; its cleanup also failed: stdio child \d+ of MCP server "legacy" did not exit within \d+ms of SIGKILL; it could not be reaped$/); + const owner = error.connection; + assert.equal(owner.isConnected, false); + assert.equal(owner.willReconnect, false); + await assert.rejects(owner.close(), /could not be reaped/, 'still running: the reap fails again'); + await killAndWait(stuck[0]!); + await owner.close(); +}); + +test('a handshake fails when its answer is read in the same callback as a ping whose answer met EPIPE, and its cleanup reaps the child', async () => { + const log = scratchLog(); + const launch = firstLaunch(); + const connecting = McplServerConnection.connect(config('deaf-ping-answered', log), HOST_CAPS).then(() => null, (e: unknown) => e); + // The server has closed its stdin and said so on stderr, and that line is + // in this host's stderr tail. Only then does it write a ping and its answer + // together, so this host reads both in one callback: its answer to the ping + // meets EPIPE, which is reported only once that callback has returned. + await until(() => /deaf-ping: stdin closed/.test(launch.transport()?.stderrTail ?? ''), "the server's stderr line"); + const [pid] = starts(log); + process.kill(pid!, 'SIGUSR2'); + const error = await connecting; + assert.deepEqual(launch.sameCallback, [false, true], 'the ping and the answer were read in one callback'); + assert.ok(error instanceof Error && !(error instanceof McplUnreapedLaunchError), String(error)); + assert.match(String(error), /^Error: MCPL server "legacy" stdio pipe failed with write EPIPE during the handshake$/); + assert.match(stderrTailOf(error) ?? '', /deaf-ping: stdin closed/, "the server's stderr goes with it, as with the handshake's other failures"); + assert.equal(alive(pid!), false, "gone once the connect failed: the handshake's cleanup reaped it"); +}); + test('with reconnect too, a failed first launch that cannot be reaped throws, handing over its owner, as without', async () => { const log = scratchLog(); const stuck = unreapableFrom(1);