fix: subscribe again after the broker cancels a consumer (DS-3536)

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 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
 
 `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,11 @@ 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 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
 
 - `cancelConsumersOnSigterm` option. Set it to `false` to keep consuming after `SIGTERM`, for processes that drain work through their own `SIGTERM` hook. On by default.

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();

test/queue.test.ts

@@ -111,6 +111,108 @@ 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 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 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;
+    (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;