fix: close the surviving connection when reconnecting (DS-3525)

README.md

@@ -293,6 +293,7 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: fa
 ### 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.
+- v5.11.1: when only one of the two connections (consume or publish) dropped, reconnecting left the other one open, so its consumers kept running next to the ones created again on `connected`. The reconnect now closes it first, and one outage emits `disconnected` once instead of once per connection and channel.
 
 ### New in v5.10.x
 

test/rabbit.test.ts

@@ -55,6 +55,18 @@ describe('Test rabbit class', function() {
     });
   });
 
+  it('should not emit "connected" when closed while connecting', async function() {
+    rabbit = new Rabbit(this.url, { replyPattern: false });
+    const connected = sandbox.spy();
+    rabbit.on('connected', connected);
+
+    await rabbit.close();
+    await rabbit.connected.should.be.rejectedWith(/closed while connecting/);
+
+    connected.called.should.be.false();
+    rabbit = null;
+  });
+
   it('should call amqp connect with socketOptions', async function() {
     const connectStub = sandbox.spy(amqp, 'connect');
     rabbit = new Rabbit(this.url, { socketOptions: 'socketOptions' });
@@ -558,4 +570,89 @@ describe('Test rabbit class', function() {
       disconnected.called.should.be.false();
     });
   });
+
+  describe('when one of the two connections drops', function() {
+    let admin: amqp.ChannelModel;
+    let queueName: string;
+
+    beforeEach(async function() {
+      admin = await amqp.connect(this.url);
+      queueName = `reconnect_${randomUUID()}`;
+      const channel = await admin.createChannel();
+      await channel.assertQueue(queueName, { durable: true });
+      await channel.close();
+    });
+
+    afterEach(async function() {
+      rabbit?.removeAllListeners('connected');
+      rabbit?.removeAllListeners('disconnected');
+      const channel = await admin.createChannel();
+      await channel.deleteQueue(queueName);
+      await admin.close();
+    });
+
+    async function consumerCount() {
+      const channel = await admin.createChannel();
+      const { consumerCount: count } = await channel.checkQueue(queueName);
+      await channel.close();
+      return count;
+    }
+
+    async function waitForConsumers(expected: number) {
+      for (let attempt = 0; attempt < 50 && (await consumerCount()) !== expected; attempt++) {
+        await new Promise(r => setTimeout(r, 20));
+      }
+      (await consumerCount()).should.equal(expected);
+    }
+
+    // What orka does: reconnect on every "disconnected" and create the handlers' queues again on "connected".
+    function resubscribeOnReconnect() {
+      rabbit.on('disconnected', () => setTimeout(() => rabbit.reconnect(), 10));
+      rabbit.on('connected', () => rabbit.createQueue(queueName, {}, () => ({})));
+    }
+
+    it('should keep one consumer per queue after the publish connection drops', async function() {
+      rabbit = new Rabbit(this.url, { replyPattern: false });
+      resubscribeOnReconnect();
+      await rabbit.connected;
+      await waitForConsumers(1);
+
+      const reconnected = new Promise(resolve => rabbit.once('connected', resolve));
+      await rabbit.publishConnection.close();
+      await reconnected;
+      await new Promise(r => setTimeout(r, 200));
+
+      (await consumerCount()).should.equal(1);
+    });
+
+    it('should close the surviving publish connection after the consume connection drops', async function() {
+      rabbit = new Rabbit(this.url, { replyPattern: false });
+      resubscribeOnReconnect();
+      await rabbit.connected;
+      await waitForConsumers(1);
+      const survivorClose = sandbox.spy(rabbit.publishConnection, 'close');
+
+      const reconnected = new Promise(resolve => rabbit.once('connected', resolve));
+      await rabbit.consumeConnection.close();
+      await reconnected;
+      await new Promise(r => setTimeout(r, 200));
+
+      survivorClose.called.should.be.true();
+      (await consumerCount()).should.equal(1);
+    });
+
+    it('should emit "disconnected" once when a connection and its channel close', async function() {
+      rabbit = new Rabbit(this.url, { replyPattern: false });
+      await rabbit.connected;
+      const disconnected = sandbox.spy();
+      rabbit.on('disconnected', disconnected);
+
+      await rabbit.publishConnection.close();
+      await new Promise(r => setTimeout(r, 100));
+
+      disconnected.callCount.should.equal(1);
+      await rabbit.consumeConnection.close();
+      rabbit = null;
+    });
+  });
 });

ts/rabbit.ts

@@ -20,6 +20,11 @@ export default class Rabbit extends EventEmitter {
   public lock: Promise<void>;
   public queues: { [s: string]: Queue } = {};
   public connecting = false;
+  private attempt: Promise<void>;
+  private connections = new Set<amqp.ChannelModel>();
+  private lossReported = false;
+  private attemptError: Error | null = null;
+  private closed = false;
   public prefetch: number;
   public replyPattern: boolean;
   public prefix: string;
@@ -77,48 +82,105 @@ export default class Rabbit extends EventEmitter {
     }
   }
 
-  private async connect() {
-    if (this.connecting) {
-      return;
+  private connect() {
+    if (!this.connecting) {
+      this.connecting = true;
+      this.attempt = this.openConnections();
+    }
+    return this.attempt;
+  }
+
+  private async openConnections() {
+    // Losing either connection triggers a reconnect, so the one that survived is closed here: otherwise its
+    // consumers keep running next to the ones the handlers create again on "connected". Not awaited, so a
+    // half-open connection still waiting for its close-ok doesn't hold up the new pair.
+    this.closeAll(this.connections);
+    const connections = new Set<amqp.ChannelModel>();
+    this.connections = connections;
+    this.lossReported = false;
+    this.attemptError = null;
+
+    try {
+      const consumeConnection = await this.openConnection(connections);
+      const consumeChannel = await this.createChannel(consumeConnection);
+      await this.initChannel(consumeChannel);
+
+      const publishConnection = await this.openConnection(connections);
+      const publishChannel = await this.createChannel(publishConnection);
+      await this.initChannel(publishChannel, true);
+
+      if (this.attemptError) {
+        throw this.attemptError;
+      }
+      this.consumeConnection = consumeConnection;
+      this.consumeChannel = consumeChannel;
+      this.publishConnection = publishConnection;
+      this.publishChannel = publishChannel;
+    } catch (error) {
+      this.closeAll(connections);
+      this.connecting = false;
+      this.reportLoss(error);
+      throw error;
     }
-    this.connecting = true;
-    this.consumeConnection = await amqp.connect(this.url, this.socketOptions);
-    this.consumeChannel = await this.createChannel(this.consumeConnection);
-    await this.initChannel(this.consumeChannel);
-
-    this.publishConnection = await amqp.connect(this.url, this.socketOptions);
-    this.publishChannel = await this.createChannel(this.publishConnection);
-    await this.initChannel(this.publishChannel, true);
-    this.emit('connected');
     this.connecting = false;
+    this.emit('connected');
   }
 
   async reconnect() {
+    this.closed = false;
     this.connected = this.connect();
-    await this.connected.catch(error => this.emitDisconnected(error));
+    await this.connected.catch(() => undefined);
   }
 
-  private emitDisconnected(error) {
-    this.connecting = false;
+  private async openConnection(connections: Set<amqp.ChannelModel>) {
+    const connection = await amqp.connect(this.url, this.socketOptions);
+    connections.add(connection);
+    if (this.closed) {
+      throw new Error('Rabbit was closed while connecting');
+    }
+    return connection;
+  }
+
+  private closeAll(connections: Iterable<amqp.ChannelModel>) {
+    return Promise.all([...connections].map(connection => connection.close().catch(() => undefined)));
+  }
+
+  private connectionLost(connection: amqp.ChannelModel, error) {
+    if (!this.connections.has(connection)) {
+      return;
+    }
+    if (this.connecting) {
+      this.attemptError = this.attemptError || error || new Error('Connection closed while connecting');
+      return;
+    }
+    this.reportLoss(error);
+  }
+
+  // One outage closes a connection and its channel, and may raise errors too: report it once.
+  private reportLoss(error) {
+    if (this.lossReported) {
+      return;
+    }
+    this.lossReported = true;
     this.emit('disconnected', error);
   }
 
   async createChannel(connection: amqp.ChannelModel) {
-    connection.once('close', error => this.emitDisconnected(error));
-    connection.on('error', error => this.emitDisconnected(error));
+    connection.once('close', error => this.connectionLost(connection, error));
+    connection.on('error', error => this.connectionLost(connection, error));
     const channel: Channel = await connection.createConfirmChannel();
     channel.channelModel = connection;
     return channel;
   }
 
   async initChannel(channel: Channel, publish = false) {
     channel.prefetch(this.prefetch);
-    channel.on('close', error => this.emitDisconnected(error));
+    channel.on('close', error => this.connectionLost(channel.channelModel, error));
     if (!publish && this.replyPattern) {
-      await createReplyQueue(this.consumeChannel);
+      await createReplyQueue(channel);
     }
     if (!publish && this.scheduledPublish) {
-      await createDelayQueueReply(this.consumeChannel, this.updateName('delay'), this.scheduledPublishQueuesAsQuorum);
+      await createDelayQueueReply(channel, this.updateName('delay'), this.scheduledPublishQueuesAsQuorum);
     }
   }
 
@@ -262,9 +324,9 @@ export default class Rabbit extends EventEmitter {
 
   async close() {
     process.off('SIGTERM', this.sigtermHandler);
-
-    await this.consumeConnection.close();
-    await this.publishConnection.close();
+    this.closed = true;
+    // Covers a connect still in flight, whose connections aren't in consumeConnection/publishConnection yet.
+    await this.closeAll(this.connections);
   }
 
   private handleSigterm() {