diff --git a/README.md b/README.md
index 27e24ea..2f5f01e 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 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.
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();
diff --git a/test/queue.test.ts b/test/queue.test.ts
index 2ecd49c..4891e20 100644
--- a/test/queue.test.ts
+++ b/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;
diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts
index fe19c8f..8dfe0dc 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 = [];
@@ -137,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);
@@ -593,13 +611,21 @@ 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) {
-      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 +667,249 @@ 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 waitUntil(condition: () => boolean | Promise<boolean>, 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 channel.close();
+      }
+
+      async function declareQueue(args: object = {}) {
+        const channel = await admin.createChannel();
+        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;
+        }
+      }
+
+      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();
+        rabbit.on('disconnected', disconnected);
+        let onMessage: (content: string) => void;
+        const received = new Promise<string>(resolve => (onMessage = resolve));
+        await rabbit.createQueue(queueName, {}, (msg, ack) => {
+          ack();
+          onMessage(msg.content.toString());
+        });
+
+        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 }));
+        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, {}, () => ({}));
+
+        await deleteQueue();
+        await new Promise(r => setTimeout(r, 1500));
+        (await queueExists()).should.be.false();
+
+        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 });
+        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 consumeAgain(queue);
+
+        prefetch.calledWith(5).should.be.true();
+      });
+
+      it('should not subscribe again after unsubscribing', 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.unsubscribe(queueName);
+        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 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<void>(resolve => (checked = resolve));
+        sandbox.stub(queue, 'exists').callsFake(async () => {
+          const found = await exists();
+          await rabbit.unsubscribe(queueName);
+          checked();
+          return found;
+        });
+
+        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<void>(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));
+
+        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);
+      });
+
+      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() {
       rabbit = new Rabbit(this.url, { replyPattern: false });
       await rabbit.connected;
diff --git a/ts/queue.ts b/ts/queue.ts
index e7f2cbe..7ff557e 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<any>;
   public handler: Function;
+  public onCancel?: () => void;
   private tag: amqp.Replies.Consume;
 
   constructor(public channel: Channel, public name: string, options: QueueOptions) {
@@ -150,10 +152,43 @@ export default class Queue {
     }
   }
 
+  get subscribed() {
+    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;
+    // Before any await, so the check after the consume also catches an unsubscribe while the queue is being created.
     this.handler = handler;
-    this.tag = await this.channel.consume(this.name, this.onMessage.bind(this), { noAck: this.options.noAck });
+    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.
+    // 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) {
+      // 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() {
@@ -164,6 +199,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);
   }
@@ -181,6 +228,15 @@ export default class Queue {
     if (!msg) {
       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);
+      }
+      // 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) => {
       const { replyTo, correlationId, headers } = msg.properties;
       if (error && reply !== Queue.STOP_PROPAGATION) {
diff --git a/ts/rabbit.ts b/ts/rabbit.ts
index bbae754..ba8108f 100644
--- a/ts/rabbit.ts
+++ b/ts/rabbit.ts
@@ -8,6 +8,10 @@ import Exchange from './exchange';
 import * as assert from 'assert';
 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;
   static STOP_STREAM = Queue.STOP_STREAM;
@@ -25,6 +29,7 @@ export default class Rabbit extends EventEmitter {
   private lossReported = false;
   private attemptError: Error | null = null;
   private closed = false;
+  private resubscribing = new Set<Queue>();
   public prefetch: number;
   public replyPattern: boolean;
   public prefix: string;
@@ -176,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);
     }
@@ -211,34 +219,111 @@ export default class Rabbit extends EventEmitter {
       tolerateQueueTypeMismatch: this.tolerateQueueTypeMismatch
     });
     this.queues[name] = queue;
+    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) {
-      // 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 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 subscribe might be waiting. Only one can pass this check.
+    } while (this.lock !== localLock);
+
+    if (!stillWanted()) {
+      return false;
+    }
+    // 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,
+      () => 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 doesn't recreate a queue someone is replacing.
+  private async resubscribe(queue: Queue, prefetch: number) {
+    const handler = queue.handler;
+    if (!handler || this.resubscribing.has(queue)) {
+      return;
+    }
+    // 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 &&
+      this.queues[queue.name] === queue &&
+      queue.channel === this.consumeChannel &&
+      queue.handler === handler &&
+      !queue.subscribed;
+    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`);
+        }
+      }
+    } finally {
+      this.resubscribing.delete(queue);
+    }
+  }
+
   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.
+    // 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);
   }