From 9b9d52445062a5069f80277f873d2aa6c44f0f2e Mon Sep 17 00:00:00 2001
From: Uladzimir <uladzimir.danko@gmail.com>
Date: Wed, 28 Aug 2024 08:41:35 +0200
Subject: [PATCH] fix: memory leak in connected clients counter (#979)

Co-authored-by: Uladzimir Danko <uladzimirdanko@uladzimirsmbp14.home>
---
 aedes.js                  | 10 +++++++---
 lib/handlers/subscribe.js |  7 +++++++
 test/events.js            | 14 ++++++++------
 3 files changed, 22 insertions(+), 9 deletions(-)

diff --git a/aedes.js b/aedes.js
index 956b7409..283351b2 100644
--- a/aedes.js
+++ b/aedes.js
@@ -172,7 +172,7 @@ function Aedes (opts) {
     if (that.clients[clientId] && serverId !== that.id) {
       if (that.clients[clientId].closed) {
         // remove the client from the list if it is already closed
-        delete that.clients[clientId]
+        that.deleteClient(clientId)
         done()
       } else {
         that.clients[clientId].close(done)
@@ -316,8 +316,7 @@ Aedes.prototype._finishRegisterClient = function (client) {
 }
 
 Aedes.prototype.unregisterClient = function (client) {
-  this.connectedClients--
-  delete this.clients[client.id]
+  this.deleteClient(client.id)
   this.emit('clientDisconnect', client)
   this.publish({
     topic: $SYS_PREFIX + this.id + '/disconnect/clients',
@@ -325,6 +324,11 @@ Aedes.prototype.unregisterClient = function (client) {
   }, noop)
 }
 
+Aedes.prototype.deleteClient = function (clientId) {
+  this.connectedClients--
+  delete this.clients[clientId]
+}
+
 function closeClient (client, cb) {
   this.clients[client].close(cb)
 }
diff --git a/lib/handlers/subscribe.js b/lib/handlers/subscribe.js
index 24704274..2ae0a467 100644
--- a/lib/handlers/subscribe.js
+++ b/lib/handlers/subscribe.js
@@ -155,6 +155,13 @@ function addSubs (sub, done) {
     func = blockDollarSignTopics(func)
   }
 
+  if (client.closed || client.broker.closed) {
+    // a hack, sometimes client.close() or broker.close() happened
+    // before authenticate() comes back
+    // we don't continue subscription here
+    return
+  }
+
   if (!client.subscriptions[topic]) {
     client.subscriptions[topic] = new Subscription(qos, func, rh, rap, nl)
     broker.subscribe(topic, func, done)
diff --git a/test/events.js b/test/events.js
index cd2137a1..76ee9ad1 100644
--- a/test/events.js
+++ b/test/events.js
@@ -223,18 +223,20 @@ test('Test backpressure aedes published function', function (t) {
 })
 
 test('clear closed clients when the same clientId is managed by another broker', function (t) {
-  t.plan(1)
+  t.plan(2)
 
   const clientId = 'closed-client'
-  const broker = aedes()
+  const aedesBroker = aedes()
 
   // simulate a closed client on the broker
-  broker.clients[clientId] = { closed: true }
+  aedesBroker.clients[clientId] = { closed: true, broker: aedesBroker }
+  aedesBroker.connectedClients = 1
 
   // simulate the creation of the same client on another broker of the cluster
-  broker.publish({ topic: '$SYS/anotherbroker/new/clients', payload: clientId }, () => {
-    t.equal(broker.clients[clientId], undefined) // check that the closed client was removed
+  aedesBroker.publish({ topic: '$SYS/anotherbroker/new/clients', payload: clientId }, () => {
+    t.equal(aedesBroker.clients[clientId], undefined) // check that the closed client was removed
+    t.equal(aedesBroker.connectedClients, 0)
   })
 
-  t.teardown(broker.close.bind(broker))
+  t.teardown(aedesBroker.close.bind(aedesBroker))
 })