fix(mqtt): Use semaphore to avoid reconfigure race condition

Hypfer · May 30, 2021 · 5b4b377 · 5b4b377
1 parent 0d7d836
commit 5b4b377
Show file tree

Hide file tree

Showing 3 changed files with 73 additions and 25 deletions.
diff --git a/backend/lib/mqtt/MqttController.js b/backend/lib/mqtt/MqttController.js
@@ -7,6 +7,7 @@ const Logger = require("../Logger");
 const mqtt = require("mqtt");
 const MqttCommonAttributes = require("./MqttCommonAttributes");
 const RobotMqttHandle = require("./handles/RobotMqttHandle");
+const Semaphore = require("semaphore");
 
 /**
  * @typedef {object} DeconfigureOptions
@@ -26,6 +27,10 @@ class MqttController {
         this.config = options.config;
         this.robot = options.robot;
 
+        this.semaphores = {
+            reconfigure: Semaphore(1)
+        };
+
         this.client = null;
         this.refreshIntervalID = null;
 
@@ -428,34 +433,46 @@ class MqttController {
      * @param {string} [options.errorState]
      * @return {Promise<void>}
      */
-    async reconfigure(cb, options) {
-        const reconfOptions = {
-            reconfigState: HomieCommonAttributes.STATE.INIT,
-            targetState: HomieCommonAttributes.STATE.READY,
-            errorState: HomieCommonAttributes.STATE.ALERT
-        };
-        if (options !== undefined) {
-            Object.assign(reconfOptions, options);
-        }
+    reconfigure(cb, options) {
+        return new Promise((resolve, reject) => {
+            this.semaphores.reconfigure.take(async () => {
+                const reconfOptions = {
+                    reconfigState: HomieCommonAttributes.STATE.INIT,
+                    targetState: HomieCommonAttributes.STATE.READY,
+                    errorState: HomieCommonAttributes.STATE.ALERT
+                };
+                if (options !== undefined) {
+                    Object.assign(reconfOptions, options);
+                }
 
-        // Nested reconfiguration, may occur i.e. if consumables are already available during first configuration.
-        // In this case, just force the target state to be init as well.
-        // on("connect") will handle the special case where this is triggered on first connection.
-        if (this.state === reconfOptions.reconfigState && this.state === HomieCommonAttributes.STATE.INIT &&
-            reconfOptions.targetState === HomieCommonAttributes.STATE.READY) {
+                // TODO: Is this still required?
 
-            reconfOptions.targetState = HomieCommonAttributes.STATE.INIT;
-        }
+                // Nested reconfiguration, may occur i.e. if consumables are already available during first configuration.
+                // In this case, just force the target state to be init as well.
+                // on("connect") will handle the special case where this is triggered on first connection.
+                if (this.state === reconfOptions.reconfigState && this.state === HomieCommonAttributes.STATE.INIT &&
+                    reconfOptions.targetState === HomieCommonAttributes.STATE.READY) {
 
-        try {
-            await this.setState(reconfOptions.reconfigState);
-            await cb();
-            await this.setState(reconfOptions.targetState);
-        } catch (err) {
-            Logger.error("MQTT reconfiguration error", err);
-            await this.setState(reconfOptions.errorState);
-            throw err;
-        }
+                    reconfOptions.targetState = HomieCommonAttributes.STATE.INIT;
+                }
+
+                try {
+                    await this.setState(reconfOptions.reconfigState);
+                    await cb();
+                    await this.setState(reconfOptions.targetState);
+
+                    this.semaphores.reconfigure.leave();
+                    resolve();
+                } catch (err) {
+                    Logger.error("MQTT reconfiguration error", err);
+                    await this.setState(reconfOptions.errorState);
+
+                    this.semaphores.reconfigure.leave();
+
+                    reject(err);
+                }
+            });
+        });
     }
 
     /**

diff --git a/backend/package.json b/backend/package.json
@@ -44,6 +44,7 @@
     "jetsons": "^1.2.2",
     "mqtt": "^4.0.0",
     "multer": "^1.4.1",
+    "semaphore": "^1.1.0",
     "uuid": "^8.3.0",
     "ws": "^6.1.4"
   },
@@ -57,6 +58,7 @@
     "@types/multer": "^1.4.2",
     "@types/node": "^13.9.1",
     "@types/on-headers": "^1.0.0",
+    "@types/semaphore": "^1.1.1",
     "@types/uuid": "^8.3.0",
     "@types/ws": "^7.2.2",
     "cross-env": "7.0.2",

diff --git a/package-lock.json b/package-lock.json