From b65e9e5e304fdce7db6fb5321f479f2516545ebc Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Wed, 21 Oct 2026 23:57:24 +0300 Subject: [PATCH 1/4] fix: subscribe again after the broker cancels a consumer (DS-3536) Deleting a queue makes the broker cancel its consumers, and the cancel was ignored, so the consumer stayed gone until the process restarted. Converting a queue to quorum deletes and redeclares it. Wait until the queue exists again, without declaring it, then subscribe again with its own prefetch. Co-Authored-By: Claude Opus 5.5 --- README.md | 6 +++ test/rabbit.test.ts | 107 +++++++++++++++++++++++++++++++++++++++++++- ts/queue.ts | 29 +++++++++++- ts/rabbit.ts | 83 ++++++++++++++++++++++++++-------- 4 files changed, 204 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index 27e24ea..23d8448 100644 --- a/README.md +++ b/README.md @@ -270,6 +270,8 @@ While queues move between classic and quorum, old and new declares disagree on t const rabbit = new Rabbit(process.env.RABBIT_URL, { tolerateQueueTypeMismatch: true }); ``` +Changing a queue's type means deleting it and declaring it again, and deleting a queue makes the broker cancel its consumers. rabbit-queue then subscribes again on its own, once the queue exists again: it checks every 1 s, backing off to every 15 s, and never declares the queue itself, so it can't recreate a queue with the old arguments while someone is replacing it. It stops when the queue is unsubscribed, the connection is lost (the reconnect subscribes every queue again anyway) or `close()` is called. Exclusive and server-named queues are left alone, since only the connection that owned them could declare them again. + ### Scheduled publish queues as quorum `publishWithDelay` uses internal queues: a `delay_{expiration}` queue for each expiration, whose expired messages are dead-lettered to `delay_reply` and republished from there (both prefixed with `prefix_` when a prefix is set). They are declared without a queue type, so they get the virtual host's default type. With `scheduledPublishQueuesAsQuorum: true` they are declared as quorum queues, named `delay_quorum_{expiration}` and `delay_quorum_reply`, so they don't conflict with the existing classic ones. Off by default. @@ -290,6 +292,10 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: fa ### Changelog +### New in v5.12.x + +- A consumer the broker cancels, e.g. because its queue was deleted, subscribes again once the queue exists again, with the queue's own prefetch. Before, it stayed cancelled until the process restarted. + ### 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. diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index fe19c8f..f23b548 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -598,8 +598,8 @@ describe('Test rabbit class', function() { return count; } - async function waitForConsumers(expected: number) { - for (let attempt = 0; attempt < 50 && (await consumerCount()) !== expected; attempt++) { + async function waitForConsumers(expected: number, attempts = 50) { + for (let attempt = 0; attempt < attempts && (await consumerCount()) !== expected; attempt++) { await new Promise(r => setTimeout(r, 20)); } (await consumerCount()).should.equal(expected); @@ -641,6 +641,109 @@ describe('Test rabbit class', function() { (await consumerCount()).should.equal(1); }); + describe('when the broker cancels a consumer because its queue was deleted', function() { + this.timeout(10000); + + async function replaceQueue(args: object, afterMs: number) { + const channel = await admin.createChannel(); + await channel.deleteQueue(queueName); + await new Promise(r => setTimeout(r, afterMs)); + await channel.assertQueue(queueName, { durable: true, arguments: args }); + await channel.close(); + } + + async function queueExists() { + const channel = await admin.createChannel(); + channel.on('error', () => undefined); + try { + await channel.checkQueue(queueName); + await channel.close(); + return true; + } catch (error) { + return false; + } + } + + it('should consume again once the queue is declared again', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); + let onMessage: (content: string) => void; + const received = new Promise(resolve => (onMessage = resolve)); + await rabbit.createQueue(queueName, {}, (msg, ack) => { + ack(); + onMessage(msg.content.toString()); + }); + + await replaceQueue({ 'x-queue-type': 'quorum' }, 300); + await waitForConsumers(0); + await waitForConsumers(1, 250); + await rabbit.publish(queueName, { test: true }); + + (await received).should.equal(JSON.stringify({ test: true })); + (await consumerCount()).should.equal(1); + disconnected.called.should.be.false(); + }); + + it('should wait for the queue instead of declaring it', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + await rabbit.createQueue(queueName, {}, () => ({})); + + const channel = await admin.createChannel(); + await channel.deleteQueue(queueName); + await channel.close(); + await new Promise(r => setTimeout(r, 1500)); + (await queueExists()).should.be.false(); + + await replaceQueue({}, 0); + await waitForConsumers(1, 250); + }); + + it('should subscribe again with the queue\'s own prefetch', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + await rabbit.createQueue(queueName, { prefetch: 5 }, () => ({})); + await rabbit.createQueue(`other_${randomUUID()}`, { exclusive: true, prefetch: 1 }, () => ({})); + const prefetch = sandbox.spy(rabbit.consumeChannel, 'prefetch'); + + await replaceQueue({}, 0); + await waitForConsumers(1, 250); + + (await consumerCount()).should.equal(1); + prefetch.calledWith(5).should.be.true(); + }); + + it('should not subscribe again after unsubscribing', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + await rabbit.createQueue(queueName, {}, () => ({})); + + const channel = await admin.createChannel(); + await channel.deleteQueue(queueName); + await channel.close(); + await rabbit.unsubscribe(queueName); + await replaceQueue({}, 0); + await new Promise(r => setTimeout(r, 1500)); + + (await consumerCount()).should.equal(0); + }); + + it('should not subscribe again when unsubscribed while subscribing again', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const queue = await rabbit.createQueue(queueName, {}, () => ({})); + const exists = queue.exists.bind(queue); + sandbox.stub(queue, 'exists').callsFake(async () => { + const found = await exists(); + await rabbit.unsubscribe(queueName); + return found; + }); + + await replaceQueue({}, 0); + await new Promise(r => setTimeout(r, 1500)); + + (queue.exists as sinon.SinonStub).called.should.be.true(); + (await consumerCount()).should.equal(0); + }); + }); + it('should emit "disconnected" once when a connection and its channel close', async function() { rabbit = new Rabbit(this.url, { replyPattern: false }); await rabbit.connected; diff --git a/ts/queue.ts b/ts/queue.ts index e7f2cbe..03b9a37 100644 --- a/ts/queue.ts +++ b/ts/queue.ts @@ -8,6 +8,7 @@ import { encode } from './encode-decode'; import getLogger from './logger'; const logger = getLogger('rabbit-queue'); +const NOT_FOUND = 404; const PRECONDITION_FAILED = 406; // What changes when a queue moves between classic and quorum: quorum queues have no x-max-priority. const TOLERATED_MISMATCHES = ['x-queue-type', 'x-max-priority']; @@ -76,6 +77,7 @@ export default class Queue { public options: QueueOptions; public created: Promise; public handler: Function; + public onCancel?: () => void; private tag: amqp.Replies.Consume; constructor(public channel: Channel, public name: string, options: QueueOptions) { @@ -150,10 +152,20 @@ export default class Queue { } } + get subscribed() { + return !!this.tag; + } + async subscribe(handler: (msg: any, ack: (error?, reply?) => any) => any) { await this.created; this.handler = handler; - this.tag = await this.channel.consume(this.name, this.onMessage.bind(this), { noAck: this.options.noAck }); + const tag = await this.channel.consume(this.name, this.onMessage.bind(this), { noAck: this.options.noAck }); + // An unsubscribe while the consume was in flight found no tag to cancel. + if (this.handler !== handler) { + await this.channel.cancel(tag.consumerTag); + return; + } + this.tag = tag; } async unsubscribe() { @@ -164,6 +176,18 @@ export default class Queue { this.tag = null; } + async exists() { + try { + await withDeclareChannel(this.channel, this.name, channel => channel.checkQueue(this.name)); + return true; + } catch (error) { + if (error.code === NOT_FOUND) { + return false; + } + throw error; + } + } + static async destroy(channel: Channel, name: string) { await channel.deleteQueue(name); } @@ -178,7 +202,10 @@ export default class Queue { this.channel.ack(msg); } }; + // amqplib passes null only when the broker cancels the consumer, e.g. because the queue was deleted. if (!msg) { + this.tag = null; + this.onCancel?.(); return; } this.handler(msg, async (error, reply) => { diff --git a/ts/rabbit.ts b/ts/rabbit.ts index bbae754..9238c9a 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -8,6 +8,9 @@ import Exchange from './exchange'; import * as assert from 'assert'; import getLogger from './logger'; +const RESUBSCRIBE_DELAY_MS = 1000; +const MAX_RESUBSCRIBE_DELAY_MS = 15000; + export default class Rabbit extends EventEmitter { static STOP_PROPAGATION = Queue.STOP_PROPAGATION; static STOP_STREAM = Queue.STOP_STREAM; @@ -211,31 +214,75 @@ export default class Rabbit extends EventEmitter { tolerateQueueTypeMismatch: this.tolerateQueueTypeMismatch }); this.queues[name] = queue; + queue.onCancel = () => this.resubscribe(queue, options.prefetch); await queue.created; this.logger.debug(`created queue ${name}`); if (handler) { - // Handle race condition. Another queue might be created at the same time with diferrent prefetch. - let localLock; - do { - localLock = this.lock; - await this.lock; - // More than one queues might be waiting to be created. Only one can pass this check. - } while (this.lock !== localLock); - - if (this.prefetch !== options.prefetch) { - this.prefetch = options.prefetch; - this.lock = Promise.resolve(this.consumeChannel.prefetch(options.prefetch)).then(() => - queue.subscribe(handler) - ); - await this.lock; - } else { - this.lock = queue.subscribe(handler); - await this.lock; - } + await this.subscribeWithPrefetch(queue, handler, options.prefetch); } return queue; } + private async subscribeWithPrefetch(queue: Queue, handler, prefetch: number, stillWanted = () => true) { + // Handle race condition. Another queue might be created at the same time with diferrent prefetch. + let localLock; + do { + localLock = this.lock; + await this.lock; + // More than one queues might be waiting to be created. Only one can pass this check. + } while (this.lock !== localLock); + + if (!stillWanted()) { + return false; + } + if (this.prefetch !== prefetch) { + this.prefetch = prefetch; + this.lock = Promise.resolve(this.consumeChannel.prefetch(prefetch)).then(() => queue.subscribe(handler)); + } else { + this.lock = queue.subscribe(handler); + } + await this.lock; + return true; + } + + // The broker cancels a consumer when its queue is deleted, e.g. while the queue is converted to quorum. Waits for + // the queue to exist again instead of declaring it, so a consumer never recreates a queue someone is replacing. + // A queue that only this connection can declare (exclusive or server-named) won't come back, so it isn't awaited. + private async resubscribe(queue: Queue, prefetch: number) { + const handler = queue.handler; + if (!handler || !queue.name || queue.options.exclusive) { + return; + } + // Stops on unsubscribe, close() and connection loss (whose reconnect subscribes afresh), or once subscribed again. + const stillWanted = () => + !this.closed && + !this.lossReported && + this.queues[queue.name] === queue && + queue.channel === this.consumeChannel && + queue.handler === handler && + !queue.subscribed; + this.logger.warn(`Consumer for queue ${queue.name} was cancelled by the broker, subscribing again once it exists`); + let delay = RESUBSCRIBE_DELAY_MS; + while (stillWanted()) { + await new Promise(resolve => setTimeout(resolve, delay).unref()); + delay = Math.min(delay * 2, MAX_RESUBSCRIBE_DELAY_MS); + try { + // A queue deleted again between the check and the consume closes the consume channel with a 404, and the + // reconnect that follows subscribes every queue afresh. + const subscribed = + stillWanted() && + (await queue.exists()) && + (await this.subscribeWithPrefetch(queue, handler, prefetch, stillWanted)); + if (subscribed) { + this.logger.info(`Subscribed to queue ${queue.name} again`); + return; + } + } catch (error) { + this.logger.warn(`Failed to subscribe to queue ${queue.name} again: ${error.message}`); + } + } + } + async destroyQueue(name: string, prefix?: string) { name = this.updateName(name, prefix); await this.connected; From e4d869851a7b174ae75b88182da7774e85ace94e Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Thu, 22 Oct 2026 00:17:12 +0300 Subject: [PATCH 2/4] fix: harden the re-subscribe against review findings (DS-3536) - Keep the subscribe lock usable after a subscribe fails, so a later createQueue (e.g. after the reconnect) doesn't reject with a stale error. - Check again after the prefetch, so an unsubscribe or SIGTERM that lands meanwhile isn't undone. - Match a broker cancel to the consume it belongs to, so one that arrives before the consume resolves isn't lost. - Hand deliveries back to the broker when there's no handler. - Listen for 'error' on the consume and publish channels: a channel the broker closes threw without a listener. - destroyQueue unsubscribes first, so deleting our own queue doesn't wait for it to come back. - Log exclusive and server-named cancels as errors, a warning every 5 min while waiting, and the error code on failures; one loop per queue. Co-Authored-By: Claude Opus 5.5 --- README.md | 4 +- test/queue.test.ts | 71 +++++++++++++++++ test/rabbit.test.ts | 188 ++++++++++++++++++++++++++++++++++++++------ ts/queue.ts | 40 ++++++++-- ts/rabbit.ts | 95 +++++++++++++++------- 5 files changed, 335 insertions(+), 63 deletions(-) diff --git a/README.md b/README.md index 23d8448..32c21c9 100644 --- a/README.md +++ b/README.md @@ -270,7 +270,7 @@ While queues move between classic and quorum, old and new declares disagree on t const rabbit = new Rabbit(process.env.RABBIT_URL, { tolerateQueueTypeMismatch: true }); ``` -Changing a queue's type means deleting it and declaring it again, and deleting a queue makes the broker cancel its consumers. rabbit-queue then subscribes again on its own, once the queue exists again: it checks every 1 s, backing off to every 15 s, and never declares the queue itself, so it can't recreate a queue with the old arguments while someone is replacing it. It stops when the queue is unsubscribed, the connection is lost (the reconnect subscribes every queue again anyway) or `close()` is called. Exclusive and server-named queues are left alone, since only the connection that owned them could declare them again. +Changing a queue's type means deleting it and declaring it again, and deleting a queue makes the broker cancel its consumers. rabbit-queue then subscribes again on its own, once the queue exists again: it checks every 1 s, backing off to every 15 s, and logs a warning every 5 minutes while it waits. It doesn't declare the queue while waiting, so it won't recreate a queue with the old arguments while someone is replacing it. It stops when the queue is unsubscribed, the connection is lost (your `connected` handler creates the queues again, as above) or `close()` is called. Exclusive and server-named queues belong to the connection, so nothing else will declare them again: their cancel is logged as an error instead. `destroyQueue` unsubscribes the queue first, so deleting your own queue doesn't start the wait. ### Scheduled publish queues as quorum @@ -294,7 +294,7 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: fa ### New in v5.12.x -- A consumer the broker cancels, e.g. because its queue was deleted, subscribes again once the queue exists again, with the queue's own prefetch. Before, it stayed cancelled until the process restarted. +- A consumer the broker cancels, e.g. because its queue was deleted, subscribes again once the queue exists again, with the queue's own prefetch. Before, it stayed cancelled until the process restarted or reconnected. `destroyQueue` now unsubscribes the queue before deleting it. ### New in v5.11.x diff --git a/test/queue.test.ts b/test/queue.test.ts index 2ecd49c..7407d01 100644 --- a/test/queue.test.ts +++ b/test/queue.test.ts @@ -111,6 +111,77 @@ describe('Test Queue class', function () { this.spyCancel.calledOnce.should.be.true(); }); + it('should call onCancel when the broker cancels the consumer', async function () { + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + queue.onCancel = sandbox.spy(); + (await queue.subscribe(() => ({}))).should.be.true(); + + await rabbit.consumeChannel.deleteQueue(this.name); + for (let attempt = 0; attempt < 50 && queue.subscribed; attempt++) { + await new Promise(r => setTimeout(r, 20)); + } + + queue.subscribed.should.be.false(); + (queue.onCancel as sinon.SinonSpy).calledOnce.should.be.true(); + }); + + it('should report a broker cancel that arrives before the consume resolves', async function () { + this.spyConsume.restore(); + const consume = rabbit.consumeChannel.consume.bind(rabbit.consumeChannel); + sandbox.stub(rabbit.consumeChannel, 'consume').callsFake(async (name, onMessage, options) => { + const tag = await consume(name, onMessage, options); + onMessage(null); + return tag; + }); + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + queue.onCancel = sandbox.spy(); + + (await queue.subscribe(() => ({}))).should.be.false(); + + queue.subscribed.should.be.false(); + (queue.onCancel as sinon.SinonSpy).calledOnce.should.be.true(); + }); + + it('should cancel a consumer whose consume resolves after an unsubscribe', async function () { + this.spyConsume.restore(); + const consume = rabbit.consumeChannel.consume.bind(rabbit.consumeChannel); + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + let consumerTag: string; + sandbox.stub(rabbit.consumeChannel, 'consume').callsFake(async (name, onMessage, options) => { + const tag = await consume(name, onMessage, options); + consumerTag = tag.consumerTag; + await queue.unsubscribe(); + return tag; + }); + + (await queue.subscribe(() => ({}))).should.be.false(); + + queue.subscribed.should.be.false(); + this.spyCancel.calledWith(consumerTag).should.be.true(); + }); + + it('should hand deliveries back when there is no handler', async function () { + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + await queue.created; + const nack = sandbox.stub(rabbit.consumeChannel, 'nack'); + const msg = { content: Buffer.from('{}'), properties: { headers: {} }, fields: {} } as any; + + queue.onMessage(msg); + + nack.calledOnceWith(msg, false, true).should.be.true(); + }); + + it('should tell whether the queue exists', async function () { + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + await queue.created; + (await queue.exists()).should.be.true(); + + const missing = new Queue(rabbit.consumeChannel, `missing_${Date.now()}`, { exclusive: true }); + await missing.created; + await rabbit.consumeChannel.deleteQueue(missing.name); + (await missing.exists()).should.be.false(); + }); + it('should destroy queue', async function () { const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); await queue.created; diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index f23b548..bcc8186 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -124,6 +124,7 @@ describe('Test rabbit class', function() { const prefetchAtSubscribe = {}; stub.callsFake(async function() { prefetchAtSubscribe[this.name] = spy.lastCall.args[0]; + return true; }); const handler = () => ({}); let promises = []; @@ -593,9 +594,17 @@ describe('Test rabbit class', function() { async function consumerCount() { const channel = await admin.createChannel(); - const { consumerCount: count } = await channel.checkQueue(queueName); - await channel.close(); - return count; + channel.on('error', () => undefined); + try { + const { consumerCount: count } = await channel.checkQueue(queueName); + await channel.close(); + return count; + } catch (error) { + if (error.code === 404) { + return 0; + } + throw error; + } } async function waitForConsumers(expected: number, attempts = 50) { @@ -644,10 +653,22 @@ describe('Test rabbit class', function() { describe('when the broker cancels a consumer because its queue was deleted', function() { this.timeout(10000); - async function replaceQueue(args: object, afterMs: number) { + async function waitUntil(condition: () => boolean | Promise, timeoutMs = 5000) { + const deadline = Date.now() + timeoutMs; + while (!(await condition()) && Date.now() < deadline) { + await new Promise(r => setTimeout(r, 20)); + } + (await condition()).should.be.true(); + } + + async function deleteQueue() { const channel = await admin.createChannel(); await channel.deleteQueue(queueName); - await new Promise(r => setTimeout(r, afterMs)); + await channel.close(); + } + + async function declareQueue(args: object = {}) { + const channel = await admin.createChannel(); await channel.assertQueue(queueName, { durable: true, arguments: args }); await channel.close(); } @@ -664,6 +685,13 @@ describe('Test rabbit class', function() { } } + async function consumeAgain(queue: Queue) { + await deleteQueue(); + await waitUntil(() => !queue.subscribed); + await declareQueue(); + await waitUntil(async () => (await consumerCount()) === 1); + } + it('should consume again once the queue is declared again', async function() { rabbit = new Rabbit(this.url, { replyPattern: false }); const disconnected = sandbox.spy(); @@ -675,71 +703,179 @@ describe('Test rabbit class', function() { onMessage(msg.content.toString()); }); - await replaceQueue({ 'x-queue-type': 'quorum' }, 300); - await waitForConsumers(0); - await waitForConsumers(1, 250); + await deleteQueue(); + await new Promise(r => setTimeout(r, 300)); + await declareQueue({ 'x-queue-type': 'quorum' }); + await waitUntil(async () => (await consumerCount()) === 1); await rabbit.publish(queueName, { test: true }); (await received).should.equal(JSON.stringify({ test: true })); - (await consumerCount()).should.equal(1); disconnected.called.should.be.false(); }); + it('should consume again after every cancel', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const queue = await rabbit.createQueue(queueName, {}, () => ({})); + + await consumeAgain(queue); + await consumeAgain(queue); + + queue.subscribed.should.be.true(); + }); + it('should wait for the queue instead of declaring it', async function() { rabbit = new Rabbit(this.url, { replyPattern: false }); + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); await rabbit.createQueue(queueName, {}, () => ({})); - const channel = await admin.createChannel(); - await channel.deleteQueue(queueName); - await channel.close(); + await deleteQueue(); await new Promise(r => setTimeout(r, 1500)); (await queueExists()).should.be.false(); - await replaceQueue({}, 0); - await waitForConsumers(1, 250); + await declareQueue(); + await waitUntil(async () => (await consumerCount()) === 1); + disconnected.called.should.be.false(); }); it('should subscribe again with the queue\'s own prefetch', async function() { rabbit = new Rabbit(this.url, { replyPattern: false }); - await rabbit.createQueue(queueName, { prefetch: 5 }, () => ({})); + const queue = await rabbit.createQueue(queueName, { prefetch: 5 }, () => ({})); await rabbit.createQueue(`other_${randomUUID()}`, { exclusive: true, prefetch: 1 }, () => ({})); const prefetch = sandbox.spy(rabbit.consumeChannel, 'prefetch'); - await replaceQueue({}, 0); - await waitForConsumers(1, 250); + await consumeAgain(queue); - (await consumerCount()).should.equal(1); prefetch.calledWith(5).should.be.true(); }); it('should not subscribe again after unsubscribing', async function() { rabbit = new Rabbit(this.url, { replyPattern: false }); - await rabbit.createQueue(queueName, {}, () => ({})); + const queue = await rabbit.createQueue(queueName, {}, () => ({})); + const exists = sandbox.spy(queue, 'exists'); - const channel = await admin.createChannel(); - await channel.deleteQueue(queueName); - await channel.close(); + await deleteQueue(); + await waitUntil(() => !queue.subscribed); await rabbit.unsubscribe(queueName); - await replaceQueue({}, 0); + await declareQueue(); await new Promise(r => setTimeout(r, 1500)); + exists.called.should.be.false(); (await consumerCount()).should.equal(0); }); - it('should not subscribe again when unsubscribed while subscribing again', async function() { + it('should not subscribe again when unsubscribed while checking the queue', async function() { rabbit = new Rabbit(this.url, { replyPattern: false }); const queue = await rabbit.createQueue(queueName, {}, () => ({})); const exists = queue.exists.bind(queue); + let checked: () => void; + const wasChecked = new Promise(resolve => (checked = resolve)); sandbox.stub(queue, 'exists').callsFake(async () => { const found = await exists(); await rabbit.unsubscribe(queueName); + checked(); return found; }); - await replaceQueue({}, 0); + await deleteQueue(); + await declareQueue(); + await wasChecked; + await new Promise(r => setTimeout(r, 200)); + + (await consumerCount()).should.equal(0); + }); + + it('should not subscribe again when unsubscribed while setting the prefetch', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + await rabbit.createQueue(queueName, { prefetch: 5 }, () => ({})); + await rabbit.createQueue(`other_${randomUUID()}`, { exclusive: true, prefetch: 1 }, () => ({})); + const prefetch = rabbit.consumeChannel.prefetch.bind(rabbit.consumeChannel); + let prefetched: () => void; + const wasPrefetched = new Promise(resolve => (prefetched = resolve)); + sandbox.stub(rabbit.consumeChannel, 'prefetch').callsFake(async (count: number) => { + const result = await prefetch(count); + await rabbit.unsubscribe(queueName); + prefetched(); + return result; + }); + + await deleteQueue(); + await declareQueue(); + await wasPrefetched; + await new Promise(r => setTimeout(r, 200)); + + (await consumerCount()).should.equal(0); + }); + + it('should create queues again after a subscribe failed and the app reconnected', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + resubscribeOnReconnect(); + await rabbit.connected; + await waitForConsumers(1); + const queue = rabbit.queues[queueName]; + // The queue is deleted again between the check and the consume: the consume fails with a 404. + sandbox.stub(queue, 'exists').resolves(true); + const reconnected = new Promise(resolve => rabbit.once('connected', resolve)); + + await deleteQueue(); + await reconnected; + + await waitUntil(async () => (await consumerCount()) === 1); + }); + + it('should stop waiting on close()', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const queue = await rabbit.createQueue(queueName, {}, () => ({})); + const exists = sandbox.spy(queue, 'exists'); + + await deleteQueue(); + await waitUntil(() => !queue.subscribed); + await rabbit.close(); + rabbit = null; await new Promise(r => setTimeout(r, 1500)); - (queue.exists as sinon.SinonStub).called.should.be.true(); + exists.called.should.be.false(); + }); + + it('should stop waiting when the connection is lost', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const queue = await rabbit.createQueue(queueName, {}, () => ({})); + const exists = sandbox.spy(queue, 'exists'); + + await deleteQueue(); + await waitUntil(() => !queue.subscribed); + await rabbit.consumeConnection.close(); + await new Promise(r => setTimeout(r, 1500)); + + exists.called.should.be.false(); + }); + + it('should log an error and not wait for an exclusive queue', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const name = `exclusive_${randomUUID()}`; + const queue = await rabbit.createQueue(name, { exclusive: true }, () => ({})); + const exists = sandbox.spy(queue, 'exists'); + const error = sandbox.spy(rabbit.logger, 'error'); + + await rabbit.consumeChannel.deleteQueue(name); + await waitUntil(() => !queue.subscribed); + await new Promise(r => setTimeout(r, 1200)); + + error.calledWithMatch(/won't subscribe again/).should.be.true(); + exists.called.should.be.false(); + }); + + it('should not wait for a queue it destroyed itself', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const queue = await rabbit.createQueue(queueName, {}, () => ({})); + const warn = sandbox.spy(rabbit.logger, 'warn'); + + await rabbit.destroyQueue(queueName); + await declareQueue(); + await new Promise(r => setTimeout(r, 1200)); + + warn.called.should.be.false(); + queue.subscribed.should.be.false(); (await consumerCount()).should.equal(0); }); }); diff --git a/ts/queue.ts b/ts/queue.ts index 03b9a37..c6997b1 100644 --- a/ts/queue.ts +++ b/ts/queue.ts @@ -156,16 +156,38 @@ export default class Queue { return !!this.tag; } + // Resolves to whether the consumer is running: an unsubscribe or a broker cancel can land before the consume resolves. async subscribe(handler: (msg: any, ack: (error?, reply?) => any) => any) { await this.created; this.handler = handler; - const tag = await this.channel.consume(this.name, this.onMessage.bind(this), { noAck: this.options.noAck }); - // An unsubscribe while the consume was in flight found no tag to cancel. + let tag: amqp.Replies.Consume; + let cancelled = false; + // amqplib calls back with null only when the broker cancels the consumer, e.g. because the queue was deleted. + // A callback per consume tells this consumer's cancel apart from an older one's. + const onMessage = (msg: amqp.Message) => { + if (msg) { + return this.onMessage(msg); + } + cancelled = true; + if (tag && this.tag === tag) { + this.tag = null; + this.onCancel?.(); + } + }; + tag = await this.channel.consume(this.name, onMessage, { noAck: this.options.noAck }); if (this.handler !== handler) { - await this.channel.cancel(tag.consumerTag); - return; + // The unsubscribe found no tag to cancel. + if (!cancelled) { + await this.channel.cancel(tag.consumerTag); + } + return false; + } + if (cancelled) { + this.onCancel?.(); + return false; } this.tag = tag; + return true; } async unsubscribe() { @@ -202,10 +224,14 @@ export default class Queue { this.channel.ack(msg); } }; - // amqplib passes null only when the broker cancels the consumer, e.g. because the queue was deleted. if (!msg) { - this.tag = null; - this.onCancel?.(); + return; + } + // Deliveries can arrive between the consume-ok and an unsubscribe cancelling it; hand them back to the broker. + if (!this.handler) { + if (!this.options.noAck) { + this.channel.nack(msg, false, true); + } return; } this.handler(msg, async (error, reply) => { diff --git a/ts/rabbit.ts b/ts/rabbit.ts index 9238c9a..c53b8b5 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -10,6 +10,7 @@ import getLogger from './logger'; const RESUBSCRIBE_DELAY_MS = 1000; const MAX_RESUBSCRIBE_DELAY_MS = 15000; +const STILL_WAITING_LOG_INTERVAL_MS = 5 * 60 * 1000; export default class Rabbit extends EventEmitter { static STOP_PROPAGATION = Queue.STOP_PROPAGATION; @@ -28,6 +29,7 @@ export default class Rabbit extends EventEmitter { private lossReported = false; private attemptError: Error | null = null; private closed = false; + private resubscribing = new Set(); public prefetch: number; public replyPattern: boolean; public prefix: string; @@ -179,6 +181,9 @@ export default class Rabbit extends EventEmitter { async initChannel(channel: Channel, publish = false) { channel.prefetch(this.prefetch); channel.on('close', error => this.connectionLost(channel.channelModel, error)); + // A channel the broker closes (e.g. a consume of a queue deleted a moment before) emits 'error' first, which + // would throw without a listener. + channel.on('error', error => this.connectionLost(channel.channelModel, error)); if (!publish && this.replyPattern) { await createReplyQueue(channel); } @@ -214,7 +219,12 @@ export default class Rabbit extends EventEmitter { tolerateQueueTypeMismatch: this.tolerateQueueTypeMismatch }); this.queues[name] = queue; - queue.onCancel = () => this.resubscribe(queue, options.prefetch); + queue.onCancel = () => { + // Nothing awaits this; console in case the failure came from a 'log' listener. + this.resubscribe(queue, options.prefetch).catch(error => + console.error(`[error] rabbit-queue Subscribing to queue ${name} again failed`, error) + ); + }; await queue.created; this.logger.debug(`created queue ${name}`); if (handler) { @@ -224,36 +234,51 @@ export default class Rabbit extends EventEmitter { } private async subscribeWithPrefetch(queue: Queue, handler, prefetch: number, stillWanted = () => true) { - // Handle race condition. Another queue might be created at the same time with diferrent prefetch. + // Handle race condition. Another subscribe (a new queue or a resubscribe) might run at the same time with a + // different prefetch. let localLock; do { localLock = this.lock; await this.lock; - // More than one queues might be waiting to be created. Only one can pass this check. + // More than one subscribe might be waiting. Only one can pass this check. } while (this.lock !== localLock); if (!stillWanted()) { return false; } + let subscribing: Promise; if (this.prefetch !== prefetch) { this.prefetch = prefetch; - this.lock = Promise.resolve(this.consumeChannel.prefetch(prefetch)).then(() => queue.subscribe(handler)); + subscribing = Promise.resolve(this.consumeChannel.prefetch(prefetch)).then( + () => stillWanted() && queue.subscribe(handler) + ); } else { - this.lock = queue.subscribe(handler); + subscribing = Promise.resolve(queue.subscribe(handler)); } - await this.lock; - return true; + // The lock only orders subscribes: one that fails mustn't fail every one after it, even after a reconnect. + this.lock = subscribing.then( + () => undefined, + () => undefined + ); + return subscribing; } // The broker cancels a consumer when its queue is deleted, e.g. while the queue is converted to quorum. Waits for - // the queue to exist again instead of declaring it, so a consumer never recreates a queue someone is replacing. - // A queue that only this connection can declare (exclusive or server-named) won't come back, so it isn't awaited. + // the queue to exist again instead of declaring it, so a consumer doesn't recreate a queue someone is replacing. private async resubscribe(queue: Queue, prefetch: number) { const handler = queue.handler; - if (!handler || !queue.name || queue.options.exclusive) { + if (!handler || this.resubscribing.has(queue)) { return; } - // Stops on unsubscribe, close() and connection loss (whose reconnect subscribes afresh), or once subscribed again. + // Exclusive and server-named queues belong to this connection: nothing else will declare them again. + if (!queue.name || queue.options.exclusive) { + this.logger.error( + `Consumer for queue ${queue.name || '(server-named)'} was cancelled by the broker and won't subscribe again` + ); + return; + } + // Stops on unsubscribe, close() and connection loss (the app's 'connected' handler creates the queues again), or + // once subscribed again. const stillWanted = () => !this.closed && !this.lossReported && @@ -261,31 +286,45 @@ export default class Rabbit extends EventEmitter { queue.channel === this.consumeChannel && queue.handler === handler && !queue.subscribed; - this.logger.warn(`Consumer for queue ${queue.name} was cancelled by the broker, subscribing again once it exists`); - let delay = RESUBSCRIBE_DELAY_MS; - while (stillWanted()) { - await new Promise(resolve => setTimeout(resolve, delay).unref()); - delay = Math.min(delay * 2, MAX_RESUBSCRIBE_DELAY_MS); - try { - // A queue deleted again between the check and the consume closes the consume channel with a 404, and the - // reconnect that follows subscribes every queue afresh. - const subscribed = - stillWanted() && - (await queue.exists()) && - (await this.subscribeWithPrefetch(queue, handler, prefetch, stillWanted)); - if (subscribed) { - this.logger.info(`Subscribed to queue ${queue.name} again`); - return; + this.resubscribing.add(queue); + try { + this.logger.warn(`Consumer for queue ${queue.name} was cancelled by the broker, subscribing again once it exists`); + const cancelledAt = Date.now(); + let loggedAt = cancelledAt; + let delay = RESUBSCRIBE_DELAY_MS; + while (stillWanted()) { + await new Promise(resolve => setTimeout(resolve, delay).unref()); + delay = Math.min(delay * 2, MAX_RESUBSCRIBE_DELAY_MS); + try { + // A queue deleted again between the check and the consume closes the consume channel with a 404, which + // emits 'disconnected' like any other connection loss. + const subscribed = + stillWanted() && + (await queue.exists()) && + (await this.subscribeWithPrefetch(queue, handler, prefetch, stillWanted)); + if (subscribed) { + this.logger.info(`Subscribed to queue ${queue.name} again`); + return; + } + } catch (error) { + this.logger.warn(`Failed to subscribe to queue ${queue.name} again (${error.code}): ${error.message}`); + } + if (Date.now() - loggedAt >= STILL_WAITING_LOG_INTERVAL_MS && stillWanted()) { + loggedAt = Date.now(); + const minutes = Math.round((loggedAt - cancelledAt) / 60000); + this.logger.warn(`Still waiting for queue ${queue.name} to exist, ${minutes} min after its consumer was cancelled`); } - } catch (error) { - this.logger.warn(`Failed to subscribe to queue ${queue.name} again: ${error.message}`); } + } finally { + this.resubscribing.delete(queue); } } async destroyQueue(name: string, prefix?: string) { name = this.updateName(name, prefix); await this.connected; + // Deleting the queue would make the broker cancel our own consumer, which would then wait for the queue to come back. + await this.queues[name]?.unsubscribe(); await Queue.destroy(this.consumeChannel, name); } From 1a8d819a2547665c37288aac855473d76421bb0a Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Thu, 22 Oct 2026 10:22:34 +0300 Subject: [PATCH 3/4] fix: keep the default prefetch and close re-subscribe gaps (DS-3536) - Set the prefetch on every subscribe instead of storing it in prefetch, which is also the default for queues created without one. A re-subscribe left its queue's prefetch as the default, so after the next reconnect maestro's default queues would come back at 3 instead of 100. - Cancel a consumer that gets a delivery with no handler. One left behind by a second subscribe got the message back on every delivery, until a quorum queue's delivery limit dropped it. - destroyQueue skips the unsubscribe for a queue from before a reconnect: the cancel on its closed channel threw, so the queue was never deleted. - Set the handler before awaiting the queue's creation, so an unsubscribe during that wait is caught after the consume too. Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + test/queue.test.ts | 33 ++++++++++++++++++++++++++++++++- test/rabbit.test.ts | 30 ++++++++++++++++++++++++++++++ ts/queue.ts | 5 ++++- ts/rabbit.ts | 19 +++++++++---------- 5 files changed, 76 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 32c21c9..2f5f01e 100644 --- a/README.md +++ b/README.md @@ -295,6 +295,7 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: fa ### New in v5.12.x - A consumer the broker cancels, e.g. because its queue was deleted, subscribes again once the queue exists again, with the queue's own prefetch. Before, it stayed cancelled until the process restarted or reconnected. `destroyQueue` now unsubscribes the queue before deleting it. +- `prefetch` stays the default for queues created without their own. Before, subscribing a queue with its own prefetch made it the default for every queue created after it, e.g. again after a reconnect. ### New in v5.11.x diff --git a/test/queue.test.ts b/test/queue.test.ts index 7407d01..4891e20 100644 --- a/test/queue.test.ts +++ b/test/queue.test.ts @@ -160,17 +160,48 @@ describe('Test Queue class', function () { this.spyCancel.calledWith(consumerTag).should.be.true(); }); + it('should not consume when unsubscribed while waiting for the queue', async function () { + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + queue.created = queue.created.then(() => queue.unsubscribe()); + + (await queue.subscribe(() => ({}))).should.be.false(); + + queue.subscribed.should.be.false(); + }); + it('should hand deliveries back when there is no handler', async function () { const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); await queue.created; const nack = sandbox.stub(rabbit.consumeChannel, 'nack'); - const msg = { content: Buffer.from('{}'), properties: { headers: {} }, fields: {} } as any; + const fields = { + consumerTag: 'amq.ctag-orphan', + deliveryTag: 1, + redelivered: false, + exchange: '', + routingKey: this.name + }; + const msg = { content: Buffer.from('{}'), properties: { headers: {} }, fields } as any; queue.onMessage(msg); nack.calledOnceWith(msg, false, true).should.be.true(); }); + it('should cancel a consumer a second subscribe left behind once it gets a message', async function () { + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + await queue.subscribe(() => ({})); + await queue.subscribe(() => ({})); + await queue.unsubscribe(); + const consumerCount = async () => (await rabbit.consumeChannel.checkQueue(this.name)).consumerCount; + + await rabbit.publish(this.name, {}); + for (let attempt = 0; attempt < 50 && (await consumerCount()) !== 0; attempt++) { + await new Promise(r => setTimeout(r, 20)); + } + + (await consumerCount()).should.equal(0); + }); + it('should tell whether the queue exists', async function () { const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); await queue.created; diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index bcc8186..8dfe0dc 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -138,6 +138,23 @@ describe('Test rabbit class', function() { stub.calledWith(handler).should.be.true(); }); + it('should use the default prefetch for a queue created after one with its own', async function() { + rabbit = new Rabbit(this.url, { prefetch: 10, replyPattern: false }); + await rabbit.createQueue(`own_${randomUUID()}`, { exclusive: true, prefetch: 5 }, () => ({})); + const name = `default_${randomUUID()}`; + let delivered = 0; + await rabbit.createQueue(name, { exclusive: true }, () => delivered++); + + for (let i = 0; i < 10; i++) { + await rabbit.publish(name, { i }); + } + for (let attempt = 0; attempt < 50 && delivered < 10; attempt++) { + await new Promise(r => setTimeout(r, 20)); + } + + delivered.should.equal(10); + }); + it('should create queue with max priority when priority is specified', async function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url); @@ -878,6 +895,19 @@ describe('Test rabbit class', function() { queue.subscribed.should.be.false(); (await consumerCount()).should.equal(0); }); + + it('should destroy a queue created before a reconnect', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + rabbit.on('disconnected', () => setTimeout(() => rabbit.reconnect(), 10)); + await rabbit.createQueue(queueName, {}, () => ({})); + const reconnected = new Promise(resolve => rabbit.once('connected', resolve)); + await rabbit.consumeConnection.close(); + await reconnected; + + await rabbit.destroyQueue(queueName); + + (await queueExists()).should.be.false(); + }); }); it('should emit "disconnected" once when a connection and its channel close', async function() { diff --git a/ts/queue.ts b/ts/queue.ts index c6997b1..7ff557e 100644 --- a/ts/queue.ts +++ b/ts/queue.ts @@ -158,8 +158,9 @@ export default class Queue { // Resolves to whether the consumer is running: an unsubscribe or a broker cancel can land before the consume resolves. async subscribe(handler: (msg: any, ack: (error?, reply?) => any) => any) { - await this.created; + // Before any await, so the check after the consume also catches an unsubscribe while the queue is being created. this.handler = handler; + await this.created; let tag: amqp.Replies.Consume; let cancelled = false; // amqplib calls back with null only when the broker cancels the consumer, e.g. because the queue was deleted. @@ -232,6 +233,8 @@ export default class Queue { if (!this.options.noAck) { this.channel.nack(msg, false, true); } + // A consumer a second subscribe left behind would otherwise keep getting the message back. + this.channel.cancel(msg.fields.consumerTag).catch(() => undefined); return; } this.handler(msg, async (error, reply) => { diff --git a/ts/rabbit.ts b/ts/rabbit.ts index c53b8b5..ba8108f 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -246,15 +246,10 @@ export default class Rabbit extends EventEmitter { if (!stillWanted()) { return false; } - let subscribing: Promise; - if (this.prefetch !== prefetch) { - this.prefetch = prefetch; - subscribing = Promise.resolve(this.consumeChannel.prefetch(prefetch)).then( - () => stillWanted() && queue.subscribe(handler) - ); - } else { - subscribing = Promise.resolve(queue.subscribe(handler)); - } + // Set on every subscribe: this.prefetch is the default for queues without their own, not the channel's current one. + const subscribing = Promise.resolve(this.consumeChannel.prefetch(prefetch)).then( + () => stillWanted() && queue.subscribe(handler) + ); // The lock only orders subscribes: one that fails mustn't fail every one after it, even after a reconnect. this.lock = subscribing.then( () => undefined, @@ -323,8 +318,12 @@ export default class Rabbit extends EventEmitter { async destroyQueue(name: string, prefix?: string) { name = this.updateName(name, prefix); await this.connected; + const queue = this.queues[name]; // Deleting the queue would make the broker cancel our own consumer, which would then wait for the queue to come back. - await this.queues[name]?.unsubscribe(); + // A queue from before a reconnect is on a closed channel, with no consumer left to cancel. + if (queue?.channel === this.consumeChannel) { + await queue.unsubscribe(); + } await Queue.destroy(this.consumeChannel, name); } From 03d56cb6aed8bb02ad1ec2c9265bdc2d9fa7d0ac Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Thu, 22 Oct 2026 10:34:46 +0300 Subject: [PATCH 4/4] test: stub publish before publishing in the DLQ tests The handler can get the message before the test's own publish is confirmed, but the stub was only installed after that confirm. When the confirm lagged by more than the 10 ms retry delay, the DLQ publishes went through the real publish and the stub saw no calls. This failed intermittently on RabbitMQ 3.12.5 in CI. Co-Authored-By: Claude Opus 5.5 --- test/base-queue-handler.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/test/base-queue-handler.test.ts b/test/base-queue-handler.test.ts index 0a72dd8..5603d6c 100644 --- a/test/base-queue-handler.test.ts +++ b/test/base-queue-handler.test.ts @@ -205,8 +205,10 @@ describe('Test baseQueueHandler', function () { ack(null); }); - await rabbit.publish(this.name, { test: 'data' }, { correlationId: '3' }); + const originalPublish = handler.rabbit.publish.bind(handler.rabbit); const publish = sandbox.spy(handler.rabbit, 'publish'); + // Spied before publishing: the handler can get the message before the publish is confirmed. + await originalPublish(this.name, { test: 'data' }, { correlationId: '3' }); await new Promise(resolve => setTimeout(resolve, 100)); afterDlq.calledOnce.should.be.true(); @@ -232,7 +234,6 @@ describe('Test baseQueueHandler', function () { ack(null); }); - await rabbit.publish(this.name, { test: 'data' }, { correlationId: '5' }); const originalPublish = handler.rabbit.publish.bind(handler.rabbit); let publishCallCount = 0; const publish = sandbox.stub(handler.rabbit, 'publish').callsFake((...args) => { @@ -242,6 +243,8 @@ describe('Test baseQueueHandler', function () { } return originalPublish(...args); }); + // Stubbed before publishing: the handler can get the message before the publish is confirmed. + await originalPublish(this.name, { test: 'data' }, { correlationId: '5' }); await new Promise(resolve => setTimeout(resolve, 100)); publish.calledTwice.should.be.true();