MQTT.js Bug QoS Greater Than Zero Messages Not Saved During Reconnection

Jul 29, 2025 by ADMIN 73 views

Bug QoS Greater Than Zero Messages Not Saved in Outgoing Storage MQTT.js Discussion

Hey guys, let's dive into a quirky issue reported in the MQTT.js library that can cause headaches when dealing with Quality of Service (QoS) levels greater than zero. This article will break down the problem, explain the context, and explore potential solutions. If you're working with MQTT and message persistence, this is definitely something you'll want to understand.

Understanding the Issue: QoS Messages and Outgoing Storage

At the heart of the matter is a scenario where MQTT messages with QoS greater than zero aren't being reliably saved in the outgoing storage during reconnections. To put it simply, QoS levels 1 and 2 ensure that messages are delivered at least once (QoS 1) or exactly once (QoS 2). This is crucial for applications where data loss is unacceptable. The outgoing storage acts as a buffer, holding messages until they're successfully sent to the broker. The problem arises when messages published during a reconnection phase, specifically while the outgoing store is being processed, get lost in the shuffle.

This problem was observed in MQTT.js version 5.13.2, using the Mosquitto broker in a NodeJS environment. The core issue lies in how the library handles message queuing during reconnections. When a disconnection occurs, messages that are in the process of being sent are queued in _storeProcessingQueue. However, if another disconnection happens while this queue is being processed, these messages are inadvertently discarded by the _flushStoreProcessingQueue function. This behavior contradicts the expected outcome, which is for these messages to be saved in the outgoing storage for later delivery, just like messages published during a disconnection.

For developers using MQTT in applications that require guaranteed message delivery, this bug can lead to significant data loss. Imagine a sensor network where readings must be reliably transmitted; losing messages due to this issue could have serious consequences. The fact that this was observed in a React Native project suggests that the problem isn't limited to a specific environment and might affect other NodeJS-based MQTT applications as well.

Diving Deeper: How the Bug Manifests

To truly grasp the issue, let's break down the sequence of events. When a client reconnects, it starts processing messages from its outgoing store. This process, managed by startStreamProcess, ensures that messages that weren't delivered in the previous session are resent. However, during this process, new messages might be published. These new messages, if they have QoS > 0, are queued in _storeProcessingQueue. The problem occurs when a disconnection happens while this queue is still active. The _flushStoreProcessingQueue function, designed to clear the queue, ends up discarding these messages instead of saving them in the outgoing store.

The critical point is the timing. If a disconnection occurs while messages are sitting in the _storeProcessingQueue, they're lost. This is not the intended behavior, as the expectation is that all messages with QoS > 0 should be persisted until they're successfully delivered. The debug logs provided in the original report clearly illustrate this. Messages are published, but then, due to the disconnection, they're discarded, leading to errors like "Connection closed." This sequence of events highlights a race condition where the disconnection interrupts the proper handling of queued messages.

The debug logs show a clear sequence of events leading to the issue. First, a series of messages are published, indicated by the "publish :: message" logs. Then, a keep-alive timeout occurs, triggering a cleanup process. This leads to the disconnection and the call to _flushStoreProcessingQueue. The logs then show a series of "store cb put" and "publish cb" errors, indicating that the messages were not properly stored or delivered. Finally, new messages are published, but because the client is disconnected, they are correctly stored offline. The problem lies in the messages that were in the processing queue during the disconnection—they vanish without a trace.

Replicating the Issue: A Manual Approach

Reproducing this bug isn't straightforward, as it requires precise timing and a controlled environment. The reporter managed to reproduce it manually by simulating a disconnection during the outgoing store processing. The steps involved having a substantial number of messages in the outgoing store and then intentionally blocking the connection between the app and the broker using a firewall rule. This forces a disconnection while the messages are being processed, triggering the bug.

The challenge lies in automating this process or creating a unit test. The manual reproduction method highlights the difficulty in reliably triggering the race condition. It requires a specific sequence of events: a backlog of messages, active processing of the outgoing store, and an abrupt disconnection. Creating a test that accurately mimics these conditions is a complex task. It would likely involve simulating network interruptions and carefully timing message publications to coincide with the processing of the outgoing queue. This complexity explains why a straightforward automated test hasn't been developed yet.

Despite the difficulty in automating the reproduction, the manual method provides valuable insights into the bug's behavior. It confirms that the issue is timing-sensitive and occurs under specific circumstances. This understanding is crucial for developing potential fixes and designing future tests. It also underscores the importance of robust error handling and message persistence mechanisms in MQTT clients, especially in environments where network connectivity is unreliable.

Potential Solutions and Workarounds

Addressing this issue requires a careful examination of the message queuing and persistence logic within MQTT.js. One potential solution is to ensure that messages in the _storeProcessingQueue are properly saved to the outgoing store before _flushStoreProcessingQueue is called during a disconnection. This would prevent messages from being discarded and ensure they are retried upon reconnection.

Another approach could involve refining the timing and synchronization of the queue processing and disconnection handling. Ensuring that the outgoing store is fully persisted before initiating the disconnection sequence could mitigate the race condition. This might involve adding additional checks or synchronization mechanisms to prevent messages from being lost during the transition.

In the meantime, there are potential workarounds that developers can implement in their applications. One strategy is to implement an application-level acknowledgment mechanism. This involves tracking messages and their delivery status at the application level, providing an additional layer of assurance. If a message isn't acknowledged within a certain timeframe, it can be republished, mitigating the risk of data loss.

Another workaround is to reduce the likelihood of disconnections during the outgoing store processing. This could involve implementing more robust network connectivity checks and handling disconnections more gracefully. By minimizing the number of interruptions during the critical processing phase, the chances of encountering the bug can be reduced.

Key Takeaways and Future Directions

This bug in MQTT.js highlights the complexities of ensuring reliable message delivery in MQTT applications, especially when dealing with reconnections and varying QoS levels. The issue underscores the importance of robust error handling, message persistence, and careful management of message queues. While the bug is tricky to reproduce and fix, understanding its root cause is the first step towards a solution.

Moving forward, the MQTT.js project could benefit from more comprehensive testing around reconnection scenarios and message persistence. Developing automated tests that simulate network interruptions and message queuing complexities would help identify and prevent similar issues in the future. Additionally, enhancing the library's internal mechanisms for handling disconnections and message queues could improve its overall reliability.

For developers using MQTT.js, this bug serves as a reminder to carefully consider the implications of network interruptions and message persistence in their applications. Implementing workarounds, such as application-level acknowledgments, can provide an extra layer of protection against data loss. Staying informed about potential issues and actively participating in the MQTT.js community can also help ensure the robustness of your MQTT solutions.

In conclusion, the QoS message bug in MQTT.js is a subtle but significant issue that can lead to data loss in certain scenarios. By understanding the bug's root cause and implementing appropriate workarounds, developers can mitigate its impact and build more reliable MQTT applications. Continued efforts in testing and improving the library's internal mechanisms will further enhance its robustness and ensure the delivery of critical messages.

MQTT.js Version

5.13.2

Broker

Mosquitto

Environment

NodeJS

Bug Description

If messages are published with QoS > 0 during a reconnection, while the messages in the outgoing store are being sent (with startStreamProcess), the processing of the messages is queued in _storeProcessingQueue.

If a disconnection occurs during this phase, all the messages are discarded here with _flushStoreProcessingQueue.

The expected behavior should be having them saved in the outgoing storage, like it happens when messages are published during a disconnection.

This library is being used in a React-Native project, but the problem should also be present in other environments.

Reproduction

Unfortunately, this can only be reproduced manually. To reproduce this, have a good number of messages in the outgoing store, and during the processing of the outgoing store stream, block the connection from the app to the broker with a firewall rule. It's unclear how to automate this or write a test for it.

Debug logs

...
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  publish :: message `...` to topic `...`
LOG  onKeepaliveTimeout :: calling _cleanUp with force true
LOG  _cleanUp :: forced? true
LOG  _cleanUp :: (927d2004-8549-4345-b861-d0cb0d44dde0) :: destroying stream
LOG  _cleanUp :: client not disconnecting/reconnecting. Clearing and resetting reconnect.
LOG  _clearReconnect : clearing reconnect timer
LOG  _setupReconnect :: emit `offline` state
LOG  _setupReconnect :: set `reconnecting` to `true`
LOG  _setupReconnect :: setting reconnectTimer for 15000 ms
LOG  _destroyKeepaliveManager :: destroying keepalive manager
LOG  (927d2004-8549-4345-b861-d0cb0d44dde0)stream :: on close
LOG  _flushVolatile :: deleting volatile messages from the queue and setting their callbacks as error function
LOG  stream: emit close to MqttClient
LOG  close :: connected set to `false`
LOG  close :: clearing connackTimer
LOG  close :: calling _setupReconnect
LOG  _setupReconnect :: doing nothing...

# Here the remove and _flushStoreProcessingQueue are being called

LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
LOG  store cb put [Error: Connection closed]
LOG  publish cb [Error: Connection closed undefined]
...
LOG  publish :: message `...` to topic `...`
LOG  publish :: qos 1
LOG  MqttClient:publish: packet cmd: publish
LOG  _sendPacket :: (927d2004-8549-4345-b861-d0cb0d44dde0) ::  start
LOG  applyTopicAlias :: auto assign(use) topic: ... - alias: 1
LOG  _sendPacket :: client not connected. Storing packet offline.
LOG  _storePacket :: packet: '...'
LOG  _storePacket :: cb? true