Repository: moscajs/aedes
Pull request #508 (2020-06-19 12:51:59)

fix: write leaks
Actually [write](https://github.com/moscajs/aedes/blob/master/lib/write.js#L5) doesn't return errors to done callback but insteed it calls `client._onError` that will emit a clientError/connectionError event and disconnect the client. This could cause some memory leaks as when an error occurs the write callback is never called

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

* **-1** of **25**   **(96.0%)**  changed or added relevant lines in **5** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage decreased (**-0.09%**) to **99.549%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [lib/handlers/publish.js](https://coveralls.io/builds/31561446/source?filename=lib%2Fhandlers%2Fpublish.js#L44) | 7 | 8 | 87.5%
<!-- | **Total:** | **24** | **25** | **96.0%** | -->


|  Totals | [![Coverage Status](https://coveralls.io/builds/31561446/badge)](https://coveralls.io/builds/31561446) |
| :-- | --: |
| Change from base [Build 133185632](https://coveralls.io/builds/31412185): |  -0.09% |
| Covered Lines: | 788 |
| Relevant Lines: | 789 |

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


Patch:
diff --git a/lib/client.js b/lib/client.js
index 78a0888..534fdd6 100644
--- a/lib/client.js
+++ b/lib/client.js
@@ -105,7 +105,10 @@ function Client (broker, conn, req) {
       setImmediate(() => {
         var packet = new Packet(toForward, broker)
         packet.qos = 0
-        write(that, packet, cb)
+        write(that, packet, function (err) {
+          that._onError(err)
+          cb() // don't pass the error here or it will be thrown be mqemitter
+        })
       })
     } else {
       setImmediate(cb)
@@ -171,6 +174,7 @@ function writeQoS (err, client, packet) {
   if (err) {
     // is this right, or we should ignore thins?
     client.emit('error', err)
+    packet.writeCallback(err)
   } else {
     write(client, packet, packet.writeCallback)
   }
@@ -181,6 +185,8 @@ function drainRequest (req) {
 }
 
 function onError (err) {
+  if (!err) return
+
   this.errored = true
   this.conn.removeAllListeners('error')
   this.conn.on('error', noop)
diff --git a/lib/handlers/connect.js b/lib/handlers/connect.js
index 66c3ba0..a48a312 100644
--- a/lib/handlers/connect.js
+++ b/lib/handlers/connect.js
@@ -213,10 +213,12 @@ function registerClient (arg, done) {
 function doConnack (arg, done) {
   const client = arg.client || this.client
   const connack = new Connack(arg)
-  write(client, connack, function () {
-    client.broker.emit('connackSent', connack, client)
-    client.connackSent = true
-    done()
+  write(client, connack, function (err) {
+    if (!err) {
+      client.broker.emit('connackSent', connack, client)
+      client.connackSent = true
+    }
+    done(err)
   })
 }
 
diff --git a/lib/handlers/publish.js b/lib/handlers/publish.js
index 229d49a..398e2a8 100644
--- a/lib/handlers/publish.js
+++ b/lib/handlers/publish.js
@@ -39,13 +39,21 @@ function enqueuePublish (packet, done) {
 
   switch (packet.qos) {
     case 2:
-      write(client, new PubRec(packet), function () {
-        client.broker.persistence.incomingStorePacket(client, packet, done)
+      write(client, new PubRec(packet), function (err) {
+        if (err) {
+          done(err)
+        } else {
+          client.broker.persistence.incomingStorePacket(client, packet, done)
+        }
       })
       break
     case 1:
-      write(client, new PubAck(packet), function () {
-        client.broker.publish(packet, client, done)
+      write(client, new PubAck(packet), function (err) {
+        if (err) {
+          done(err)
+        } else {
+          client.broker.publish(packet, client, done)
+        }
       })
       break
     case 0:
diff --git a/lib/qos-packet.js b/lib/qos-packet.js
index fee554b..5527fe1 100644
--- a/lib/qos-packet.js
+++ b/lib/qos-packet.js
@@ -6,7 +6,7 @@ const util = require('util')
 function QoSPacket (original, client) {
   Packet.call(this, original, client.broker)
 
-  this.writeCallback = noop
+  this.writeCallback = client._onError.bind(client)
 
   if (!original.messageId) {
     this.messageId = client._nextId
@@ -22,6 +22,4 @@ function QoSPacket (original, client) {
 
 util.inherits(QoSPacket, Packet)
 
-function noop () {}
-
 module.exports = QoSPacket
diff --git a/lib/write.js b/lib/write.js
index 06f0430..2eb31a2 100644
--- a/lib/write.js
+++ b/lib/write.js
@@ -3,22 +3,22 @@
 const mqtt = require('mqtt-packet')
 
 function write (client, packet, done) {
-  if (client.conn.writable && (client.connecting || client.connected)) {
+  var error = null
+  if (client.connecting || client.connected) {
     try {
       const result = mqtt.writeToStream(packet, client.conn)
-      if (!result && !client.errored && done) {
+      if (!result && !client.errored) {
         client.conn.once('drain', done)
         return
       }
-      if (done) {
-        setImmediate(done)
-      }
-    } catch (err) {
-      setImmediate(client._onError.bind(client, new Error('packet received not valid')))
+    } catch (e) {
+      error = new Error('packet received not valid')
     }
   } else {
-    setImmediate(client._onError.bind(client, new Error('connection closed')))
+    error = new Error('connection closed')
   }
+
+  setImmediate(done, error, client)
 }
 
 module.exports = write


Test patch:
diff --git a/test/basic.js b/test/basic.js
index b38151e..893ffc0 100644
--- a/test/basic.js
+++ b/test/basic.js
@@ -132,7 +132,7 @@ test('publish empty topic throws error', function (t) {
 })
 
 // Catch invalid packet writeToStream errors
-test('catch write errors', function (t) {
+test('return write errors to callback', function (t) {
   t.plan(1)
 
   const write = proxyquire('../lib/write.js', {
@@ -147,13 +147,12 @@ test('catch write errors', function (t) {
     conn: {
       writable: true
     },
-    connecting: true,
-    _onError: (err) => {
-      t.equal(err.message, 'packet received not valid', 'should catch the error')
-    }
+    connecting: true
   }
 
-  write(client, {})
+  write(client, {}, function (err) {
+    t.equal(err.message, 'packet received not valid', 'should return the error to callback')
+  })
 })
 
 ;[{ qos: 0, clean: false }, { qos: 0, clean: true }, { qos: 1, clean: false }, { qos: 1, clean: true }].forEach(function (ele) {
diff --git a/test/qos1.js b/test/qos1.js
index 1eba58a..202271a 100644
--- a/test/qos1.js
+++ b/test/qos1.js
@@ -56,6 +56,30 @@ test('publish QoS 1 throws error', function (t) {
   })
 })
 
+test('publish QoS 1 throws error on write', function (t) {
+  t.plan(1)
+
+  const s = connect(setup())
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  s.broker.on('client', function (client) {
+    client.connected = false
+    client.connecting = false
+
+    s.inStream.write({
+      cmd: 'publish',
+      topic: 'hello',
+      payload: 'world',
+      qos: 1,
+      messageId: 42
+    })
+  })
+
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'connection closed', 'throws error')
+  })
+})
+
 test('publish QoS 1 and check offline queue', function (t) {
   t.plan(13)
 
diff --git a/test/qos2.js b/test/qos2.js
index 61acaac..119a54c 100644
--- a/test/qos2.js
+++ b/test/qos2.js
@@ -119,6 +119,30 @@ test('subscribe QoS 2', function (t) {
   })
 })
 
+test('publish QoS 2 throws error on write', function (t) {
+  t.plan(1)
+
+  const s = connect(setup())
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  s.broker.on('client', function (client) {
+    client.connected = false
+    client.connecting = false
+
+    s.inStream.write({
+      cmd: 'publish',
+      topic: 'hello',
+      payload: 'world',
+      qos: 2,
+      messageId: 42
+    })
+  })
+
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'connection closed', 'throws error')
+  })
+})
+
 test('client.publish with clean=true subscribption QoS 2', function (t) {
   t.plan(8)