Repository: moscajs/aedes
Pull request #979 (2024-08-25 16:31:02)

fix: memory leak in connected clients counter
This pr is related to the https://github.com/moscajs/aedes/issues/928 
The made fix **didn't decrease connectedClients** counter, which lead to incorrect amount of clients provided by broker.

One more issue is related to constant increase of subscriptions. My broker has a lot of unique clients per day with short keepAliveTimeout and frequent reconnection and subscription rate. I suspect that the problem is in the **authorizeSubscribe** method which can be async. This may lead to subscription for closed clients. In this case broker will never unsub them. 

Unfortunately, I am limited in experiments on production to prove my idea, but the proposed fix seems reasonable and harmless to me. 

coveralls: ## Pull Request Test Coverage Report for [Build 10548488946](https://coveralls.io/builds/69470977)


### Details

* **6** of **7**   **(85.71%)**  changed or added relevant lines in **2** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage decreased (**-0.2%**) to **99.575%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [lib/handlers/subscribe.js](https://coveralls.io/builds/69470977/source?filename=lib%2Fhandlers%2Fsubscribe.js#L162) | 1 | 2 | 50.0%
<!-- | **Total:** | **6** | **7** | **85.71%** | -->


|  Totals | [![Coverage Status](https://coveralls.io/builds/69470977/badge)](https://coveralls.io/builds/69470977) |
| :-- | --: |
| Change from base [Build 9447929816](https://coveralls.io/builds/67987177): |  -0.2% |
| Covered Lines: | 827 |
| Relevant Lines: | 828 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/aedes.js b/aedes.js
index 956b740..283351b 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 2470427..2ae0a46 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)


Test patch:
diff --git a/test/events.js b/test/events.js
index cd2137a..76ee9ad 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))
 })