Repository: moscajs/aedes (js)

Issue:
subscribe/unsubscribe processed out of order when persistence is used
When a client subscribes and unsubscribes repeatedly to the same topic in quick succession, sometimes these are processed out of order, resulting in the client believing it has subscribed but it is actually not subscribed.

It looks like the root cause is that the subscribe handler may [complete synchronously](https://github.com/moscajs/aedes/blob/295538898deaa57edcd6f62c469a55dd02a3a58c/lib/handlers/subscribe.js#L104) but the unsubscribe handler always [completes asynchronously](https://github.com/moscajs/aedes/blob/295538898deaa57edcd6f62c469a55dd02a3a58c/lib/handlers/unsubscribe.js#L30), so it's possible for the unsubscription to complete after the subscription.

As a workaround, I have implemented a modified persistence which synchronously completes the unsubscribe call when the client is "clean".

Pull request description:
fix: remove subs only when clean flag is false
Fixes #469 

Patch:
diff --git a/lib/handlers/unsubscribe.js b/lib/handlers/unsubscribe.js
index 6c75d9e..c616fc4 100644
--- a/lib/handlers/unsubscribe.js
+++ b/lib/handlers/unsubscribe.js
@@ -27,6 +27,10 @@ function handleUnsubscribe (client, packet, done) {
   }
 
   if (packet.messageId) {
+    if (client.clean) {
+      return actualUnsubscribe(client, packet, done)
+    }
+
     broker.persistence.removeSubscriptions(client, unsubscriptions, function (err) {
       if (err) {
         return done(err)


Test patch:
diff --git a/test/client-pub-sub.js b/test/client-pub-sub.js
index 92b9147..a01c21e 100644
--- a/test/client-pub-sub.js
+++ b/test/client-pub-sub.js
@@ -514,6 +514,36 @@ test('unsubscribe a client', function (t) {
   connect(setup(broker))
 })
 
+test('unsubscribe should not call removeSubscriptions when [clean=true]', function (t) {
+  t.plan(2)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  broker.persistence.removeSubscriptions = function (client, subs, cb) {
+    cb(Error('remove subscription is called'))
+  }
+
+  broker.on('client', function (client) {
+    client.subscribe({
+      topic: 'hello',
+      qos: 1
+    }, function (err) {
+      t.error(err, 'no error')
+      client.unsubscribe({
+        unsubscriptions: [{
+          topic: 'hello',
+          qos: 1
+        }],
+        messageId: 42
+      }, function (err) {
+        t.error(err, 'no error')
+      })
+    })
+  })
+  connect(setup(broker), { clean: true })
+})
+
 test('unsubscribe throws error', function (t) {
   t.plan(2)