From 9cc02ec6eafaec5ec5f8c4bed3d4832d8dc3d4fe Mon Sep 17 00:00:00 2001 From: Ioannis Karachristos Date: Tue, 20 Oct 2026 11:30:09 +0300 Subject: [PATCH] fix: close the surviving connection when reconnecting (DS-3525) Losing either the consume or the publish connection emits "disconnected", and the reconnect opened a fresh pair without closing the one that survived. When the consume connection survived, its consumers kept running next to the ones the handlers create again on "connected", doubling the process's consumers and prefetch on every partial outage (seen on officedroid after a node restart). The reconnect now closes the previous pair first, ignores events from connections it has replaced, reports one outage as one "disconnected", and fails an attempt that loses a connection midway instead of emitting "connected" on it. close() also covers a connect still in flight. Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + test/rabbit.test.ts | 97 +++++++++++++++++++++++++++++++++++++++ ts/rabbit.ts | 108 ++++++++++++++++++++++++++++++++++---------- 3 files changed, 183 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 6f24401..27e24ea 100644 --- a/README.md +++ b/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 diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index be4bd11..fe19c8f 100644 --- a/test/rabbit.test.ts +++ b/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; + }); + }); }); diff --git a/ts/rabbit.ts b/ts/rabbit.ts index 96216fc..bbae754 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -20,6 +20,11 @@ export default class Rabbit extends EventEmitter { public lock: Promise; public queues: { [s: string]: Queue } = {}; public connecting = false; + private attempt: Promise; + private connections = new Set(); + private lossReported = false; + private attemptError: Error | null = null; + private closed = false; public prefetch: number; public replyPattern: boolean; public prefix: string; @@ -77,35 +82,92 @@ 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(); + 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) { + 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) { + 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; @@ -113,12 +175,12 @@ export default class Rabbit extends EventEmitter { 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() {