Repository: node-red/node-red (js)

Issue:
Persistent MQTT Session not working as expected with Mosquitto
### Current Behavior

As mentioned in my [Node-RED forum post](https://discourse.nodered.org/t/persistent-client-with-mosquitto-not-working-as-intended/66278) i'm having issues getting persistent MQTT sessions to work with Mosquitto.

I have a MQTT broker node in Node-RED that is connecting to my Mosquitto broker. I'm using MQTT v5 and a selfsigned OpenSSL certificate that I don't check on my client.

The connection, birth message, last will, subscriptions and publishes all are working perfectly fine. Also i have disabled "Use clean start" and i have set a `session expiry` of 2592000 seconds (=30 days).

When i check my Mosquitto persistence file, the client is indeed stored in that file, but the `Session Expiry Interval` is set to 0 instead of 2592000. When i configure a mqtt.js-Client, set `clean: false` and give that client a `sessionExpiryInterval: 2592000`, that client is persisted correctly.

The problem with this is, that if my Raspberry Pi running Node-RED is getting disconnected for a time longer than the keep alive for whatever reason, all messages sent to my Node-RED are lost even though they are sent with `qos: 2`. I think the reason is, that Mosquitto deletes the session of the client, since its set to persist for 0 seconds.

### Expected Behavior

The expected behavior would be, that the client gets persisted in Mosquitto with the correct `Session Expiry Interval` of 2592000.

### Steps To Reproduce

Set up a Mosquitto broker. The important settings for reproduction of the issue should be the following:
```
persistence true
persistence_location /var/lib/mosquitto/
persistent_client_expiration 14d
autosave_interval 300

log_dest file /var/log/mosquitto/mosquitto.log
log_type error
log_type warning
log_type notice
log_type information
log_type debug
log_type subscribe
log_type unsubscribe

listener 8883 0.0.0.0
require_certificate false
```

Now use my example flow and let Node-RED connect to Mosquitto. The node should subscribe just fine, everything works as expected and the birth message should be sent.

If you check the Mosquitto persistence file in the `persistence location` the client should be registered with `Session Expiry Interval: 0`, even tho it is set to 2592000 in Node-RED. Also you should see that the subscription of the client to that topic is also stored there, which is as expected.

Now open the Mosquitto log file with `tail -f` or similar.

Next you want to disconnect the Node-RED client from Mosquitto. Depending on how you disconnect you should see the last will or the disconnection message sent. Also there should be a `Client closed connection` and unsubscribes in the log file.

Now send a message with `qos: 2` to the Topic that Node-RED was subscribed to and reconnect Node-RED afterwards. You should see in the log file, that the client is getting subscribed to the topic again, but no message is sent. Also the debug node in Node-RED doesn't receive anything.

### Example flow

```
[
    {
        "id": "66665620ee0f942d",
        "type": "tab",
        "label": "Flow 1",
        "disabled": false,
        "info": "",
        "env": []
    },
    {
        "id": "10e93a9230f786ee",
        "type": "mqtt in",
        "z": "66665620ee0f942d",
        "name": "Subscribe",
        "topic": "",
        "qos": "2",
        "datatype": "auto",
        "broker": "652b515d2702868a",
        "nl": false,
        "rap": true,
        "rh": 0,
        "inputs": 1,
        "x": 680,
        "y": 480,
        "wires": [
            [
                "6077501aadbc3fd5"
            ]
        ]
    },
    {
        "id": "6ddf714390eb2e63",
        "type": "inject",
        "z": "66665620ee0f942d",
        "name": "inject",
        "props": [],
        "repeat": "",
        "crontab": "",
        "once": true,
        "onceDelay": "0.1",
        "topic": "",
        "x": 370,
        "y": 480,
        "wires": [
            [
                "cc85799bd023dbb2"
            ]
        ]
    },
    {
        "id": "cc85799bd023dbb2",
        "type": "function",
        "z": "66665620ee0f942d",
        "name": "Set MQTT Topic",
        "func": "msg.action = \"subscribe\";\nmsg.topic = {topic:\"some/random/topic\", qos:2, nl:false, rap:true, rh:true};\nreturn msg;",
        "outputs": 1,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "libs": [],
        "x": 520,
        "y": 480,
        "wires": [
            [
                "10e93a9230f786ee"
            ]
        ]
    },
    {
        "id": "6077501aadbc3fd5",
        "type": "debug",
        "z": "66665620ee0f942d",
        "name": "",
        "active": true,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "false",
        "statusVal": "",
        "statusType": "auto",
        "x": 830,
        "y": 480,
        "wires": []
    },
    {
        "id": "652b515d2702868a",
        "type": "mqtt-broker",
        "name": "",
        "broker": "",
        "port": "8883",
        "tls": "12e956d8a888d3bb",
        "clientid": "clientid",
        "autoConnect": true,
        "usetls": true,
        "protocolVersion": "5",
        "keepalive": "60",
        "cleansession": false,
        "birthTopic": "some/other/topic",
        "birthQos": "2",
        "birthRetain": "true",
        "birthPayload": "Client online",
        "birthMsg": {},
        "closeTopic": "some/other/topic",
        "closeQos": "2",
        "closeRetain": "true",
        "closePayload": "Client offline",
        "closeMsg": {},
        "willTopic": "some/other/topic",
        "willQos": "2",
        "willRetain": "true",
        "willPayload": "Client offline",
        "willMsg": {},
        "userProps": "",
        "sessionExpiry": "2592000",
        "credentials": {
            "user": "",
            "password": ""
        }
    },
    {
        "id": "12e956d8a888d3bb",
        "type": "tls-config",
        "name": "",
        "cert": "",
        "key": "",
        "ca": "",
        "certname": "",
        "keyname": "",
        "caname": "",
        "servername": "",
        "verifyservercert": false,
        "alpnprotocol": "",
        "credentials": {}
    }
]
```


### Environment

- Node-RED version: v2.2.3
- Node.js version: v14.19.1
- npm version: v6.14.16
- Platform/OS: Raspberry Pi -> Raspbian Lite 
- Browser: Google Chrome


Pull request:
ensure sessionExpiry(Interval) is applied
fixes #3838 

## Types of changes

- [x] Bugfix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)

## Proposed changes

Ensure MQTT v5 `options.property.sessionExpiryInterval` is set if configured in UI or passed in as an option with the "connect" action

NOTE: This PR also addresses some existing issues with the local MQTT tests namely...
* `done` being called more than once
* MQTT `birth` being published BEFORE `connect` is complete, making it impossible to test birth msg correctly


## Local MQTT Test results (MQTT Tests are skipped in GitHub)...
```
  MQTT Nodes
    √ should be loaded and have default values (MQTT V4) (108ms)  << modified to check MQTT v5 properties are not present
    √ should be loaded and have default values (MQTT V5)  << new test to check MQTT v5 properties are present
    √ basic send and receive tests (64ms)
    √ should send JSON and receive string (auto mode)
    √ should send JSON and receive object (auto-detect mode) (43ms)
    √ should send invalid JSON and receive string (auto mode)
    √ should send invalid JSON and receive string (auto-detect mode)
    √ should send JSON and receive string (utf8 mode)
    √ should send JSON and receive Object (json mode)
    √ should send invalid JSON and raise error (json mode)
    √ should send String and receive Buffer (buffer mode)
    √ should send utf8 Buffer and receive String (auto mode) (42ms)
    √ should send non utf8 Buffer and receive Buffer (auto mode)
    √ should send/receive all v5 flags and settings
    √ should send regular string with v5 media type "text/plain" and receive a string (auto mode) (40ms)
    √ should send JSON with v5 media type "text/plain" and receive a string (auto mode)
    √ should send JSON with v5 media type "text/plain" and receive a string (auto-detect mode)
    √ should send JSON with v5 media type "application/json" and receive an object (auto-detect mode)
    √ should send invalid JSON with v5 media type "application/json" and raise an error (auto mode)
    √ should send buffer with v5 media type "application/json" and receive an object (auto-detect mode)
    √ should send buffer with v5 media type "text/plain" and receive a string (auto mode)
    √ should send buffer with v5 media type "application/zip" and receive a buffer (auto mode)
    √ should subscribe dynamically via action
    √ should connect via "connect" action
    √ should disconnect via "disconnect" action
    √ should publish birth message
  29 passing (1s)
```

## Checklist
- [x] I have read the [contribution guidelines](https://github.com/node-red/node-red/blob/master/CONTRIBUTING.md)
- [ ] For non-bugfix PRs, I have discussed this change on the forum/slack team.
- [x] I have run `grunt` to verify the unit tests pass
- [ ] I have added suitable unit tests to cover the new/changed functionality


Patch:
diff --git a/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js b/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js
index 9e6195b5d..50b1d690f 100644
--- a/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js
+++ b/packages/node_modules/@node-red/nodes/core/network/10-mqtt.js
@@ -474,7 +474,6 @@ module.exports = function(RED) {
             setIfHasProperty(opts, node, "protocolVersion", init);
             setIfHasProperty(opts, node, "keepalive", init);
             setIfHasProperty(opts, node, "cleansession", init);
-            setIfHasProperty(opts, node, "sessionExpiry", init);
             setIfHasProperty(opts, node, "topicAliasMaximum", init);
             setIfHasProperty(opts, node, "maximumPacketSize", init);
             setIfHasProperty(opts, node, "receiveMaximum", init);
@@ -484,6 +483,11 @@ module.exports = function(RED) {
             } else if (hasProperty(opts, "userProps")) {
                 node.userProperties = opts.userProps;
             }
+            if (hasProperty(opts, "sessionExpiry")) {
+                node.sessionExpiryInterval = opts.sessionExpiry;
+            } else if (hasProperty(opts, "sessionExpiryInterval")) {
+                node.sessionExpiryInterval = opts.sessionExpiryInterval
+            }
 
             function createLWT(topic, payload, qos, retain, v5opts, v5SubPropName) {
                 let message = undefined;
@@ -782,7 +786,9 @@ module.exports = function(RED) {
 
                         // Send any birth message
                         if (node.birthMessage) {
-                            node.publish(node.birthMessage);
+                            setTimeout(() => {
+                                node.publish(node.birthMessage);
+                            }, 1);
                         }
                     });
                     node._clientOn("reconnect", function() {


Test patch:
diff --git a/test/nodes/core/network/21-mqtt_spec.js b/test/nodes/core/network/21-mqtt_spec.js
index 655cc9c73..fd1309b74 100644
--- a/test/nodes/core/network/21-mqtt_spec.js
+++ b/test/nodes/core/network/21-mqtt_spec.js
@@ -27,7 +27,7 @@ describe('MQTT Nodes', function () {
         } catch (error) { }
     });
 
-    it('should be loaded and have default values', function (done) {
+    it('should be loaded and have default values (MQTT V4)', function (done) {
         this.timeout = 2000;
         const { flow, nodes } = buildBasicMQTTSendRecvFlow({ id: "mqtt.broker", name: "mqtt_broker", autoConnect: false }, { id: "mqtt.in", topic: "in_topic" }, { id: "mqtt.out", topic: "out_topic" });
         helper.load(mqttNodes, flow, function () {
@@ -61,6 +61,52 @@ describe('MQTT Nodes', function () {
                 mqttBroker.options.clientId.should.containEql('nodered_');
                 mqttBroker.options.should.have.property('keepalive').type("number");
                 mqttBroker.options.should.have.property('reconnectPeriod').type("number");
+                //as this is not a v5 connection, ensure v5 properties are not present
+                mqttBroker.options.should.not.have.property('protocolVersion', 5);
+                mqttBroker.options.should.not.have.property('properties');
+                done();
+            } catch (error) {
+                done(error)
+            }
+        });
+    });
+    it('should be loaded and have default values (MQTT V5)', function (done) {
+        this.timeout = 2000;
+        const { flow, nodes } = buildBasicMQTTSendRecvFlow({ id: "mqtt.broker", name: "mqtt_broker", autoConnect: false, cleansession: false, clientid: 'clientid', keepalive: 35, sessionExpiry: '6000', protocolVersion: '5', userProps: {"prop": "val"}}, { id: "mqtt.in", topic: "in_topic" }, { id: "mqtt.out", topic: "out_topic" });
+        helper.load(mqttNodes, flow, function () {
+            try {
+                const mqttIn = helper.getNode("mqtt.in");
+                const mqttOut = helper.getNode("mqtt.out");
+                const mqttBroker = helper.getNode("mqtt.broker");
+
+                should(mqttIn).be.type("object", "mqtt in node should be an object")
+                mqttIn.should.have.property('broker', nodes.mqtt_broker.id); //should be the id of the broker node
+                mqttIn.should.have.property('datatype', 'utf8'); //default: 'utf8'
+                mqttIn.should.have.property('isDynamic', false);  //default: false
+                mqttIn.should.have.property('inputs', 0); //default: 0 
+                mqttIn.should.have.property('qos', 2); //default: 2
+                mqttIn.should.have.property('topic', "in_topic");
+                mqttIn.should.have.property('wires', [["helper.node"]]);
+
+                should(mqttOut).be.type("object", "mqtt out node should be an object")
+                mqttOut.should.have.property('broker', nodes.mqtt_broker.id); //should be the id of the broker node
+                mqttOut.should.have.property('topic', "out_topic");
+
+                should(mqttBroker).be.type("object", "mqtt broker node should be an object")
+                mqttBroker.should.have.property('broker', BROKER_HOST);
+                mqttBroker.should.have.property('port', BROKER_PORT);
+                mqttBroker.should.have.property('brokerurl');
+                // mqttBroker.should.have.property('autoUnsubscribe', true);//default: true 
+                mqttBroker.should.have.property('autoConnect', false);//Set "autoConnect:false" in brokerOptions 
+                mqttBroker.should.have.property('options');
+                mqttBroker.options.should.have.property('clean', false);
+                mqttBroker.options.should.have.property('clientId', 'clientid');
+                mqttBroker.options.should.have.property('keepalive').type("number", 35);
+                mqttBroker.options.should.have.property('reconnectPeriod').type("number");
+                //as this IS a v5 connection, ensure v5 properties are not present
+                mqttBroker.options.should.have.property('protocolVersion', 5);
+                mqttBroker.options.should.have.property('properties');
+                mqttBroker.options.properties.should.have.property('sessionExpiryInterval');
                 done();
             } catch (error) {
                 done(error)
@@ -173,7 +219,7 @@ describe('MQTT Nodes', function () {
             topic: nextTopic(),
             payload: '{prop:"value3", "num":3}', // send invalid JSON ...
         }
-        const hooks = { done: done, beforeLoad: null, afterLoad: null, afterConnect: null }
+        const hooks = { done: null, beforeLoad: null, afterLoad: null, afterConnect: null }
         hooks.afterLoad = (helperNode, mqttBroker, mqttIn, mqttOut) => {
             helperNode.on("input", function (msg) {
                 try {
@@ -299,7 +345,7 @@ describe('MQTT Nodes', function () {
             topic: nextTopic(),
             payload: '{prop:"value3", "num":3}', contentType: "application/json", // send invalid JSON ...
         }
-        const hooks = { done: done, beforeLoad: null, afterLoad: null, afterConnect: null }
+        const hooks = { done: null, beforeLoad: null, afterLoad: null, afterConnect: null }
         hooks.afterLoad = (helperNode, mqttBroker, mqttIn, mqttOut) => {
             helperNode.on("input", function (msg) {
                 try {
@@ -385,7 +431,7 @@ describe('MQTT Nodes', function () {
         if (skipTests) { return this.skip() }
         this.timeout = 2000;
         const options = {}
-        const hooks = { done: done, beforeLoad: null, afterLoad: null, afterConnect: null }
+        const hooks = { beforeLoad: null, afterLoad: null, afterConnect: null }
         hooks.beforeLoad = (flow) => { //add a status node pointed at MQTT Out node (to watch for connection status change)
             flow.push({ "id": "status.node", "type": "status", "name": "status_node", "scope": ["mqtt.out"], "wires": [["helper.node"]] });//add status node to watch mqtt_out
         }
@@ -416,18 +462,31 @@ describe('MQTT Nodes', function () {
         this.timeout = 2000;
         const baseTopic = nextTopic();
         const brokerOptions = {
+            autoConnect: false,
             protocolVersion: 4,
             birthTopic: baseTopic + "/birth",
-            birthPayload: "broker connected",
+            birthPayload: "broker birth",
             birthQos: 2,
         }
-        const options = {};
-        const hooks = { done: done, beforeLoad: null, afterLoad: null, afterConnect: null };
-        options.expectMsg = {
+        const expectMsg = {
             topic: brokerOptions.birthTopic,
             payload: brokerOptions.birthPayload,
             qos: brokerOptions.birthQos
         };
+        const options = { };
+        const hooks = { };
+        hooks.afterLoad = (helperNode, mqttBroker, mqttIn, mqttOut) => {
+            helperNode.on("input", function (msg) {
+                try {
+                    compareMsgToExpected(msg, expectMsg);
+                    done();
+                } catch (error) {
+                    done(error)
+                }
+            })
+            mqttIn.receive({ "action": "connect" }); //now request connect action
+            return true; //handled
+        }       
         testSendRecv(brokerOptions, { topic: brokerOptions.birthTopic }, {}, options, hooks);
     });
     itConditional('should publish close message', function (done) {
@@ -666,6 +725,7 @@ function buildMQTTBrokerNode(id, name, brokerHost, brokerPort, options) {
     node.cleansession = String(options.cleansession) == "false" ? false : true;
     node.autoUnsubscribe = String(options.autoUnsubscribe) == "false" ? false : true;
     node.autoConnect = String(options.autoConnect) == "false" ? false : true;
+    node.sessionExpiry = options.sessionExpiry ? options.sessionExpiry : undefined;
 
     if (options.birthTopic) {
         node.birthTopic = options.birthTopic;
@@ -760,8 +820,8 @@ function waitBrokerConnect(broker, timeLimit) {
     let waitConnected = (broker, timeLimit) => {
         const brokers = Array.isArray(broker) ? broker : [broker];
         timeLimit = timeLimit || 1000;
-        let timer, resolved = false;
         return new Promise( (resolve, reject) => {
+            let timer, resolved = false;
             timer = wait();
             function wait() {
                 if (brokers.every(e => e.connected == true)) {