Fix: throws in user event handlers are caught and re-emitted as handler-error

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Change log for amqplib
 
+## Unreleased
+- Add `handler-error` event to connections and channels. If a user-supplied event handler (e.g. `connection.on('close', ...)`, `channel.on('error', ...)`, `channel.on('delivery', ...)` etc.) throws a synchronous error, amqplib will emit a `handler-error` event on the same emitter with the thrown error and the name of the event whose handler threw — provided a `handler-error` listener is registered. If no `handler-error` listener is registered, behaviour is unchanged from previous versions. Note: in previous versions, errors thrown in connection `close` event handlers were silently swallowed; errors thrown in channel event handlers (other than `delivery`/`return`) would kill the channel and possibly the connection (fixes #334).
+
 ## v1.0.6
 - Fix channel.get() not invoking callback with error on channel close; previously only an error event was emitted (fixes #832). **Note:** if you use the callback API, ensure your channel.get() callbacks handle errors — they will now be invoked in error cases where previously they were not. If you use the promise API, the returned promise now rejects with a proper Error object (with .code, .classId and .methodId properties) rather than a raw close frame.
 

README.md

@@ -43,10 +43,16 @@ const queue = 'tasks';
 amqplib.connect('amqp://localhost', (err, conn) => {
   if (err) throw err;
 
+  conn.on('error', (err) => { console.error('Connection error:', err); });
+  conn.on('handler-error', (err, event) => { console.error(`Uncaught exception in connection ${event} listener:`, err); });
+
   // Listener
   conn.createChannel((err, ch2) => {
     if (err) throw err;
 
+    ch2.on('error', (err) => { console.error('Channel error:', err); });
+    ch2.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });
+
     ch2.assertQueue(queue);
 
     ch2.consume(queue, (msg) => {
@@ -63,6 +69,8 @@ amqplib.connect('amqp://localhost', (err, conn) => {
   conn.createChannel((err, ch1) => {
     if (err) throw err;
 
+    ch1.on('error', (err) => { console.error('Channel error:', err); });
+    ch1.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });
     ch1.assertQueue(queue);
 
     setInterval(() => {
@@ -80,8 +88,12 @@ const amqplib = require('amqplib');
 (async () => {
   const queue = 'tasks';
   const conn = await amqplib.connect('amqp://localhost');
+  conn.on('error', (err) => { console.error('Connection error:', err); });
+  conn.on('handler-error', (err, event) => { console.error(`Uncaught exception in connection ${event} listener:`, err); });
 
   const ch1 = await conn.createChannel();
+  ch1.on('error', (err) => { console.error('Channel error:', err); });
+  ch1.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });
   await ch1.assertQueue(queue);
 
   // Listener
@@ -96,6 +108,8 @@ const amqplib = require('amqplib');
 
   // Sender
   const ch2 = await conn.createChannel();
+  ch2.on('error', (err) => { console.error('Channel error:', err); });
+  ch2.on('handler-error', (err, event) => { console.error(`Uncaught exception in channel ${event} listener:`, err); });
 
   setInterval(() => {
     ch2.sendToQueue(queue, Buffer.from('something to do'));
@@ -104,6 +118,43 @@ const amqplib = require('amqplib');
 
 ```
 
+## Error handling in event handlers
+
+If a user-supplied event handler throws a synchronous error, the throw will
+propagate into amqplib internals. Depending on where in the call stack it
+escapes, this can silently swallow the error, or close the channel or
+connection.
+
+To avoid this, register a `handler-error` listener on the connection and on
+each channel. If a listener is present, amqplib will catch any throw from a
+user event handler and deliver it there instead of letting it propagate
+internally. The listener receives the thrown error and the name of the event
+whose handler threw.
+
+Note that `handler-error` is not a replacement for the `error` event.
+The `error` event is emitted by amqplib itself when the connection or channel
+encounters a protocol-level error. The `handler-error` event is only emitted
+when *your own* event listener throws.
+
+```js
+const connection = await amqp.connect('amqp://localhost');
+
+connection.on('error', (err) => { /* handle protocol errors */ });
+connection.on('handler-error', (err, event) => {
+  console.error(`Uncaught exception in connection ${event} listener:`, err);
+});
+
+const channel = await connection.createChannel();
+
+channel.on('error', (err) => { /* handle protocol errors */ });
+channel.on('handler-error', (err, event) => {
+  console.error(`Uncaught exception in channel ${event} listener:`, err);
+});
+```
+
+If no `handler-error` listener is registered, behaviour is unchanged from
+previous versions.
+
 ## Running tests
 
     npm test

lib/channel.js

@@ -7,6 +7,7 @@ const EventEmitter = require('node:events');
 const fmt = require('node:util').format;
 const IllegalOperationError = require('./error').IllegalOperationError;
 const stackCapture = require('./error').stackCapture;
+const safeEmit = require('./safe_emit');
 
 class Channel extends EventEmitter {
   constructor(connection) {
@@ -128,7 +129,7 @@ class Channel extends EventEmitter {
     invalidateSend(this, 'Channel closed', capturedStack);
     this.accept = invalidOp('Channel closed', capturedStack);
     this.connection.releaseChannel(this.ch);
-    this.emit('close');
+    safeEmit(this, 'close');
   }
 
   // Stop being able to send and receive methods and content. Used when
@@ -188,7 +189,7 @@ class Channel extends EventEmitter {
         error.classId = defs.info(id).classId;
         error.methodId = defs.info(id).methodId;
       }
-      this.emit('error', error);
+      safeEmit(this, 'error', error);
     });
   }
 
@@ -250,7 +251,7 @@ class Channel extends EventEmitter {
   }
 
   onBufferDrain() {
-    this.emit('drain');
+    safeEmit(this, 'drain');
   }
 
   accept(f) {
@@ -264,12 +265,12 @@ class Channel extends EventEmitter {
 
       // confirmations, need to do confirm.select first
       case defs.BasicAck:
-        return this.emit('ack', f.fields);
+        return safeEmit(this, 'ack', f.fields);
       case defs.BasicNack:
-        return this.emit('nack', f.fields);
+        return safeEmit(this, 'nack', f.fields);
       case defs.BasicCancel:
         // The broker can send this if e.g., the queue is deleted.
-        return this.emit('cancel', f.fields);
+        return safeEmit(this, 'cancel', f.fields);
 
       case defs.ChannelClose: {
         // Any remote closure is an error to us. Reject the pending reply
@@ -287,7 +288,7 @@ class Channel extends EventEmitter {
         error.code = f.fields.replyCode;
         error.classId = f.fields.classId;
         error.methodId = f.fields.methodId;
-        this.emit('error', error);
+        safeEmit(this, 'error', error);
 
         const s = stackCapture(emsg);
         this.toClosed(s);
@@ -352,7 +353,7 @@ function acceptDeliveryOrReturn(f) {
   const fields = f.fields;
   return acceptMessage((message) => {
     message.fields = fields;
-    this.emit(event, message);
+    safeEmit(this, event, message);
   });
 }
 

lib/connection.js

@@ -17,6 +17,7 @@ const fmt = require('node:util').format;
 const PassThrough = require('node:stream').PassThrough;
 const IllegalOperationError = require('./error').IllegalOperationError;
 const stackCapture = require('./error').stackCapture;
+const safeEmit = require('./safe_emit');
 
 // High-water mark for channel write buffers, in 'objects' (which are
 // encoded frames as buffers).
@@ -281,7 +282,7 @@ class Connection extends EventEmitter {
   }
 
   closeWithError(reason, code, error) {
-    this.emit('error', error);
+    safeEmit(this, 'error', error);
     this.closeBecause(reason, code);
   }
 
@@ -290,7 +291,7 @@ class Connection extends EventEmitter {
       // forestall any more calls to onSocketError, since we're signed
       // up for `'error'` *and* `'end'`
       this.expectSocketClose = true;
-      this.emit('error', err);
+      safeEmit(this, 'error', err);
       const s = stackCapture('Socket error');
       this.toClosed(s, err);
     }
@@ -345,7 +346,7 @@ class Connection extends EventEmitter {
     // This is certainly true now, if it wasn't before
     this.expectSocketClose = true;
     this.stream.end();
-    this.emit('close', maybeErr);
+    safeEmit(this, 'close', maybeErr);
   }
 
   _updateSecret(newSecret, reason, cb) {
@@ -363,7 +364,7 @@ class Connection extends EventEmitter {
       const hb = new Heart(this.heartbeat, this.checkSend.bind(this), this.checkRecv.bind(this));
       hb.on('timeout', () => {
         const hberr = new Error('Heartbeat timeout');
-        this.emit('error', hberr);
+        safeEmit(this, 'error', hberr);
         const s = stackCapture('Heartbeat timeout');
         this.toClosed(s, hberr);
       });
@@ -585,17 +586,17 @@ function channel0(connection) {
       const e = new Error(emsg);
       e.code = f.fields.replyCode;
       if (isFatalError(e)) {
-        connection.emit('error', e);
+        safeEmit(connection, 'error', e);
       }
       connection.toClosed(s, e);
     } else if (f.id === defs.ConnectionBlocked) {
       connection.blocked = true;
-      connection.emit('blocked', f.fields.reason);
+      safeEmit(connection, 'blocked', f.fields.reason);
     } else if (f.id === defs.ConnectionUnblocked) {
       connection.blocked = false;
-      connection.emit('unblocked');
+      safeEmit(connection, 'unblocked');
     } else if (f.id === defs.ConnectionUpdateSecretOk) {
-      connection.emit('update-secret-ok');
+      safeEmit(connection, 'update-secret-ok');
     } else {
       connection.closeWithError(
         fmt('Unexpected frame on channel 0'),

lib/safe_emit.js

@@ -0,0 +1,13 @@
+function safeEmit(emitter, event, ...args) {
+  try {
+    emitter.emit(event, ...args);
+  } catch (e) {
+    if (emitter.listenerCount('handler-error') > 0) {
+      setImmediate(() => emitter.emit('handler-error', e, event));
+    } else {
+      throw e;
+    }
+  }
+}
+
+module.exports = safeEmit;