node-red - Units of work

Question

I have a MQTT Input node that kicks of database operations (SELECT and INSERT) for PostgreSQL database. The database operations are done with node-contrib-postgres-multi.Since these operations are separated by function nodes, I am saving portion of the message with flow.set and retrieving with flow.get later. For example,

• Function node saves with flow.set and generates SELECT query.
• Wire to PostgreSQL node.
• Parse output, use flow.get and generate INSERT statement.
• Wire to PostgreSQL node.

I can't help but imagine that the flow.set and flow.get are not in sync.

Currently, I am simulating around 20 devices to publish data every second with timestamp increased by 1 second for every publication. There is absolutely no reason why the generated messages should ever get duplicated. However, the database insert nodes fail because of unique index violation as seen in Node-RED log (.pm2/logs/red-out-0.log).

If the function nodes and database processing take say, 2 seconds and the MQTT messages (QoS=0) are received every second, would MQTT or Node-RED buffer them? Therefore, every received message is treated as a Unit of Work until it errors out or 'leaves' the flow into a database, HTTP request, MQTT publish, etc. ?

Answer 1

It's worth remembering that everything in the NodeJS world is asynchronous which means nothing blocks the event loop.

In this case it sounds like you get the first incoming message, which you store with a fixed key in the flow context, you then move on to the Select query. At this point the SQL node is going to end up doing some network IO which will block while it waits for a response from the database so while it waits it will give up the execution context.

If while it's waiting a new MQTT messages arrives it will be handled immediately and passed to the first function node which is going to overwrite what ever was stored in the flows context because the key is the same.

When the first Select statement returns the first message will move on to the second function node and when it retrieves the value from the flow context it will be the second message not the first.

The way to solve this is to not use the context to keep state but to move the information you want to keep from msg.payload to a different key on the msg object. Well behaved Node-RED nodes should always pass on the original msg object and by default only really change the msg.payload (there are exceptions, but they tend to document what they change and why).

Using the msg to hold state for a unit of work guarantees that it can only be changed in step with the message travelling through a flow.