From 83731f85bf05ccda89dd46fb78ae9e2a5dd6b778 Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Fri, 9 Oct 2026 11:44:05 +0300 Subject: [PATCH 1/4] feat: tolerate queue declares that don't match the existing queue (DS-3366) Declare each queue on a short-lived channel. On a 406 for a queue that exists with different arguments, log a warning and consume the queue as it is, instead of closing the shared consume channel and disconnecting every consumer. Co-Authored-By: Claude Opus 5.5 --- README.md | 6 ++- test/queue.test.ts | 3 +- test/rabbit.test.ts | 95 ++++++++++++++++++++++++++++++++++++++++----- ts/channel.ts | 1 + ts/queue.ts | 37 +++++++++++++++++- ts/rabbit.ts | 4 +- 6 files changed, 133 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 5391356..6d86fdf 100644 --- a/README.md +++ b/README.md @@ -260,10 +260,14 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { defaultQueueType: 'quorum' } When declaring queues, the following rules apply: - **Non-durable, exclusive, or auto-delete queues** always use **classic** queue type, since **quorum** queues and **streams** are always durable and not designed for temporary or connection-bound queues. -The type of a queue is **immutable** once it has been declared. Attempting to change it after creation will result in a **PRECONDITION_FAILED** error. +The type of a queue is **immutable** once it has been declared. When a queue already exists with different arguments (e.g. another type, another `x-max-priority`, or another DLX or TTL), the broker rejects the declare with **PRECONDITION_FAILED**. `createQueue` then logs a warning naming the argument and consumes the queue as it exists, with its existing arguments. Declares run on a short-lived channel, so the rejection doesn't close the channel other consumers are on. Any other declare error still rejects `createQueue`. ### Changelog +### New in v5.10.x + +- Tolerant queue declares. A queue that already exists with different arguments (classic vs quorum, with or without priority) is consumed as it is, with a warning, instead of the declare closing the shared consume channel and disconnecting every consumer. Old and new pods can now 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. + ### New in v5.4.x - Support for stream RPC to pass options for the Readable stream created in the caller. diff --git a/test/queue.test.ts b/test/queue.test.ts index 2d76c90..42002c8 100644 --- a/test/queue.test.ts +++ b/test/queue.test.ts @@ -4,6 +4,7 @@ import * as assert from 'assert'; import * as sinon from 'sinon'; import Queue from '../ts/queue'; import { Readable } from 'stream'; +const { Channel: AmqpChannel } = require('amqplib/lib/channel_model'); const sandbox = sinon.createSandbox(); describe('Test Queue class', function () { @@ -17,7 +18,7 @@ describe('Test Queue class', function () { }); beforeEach(function () { - this.spyAssert = sandbox.spy(rabbit.consumeChannel, 'assertQueue'); + this.spyAssert = sandbox.spy(AmqpChannel.prototype, 'assertQueue'); this.spyConsume = sandbox.spy(rabbit.consumeChannel, 'consume'); this.spyCancel = sandbox.spy(rabbit.consumeChannel, 'cancel'); this.spyDeleteQueue = sandbox.spy(rabbit.consumeChannel, 'deleteQueue'); diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index 85ac79a..aaa6749 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -9,6 +9,7 @@ import * as DelayQueue from '../ts/delay-queue'; import * as amqp from 'amqplib'; import { randomUUID } from 'crypto'; +const { Channel: AmqpChannel } = require('amqplib/lib/channel_model'); const sandbox = sinon.createSandbox(); describe('Test rabbit class', function() { @@ -108,6 +109,10 @@ describe('Test rabbit class', function() { rabbit = new Rabbit(this.url); await rabbit.connected; const spy = sandbox.spy(rabbit.consumeChannel, 'prefetch'); + const prefetchAtSubscribe = {}; + stub.callsFake(async function() { + prefetchAtSubscribe[this.name] = spy.lastCall.args[0]; + }); const handler = () => ({}); let promises = []; promises.push(rabbit.createQueue(this.name, { exclusive: true, prefetch: 20 }, handler)); @@ -115,7 +120,7 @@ describe('Test rabbit class', function() { promises.push(rabbit.createQueue(this.name3, { exclusive: true, prefetch: 10 }, handler)); promises.push(rabbit.createQueue(this.name4, { exclusive: true, prefetch: 10 }, handler)); await Promise.all(promises); - spy.args.should.eql([[20], [15], [10]]); + prefetchAtSubscribe.should.eql({ [this.name]: 20, [this.name2]: 15, [this.name3]: 10, [this.name4]: 10 }); rabbit.queues[this.name].name.should.equal(this.name); stub.calledWith(handler).should.be.true(); }); @@ -124,7 +129,7 @@ describe('Test rabbit class', function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const priority = 23; const handler = () => {}; @@ -140,7 +145,7 @@ describe('Test rabbit class', function() { const queueType = 'classic'; rabbit = new Rabbit(this.url, { defaultQueueType: queueType }); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; await rabbit.createQueue(this.name, {}, handler); @@ -154,7 +159,7 @@ describe('Test rabbit class', function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; await rabbit.createQueue(this.name, { }, handler); @@ -169,7 +174,7 @@ describe('Test rabbit class', function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url, { defaultQueueType: 'classic' }); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; const queueType = 'quorum'; @@ -184,7 +189,7 @@ describe('Test rabbit class', function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url, { defaultQueueType: 'quorum' }); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; await rabbit.createQueue(this.name, { priority: 23 }, handler); @@ -199,7 +204,7 @@ describe('Test rabbit class', function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url, { defaultQueueType: 'quorum' }); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; await rabbit.createQueue(this.name, { durable: false }, handler); @@ -214,7 +219,7 @@ describe('Test rabbit class', function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url, { defaultQueueType: 'quorum' }); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; await rabbit.createQueue(this.name, { exclusive: true }, handler); @@ -229,7 +234,7 @@ describe('Test rabbit class', function() { const subscription = sandbox.stub(Queue.prototype, 'subscribe'); rabbit = new Rabbit(this.url, { defaultQueueType: 'quorum' }); await rabbit.connected; - const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue') + const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; await rabbit.createQueue(this.name, { autoDelete: true }, handler); @@ -393,4 +398,76 @@ describe('Test rabbit class', function() { (rabbit as any).sigtermHandler(); stub.calledTwice.should.be.true(); }); + + describe('when the queue exists with different arguments', function() { + let admin: amqp.ChannelModel; + let queueName: string; + + beforeEach(async function() { + admin = await amqp.connect(this.url); + queueName = `existing_${randomUUID()}`; + }); + + afterEach(async function() { + const channel = await admin.createChannel(); + await channel.deleteQueue(queueName); + await admin.close(); + }); + + async function declareExisting(args: object) { + const channel = await admin.createChannel(); + await channel.assertQueue(queueName, { durable: true, arguments: args }); + await channel.close(); + } + + async function consumeOne(options: object) { + let onMessage: (content: string) => void; + const received = new Promise(resolve => (onMessage = resolve)); + await rabbit.createQueue(queueName, options, (msg, ack) => { + ack(); + onMessage(msg.content.toString()); + }); + await rabbit.publish(queueName, { test: true }); + return received; + } + + const cases = [ + { existing: 'a classic queue', declared: 'as quorum', args: {}, options: { arguments: { 'x-queue-type': 'quorum' } } }, + { existing: 'a quorum queue', declared: 'without a type', args: { 'x-queue-type': 'quorum' }, options: {} }, + { existing: 'a priority queue', declared: 'without priority', args: { 'x-max-priority': 10 }, options: {} }, + { existing: 'a queue without priority', declared: 'with priority', args: {}, options: { priority: 10 } } + ]; + + cases.forEach(({ existing, declared, args, options }) => { + it(`should consume ${existing} declared ${declared} without disconnecting`, async function() { + await declareExisting(args); + rabbit = new Rabbit(this.url); + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); + + (await consumeOne(options)).should.equal(JSON.stringify({ test: true })); + disconnected.called.should.be.false(); + }); + }); + + it('should reject invalid arguments for a new queue with the original error', async function() { + rabbit = new Rabbit(this.url); + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); + + await rabbit.createQueue(queueName, { arguments: { 'x-max-length-bytes': 'abc' } }) + .should.be.rejectedWith({ code: 406, message: /invalid arg/ }); + disconnected.called.should.be.false(); + }); + + it('should reject other declare errors without disconnecting', async function() { + rabbit = new Rabbit(this.url); + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); + + await rabbit.createQueue('amq.reserved').should.be.rejectedWith({ code: 403 }); + (await consumeOne({})).should.equal(JSON.stringify({ test: true })); + disconnected.called.should.be.false(); + }); + }); }); diff --git a/ts/channel.ts b/ts/channel.ts index 25cc770..f443c36 100644 --- a/ts/channel.ts +++ b/ts/channel.ts @@ -10,4 +10,5 @@ export interface Channel extends amqp.Channel { cb?: Function ): boolean; replyName?: string; + channelModel?: amqp.ChannelModel; } diff --git a/ts/queue.ts b/ts/queue.ts index 4a6498f..6af12de 100644 --- a/ts/queue.ts +++ b/ts/queue.ts @@ -8,6 +8,8 @@ import { encode } from './encode-decode'; import getLogger from './logger'; const logger = getLogger('rabbit-queue'); +const PRECONDITION_FAILED = 406; + export default class Queue { static STOP_PROPAGATION = { stopPropagation: true }; static ERROR_DURING_REPLY = { error: true, error_code: 999 }; @@ -63,7 +65,40 @@ export default class Queue { logger.warn(`Invalid x-max-priority argument for quorum queue '${this.name}'`); delete queueOptions.arguments['x-max-priority']; } - await this.channel.assertQueue(this.name, queueOptions); + await this.assertQueue(queueOptions); + } + + // A failed declare closes its channel, so declares run on throwaway channels and a mismatch with the + // existing queue (e.g. classic vs quorum during a migration) can't take down the consumers. + // Server-named queues stay on the consume channel: consume('') resolves to the last queue declared there. + private async assertQueue(queueOptions: amqp.Options.AssertQueue) { + if (!this.channel.channelModel || !this.name) { + await this.channel.assertQueue(this.name, queueOptions); + return; + } + try { + await this.withDeclareChannel(channel => channel.assertQueue(this.name, queueOptions)); + } catch (error) { + if (error.code !== PRECONDITION_FAILED) { + throw error; + } + // 406 also means invalid arguments for a queue that doesn't exist yet; report that, not the 404. + await this.withDeclareChannel(channel => channel.checkQueue(this.name)).catch(() => { + throw error; + }); + logger.warn(`Queue ${this.name} exists with different arguments, using it as it is: ${error.message}`); + } + } + + private async withDeclareChannel(declare: (channel: amqp.Channel) => Promise) { + const channel = await this.channel.channelModel.createChannel(); + // The rejected declare already carries the error; without a listener the 'error' event would throw. + channel.on('error', () => undefined); + try { + await declare(channel); + } finally { + await channel.close().catch(() => undefined); + } } async subscribe(handler: (msg: any, ack: (error?, reply?) => any) => any) { diff --git a/ts/rabbit.ts b/ts/rabbit.ts index e3c5d73..9524701 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -94,7 +94,9 @@ export default class Rabbit extends EventEmitter { async createChannel(connection: amqp.ChannelModel) { connection.once('close', error => this.emitDisconnected(error)); connection.on('error', error => this.emitDisconnected(error)); - return connection.createConfirmChannel(); + const channel: Channel = await connection.createConfirmChannel(); + channel.channelModel = connection; + return channel; } async initChannel(channel: Channel, publish = false) { From 28a4f8d3cb43443d08ef3b411af73e783d274e7c Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Mon, 12 Oct 2026 13:41:15 +0300 Subject: [PATCH 2/4] fix: bound concurrent queue-declare channels per connection (DS-3366) Startup declares run concurrently, and each one now opens its own channel, so a service with many queues could exhaust the connection's channel_max. Cap declare channels at 10 per connection; further declares wait for a slot. Co-Authored-By: Claude Opus 5.5 --- test/queue.test.ts | 31 +++++++++++++++++++++++++++++++ ts/queue.ts | 39 ++++++++++++++++++++++++++++++++++----- 2 files changed, 65 insertions(+), 5 deletions(-) diff --git a/test/queue.test.ts b/test/queue.test.ts index 42002c8..4aa373b 100644 --- a/test/queue.test.ts +++ b/test/queue.test.ts @@ -53,6 +53,37 @@ describe('Test Queue class', function () { .should.be.true(); }); + it('should open at most 10 declare channels at once per connection', async function () { + const connection = rabbit.consumeConnection; + const createChannel = connection.createChannel.bind(connection); + let open = 0; + let maxOpen = 0; + sandbox.stub(connection, 'createChannel').callsFake(async () => { + maxOpen = Math.max(maxOpen, ++open); + const channel = await createChannel(); + const close = channel.close.bind(channel); + channel.close = async () => { + try { + await close(); + } finally { + open--; + } + }; + return channel; + }); + + const queues = Array.from({ length: 25 }, (_, i) => new Queue(rabbit.consumeChannel, `${this.name}.${i}`, { exclusive: true })); + await Promise.all(queues.map(queue => queue.created)); + maxOpen.should.equal(10); + }); + + it('should free declare channels after failed declares', async function () { + for (let i = 0; i < 11; i++) { + await new Queue(rabbit.consumeChannel, 'amq.reserved', {}).created.should.be.rejectedWith({ code: 403 }); + } + await new Queue(rabbit.consumeChannel, this.name, { exclusive: true }).created; + }); + it('should create queue and add handler to it', async function () { const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true, noAck: true }); await queue.created; diff --git a/ts/queue.ts b/ts/queue.ts index 6af12de..092c8da 100644 --- a/ts/queue.ts +++ b/ts/queue.ts @@ -9,6 +9,30 @@ import getLogger from './logger'; const logger = getLogger('rabbit-queue'); const PRECONDITION_FAILED = 406; +const MAX_DECLARE_CHANNELS_PER_CONNECTION = 10; +const declareSlots = new WeakMap void)[] }>(); + +// Startup declares run concurrently; without a bound, their channels could exhaust the connection's channel_max. +async function acquireDeclareSlot(connection: amqp.ChannelModel): Promise<() => void> { + let slots = declareSlots.get(connection); + if (!slots) { + slots = { inUse: 0, waiting: [] }; + declareSlots.set(connection, slots); + } + if (slots.inUse < MAX_DECLARE_CHANNELS_PER_CONNECTION) { + slots.inUse++; + } else { + await new Promise(resolve => slots.waiting.push(resolve)); + } + return () => { + const next = slots.waiting.shift(); + if (next) { + next(); + } else { + slots.inUse--; + } + }; +} export default class Queue { static STOP_PROPAGATION = { stopPropagation: true }; @@ -91,13 +115,18 @@ export default class Queue { } private async withDeclareChannel(declare: (channel: amqp.Channel) => Promise) { - const channel = await this.channel.channelModel.createChannel(); - // The rejected declare already carries the error; without a listener the 'error' event would throw. - channel.on('error', () => undefined); + const release = await acquireDeclareSlot(this.channel.channelModel); try { - await declare(channel); + const channel = await this.channel.channelModel.createChannel(); + // The rejected declare already carries the error; without a listener the 'error' event would throw. + channel.on('error', () => undefined); + try { + await declare(channel); + } finally { + await channel.close().catch(() => undefined); + } } finally { - await channel.close().catch(() => undefined); + release(); } } From 816d1f769f6b5f73dd7269d1933c1be2b6f18b44 Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Mon, 12 Oct 2026 14:05:22 +0300 Subject: [PATCH 3/4] fix: keep declare channels within the negotiated channel limit (DS-3366) The cap of 10 ignored the consume channel already open on the connection, so a broker with channel_max of 10 or less could still run out of channels. Derive the cap from the negotiated channelMax, reserving the consume channel. Co-Authored-By: Claude Opus 5.5 --- test/queue.test.ts | 6 ++++++ ts/queue.ts | 6 +++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/test/queue.test.ts b/test/queue.test.ts index 4aa373b..614b770 100644 --- a/test/queue.test.ts +++ b/test/queue.test.ts @@ -77,6 +77,12 @@ describe('Test Queue class', function () { maxOpen.should.equal(10); }); + it('should keep declare channels within the negotiated channel limit', async function () { + sandbox.stub(rabbit.consumeConnection.connection as any, 'channelMax').value(4); + const queues = Array.from({ length: 10 }, (_, i) => new Queue(rabbit.consumeChannel, `${this.name}.limit.${i}`, { exclusive: true })); + await Promise.all(queues.map(queue => queue.created)); + }); + it('should free declare channels after failed declares', async function () { for (let i = 0; i < 11; i++) { await new Queue(rabbit.consumeChannel, 'amq.reserved', {}).created.should.be.rejectedWith({ code: 403 }); diff --git a/ts/queue.ts b/ts/queue.ts index 092c8da..0c219d1 100644 --- a/ts/queue.ts +++ b/ts/queue.ts @@ -19,7 +19,11 @@ async function acquireDeclareSlot(connection: amqp.ChannelModel): Promise<() => slots = { inUse: 0, waiting: [] }; declareSlots.set(connection, slots); } - if (slots.inUse < MAX_DECLARE_CHANNELS_PER_CONNECTION) { + // amqplib's typings omit the negotiated channelMax. One channel is reserved for the consume channel; + // at least one slot is kept so a tiny limit fails with amqplib's error instead of hanging. + const { channelMax } = connection.connection as unknown as { channelMax: number }; + const limit = Math.max(1, Math.min(MAX_DECLARE_CHANNELS_PER_CONNECTION, channelMax - 1)); + if (slots.inUse < limit) { slots.inUse++; } else { await new Promise(resolve => slots.waiting.push(resolve)); From fa2956120f18da260e97f5900cd03532ddb15d69 Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Tue, 13 Oct 2026 10:08:22 +0300 Subject: [PATCH 4/4] fix: tolerate only queue type and priority mismatches, behind a flag (DS-3366) Tolerating every 406 hid mismatches the migration doesn't need, such as a delay queue whose dead-letter routing key changed. Only x-queue-type and x-max-priority are tolerated now, and only with the new tolerateQueueTypeMismatch option, which is off by default. The broker reports one mismatch at a time and checks a quorum queue's type before its TTL and delivery limit, so after a tolerated mismatch the queue is redeclared with its current value to let the broker check the rest. Binds and unbinds also run on short-lived channels, so a failed bind rejects without closing the consume channel. Co-Authored-By: Claude Opus 5.5 --- README.md | 11 ++++-- test/queue.test.ts | 17 ++++++--- test/rabbit.test.ts | 78 +++++++++++++++++++++++++++++++++++----- ts/queue.ts | 88 ++++++++++++++++++++++++++------------------- ts/rabbit.ts | 9 ++++- 5 files changed, 152 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index 6d86fdf..6ceaceb 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,7 @@ const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost', { replyPattern: true, //if reply pattern is enabled an exclusive queue is created scheduledPublish: false, 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) socketOptions: {} // socketOptions will be passed as a second param to amqp.connect and from ther to the socket library (net or tls) }); @@ -260,13 +261,19 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { defaultQueueType: 'quorum' } When declaring queues, the following rules apply: - **Non-durable, exclusive, or auto-delete queues** always use **classic** queue type, since **quorum** queues and **streams** are always durable and not designed for temporary or connection-bound queues. -The type of a queue is **immutable** once it has been declared. When a queue already exists with different arguments (e.g. another type, another `x-max-priority`, or another DLX or TTL), the broker rejects the declare with **PRECONDITION_FAILED**. `createQueue` then logs a warning naming the argument and consumes the queue as it exists, with its existing arguments. Declares run on a short-lived channel, so the rejection doesn't close the channel other consumers are on. Any other declare error still rejects `createQueue`. +The type of a queue is **immutable** once it has been declared. When a queue already exists with different arguments, the broker rejects the declare with **PRECONDITION_FAILED** and `createQueue` rejects. Queue declares, binds and unbinds run on a short-lived channel, so a rejected one doesn't close the channel the other consumers are on. + +While queues move between classic and quorum, old and new declares disagree on the queue type and on priority, which quorum queues don't support. With `tolerateQueueTypeMismatch: true`, a mismatch on `x-queue-type` or `x-max-priority` only logs a warning, and the queue is consumed as it exists. Any other mismatch (e.g. dead-letter exchange or routing key, TTL, max length) still rejects `createQueue`, even when the type differs too. +```javascript +const rabbit = new Rabbit(process.env.RABBIT_URL, { tolerateQueueTypeMismatch: true }); +``` ### Changelog ### New in v5.10.x -- Tolerant queue declares. A queue that already exists with different arguments (classic vs quorum, with or without priority) is consumed as it is, with a warning, instead of the declare closing the shared consume channel and disconnecting every consumer. Old and new pods can now 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. +- 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. ### New in v5.4.x diff --git a/test/queue.test.ts b/test/queue.test.ts index 614b770..2ecd49c 100644 --- a/test/queue.test.ts +++ b/test/queue.test.ts @@ -466,17 +466,26 @@ describe('Test Queue class', function () { }); it('should bind to exchange', async function () { - const spy = sandbox.spy(rabbit.consumeChannel, 'bindQueue'); + const spy = sandbox.spy(AmqpChannel.prototype, 'bindQueue'); const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); await Queue.bindToExchange('amq.topic', this.name, rabbit.consumeChannel, this.name, queue); - spy.calledWith(this.name, 'amq.topic', this.name); + spy.calledWith(this.name, 'amq.topic', this.name).should.be.true(); + spy.calledOn(rabbit.consumeChannel).should.be.false(); }); it('should unbind to exchange', async function () { - const spy = sandbox.spy(rabbit.consumeChannel, 'unbindQueue'); + const spy = sandbox.spy(AmqpChannel.prototype, 'unbindQueue'); const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); await Queue.unbindFromExchange('amq.topic', this.name, rabbit.consumeChannel, this.name, queue); - spy.calledWith(this.name, 'amq.topic', this.name); + spy.calledWith(this.name, 'amq.topic', this.name).should.be.true(); + spy.calledOn(rabbit.consumeChannel).should.be.false(); + }); + + it('should reject a failed bind without closing the consume channel', async function () { + const queue = new Queue(rabbit.consumeChannel, this.name, { exclusive: true }); + await Queue.bindToExchange(`missing_${this.name}`, this.name, rabbit.consumeChannel, this.name, queue) + .should.be.rejectedWith({ code: 404 }); + await rabbit.consumeChannel.checkQueue(this.name); }); it('should getReply as a stream and stop on demand', async function () { diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index aaa6749..cbe3e98 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -420,28 +420,38 @@ describe('Test rabbit class', function() { await channel.close(); } - async function consumeOne(options: object) { + async function consumeOne(options: object, name = queueName) { let onMessage: (content: string) => void; const received = new Promise(resolve => (onMessage = resolve)); - await rabbit.createQueue(queueName, options, (msg, ack) => { + await rabbit.createQueue(name, options, (msg, ack) => { ack(); onMessage(msg.content.toString()); }); - await rabbit.publish(queueName, { test: true }); + await rabbit.publish(name, { test: true }); return received; } + async function consumeTemporaryQueue() { + return consumeOne({ exclusive: true }, `temporary_${randomUUID()}`); + } + const cases = [ { existing: 'a classic queue', declared: 'as quorum', args: {}, options: { arguments: { 'x-queue-type': 'quorum' } } }, { existing: 'a quorum queue', declared: 'without a type', args: { 'x-queue-type': 'quorum' }, options: {} }, { existing: 'a priority queue', declared: 'without priority', args: { 'x-max-priority': 10 }, options: {} }, - { existing: 'a queue without priority', declared: 'with priority', args: {}, options: { priority: 10 } } + { existing: 'a queue without priority', declared: 'with priority', args: {}, options: { priority: 10 } }, + { + existing: 'a classic priority queue', + declared: 'as quorum', + args: { 'x-max-priority': 10 }, + options: { arguments: { 'x-queue-type': 'quorum' } } + } ]; cases.forEach(({ existing, declared, args, options }) => { - it(`should consume ${existing} declared ${declared} without disconnecting`, async function() { + it(`should consume ${existing} declared ${declared} when tolerating type mismatches`, async function() { await declareExisting(args); - rabbit = new Rabbit(this.url); + rabbit = new Rabbit(this.url, { tolerateQueueTypeMismatch: true }); const disconnected = sandbox.spy(); rabbit.on('disconnected', disconnected); @@ -450,18 +460,70 @@ describe('Test rabbit class', function() { }); }); - it('should reject invalid arguments for a new queue with the original error', async function() { + it('should reject a type mismatch by default without disconnecting', async function() { + await declareExisting({ 'x-queue-type': 'quorum' }); rabbit = new Rabbit(this.url); const disconnected = sandbox.spy(); rabbit.on('disconnected', disconnected); + await rabbit.createQueue(queueName, {}).should.be.rejectedWith({ code: 406, message: /inequivalent arg 'x-queue-type'/ }); + (await consumeTemporaryQueue()).should.equal(JSON.stringify({ test: true })); + disconnected.called.should.be.false(); + }); + + const otherMismatches = [ + { + existing: 'a queue with another dead-letter routing key', + args: { 'x-dead-letter-exchange': '', 'x-dead-letter-routing-key': 'old' }, + options: { deadLetterExchange: '', deadLetterRoutingKey: 'new' }, + mismatch: 'x-dead-letter-routing-key' + }, + { + existing: 'a quorum queue with another dead-letter routing key', + args: { 'x-queue-type': 'quorum', 'x-dead-letter-exchange': '', 'x-dead-letter-routing-key': 'old' }, + options: { deadLetterExchange: '', deadLetterRoutingKey: 'new' }, + mismatch: 'x-dead-letter-routing-key' + }, + { + existing: 'a quorum queue with another TTL', + args: { 'x-queue-type': 'quorum', 'x-message-ttl': 60000 }, + options: { messageTtl: 30000 }, + mismatch: 'x-message-ttl' + }, + { + existing: 'a classic priority queue with another overflow', + args: { 'x-max-priority': 10, 'x-overflow': 'drop-head' }, + options: { arguments: { 'x-queue-type': 'quorum', 'x-overflow': 'reject-publish' } }, + mismatch: 'x-overflow' + } + ]; + + otherMismatches.forEach(({ existing, args, options, mismatch }) => { + it(`should reject ${existing} when tolerating type mismatches`, async function() { + await declareExisting(args); + rabbit = new Rabbit(this.url, { tolerateQueueTypeMismatch: true }); + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); + + await rabbit.createQueue(queueName, options) + .should.be.rejectedWith({ code: 406, message: new RegExp(`inequivalent arg '${mismatch}'`) }); + (await consumeTemporaryQueue()).should.equal(JSON.stringify({ test: true })); + disconnected.called.should.be.false(); + }); + }); + + it('should reject invalid arguments for a new queue', async function() { + rabbit = new Rabbit(this.url, { tolerateQueueTypeMismatch: true }); + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); + await rabbit.createQueue(queueName, { arguments: { 'x-max-length-bytes': 'abc' } }) .should.be.rejectedWith({ code: 406, message: /invalid arg/ }); disconnected.called.should.be.false(); }); it('should reject other declare errors without disconnecting', async function() { - rabbit = new Rabbit(this.url); + rabbit = new Rabbit(this.url, { tolerateQueueTypeMismatch: true }); const disconnected = sandbox.spy(); rabbit.on('disconnected', disconnected); diff --git a/ts/queue.ts b/ts/queue.ts index 0c219d1..e7f2cbe 100644 --- a/ts/queue.ts +++ b/ts/queue.ts @@ -9,6 +9,8 @@ import getLogger from './logger'; const logger = getLogger('rabbit-queue'); 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']; const MAX_DECLARE_CHANNELS_PER_CONNECTION = 10; const declareSlots = new WeakMap void)[] }>(); @@ -38,6 +40,29 @@ async function acquireDeclareSlot(connection: amqp.ChannelModel): Promise<() => }; } +// A failed declare closes its channel, so declares run on throwaway channels and one that fails can't take down +// the consumers. Server-named queues stay on the consume channel: consume('') resolves to the last queue declared there. +async function withDeclareChannel(channel: Channel, queueName: string, declare: (channel: amqp.Channel) => Promise) { + if (!channel.channelModel || !queueName) { + return declare(channel); + } + const release = await acquireDeclareSlot(channel.channelModel); + try { + const declareChannel = await channel.channelModel.createChannel(); + // The rejected declare already carries the error; without a listener the 'error' event would throw. + declareChannel.on('error', () => undefined); + try { + return await declare(declareChannel); + } finally { + await declareChannel.close().catch(() => undefined); + } + } finally { + release(); + } +} + +export type QueueOptions = amqp.Options.AssertQueue & amqp.Options.Consume & { tolerateQueueTypeMismatch?: boolean }; + export default class Queue { static STOP_PROPAGATION = { stopPropagation: true }; static ERROR_DURING_REPLY = { error: true, error_code: 999 }; @@ -48,12 +73,12 @@ export default class Queue { durable: true, noAck: false }; - public options: amqp.Options.AssertQueue & amqp.Options.Consume; + public options: QueueOptions; public created: Promise; public handler: Function; private tag: amqp.Replies.Consume; - constructor(public channel: Channel, public name: string, options: amqp.Options.AssertQueue & amqp.Options.Consume) { + constructor(public channel: Channel, public name: string, options: QueueOptions) { this.options = Object.assign({}, this.defaultOptions, options); this.created = this.create(); } @@ -96,41 +121,32 @@ export default class Queue { await this.assertQueue(queueOptions); } - // A failed declare closes its channel, so declares run on throwaway channels and a mismatch with the - // existing queue (e.g. classic vs quorum during a migration) can't take down the consumers. - // Server-named queues stay on the consume channel: consume('') resolves to the last queue declared there. private async assertQueue(queueOptions: amqp.Options.AssertQueue) { - if (!this.channel.channelModel || !this.name) { - await this.channel.assertQueue(this.name, queueOptions); - return; - } - try { - await this.withDeclareChannel(channel => channel.assertQueue(this.name, queueOptions)); - } catch (error) { - if (error.code !== PRECONDITION_FAILED) { - throw error; - } - // 406 also means invalid arguments for a queue that doesn't exist yet; report that, not the 404. - await this.withDeclareChannel(channel => channel.checkQueue(this.name)).catch(() => { - throw error; - }); - logger.warn(`Queue ${this.name} exists with different arguments, using it as it is: ${error.message}`); - } - } - - private async withDeclareChannel(declare: (channel: amqp.Channel) => Promise) { - const release = await acquireDeclareSlot(this.channel.channelModel); - try { - const channel = await this.channel.channelModel.createChannel(); - // The rejected declare already carries the error; without a listener the 'error' event would throw. - channel.on('error', () => undefined); + const tolerated: string[] = []; + while (true) { try { - await declare(channel); - } finally { - await channel.close().catch(() => undefined); + await withDeclareChannel(this.channel, this.name, channel => channel.assertQueue(this.name, queueOptions)); + break; + } catch (error) { + const [, arg, current] = /inequivalent arg '([^']+)'.* current is (?:none|(?:the value )?'([^']*)')/.exec(error.message) ?? []; + if (!this.options.tolerateQueueTypeMismatch || error.code !== PRECONDITION_FAILED || + !TOLERATED_MISMATCHES.includes(arg) || tolerated.includes(arg)) { + throw error; + } + tolerated.push(arg); + // The broker reports only the first mismatch it finds. Redeclaring with the existing value lets it check the + // rest, so a tolerated mismatch can't hide one that isn't (e.g. x-message-ttl is checked after x-queue-type). + const args = { ...queueOptions.arguments }; + if (current === undefined) { + delete args[arg]; + } else { + args[arg] = /^\d+$/.test(current) ? Number(current) : current; + } + queueOptions = { ...queueOptions, arguments: args }; } - } finally { - release(); + } + if (tolerated.length) { + logger.warn(`Queue ${this.name} exists with a different ${tolerated.join(' and ')}, using it as it is`); } } @@ -264,7 +280,7 @@ export default class Queue { if (queue) { await queue.created; } - await channel.bindQueue(name, exchange, routingKey); + await withDeclareChannel(channel, name, declareChannel => declareChannel.bindQueue(name, exchange, routingKey)); logger.debug(`created binding ${exchange} routingkey:${routingKey} --> queue:${name}`); } @@ -272,7 +288,7 @@ export default class Queue { if (queue) { await queue.created; } - await channel.unbindQueue(name, exchange, routingKey); + await withDeclareChannel(channel, name, declareChannel => declareChannel.unbindQueue(name, exchange, routingKey)); logger.debug(`deleted binding${exchange} routingkey:${routingKey} -X-> queue:${name}}`); } } diff --git a/ts/rabbit.ts b/ts/rabbit.ts index 9524701..0043bb6 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -26,6 +26,7 @@ export default class Rabbit extends EventEmitter { public scheduledPublish: boolean; public socketOptions; private defaultQueueType: string | undefined; + private tolerateQueueTypeMismatch: boolean; public logger: ReturnType; private readonly sigtermHandler: () => void; @@ -38,6 +39,7 @@ export default class Rabbit extends EventEmitter { scheduledPublish = false, socketOptions = {}, defaultQueueType = undefined, + tolerateQueueTypeMismatch = false, }: { prefetch?: number replyPattern?: boolean @@ -45,6 +47,7 @@ export default class Rabbit extends EventEmitter { scheduledPublish?: boolean socketOptions?: any defaultQueueType?: string + tolerateQueueTypeMismatch?: boolean } = {} ) { super(); @@ -59,6 +62,7 @@ export default class Rabbit extends EventEmitter { this.scheduledPublish = scheduledPublish; this.socketOptions = socketOptions; this.defaultQueueType = defaultQueueType; + this.tolerateQueueTypeMismatch = tolerateQueueTypeMismatch; this.sigtermHandler = this.handleSigterm.bind(this); this.reconnect(); @@ -132,7 +136,10 @@ export default class Rabbit extends EventEmitter { options.prefetch = options.prefetch || this.prefetch; name = this.updateName(name, options.prefix); await this.connected; - const queue = new Queue(this.consumeChannel, name, options); + const queue = new Queue(this.consumeChannel, name, { + ...options, + tolerateQueueTypeMismatch: this.tolerateQueueTypeMismatch + }); this.queues[name] = queue; await queue.created; this.logger.debug(`created queue ${name}`);