[Bug]: QoS > 0 messages not saved in outgoing storage

[tanoc]

MQTTjs Version

5.13.2

Broker

mosquitto

Environment

NodeJS

Description

If I try to publish messages 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 occur during this phase, all the messages are discarded here with _flushStoreProcessingQueue.

I think the expected behaviour should be having them saved in the outgoing storage, like it happen when I publish messages during a disconnection.

I'm using this library in a react-native project (but the problem should be also present in other environments).

Minimal Reproduction

Unfortunately I can only reproduce this manually. What I'm doing to reproduce this is to have a good number of messages in the outgoing store and during the processing of the outgoing store stream I block the connection from the app to the broker with a firewall rule. Not sure 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

[robertsLando]

@tanoc PR?

[tanoc]

to be honest I'm not really sure how to fix it.
One option would be to store the packets with QoS > 0 directly in the publish method, but I feel there is a reason why it isn't done there.
Another would be to store them during the _flushStoreProcessingQueue instead of calling ...cbStorePut(new Error(..) (this is how I'm fixing it on the app side, waiting for that callback, checking the error and then inserting the message manually in the store). The issue in this case is that _flushStoreProcessingQueue is called when the close event is triggered, and that could happen some time later. In case of crash the messages in the queue will be lost.
Moreover I'm not sure what are the limitation for the storage interface regarding inserting data while there is a stream reading in progress.

[tanoc]

Another thing, I think I notice that because the republish of messages from the store is probably slower than it needs to be, since the messages are sent sequentially in storeDeliver, waiting for a PUBACK before sending the next (see #1086).

[tanoc]

Regarding #1086 I tried to fix here. I'm not sure if it's the correct approach but if it's interesting I could open a PR.

[robertsLando]

Another thing, I think I notice that because the republish of messages from the store is probably slower than it needs to be, since the messages are sent sequentially in storeDeliver, waiting for a PUBACK before sending the next (see #1086).

This is to respect specs, messages should be sent in the same order they have been stored so we have to send them sequentially

[robertsLando]

I think a possible solution is to store packet in outgoing store if cmd is publish and QoS is > 0 in

MQTT.js/src/lib/client.ts

Line 2403 in 1dd6c03

private _flushStoreProcessingQueue() {

[tanoc]

Another thing, I think I notice that because the republish of messages from the store is probably slower than it needs to be, since the messages are sent sequentially in storeDeliver, waiting for a PUBACK before sending the next (see #1086).

This is to respect specs, messages should be sent in the same order they have been stored so we have to send them sequentially

in theory they are sent in the same order. it just doesn't wait for the acknowledge before sending the next one.

in the specs it says:

If both Client and Server set Receive Maximum to 1, they make sure that no more than one message is “in-flight” at any one time. In this case no QoS 1 message will be received after any later one even on re-connection. For example a subscriber might receive them in the order 1,2,3,3,4 but not 1,2,3,2,3,4. Refer to section 4.9 Flow Control for details of how the Receive Maximum is used.

for the way I see it the "if" there means that sending more than one at the time is also an option, no? (in the v3 it says more or less the same thing without the receive maximum since it's added in v5)

same thing for the mosquitto config documentation:

max_inflight_messages count
The maximum number of outgoing QoS 1 or 2 messages that can be in the process of being transmitted simultaneously. This includes messages currently going through handshakes and messages that are being retried. Defaults to 20. Set to 0 for no maximum. If set to 1, this will guarantee in-order delivery of messages.

even in my patch you can always set the maxInflightMessages to 1. it should work like the current implementation.

[tanoc]

I think a possible solution is to store packet in outgoing store if cmd is publish and QoS is > 0 in

MQTT.js/src/lib/client.ts

Line 2403 in 1dd6c03

private _flushStoreProcessingQueue() {

that would require to save the packet in _storeProcessingQueue (or something like it since the packet is created inside publishProc) . Would that be a problem?

[robertsLando]

in theory they are sent in the same order. it just doesn't wait for the acknowledge before sending the next one.

Yeah but that means that for some reason a qos 2 packet could be received after a qos 1 considering there is more messages exchanged and this might not be the desired behaviour

I wonder what other libraries like paho does in this case

[tanoc]

in theory they are sent in the same order. it just doesn't wait for the acknowledge before sending the next one.

Yeah but that means that for some reason a qos 2 packet could be received after a qos 1 considering there is more messages exchanged and this might not be the desired behaviour

tbh I didn't think about that, but I'm not sure if that's part of the standard:

An Ordered Topic is a Topic where the Client can be certain that the Application Messages in that Topic from the same Client and at the same QoS are received are in the order they were published.

it looks to me like the order has to be guaranteed for messages with the same QoS, no?

I wonder what other libraries like paho does in this case

I gave a quick look to the js implementation and to me it looks like they are just saving stuff in localstorage and resending them according to a sequence number without waiting for acks between messages. See here.