From 2a8421d7beb85bd5063aab212870e3ae57621ad2 Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Wed, 14 Oct 2026 09:31:42 +0300 Subject: [PATCH 1/2] feat: add cancelConsumersOnSigterm option to opt out of the SIGTERM consumer cancel (DS-3369) Processes with their own SIGTERM hook that waits on messages from these queues (e.g. eip-redis PubSub draining throttled events through eip-rabbit Timer queues) hang until SIGKILL once the consumers are cancelled. On by default, so existing behaviour is unchanged. Co-Authored-By: Claude Opus 5.5 --- README.md | 9 +++++++++ test/rabbit.test.ts | 9 +++++++++ ts/rabbit.ts | 6 +++++- 3 files changed, 23 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index c68e707..d530df8 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost', { scheduledPublishQueuesAsQuorum: false, //declare the publishWithDelay queues as quorum (see Scheduled publish queues as quorum) prefix: '', //prefix all queues with an application name tolerateQueueTypeMismatch: false, //consume an existing queue whose type or priority differs from the declared one (see Queue Type Behavior) + cancelConsumersOnSigterm: true, //cancel every consumer when the process receives SIGTERM (see Shutdown) socketOptions: {} // socketOptions will be passed as a second param to amqp.connect and from ther to the socket library (net or tls) }); @@ -280,10 +281,18 @@ Quorum queues need RabbitMQ 3.10 or later for per-message TTL and at-least-once Messages already in a classic `delay_{expiration}` queue are still dead-lettered to `delay_reply`. Keep at least one consumer running without the option until the classic delay queues are empty, then delete them along with `delay_reply`. +### Shutdown + +On `SIGTERM`, rabbit-queue cancels the consumers of every queue it created, so the process stops taking new messages while it shuts down. Set `cancelConsumersOnSigterm: false` when something else in the process has its own `SIGTERM` hook that waits on messages from these queues, e.g. eip-redis's PubSub waiting for throttled events that only complete when an eip-rabbit Timer queue delivers. With the consumers cancelled, that wait never ends and the process hangs until it is killed. On by default. +```javascript +const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: false }); +``` + ### Changelog ### New in v5.10.x +- `cancelConsumersOnSigterm` option. Set it to `false` to keep consuming after `SIGTERM`, for processes that drain work through their own `SIGTERM` hook. On by default. - `scheduledPublishQueuesAsQuorum` option. When enabled, the internal queues used by `publishWithDelay` are declared as quorum queues, under new names (`delay_quorum_{expiration}`, `delay_quorum_reply`). Off by default. - Queue declares, binds and unbinds run on short-lived channels. One that fails still rejects, but no longer closes the shared consume channel and disconnects every consumer. - `tolerateQueueTypeMismatch` option. When enabled, a queue that already exists with a different type or priority (classic vs quorum, with or without `x-max-priority`) is consumed as it is, with a warning. Old and new pods can then run side by side during a rolling deploy that changes a queue's type or priority, e.g. while migrating classic mirrored queues to quorum. Off by default. diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index 72a446e..81affa8 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -385,6 +385,15 @@ describe('Test rabbit class', function() { process.listeners('SIGTERM').should.not.containEql(handler); }); + it('should not register sigtermHandler when cancelConsumersOnSigterm is false', async function() { + const listenersBefore = process.listeners('SIGTERM'); + rabbit = new Rabbit(this.url, { cancelConsumersOnSigterm: false }); + await rabbit.connected; + + process.listeners('SIGTERM').should.eql(listenersBefore); + process.listeners('SIGTERM').should.not.containEql((rabbit as any).sigtermHandler); + }); + it('should call unsubscribe for all queues on SIGTERM', async function() { const stub = sandbox.stub(Queue.prototype, 'unsubscribe').resolves(); diff --git a/ts/rabbit.ts b/ts/rabbit.ts index 83104d7..96216fc 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -42,6 +42,7 @@ export default class Rabbit extends EventEmitter { defaultQueueType = undefined, scheduledPublishQueuesAsQuorum = false, tolerateQueueTypeMismatch = false, + cancelConsumersOnSigterm = true, }: { prefetch?: number replyPattern?: boolean @@ -51,6 +52,7 @@ export default class Rabbit extends EventEmitter { defaultQueueType?: string scheduledPublishQueuesAsQuorum?: boolean tolerateQueueTypeMismatch?: boolean + cancelConsumersOnSigterm?: boolean } = {} ) { super(); @@ -70,7 +72,9 @@ export default class Rabbit extends EventEmitter { this.sigtermHandler = this.handleSigterm.bind(this); this.reconnect(); - process.once('SIGTERM', this.sigtermHandler); + if (cancelConsumersOnSigterm) { + process.once('SIGTERM', this.sigtermHandler); + } } private async connect() { From aa67e038a892d6e369bb9280ec183d08f65af52e Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Wed, 14 Oct 2026 09:35:05 +0300 Subject: [PATCH 2/2] test: check consumers survive SIGTERM with cancelConsumersOnSigterm false; list option under v5.11.x Co-Authored-By: Claude Opus 5.5 --- README.md | 5 ++++- test/rabbit.test.ts | 11 ++++++++--- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index d530df8..6f24401 100644 --- a/README.md +++ b/README.md @@ -290,9 +290,12 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: fa ### Changelog -### New in v5.10.x +### New in v5.11.x - `cancelConsumersOnSigterm` option. Set it to `false` to keep consuming after `SIGTERM`, for processes that drain work through their own `SIGTERM` hook. On by default. + +### New in v5.10.x + - `scheduledPublishQueuesAsQuorum` option. When enabled, the internal queues used by `publishWithDelay` are declared as quorum queues, under new names (`delay_quorum_{expiration}`, `delay_quorum_reply`). Off by default. - Queue declares, binds and unbinds run on short-lived channels. One that fails still rejects, but no longer closes the shared consume channel and disconnects every consumer. - `tolerateQueueTypeMismatch` option. When enabled, a queue that already exists with a different type or priority (classic vs quorum, with or without `x-max-priority`) is consumed as it is, with a warning. Old and new pods can then run side by side during a rolling deploy that changes a queue's type or priority, e.g. while migrating classic mirrored queues to quorum. Off by default. diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index 81affa8..be4bd11 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -385,13 +385,18 @@ describe('Test rabbit class', function() { process.listeners('SIGTERM').should.not.containEql(handler); }); - it('should not register sigtermHandler when cancelConsumersOnSigterm is false', async function() { + it('should keep consumers on SIGTERM when cancelConsumersOnSigterm is false', async function() { + const stub = sandbox.stub(Queue.prototype, 'unsubscribe').resolves(); const listenersBefore = process.listeners('SIGTERM'); + rabbit = new Rabbit(this.url, { cancelConsumersOnSigterm: false }); await rabbit.connected; + await rabbit.createQueue(randomUUID().toString(), { exclusive: true }, () => ({})); - process.listeners('SIGTERM').should.eql(listenersBefore); - process.listeners('SIGTERM').should.not.containEql((rabbit as any).sigtermHandler); + const added = process.listeners('SIGTERM').filter(l => !listenersBefore.includes(l)); + added.should.be.empty(); + added.forEach(l => (l as () => void)()); + stub.called.should.be.false(); }); it('should call unsubscribe for all queues on SIGTERM', async function() {