diff --git a/changelog.d/391-transport-error-no-listener.fixed.md b/changelog.d/391-transport-error-no-listener.fixed.md new file mode 100644 index 000000000..92bb1d358 --- /dev/null +++ b/changelog.d/391-transport-error-no-listener.fixed.md @@ -0,0 +1,15 @@ +- **A WebSocket MCPL server that sends a malformed frame with its upgrade + response no longer ends the host** (#391). ws raised the frame's error + before the handshake had attached a listener to the new transport, and + the transport passed it on with nothing listening, which EventEmitter + throws from an I/O callback. Such an error is now logged + (`[mcpl] : a websocket transport error, with nothing listening for + it: …`), and the connect is refused: at once when the server then closes + the connection, as a WebSocket server does after answering the close frame + ws sends, and after about 30 s when it never does. +- **A transport error no longer ends a process whose `McplServerConnection` + has no `'error'` listener** (#391). A connection the framework wires always + has one, but a probe, a playtest or a harness holding a bare connection + often doesn't. With no listener the error is now logged + (`[mcpl] : transport error: …`), and a listener that throws is + caught and logged with the error it was handling. diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 7f7eb4337..c349d3f0d 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -1426,12 +1426,29 @@ export class McplServerConnection extends EventEmitter { // the log line prints the same through log-value. const error = new Error(`MCPL server "${this.id}" inbound dispatch failed${where(boundedForLog)}: ${boundedForLog(describeCause(cause))}`, { cause }); console.error(`[mcpl] ${this.id}: inbound dispatch failed${where(logValue)}: ${logDetail(describeCause(cause))}`); - // EventEmitter's unhandled 'error' behavior would defeat containment before - // framework wiring, so stderr remains the fallback when no listener exists. - if (this.listenerCount('error') > 0) { - try { this.emit('error', error); } - catch (reportError) { console.error(`[mcpl] ${this.id}: an 'error' listener failed: ${logDetail(describeCause(reportError))}`); } + this.emitToErrorListeners(error); + } + + /** + * Give an error to the connection's 'error' listeners, and say whether it + * has any. EventEmitter throws an 'error' that nothing listens for, and + * from a transport's callback, where these errors arise, nothing catches + * that throw, so it ends the process. A bare connection has no listener: + * the class is exported, and a probe, a playtest or a harness often attaches + * none. So with no listener, the error is the caller's to log, unless it + * already has. A listener that throws is caught and logged, with the error + * it was handling. + */ + private emitToErrorListeners(error: Error): boolean { + if (this.listenerCount('error') === 0) return false; + try { this.emit('error', error); } + catch (listenerError) { + console.error( + `[mcpl] ${this.id}: an 'error' listener failed: ${logDetail(describeCause(listenerError))}` + + `, handling ${logDetail(error.message)}`, + ); } + return true; } /** @@ -1527,7 +1544,8 @@ export class McplServerConnection extends EventEmitter { private setupLifecycle(transport: McplTransport): void { this.hostCloseRequested = false; transport.on('error', (err: Error) => { - this.emit('error', new Error(`MCPL server "${this.id}" transport error: ${err.message}`)); + const error = new Error(`MCPL server "${this.id}" transport error: ${err.message}`); + if (!this.emitToErrorListeners(error)) console.error(`[mcpl] ${this.id}: transport error: ${logDetail(err.message)}`); }); transport.on('close', (info: TransportCloseInfo) => { diff --git a/src/mcpl/transport.ts b/src/mcpl/transport.ts index 8be8747ed..22d699ad5 100644 --- a/src/mcpl/transport.ts +++ b/src/mcpl/transport.ts @@ -23,6 +23,7 @@ import { StringDecoder } from 'node:string_decoder'; import WebSocket from 'ws'; import type { McplServerConfig } from './types.js'; +import { logDetail } from './log-value.js'; /** Why/how a transport closed. `code`/`signal` are populated for stdio child * exits; `reason` carries a WebSocket close reason or an error message. */ @@ -55,7 +56,10 @@ export abstract class McplTransport extends EventEmitter { private lineBuffer: string[] = []; private stderrBuffer: string[] = []; - protected constructor() { + /** The server's id from its config, so a line the transport logs names it. + * The stdio and WebSocket transports always give it; a test's stand-in + * transport may leave it out. */ + protected constructor(protected readonly serverId?: string) { super(); // Flush buffered inbound lines the moment a consumer subscribes. this.on('newListener', (event) => { @@ -97,6 +101,23 @@ export abstract class McplTransport extends EventEmitter { else this.stderrBuffer.push(line); } + /** + * Report a transport error: to the 'error' listeners if there are any, else + * as a log line. EventEmitter throws an 'error' that nothing listens for, + * and a transport reports its errors from I/O callbacks, where nothing + * catches that throw, so it would end the process. A transport has no + * listener from its creation until the handshake attaches one, and a + * WebSocket server that sends a malformed frame with its upgrade response + * makes ws raise an 'error' in that window. + */ + protected reportError(err: Error): void { + if (this.listenerCount('error') > 0) this.emit('error', err); + else { + const server = this.serverId === undefined ? '' : `${this.serverId}: `; + console.error(`[mcpl] ${server}a ${this.kind} transport error, with nothing listening for it: ${logDetail(err.message)}`); + } + } + /** Emit 'close' exactly once. */ protected markClosed(info: TransportCloseInfo): void { if (this._closed) return; @@ -200,8 +221,9 @@ export class StdioTransport extends McplTransport { private constructor( private readonly child: ChildProcess, private readonly rl: ReadlineInterface, + serverId: string, ) { - super(); + super(serverId); this.rl.on('line', (line: string) => this.pushLine(line)); @@ -228,7 +250,7 @@ export class StdioTransport extends McplTransport { if (carry.length > 0) this.pushStderr(carry); }); - this.child.on('error', (err) => this.emit('error', err)); + this.child.on('error', (err) => this.reportError(err)); // The child's stdin and stdout carry the protocol, so after an error on // either, a write's EPIPE or a read error, this transport can't reach the // child even if it's still running. The handler sends a running child @@ -249,13 +271,11 @@ export class StdioTransport extends McplTransport { // listener absorbs that copy. this.child.stdout?.on('error', endOnPipeError); this.rl.on('error', () => { /* readline's copy of a stdout error, handled above */ }); - // stderr carries diagnostics only: a read error on it is reported if - // anyone listens and never thrown. It ends nothing once the connection - // is wired; during a handshake the listener is the handshake's, which - // fails it. - this.child.stderr?.on('error', (err) => { - if (this.listenerCount('error') > 0) this.emit('error', err); - }); + // stderr carries diagnostics only: a read error on it is reported, to the + // listeners or else in the log, and never thrown. It ends nothing once + // the connection is wired; during a handshake the listener is the + // handshake's, which fails it. + this.child.stderr?.on('error', (err) => this.reportError(err)); this.child.on('exit', (code, signal) => this.markClosed({ code, signal, @@ -295,7 +315,7 @@ export class StdioTransport extends McplTransport { }); }); const rl = createInterface({ input: child.stdout! }); - return new StdioTransport(child, rl); + return new StdioTransport(child, rl, config.id); } writeLine(json: string): void { @@ -334,8 +354,8 @@ const WS_OPEN_TIMEOUT_MS = 15_000; export class WebSocketTransport extends McplTransport { readonly kind = 'websocket' as const; - private constructor(private readonly ws: WebSocket) { - super(); + private constructor(private readonly ws: WebSocket, serverId: string) { + super(serverId); this.ws.on('message', (data: WebSocket.RawData, isBinary: boolean) => { // MCPL frames are UTF-8 JSON text. A frame MAY contain multiple @@ -347,7 +367,7 @@ export class WebSocketTransport extends McplTransport { } }); - this.ws.on('error', (err: Error) => this.emit('error', err)); + this.ws.on('error', (err: Error) => this.reportError(err)); this.ws.on('close', (code: number, reason: Buffer) => { const text = reason?.toString('utf8'); this.markClosed(text ? { code, reason: text, reasonFromPeer: true } : { code, reason: 'websocket closed' }); @@ -406,7 +426,7 @@ export class WebSocketTransport extends McplTransport { settled = true; clearTimeout(timer); ws.removeAllListeners('error'); // the instance re-attaches its own handlers - resolve(new WebSocketTransport(ws)); + resolve(new WebSocketTransport(ws, config.id)); }); ws.once('error', (err: Error) => { diff --git a/test/fixtures/mcpl-ws-upgrade-error-probe.mjs b/test/fixtures/mcpl-ws-upgrade-error-probe.mjs new file mode 100644 index 000000000..d27679953 --- /dev/null +++ b/test/fixtures/mcpl-ws-upgrade-error-probe.mjs @@ -0,0 +1,50 @@ +// Runs as a separate process: connects to a WebSocket server that sends a +// frame with a reserved opcode in the same write as its 101 upgrade response, +// before the host has sent initialize. ws raises that frame's 'error' before +// the handshake has attached any listener. Prints "refused after N ms: …" +// when connect() rejects, or "connected"; an 'error' thrown with nothing +// listening would end the process before it prints either. +// Mode "answers-close" (the default): the server answers the client's close +// frame and ends the socket, as most servers do. ws reads nothing more once it +// has failed the connection, so it's the socket's end, not the answering +// frame, that ends the wait. Mode "silent": it never answers or ends, so the +// connect waits out a 30 s timeout, ws's close timeout or the handshake's own. +import { createServer } from 'node:http'; +import { createHash } from 'node:crypto'; +import { McplServerConnection } from '../../src/mcpl/server-connection.js'; + +const mode = process.argv[2] ?? 'answers-close'; +const server = createServer(); +server.on('upgrade', (request, socket) => { + const accept = createHash('sha1') + .update(`${request.headers['sec-websocket-key']}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`) + .digest('base64'); + socket.write(Buffer.concat([ + Buffer.from( + 'HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n' + + `Sec-WebSocket-Accept: ${accept}\r\n\r\n`, + ), + // A frame header with opcode 3, which RFC 6455 reserves. + Buffer.from([0x83, 0x00]), + ])); + socket.on('error', () => {}); + if (mode === 'answers-close') { + // The client's next frame is its close frame: answer it with one, and end. + socket.once('data', () => socket.end(Buffer.from([0x88, 0x00]))); + } +}); +await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + +const started = Date.now(); +try { + const connection = await McplServerConnection.connect( + { id: 'probe', url: `ws://127.0.0.1:${server.address().port}` }, + { version: '0.5' }, + ); + console.log('connected'); + await connection.close(); +} catch (error) { + console.log(`refused after ${Date.now() - started} ms: ${error.message}`); +} +server.closeAllConnections?.(); +server.close(); diff --git a/test/mcpl-transport-error-listeners.test.ts b/test/mcpl-transport-error-listeners.test.ts new file mode 100644 index 000000000..7c54e068a --- /dev/null +++ b/test/mcpl-transport-error-listeners.test.ts @@ -0,0 +1,104 @@ +/** + * A transport error on a connection nothing listens to for 'error' (#391). + * EventEmitter throws an 'error' with no listener, and a transport reports its + * errors from I/O callbacks, where nothing catches the throw: it ended the + * process. McplServerConnection is exported, and a probe, a playtest or a + * harness holding one often attaches no 'error' listener (the framework + * always does). The error is logged instead, and given to listeners when + * there are any, a throwing one caught as #208's inbound failures are. + */ +import { test, type TestContext } from 'node:test'; +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { spawnSync } from 'node:child_process'; +import { fileURLToPath } from 'node:url'; + +import { McplServerConnection } from '../src/mcpl/server-connection.js'; +import { McplTransport, StdioTransport, WebSocketTransport } from '../src/mcpl/transport.js'; + +class LineTransport extends McplTransport { + constructor() { super(); } + readonly kind = 'websocket' as const; + writeLine(): void {} + async close(): Promise {} +} + +function bareConnection(t: TestContext) { + const transport = new LineTransport(); + const connection = new (McplServerConnection as any)('srv', null, transport) as McplServerConnection; + connection.ready(); + const logged: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }); + const reset = Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }); + return { connection, transport, logged, reset }; +} + +test("a transport error on a connection with no 'error' listener is logged, never thrown", (t) => { + const { connection, transport, logged, reset } = bareConnection(t); + assert.equal(connection.listenerCount('error'), 0); + assert.doesNotThrow(() => transport.emit('error', reset)); + assert.deepEqual(logged, ['[mcpl] srv: transport error: "read ECONNRESET"']); +}); + +test("with an 'error' listener, a transport error is the listener's, and nothing is logged", (t) => { + const { connection, transport, logged, reset } = bareConnection(t); + const reported: Error[] = []; + connection.on('error', (error: Error) => reported.push(error)); + transport.emit('error', reset); + assert.deepEqual(reported.map((error) => error.message), ['MCPL server "srv" transport error: read ECONNRESET']); + assert.deepEqual(logged, []); +}); + +test("an 'error' listener that throws on a transport error is caught and logged, with the error it was handling", (t) => { + const { connection, transport, logged, reset } = bareConnection(t); + connection.on('error', () => { throw new Error('listener exploded'); }); + assert.doesNotThrow(() => transport.emit('error', reset)); + assert.deepEqual(logged, [ + `[mcpl] srv: an 'error' listener failed: "listener exploded", handling ${JSON.stringify('MCPL server "srv" transport error: read ECONNRESET')}`, + ]); +}); + +// A transport reports its own errors through reportError, which gives them to +// its 'error' listeners, or logs them when it has none: from its creation +// until the handshake attaches a listener, it has none (#391's rule, at the +// transport's own forwarders). +test("a transport with nothing listening for 'error' logs its errors, and gives them to a listener once there is one", (t) => { + const logged: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }); + // Stand-ins for a spawned child and a socket, as the stdin test in + // mcpl-request-handler-errors.test.ts uses. + const child = Object.assign(new EventEmitter(), { + stdin: new EventEmitter(), stdout: new EventEmitter(), stderr: new EventEmitter(), killed: false, exitCode: null, signalCode: null, + }); + const stdio = new (StdioTransport as any)(child, new EventEmitter(), 'child-srv') as StdioTransport; + const ws = new EventEmitter(); + const socket = new (WebSocketTransport as any)(ws, 'ws-srv') as WebSocketTransport; + assert.doesNotThrow(() => child.emit('error', new Error('kill EPERM'))); + assert.doesNotThrow(() => child.stderr.emit('error', new Error('read EIO'))); + assert.doesNotThrow(() => ws.emit('error', new Error('Invalid WebSocket frame: invalid opcode 3'))); + assert.deepEqual(logged, [ + '[mcpl] child-srv: a stdio transport error, with nothing listening for it: "kill EPERM"', + '[mcpl] child-srv: a stdio transport error, with nothing listening for it: "read EIO"', + '[mcpl] ws-srv: a websocket transport error, with nothing listening for it: "Invalid WebSocket frame: invalid opcode 3"', + ]); + const reported: string[] = []; + stdio.on('error', (error: Error) => reported.push(error.message)); + socket.on('error', (error: Error) => reported.push(error.message)); + child.emit('error', new Error('kill EPERM')); + child.stderr.emit('error', new Error('read EIO')); + ws.emit('error', new Error('read ECONNRESET')); + assert.deepEqual(reported, ['kill EPERM', 'read EIO', 'read ECONNRESET']); + assert.equal(logged.length, 3, 'nothing more is logged once a listener has them'); +}); + +test('a WebSocket server that sends a malformed frame with its upgrade response is refused, and the connecting process survives', () => { + // The server's frame arrives with the 101, and ws raises its error before + // the handshake has attached a listener to the new transport. + const probe = spawnSync(process.execPath, [ + fileURLToPath(new URL('./fixtures/mcpl-ws-upgrade-error-probe.mjs', import.meta.url)), + ], { encoding: 'utf8', timeout: 60_000 }); + assert.equal(probe.error, undefined); + assert.equal(probe.status, 0, `stdout: ${probe.stdout}\nstderr: ${probe.stderr}`); + assert.match(probe.stdout, /^refused after \d+ ms: MCPL server "probe" transport closed before handshake completed/m); + assert.match(probe.stderr, /\[mcpl\] probe: a websocket transport error, with nothing listening for it: "Invalid WebSocket frame: invalid opcode 3"/); +});