feat: tolerate queue declares that don't match the existing queue (DS-3366)

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,10 +261,20 @@ 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, 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
+
+- 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
 
 - Support for stream RPC to pass options for the Readable stream created in the caller.

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');
@@ -52,6 +53,43 @@ 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 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 });
+    }
+    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;
@@ -428,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 () {

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,14 +109,18 @@ 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));
     promises.push(rabbit.createQueue(this.name2, { exclusive: true, prefetch: 15 }, handler));
     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,138 @@ 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, name = queueName) {
+      let onMessage: (content: string) => void;
+      const received = new Promise<string>(resolve => (onMessage = resolve));
+      await rabbit.createQueue(name, options, (msg, ack) => {
+        ack();
+        onMessage(msg.content.toString());
+      });
+      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 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} 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 consumeOne(options)).should.equal(JSON.stringify({ test: true }));
+        disconnected.called.should.be.false();
+      });
+    });
+
+    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, { tolerateQueueTypeMismatch: true });
+      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();
+    });
+  });
 });

ts/channel.ts

@@ -10,4 +10,5 @@ export interface Channel extends amqp.Channel {
     cb?: Function
   ): boolean;
   replyName?: string;
+  channelModel?: amqp.ChannelModel;
 }