NR stalls with this specific MQTT input stream

General
1

This is the first time that NR actually crashes on a specific flow. I guess that will be tricky to debug..

I got a data stream for my BMW car via an external aggregator
(Project BMW CarData → MQTT Bridge) . There had been an equivalent NR node but it's no longer functional.

When the car sends the data it will be a stream of about 10-20 MQTT messages, sometimes even double messages with same topics. I am catching these messages with a function node and "distribute" and sort the parameters in different flow variables for further processing:

[
    {
        "id": "c727cd363836fe5c",
        "type": "tab",
        "label": "Flow 1",
        "disabled": false,
        "info": "",
        "env": []
    },
    {
        "id": "5b67aa09c7333225",
        "type": "mqtt in",
        "z": "c727cd363836fe5c",
        "name": "BWM CarDataStream",
        "topic": "bmw/WTT5A110607J29109/#",
        "qos": "0",
        "datatype": "json",
        "broker": "43078758.727338",
        "nl": false,
        "rap": true,
        "rh": 0,
        "inputs": 0,
        "x": 280,
        "y": 300,
        "wires": [
            [
                "9a15f9a3fe6a7205"
            ]
        ]
    },
    {
        "id": "9a15f9a3fe6a7205",
        "type": "function",
        "z": "c727cd363836fe5c",
        "name": "set flow variables (single topic)",
        "func": "// proper parameter name now in payload.data\n\nfunction mergeFlatKeysIntoNestedDict(newDict, inputObj, { omitFirst = true, maxDepth = 50 } = {}) {\n    for (const fullKey of Object.keys(inputObj)) {\n        const parts = fullKey.split(\".\").filter(Boolean);\n        let path = omitFirst ? parts.slice(2) : parts;\n\n        if (path.length === 0) continue;\n\n        // target is a pointer for newDict (avoid shadowing Node-RED's `node`)\n        let target = newDict;\n\n        for (let i = 0; i < path.length; i++) {\n            const seg = path[i];\n            const isLeaf = i === path.length - 1;\n\n            if (isLeaf) { // final Leaf of the tree, now assign value\n//                node.warn(newDict);\n                target[seg] = inputObj[fullKey]; // add/overwrite only this leaf\n            } else {\n                // ensure branch exists; if something non-object is there, replace it\n                if (typeof target[seg] !== \"object\" || target[seg] === null || Array.isArray(target[seg])) {\n                    target[seg] = {};\n                }\n                // move pointer down to new child\n                target = target[seg];\n            }\n        }\n    }\n    return newDict;\n}\n\nconst para = msg.payload.data\n\n// get name of flow variable \nif (!para || typeof para !== \"object\") {\n    node.warn(\"Expected msg.payload.data to be an object\");\n    return { payload: {} };\n}\n\nconst key = Object.keys(para)[0];\nif (!key) {\n    node.warn(\"msg.payload.data had no keys\");\n    return { payload: {} };\n}\n\nconst nameParts = key.split('.');\nconst name = nameParts[1] ?? nameParts[0];\nnode.warn(`flow variable name: ${name}`)\n\n// read old values if present\nvar para_field = flow.get(name)\nif (para_field == undefined) {\n    para_field = {}; //define first\n}\nmergeFlatKeysIntoNestedDict(para_field, para);\nflow.set(name, para_field)\nreturn { \"payload\": para_field }\n",
        "outputs": 1,
        "timeout": 0,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "libs": [],
        "x": 590,
        "y": 300,
        "wires": [
            [
                "c4801b6a0f3040ef",
                "e1d306ae278ad009"
            ]
        ],
        "info": "still not complete\n.env file should better include the option SPLIT_TOPICS=1"
    },
    {
        "id": "c4801b6a0f3040ef",
        "type": "change",
        "z": "c727cd363836fe5c",
        "name": "Flow.drivetrain",
        "rules": [
            {
                "t": "set",
                "p": "payload",
                "pt": "msg",
                "to": "drivetrain",
                "tot": "flow",
                "dc": true
            }
        ],
        "action": "",
        "property": "",
        "from": "",
        "to": "",
        "reg": false,
        "x": 900,
        "y": 300,
        "wires": [
            []
        ]
    },
    {
        "id": "e1d306ae278ad009",
        "type": "change",
        "z": "c727cd363836fe5c",
        "name": "Flow.cabin",
        "rules": [
            {
                "t": "set",
                "p": "payload",
                "pt": "msg",
                "to": "cabin",
                "tot": "flow",
                "dc": true
            }
        ],
        "action": "",
        "property": "",
        "from": "",
        "to": "",
        "reg": false,
        "x": 890,
        "y": 380,
        "wires": [
            []
        ]
    },
    {
        "id": "43078758.727338",
        "type": "mqtt-broker",
        "name": "Smarthome",
        "broker": "192.168.178.40",
        "port": "1883",
        "clientid": "SMARTHOME",
        "autoConnect": true,
        "usetls": false,
        "protocolVersion": "4",
        "keepalive": "60",
        "cleansession": true,
        "autoUnsubscribe": true,
        "birthTopic": "",
        "birthQos": "0",
        "birthRetain": "false",
        "birthPayload": "",
        "birthMsg": {},
        "closeTopic": "",
        "closeQos": "0",
        "closeRetain": "false",
        "closePayload": "",
        "closeMsg": {},
        "willTopic": "",
        "willQos": "0",
        "willRetain": "false",
        "willPayload": "",
        "willMsg": {},
        "userProps": "",
        "sessionExpiry": ""
    }
]

Every time a data streams arrives NR dashboard gets "offline", and according to the node.warn() messages visible in the log file the function node processes some messages even minutes(!) after receiving.

I observe 2 specific error messages in the log:

3 Apr 16:33:42 - [info] [mqtt-broker:Smarthome] Disconnected from broker: SMARTHOME@mqtt://192.168.178.40:1883
3 Apr 16:33:42 - [info] [zigbee2mqtt-server:Z2M Broker] MQTT Reconnect
3 Apr 16:36:14 - [info] [zigbee2mqtt-server:Z2M Broker] MQTT Error

(Smarthome is my local NR installation) So has my MQTT Broker an "overload" with this particular message stream? Is my ZigBee data stream interfering?

3 Apr 16:47:44 - [error] [function:set flow variables (single topic)] RangeError: Maximum call stack size exceeded

Something is wrong with my javascript function? The message to process looks like:

{"vin":"XXX6T910607JXXXX","entityId":"58b0daa7-9daa-46f1-859a-55d5169ee206","topic":"XXX6T910607JXXXX","timestamp":"2026-03-15T14:57:33.378Z","data":{"vehicle.cabin.door.status":{"timestamp":"2026-03-15T14:57:29Z","value":"UNLOCKED"}}}

2

This is usually due to a loop somewhere in your flows. Possibly even in the function code itself.

Probably a timeout due to the issue.

Unless you have a heavily overloaded host computer or something is sending 10's of thousands of complex MQTT messages per second, unlikely.

This appears to be the core of the issue. I suspect an error in your function node code. Likely either a loop, a recursive call to a function that is not properly bound.

You should probably share your function code.

3

Sure! The provided flow is probably useless for someone not having the same MQTT input:
The MQTT message topic like vehicle.cabin.window.row2.driver.status is converted in a nested JSON variable but there are no more than 5 levels. Where else could the stack size be exceeded?

// proper parameter name now in payload.data

function mergeFlatKeysIntoNestedDict(newDict, inputObj, { omitFirst = true, maxDepth = 50 } = {}) {
    for (const fullKey of Object.keys(inputObj)) {
        const parts = fullKey.split(".").filter(Boolean);
        let path = omitFirst ? parts.slice(2) : parts;

        if (path.length === 0) continue;

        // target is a pointer for newDict
        let target = newDict;

        for (let i = 0; i < path.length; i++) {
            const seg = path[i];
            const isLeaf = i === path.length - 1;

            if (isLeaf) { // final Leaf of the tree, now assign value
//                node.warn(newDict);
                target[seg] = inputObj[fullKey]; // add/overwrite only this leaf
            } else {
                // ensure branch exists; if something non-object is there, replace it
                if (typeof target[seg] !== "object" || target[seg] === null || Array.isArray(target[seg])) {
                    target[seg] = {};
                }
                // move pointer down to new child
                target = target[seg];
            }
        }
    }
    return newDict;
}

const para = msg.payload.data

// get name of flow variable 
if (!para || typeof para !== "object") {
    node.warn("Expected msg.payload.data to be an object");
    return { payload: {} };
}

const key = Object.keys(para)[0];
if (!key) {
    node.warn("msg.payload.data had no keys");
    return { payload: {} };
}

const nameParts = key.split('.');
const name = nameParts[1] ?? nameParts[0];
node.warn(`flow variable name: ${name}`)

// read old values if present
var para_field = flow.get(name)
if (para_field == undefined) {
    para_field = {}; //define first
}
mergeFlatKeysIntoNestedDict(para_field, para);
flow.set(name, para_field)
return { "payload": para_field }
4

If you feed in a single message with an inject node what happens?

5

Correct processing, at least for my few constructed simulated input test data. I probably need to simulate the complete burst of messages arriving from the car.. :face_with_peeking_eye:

6

How many of these variables are being created and how large are they?

7

What rate do they arrive at, or how many are there if they arrive in a burst, and how often are the bursts?

Is there anything else in your flows? Is there something monitoring the context vars you are setting?

8

No idea how/if this could help fix your issue however my OCD has a couple comments on your code:

It might well be better to return the initial message from the function:

msg.payload = para_field
return msg

seems to be the convention. In that way you can also add that code to a request-response node as the context is maintained in the msg (i.e. msg.req and msg.res) object and would be lost if you returned { payload: para_field } from your function.

It might not be relevant today, but tomorrow it will be the next bug to debug.

Generally it is good practice to use the return value:

para_field = mergeFlatKeysIntoNestedDict(para_field, para);

this is a general coding convention: don't alter values you pass into a function. Instead use these values in the function and return an alter version of these from function. This avoids unwanted side effects. This is the main difference between "procedures" and "functions" - functions return something and procedures can alter input parameters (i.e. have side effects).

para_field_one = mergeFlatKeysIntoNestedDict(para_field, para);
para_field_two = mergeFlatKeysIntoNestedDict(para_field, para);

generally two calls with the same parameters should return the same values. Because you alter the para_field redirectly, this isn't the case and can be confusing.

I realise that in this initial version, mergeFlatKeysIntoNestedDict is only called once and this isn't an issue but tomorrow will bring more features and changes.

This can be shortcutted to var para_field = flow.get(name) || {} - undefined and false will cause the right handside of the || (i.e. or operator) to be evaluated.

1 Like
9

Sorry just back from Eastern break..

It is processed as expected. It is difficult to simulate the arriving data stream w.r.t timing. Assuming this is a "hiccup" related to this MQTT data.

Its about 6 flow variables sorted by the MQTT topics.

I will follow the hints from @gregorius to polish my javascript function. What I get so far, there is no major flaw in the code itself?

Thanks to all for the support!

10

You haven't answered any of those questions, or have I missed a post?

11

You are right, sorry!

Looking at the MQTT timestamps all messages (about 20) arrive in about 4 seconds. Thus the data rate should be not an issue. (MQTT Broker and NR run on a thin server Siemens-Fujitsu 720)

Of course there are other data streams, mostly from my Zigbee network and others. But NR only stalls when this particular car data stream is received.
As @TotallyInformation already pointed out the "call stack size error" must be further analysed. I will try to debug my function node.