From 80158bb8ec9d7522375ebe4ae3e33c7340b197a0 Mon Sep 17 00:00:00 2001 From: Stephen Cresswell <229672+cressie176@users.noreply.github.com> Date: Thu, 7 May 2026 20:46:12 +0100 Subject: [PATCH 1/3] Fix: throws in user event handlers are caught and re-emitted as handler-error (#334) Wraps all user-facing emit() calls on both Connection and Channel with safeEmit(). If a handler throws and a handler-error listener is registered, the error is delivered there via setImmediate (preventing re-entry into amqplib internals). If no handler-error listener is registered, behaviour is unchanged. Previously, throws in channel event handlers (ack, nack, cancel, error, close) would propagate through acceptLoop and kill the connection. Throws in connection close handlers were silently swallowed. Co-Authored-By: Claude Opus 4.6 --- CHANGELOG.md | 3 + lib/channel.js | 17 +-- lib/connection.js | 11 +- lib/safe_emit.js | 13 ++ test/channel.test.js | 300 +++++++++++++++++++++++++++++++++++++++- test/connection.test.js | 223 +++++++++++++++++++++++++++++ 6 files changed, 553 insertions(+), 14 deletions(-) create mode 100644 lib/safe_emit.js diff --git a/CHANGELOG.md b/CHANGELOG.md index 33055912..aa28b0ec 100644 --- a/CHANGELOG.md +++ b/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 — 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. diff --git a/lib/channel.js b/lib/channel.js index 1c62d731..20e33e36 100644 --- a/lib/channel.js +++ b/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); }); } diff --git a/lib/connection.js b/lib/connection.js index a0bd782f..5000da6c 100644 --- a/lib/connection.js +++ b/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). @@ -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) { @@ -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'), diff --git a/lib/safe_emit.js b/lib/safe_emit.js new file mode 100644 index 00000000..cab7af38 --- /dev/null +++ b/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)); + } else { + throw e; + } + } +} + +module.exports = safeEmit; diff --git a/test/channel.test.js b/test/channel.test.js index 92de04e3..c6fba1ae 100644 --- a/test/channel.test.js +++ b/test/channel.test.js @@ -1,4 +1,4 @@ -const { describe, it } = require('node:test'); +const { describe, it, beforeEach, afterEach } = require('node:test'); const assert = require('node:assert'); const promisify = require('node:util').promisify; const Channel = require('../lib/channel').Channel; @@ -609,4 +609,302 @@ describe('Channel', () => { cb(); })); }); + + describe('Event handler errors - without handler-error listener', () => { + let prevUncaughtExceptionListeners; + + beforeEach(() => { + prevUncaughtExceptionListeners = process.rawListeners('uncaughtException').slice(); + process.removeAllListeners('uncaughtException'); + }); + + afterEach(() => { + prevUncaughtExceptionListeners.forEach((h) => process.on('uncaughtException', h)); + }); + + it('throw in close handler kills the connection without handler-error listener', channelTest((ch, cb, conn) => { + const expectedErr = new Error('user close handler explodes'); + let connErrorSeen = false; + // The throw propagates to acceptLoop → frameError → onSocketError → connection.emit('error') + conn.on('error', (err) => { + assert.strictEqual(err, expectedErr); + connErrorSeen = true; + }); + // The ChannelCloseOk buffered before the kill is flushed by the mux after the + // stream is ended, producing ERR_STREAM_PUSH_AFTER_EOF. Absorb it and use it + // as the signal to end the test, since it fires after the conn error. + process.once('uncaughtException', (err) => { + assert.strictEqual(err.code, 'ERR_STREAM_PUSH_AFTER_EOF'); + assert.ok(connErrorSeen); + cb(); + }); + ch.once('error', (err) => { assert.match(err.message, /Channel closed by server/); }); + ch.on('close', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.ChannelClose, { + replyText: 'Forced close', + replyCode: defs.constants.CHANNEL_ERROR, + classId: 0, + methodId: 0, + }, ch); + }, cb); + })); + + it('throw in error handler becomes uncaught exception', channelTest((ch, cb) => { + const expectedErr = new Error('user error handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('error', () => { throw expectedErr; }); + open(ch); + }, (send, wait, cb, ch) => { + send(defs.ChannelClose, { + replyText: 'Forced close', + replyCode: defs.constants.CHANNEL_ERROR, + classId: 0, + methodId: 0, + }, ch); + wait(defs.ChannelCloseOk)().then(cb, cb); + })); + + it('throw in drain handler becomes uncaught exception', channelTest((ch, cb) => { + const expectedErr = new Error('user drain handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('drain', () => { throw expectedErr; }); + // Call outside a Promise chain so the throw becomes an uncaughtException + open(ch).then(() => setImmediate(() => ch.onBufferDrain())); + }, (_send, _wait, cb) => { cb(); })); + + it('throw in ack handler becomes uncaught exception', channelTest((ch, cb) => { + const expectedErr = new Error('user ack handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('ack', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicAck, { deliveryTag: 1, multiple: false }, ch); + }, cb); + })); + + it('throw in nack handler becomes uncaught exception', channelTest((ch, cb) => { + const expectedErr = new Error('user nack handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('nack', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicNack, { deliveryTag: 1, multiple: false, requeue: false }, ch); + }, cb); + })); + + it('throw in cancel handler becomes uncaught exception', channelTest((ch, cb) => { + const expectedErr = new Error('user cancel handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('cancel', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicCancel, { consumerTag: 'tag', nowait: false }, ch); + }, cb); + })); + + it('throw in delivery handler closes channel with error', channelTest((ch, cb) => { + // acceptMessageFrame catches the throw and routes it to closeWithError, + // which fires channel 'error' then 'close' — it does NOT become an uncaughtException. + const expectedErr = new Error('user delivery handler explodes'); + const decrementLatch = latch(2, cb); + ch.on('error', (err) => { + assert.strictEqual(err, expectedErr); + decrementLatch(); + }); + ch.on('close', decrementLatch); + ch.on('delivery', () => { throw expectedErr; }); + open(ch); + }, (send, wait, cb, ch) => { + send(defs.BasicDeliver, DELIVER_FIELDS, ch, Buffer.from('hello')); + wait(defs.ChannelClose)() + .then(() => send(defs.ChannelCloseOk, {}, ch)) + .then(cb, cb); + })); + + it('throw in return handler closes channel with error', channelTest((ch, cb) => { + // acceptMessageFrame catches the throw and routes it to closeWithError. + const expectedErr = new Error('user return handler explodes'); + const decrementLatch = latch(2, cb); + ch.on('error', (err) => { + assert.strictEqual(err, expectedErr); + decrementLatch(); + }); + ch.on('close', decrementLatch); + ch.on('return', () => { throw expectedErr; }); + open(ch); + }, (send, wait, cb, ch) => { + send(defs.BasicReturn, DELIVER_FIELDS, ch, Buffer.from('hello')); + wait(defs.ChannelClose)() + .then(() => send(defs.ChannelCloseOk, {}, ch)) + .then(cb, cb); + })); + }); + + describe('Event handler errors - with handler-error listener', () => { + let prevUncaughtExceptionListeners; + + beforeEach(() => { + prevUncaughtExceptionListeners = process.rawListeners('uncaughtException').slice(); + process.removeAllListeners('uncaughtException'); + }); + + afterEach(() => { + prevUncaughtExceptionListeners.forEach((h) => process.on('uncaughtException', h)); + }); + + it('throw in close handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user close handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + // A server-initiated ChannelClose always fires 'error' before 'close' (line 291 in channel.js). + // Handle it so only the throw from the close handler reaches handler-error. + ch.once('error', (err) => { assert.match(err.message, /Channel closed by server/); }); + ch.on('close', () => { throw expectedErr; }); + open(ch); + }, (send, wait, cb, ch) => { + send(defs.ChannelClose, { + replyText: 'Forced close', + replyCode: defs.constants.CHANNEL_ERROR, + classId: 0, + methodId: 0, + }, ch); + wait(defs.ChannelCloseOk)().then(cb, cb); + })); + + it('throw in error handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user error handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('error', () => { throw expectedErr; }); + open(ch); + }, (send, wait, cb, ch) => { + send(defs.ChannelClose, { + replyText: 'Forced close', + replyCode: defs.constants.CHANNEL_ERROR, + classId: 0, + methodId: 0, + }, ch); + wait(defs.ChannelCloseOk)().then(cb, cb); + })); + + it('throw in drain handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user drain handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('drain', () => { throw expectedErr; }); + open(ch).then(() => setImmediate(() => ch.onBufferDrain())); + }, (_send, _wait, cb) => { cb(); })); + + it('throw in ack handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user ack handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('ack', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicAck, { deliveryTag: 1, multiple: false }, ch); + }, cb); + })); + + it('throw in nack handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user nack handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('nack', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicNack, { deliveryTag: 1, multiple: false, requeue: false }, ch); + }, cb); + })); + + it('throw in cancel handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user cancel handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('cancel', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicCancel, { consumerTag: 'tag', nowait: false }, ch); + }, cb); + })); + + it('throw in delivery handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user delivery handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('delivery', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicDeliver, DELIVER_FIELDS, ch, Buffer.from('hello')); + }, cb); + })); + + it('throw in return handler is delivered via handler-error event', channelTest((ch, cb) => { + const expectedErr = new Error('user return handler explodes'); + ch.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('return', () => { throw expectedErr; }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicReturn, DELIVER_FIELDS, ch, Buffer.from('hello')); + }, cb); + })); + + it('throw in handler-error handler becomes uncaught exception', channelTest((ch, cb) => { + const expectedErr = new Error('handler-error handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + ch.on('handler-error', () => { throw expectedErr; }); + ch.on('ack', () => { throw new Error('user ack handler explodes'); }); + open(ch); + }, (send, _wait, cb, ch) => { + completes(() => { + send(defs.BasicAck, { deliveryTag: 1, multiple: false }, ch); + }, cb); + })); + }); }); diff --git a/test/connection.test.js b/test/connection.test.js index 2edb2ed3..62932ab8 100644 --- a/test/connection.test.js +++ b/test/connection.test.js @@ -300,6 +300,229 @@ describe('Connection', () => { })); }); + describe('Event handler errors - without handler-event listener', () => { + let prevUncaughtExceptionListeners; + + beforeEach(() => { + prevUncaughtExceptionListeners = process.rawListeners('uncaughtException').slice(); + process.removeAllListeners('uncaughtException'); + }); + + afterEach(() => { + prevUncaughtExceptionListeners.forEach((h) => process.on('uncaughtException', h)); + }); + + it('throw in close handler from server-initiated close is swallowed without handler-error listener', connectionTest((c, cb) => { + c.on('close', () => { throw new Error('user handler explodes'); }); + c.open(OPEN_OPTS); + cb(); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionClose, { + replyText: 'Begone', + replyCode: defs.constants.CONNECTION_FORCED, + methodId: 0, + classId: 0, + })) + .then(wait(defs.ConnectionCloseOk)) + .then(cb, cb); + })); + + it('throw in close handler from client-initiated close is swallowed without handler-error listener', connectionTest((c, cb) => { + c.open(OPEN_OPTS, (err) => { + assert.ifError(err); + c.on('close', () => { throw new Error('user handler explodes on client close'); }); + c.close(); + cb(); + }); + }, (send, wait, cb) => { + handshake(send, wait) + .then(wait(defs.ConnectionClose)) + .then(() => send(defs.ConnectionCloseOk, {})) + .then(cb, cb); + })); + + it('throw in error handler becomes uncaught exception', connectionTest((c, cb) => { + const expectedErr = new Error('user error handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('error', () => { throw expectedErr; }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionClose, { + replyText: 'Begone', + replyCode: defs.constants.INTERNAL_ERROR, + methodId: 0, + classId: 0, + })) + .then(wait(defs.ConnectionCloseOk)) + .then(cb, cb); + })); + + it('throw in blocked handler becomes uncaught exception', connectionTest((c, cb) => { + const expectedErr = new Error('user blocked handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('blocked', () => { throw expectedErr; }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionBlocked, { reason: 'memory' }, 0)) + .then(cb, cb); + })); + + it('throw in unblocked handler becomes uncaught exception', connectionTest((c, cb) => { + const expectedErr = new Error('user unblocked handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('unblocked', () => { throw expectedErr; }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionUnblocked, {}, 0)) + .then(cb, cb); + })); + + it('throw in update-secret-ok handler becomes uncaught exception', connectionTest((c, cb) => { + const expectedErr = new Error('user update-secret-ok handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.open(OPEN_OPTS, (err) => { + assert.ifError(err); + c.on('update-secret-ok', () => { throw expectedErr; }); + c._updateSecret(Buffer.from('new secret'), 'reason', () => {}); + }); + }, (send, wait, cb) => { + handshake(send, wait) + .then(wait(defs.ConnectionUpdateSecret)) + .then(() => send(defs.ConnectionUpdateSecretOk, {}, 0)) + .then(cb, cb); + })); + }); + + describe('Event handler errors - with handler-error listener', () => { + let prevUncaughtExceptionListeners; + + beforeEach(() => { + prevUncaughtExceptionListeners = process.rawListeners('uncaughtException').slice(); + process.removeAllListeners('uncaughtException'); + }); + + afterEach(() => { + prevUncaughtExceptionListeners.forEach((h) => process.on('uncaughtException', h)); + }); + + it('throw in close handler is delivered via handler-error event', connectionTest((c, cb) => { + const expectedErr = new Error('user close handler explodes'); + c.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('close', () => { throw expectedErr; }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionClose, { + replyText: 'Begone', + replyCode: defs.constants.CONNECTION_FORCED, + methodId: 0, + classId: 0, + })) + .then(wait(defs.ConnectionCloseOk)) + .then(cb, cb); + })); + + it('throw in error handler is delivered via handler-error event', connectionTest((c, cb) => { + const expectedErr = new Error('user error handler explodes'); + c.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('error', () => { throw expectedErr; }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionClose, { + replyText: 'Begone', + replyCode: defs.constants.INTERNAL_ERROR, + methodId: 0, + classId: 0, + })) + .then(wait(defs.ConnectionCloseOk)) + .then(cb, cb); + })); + + it('throw in blocked handler is delivered via handler-error event', connectionTest((c, cb) => { + const expectedErr = new Error('user blocked handler explodes'); + c.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('blocked', () => { throw expectedErr; }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionBlocked, { reason: 'memory' }, 0)) + .then(cb, cb); + })); + + it('throw in unblocked handler is delivered via handler-error event', connectionTest((c, cb) => { + const expectedErr = new Error('user unblocked handler explodes'); + c.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('unblocked', () => { throw expectedErr; }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionUnblocked, {}, 0)) + .then(cb, cb); + })); + + it('throw in update-secret-ok handler is delivered via handler-error event', connectionTest((c, cb) => { + const expectedErr = new Error('user update-secret-ok handler explodes'); + c.on('handler-error', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.open(OPEN_OPTS, (err) => { + assert.ifError(err); + c.on('update-secret-ok', () => { throw expectedErr; }); + c._updateSecret(Buffer.from('new secret'), 'reason', () => {}); + }); + }, (send, wait, cb) => { + handshake(send, wait) + .then(wait(defs.ConnectionUpdateSecret)) + .then(() => send(defs.ConnectionUpdateSecretOk, {}, 0)) + .then(cb, cb); + })); + + it('throw in handler-error handler becomes uncaught exception', connectionTest((c, cb) => { + const expectedErr = new Error('handler-error handler explodes'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.on('handler-error', () => { throw expectedErr; }); + c.on('blocked', () => { throw new Error('user blocked handler explodes'); }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ConnectionBlocked, { reason: 'memory' }, 0)) + .then(cb, cb); + })); + }); + describe('heartbeats', () => { beforeEach(() => { From 8a476c5f0267b503e34ebb686b72a998ece662f8 Mon Sep 17 00:00:00 2001 From: Stephen Cresswell <229672+cressie176@users.noreply.github.com> Date: Thu, 7 May 2026 22:57:29 +0100 Subject: [PATCH 2/3] Docs and API: pass event name as second arg to handler-error; document in README handler-error listeners now receive (err, event) so the listener knows which event handler threw. Update README with dedicated handler-error section, error handler examples, and clearer log messages. Update CHANGELOG to reflect the new signature. Co-Authored-By: Claude Opus 4.6 --- CHANGELOG.md | 2 +- README.md | 51 +++++++++++++++++++++++++++++++++++++++++ lib/safe_emit.js | 2 +- test/channel.test.js | 24 ++++++++++++------- test/connection.test.js | 15 ++++++++---- 5 files changed, 79 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aa28b0ec..dab883a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,7 @@ # 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 — 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). +- 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. diff --git a/README.md b/README.md index 644d77f7..0b492d12 100644 --- a/README.md +++ b/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 diff --git a/lib/safe_emit.js b/lib/safe_emit.js index cab7af38..d5be07fd 100644 --- a/lib/safe_emit.js +++ b/lib/safe_emit.js @@ -3,7 +3,7 @@ function safeEmit(emitter, event, ...args) { emitter.emit(event, ...args); } catch (e) { if (emitter.listenerCount('handler-error') > 0) { - setImmediate(() => emitter.emit('handler-error', e)); + setImmediate(() => emitter.emit('handler-error', e, event)); } else { throw e; } diff --git a/test/channel.test.js b/test/channel.test.js index c6fba1ae..e031cdab 100644 --- a/test/channel.test.js +++ b/test/channel.test.js @@ -775,8 +775,9 @@ describe('Channel', () => { it('throw in close handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user close handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'close'); cb(); }); // A server-initiated ChannelClose always fires 'error' before 'close' (line 291 in channel.js). @@ -796,8 +797,9 @@ describe('Channel', () => { it('throw in error handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user error handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'error'); cb(); }); ch.on('error', () => { throw expectedErr; }); @@ -814,8 +816,9 @@ describe('Channel', () => { it('throw in drain handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user drain handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'drain'); cb(); }); ch.on('drain', () => { throw expectedErr; }); @@ -824,8 +827,9 @@ describe('Channel', () => { it('throw in ack handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user ack handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'ack'); cb(); }); ch.on('ack', () => { throw expectedErr; }); @@ -838,8 +842,9 @@ describe('Channel', () => { it('throw in nack handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user nack handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'nack'); cb(); }); ch.on('nack', () => { throw expectedErr; }); @@ -852,8 +857,9 @@ describe('Channel', () => { it('throw in cancel handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user cancel handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'cancel'); cb(); }); ch.on('cancel', () => { throw expectedErr; }); @@ -866,8 +872,9 @@ describe('Channel', () => { it('throw in delivery handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user delivery handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'delivery'); cb(); }); ch.on('delivery', () => { throw expectedErr; }); @@ -880,8 +887,9 @@ describe('Channel', () => { it('throw in return handler is delivered via handler-error event', channelTest((ch, cb) => { const expectedErr = new Error('user return handler explodes'); - ch.on('handler-error', (err) => { + ch.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'return'); cb(); }); ch.on('return', () => { throw expectedErr; }); diff --git a/test/connection.test.js b/test/connection.test.js index 62932ab8..a4cd2640 100644 --- a/test/connection.test.js +++ b/test/connection.test.js @@ -423,8 +423,9 @@ describe('Connection', () => { it('throw in close handler is delivered via handler-error event', connectionTest((c, cb) => { const expectedErr = new Error('user close handler explodes'); - c.on('handler-error', (err) => { + c.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'close'); cb(); }); c.on('close', () => { throw expectedErr; }); @@ -443,8 +444,9 @@ describe('Connection', () => { it('throw in error handler is delivered via handler-error event', connectionTest((c, cb) => { const expectedErr = new Error('user error handler explodes'); - c.on('handler-error', (err) => { + c.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'error'); cb(); }); c.on('error', () => { throw expectedErr; }); @@ -463,8 +465,9 @@ describe('Connection', () => { it('throw in blocked handler is delivered via handler-error event', connectionTest((c, cb) => { const expectedErr = new Error('user blocked handler explodes'); - c.on('handler-error', (err) => { + c.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'blocked'); cb(); }); c.on('blocked', () => { throw expectedErr; }); @@ -477,8 +480,9 @@ describe('Connection', () => { it('throw in unblocked handler is delivered via handler-error event', connectionTest((c, cb) => { const expectedErr = new Error('user unblocked handler explodes'); - c.on('handler-error', (err) => { + c.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'unblocked'); cb(); }); c.on('unblocked', () => { throw expectedErr; }); @@ -491,8 +495,9 @@ describe('Connection', () => { it('throw in update-secret-ok handler is delivered via handler-error event', connectionTest((c, cb) => { const expectedErr = new Error('user update-secret-ok handler explodes'); - c.on('handler-error', (err) => { + c.on('handler-error', (err, event) => { assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'update-secret-ok'); cb(); }); c.open(OPEN_OPTS, (err) => { From 721657acfe5bcbb248e86341fb56b144323d600e Mon Sep 17 00:00:00 2001 From: Stephen Cresswell <229672+cressie176@users.noreply.github.com> Date: Fri, 8 May 2026 08:23:57 +0100 Subject: [PATCH 3/3] Fix: wrap remaining connection error emits with safeEmit Apply safeEmit to the three remaining bare this.emit('error') calls in Connection: closeWithError, onSocketError, and startHeartbeater. Adds tests for all three paths in both the without and with handler-error listener describe blocks. Co-Authored-By: Claude Opus 4.6 --- lib/connection.js | 6 +- test/connection.test.js | 123 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 3 deletions(-) diff --git a/lib/connection.js b/lib/connection.js index 5000da6c..44a6be8e 100644 --- a/lib/connection.js +++ b/lib/connection.js @@ -282,7 +282,7 @@ class Connection extends EventEmitter { } closeWithError(reason, code, error) { - this.emit('error', error); + safeEmit(this, 'error', error); this.closeBecause(reason, code); } @@ -291,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); } @@ -364,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); }); diff --git a/test/connection.test.js b/test/connection.test.js index a4cd2640..4187995d 100644 --- a/test/connection.test.js +++ b/test/connection.test.js @@ -310,6 +310,7 @@ describe('Connection', () => { afterEach(() => { prevUncaughtExceptionListeners.forEach((h) => process.on('uncaughtException', h)); + heartbeat.UNITS_TO_MS = 1000; }); it('throw in close handler from server-initiated close is swallowed without handler-error listener', connectionTest((c, cb) => { @@ -407,6 +408,65 @@ describe('Connection', () => { .then(() => send(defs.ConnectionUpdateSecretOk, {}, 0)) .then(cb, cb); })); + + it('throw in error handler from closeWithError becomes uncaught exception', connectionTest((c, cb) => { + const expectedErr = new Error('user error handler explodes on closeWithError'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.once('error', (err) => { + assert.match(err.message, /Unexpected frame on channel 0/); + throw expectedErr; + }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ChannelOpenOk, { channelId: Buffer.from('') }, 0)) + .then(cb, cb); + })); + + it('throw in error handler from onSocketError', connectionTest((c, cb) => { + const expectedErr = new Error('user error handler explodes on socket error'); + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + cb(); + }); + c.once('error', (err) => { + assert.match(err.message, /Unexpected close/); + throw expectedErr; + }); + c.open(OPEN_OPTS, (err) => { + assert.ifError(err); + c.sendHeartbeat(); + }); + }, (send, wait, cb, socket) => { + handshake(send, wait) + .then(wait()) + .then(() => socket.end()) + .then(cb, cb); + })); + + it('throw in error handler from heartbeat timeout', connectionTest((c, cb) => { + heartbeat.UNITS_TO_MS = 20; + const expectedErr = new Error('user error handler explodes on heartbeat timeout'); + const opts = Object.create(OPEN_OPTS); + opts.heartbeat = 1; + process.once('uncaughtException', (err) => { + assert.strictEqual(err, expectedErr); + c.heartbeater.clear(); + cb(); + }); + c.on('error', (err) => { + assert.match(err.message, /Heartbeat timeout/); + throw expectedErr; + }); + c.open(opts); + }, (send, wait, cb) => { + handshake(send, wait) + .then(cb, cb); + // conspicuously not sending anything ... + })); }); describe('Event handler errors - with handler-error listener', () => { @@ -419,6 +479,7 @@ describe('Connection', () => { afterEach(() => { prevUncaughtExceptionListeners.forEach((h) => process.on('uncaughtException', h)); + heartbeat.UNITS_TO_MS = 1000; }); it('throw in close handler is delivered via handler-error event', connectionTest((c, cb) => { @@ -512,6 +573,68 @@ describe('Connection', () => { .then(cb, cb); })); + it('throw in error handler from heartbeat timeout is delivered via handler-error event', connectionTest((c, cb) => { + heartbeat.UNITS_TO_MS = 20; + const expectedErr = new Error('user error handler explodes on heartbeat timeout'); + const opts = Object.create(OPEN_OPTS); + opts.heartbeat = 1; + c.on('handler-error', (err, event) => { + assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'error'); + c.heartbeater.clear(); + cb(); + }); + c.on('error', (err) => { + assert.match(err.message, /Heartbeat timeout/); + throw expectedErr; + }); + c.open(opts); + }, (send, wait, cb) => { + handshake(send, wait) + .then(cb, cb); + // conspicuously not sending anything ... + })); + + it('throw in error handler from closeWithError is delivered via handler-error event', connectionTest((c, cb) => { + const expectedErr = new Error('user error handler explodes on closeWithError'); + c.on('handler-error', (err, event) => { + assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'error'); + cb(); + }); + c.once('error', (err) => { + assert.match(err.message, /Unexpected frame on channel 0/); + throw expectedErr; + }); + c.open(OPEN_OPTS); + }, (send, wait, cb) => { + handshake(send, wait) + .then(() => send(defs.ChannelOpenOk, { channelId: Buffer.from('') }, 0)) + .then(cb, cb); + })); + + it('throw in error handler from onSocketError is delivered via handler-error event', connectionTest((c, cb) => { + const expectedErr = new Error('user error handler explodes on socket error'); + c.on('handler-error', (err, event) => { + assert.strictEqual(err, expectedErr); + assert.strictEqual(event, 'error'); + cb(); + }); + c.once('error', (err) => { + assert.match(err.message, /Unexpected close/); + throw expectedErr; + }); + c.open(OPEN_OPTS, (err) => { + assert.ifError(err); + c.sendHeartbeat(); + }); + }, (send, wait, cb, socket) => { + handshake(send, wait) + .then(wait()) + .then(() => socket.end()) + .then(cb, cb); + })); + it('throw in handler-error handler becomes uncaught exception', connectionTest((c, cb) => { const expectedErr = new Error('handler-error handler explodes'); process.once('uncaughtException', (err) => {