fix(mcpl): a transport error with nothing listening is logged, not thrown, so a malformed frame at upgrade no longer ends the host (#391)

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] <server>: 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] <server>: transport error: …`), and a listener that throws is
+  caught and logged with the error it was handling.

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) => {

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) => {

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();

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<void> {}
+}
+
+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"/);
+});