Fix resubscribe message id allocate twice #1337

redboltz · 2021-10-06T23:35:34Z

No description provided.

redboltz · 2021-10-07T02:58:29Z

Please don't merge this PR.
I'm checking the deliver logic.

resubscribe is out of MQTT spec. It is MQTT.js expansion. On connect sequence, the following three steps are defined by the MQTT Spec. 1. The client sends CONNECT to the broker with CleanStart:false 2. The broker sends CONNACK to the client with SessionPresent:true if session exists 3. The client re-sends in-flight PUBLISH and PUBREL messages resubscribe was processed between the step 2 and step 3. It's too early. The resubscribe might allocate messageId that is the same as PUBLISH or PUBREL packet. It is not good. So I moved resubscribe process to after the step 3.

redboltz · 2021-10-07T03:37:54Z

I updated the PR.
The root issue is that resubscribe process does too early.
So I moved the process to after stored PUBLISH/PUBREL sending finished.

I think that now the PR is ready to merge.

YoDaMa · 2021-10-07T15:09:17Z

lib/client.js

@@ -335,6 +338,7 @@ function MqttClient (streamBuilder, options) {
      var packet = null

      if (!entry) {
+        that._resubscribe()


@redboltz I'm curious what the reason for validating there is no entry. Is that so if there are no queued packets left, then it will resubscribe?

Yes, exactly.

YoDaMa · 2021-10-07T15:54:00Z

lib/client.js

@@ -343,7 +347,7 @@ function MqttClient (streamBuilder, options) {
      var send = true
      if (packet.messageId && packet.messageId !== 0) {
        if (!that.messageIdProvider.register(packet.messageId)) {
-          packet.messageeId = that.messageIdProvider.allocate()
+          packet.messageId = that.messageIdProvider.allocate()


so why did this get through the tests? it's cause it was a double typo in here and in the test?

No. There was no double typo. It is a fallback code for registration failure.

The failed test case was a combination of UniqueMessageIdProvider and resubscribe.
The error was when suback was received, no matched messageId was found in outgoing.

Before fixed the typo, packet has already been allocated messageId on resubscribe().
The messageId was 1. UniqueMessageIdProvider marked 1 as used.
The packet was pushed into queue because connected is false.
And in subscribeProc that.outgoing[packet.messageId] was added. The message id was 1.

See

Call resubscribe

MQTT.js/lib/client.js

Lines 1709 to 1721 in c92b877

MqttClient.prototype._onConnect = function (packet) {

if (this.disconnected) {

this.emit('connect', packet)

return

}

var that = this

this.messageIdProvider.clear()

this._setupPingTimer()

this._resubscribe(packet)

this.connected = true

resubscribe

MQTT.js/lib/client.js

Line 1677 in c92b877

MqttClient.prototype._resubscribe = function (connack) {

MQTT.js/lib/client.js

Line 1690 in c92b877

this.subscribe(resubscribeTopic, {properties: resubscribeTopic[_resubscribeTopicsKeys[topicI]].properties})

subscribe

MQTT.js/lib/client.js

Line 693 in c92b877

MqttClient.prototype.subscribe = function () {

MQTT.js/lib/client.js

Line 853 in c92b877

subscribeProc()

subscribeProc

MQTT.js/lib/client.js

Line 787 in c92b877

var subscribeProc = function () {

MQTT.js/lib/client.js

Line 827 in c92b877

that.outgoing[packet.messageId] = {

MQTT.js/lib/client.js

Line 841 in c92b877

that._sendPacket(packet)

_sendPacket

MQTT.js/lib/client.js

Line 1203 in c92b877

MqttClient.prototype._sendPacket = function (packet, cb, cbStorePut) {

MQTT.js/lib/client.js

Lines 1214 to 1218 in c92b877

if (!this.connected) {

debug('_sendPacket :: client not connected. Storing packet offline.')

this._storePacket(packet, cb, cbStorePut)

return

}

_storePacket

MQTT.js/lib/client.js

Line 1260 in c92b877

MqttClient.prototype._storePacket = function (packet, cb, cbStorePut) {

MQTT.js/lib/client.js

Line 1278 in c92b877

this.queue.push({ packet: storePacket, cb: cb })

_handleAck

MQTT.js/lib/client.js

Lines 1535 to 1548 in c92b877

MqttClient.prototype._handleAck = function (packet) {

/* eslint no-fallthrough: "off" */

var messageId = packet.messageId

var type = packet.cmd

var response = null

var cb = this.outgoing[messageId] ? this.outgoing[messageId].cb : null

var that = this

var err

if (!cb) {

debug('_handleAck :: Server sent an ack in error. Ignoring.')

// Server sent an ack in error, ignore it.

return

}

At the following point, try to register the messageId 1 but failed because UniqueMessageIdProvider had marked 1 as userd.

MQTT.js/lib/client.js

Lines 344 to 346 in c92b877

if (packet.messageId && packet.messageId !== 0) {

if (!that.messageIdProvider.register(packet.messageId)) {

packet.messageeId = that.messageIdProvider.allocate()

Then tried to allocate the new messageId. It is a fallback code. I just noticed that the fallback code should be removed because there is no registration failed case. In addition allocating the new messageId doesn't help the situation if it happens by bug. I will update the PR.

Anyway the new allocated messageId was 2 but it was written as messageeId.
So the packet filed was as follows:

{ messageId:1 messageeId:2 }

The packet was sent with messageId 1 and suback is received messageId 1. That matched outgoing. So the test was passed.

After the typo fixed, the packet filed became as follows:

{ messageId:2 }

messageId was overwritten. But outgoing still has messageId 1. So it didn't mached.

YoDaMa · 2021-10-07T15:54:40Z

lib/client.js

  debug('_resubscribe')
  var _resubscribeTopicsKeys = Object.keys(this._resubscribeTopics)
  if (!this._firstConnection &&
-      (this.options.clean || (this.options.protocolVersion === 5 && !connack.sessionPresent)) &&
+      (this.options.clean || (this.options.protocolVersion === 5 && !this.sessionPresent)) &&


so resubscribe no longer uses CONNACK. this makes sense...

YoDaMa · 2021-10-07T15:55:39Z

lib/client.js

@@ -1714,9 +1718,9 @@ MqttClient.prototype._onConnect = function (packet) {

  var that = this

+  this.sessionPresent = packet.sessionPresent


but why do this rather than just pass the packet with the .sessionPresent to the reconnect? This introduces the possibility of less well defined behavior if the this.sessionPresent field is manipulated elsewhere.

Do you mean use this.connackPacket = packet here?
And at #1337 (review)

(this.options.clean || (this.options.protocolVersion === 5 && !this.connackPacket.sessionPresent)) &&

It still has possibility that someone uses undefined this.connackPacket. However I think the possibility is lower than this.sessionPresent.

fix: messageeId

11f6321

redboltz mentioned this pull request Oct 6, 2021

fix: messageeId #1336

Closed

redboltz force-pushed the fix_resubscribe_message_id_allocate_twice branch from 1a9518f to 5212022 Compare October 7, 2021 03:32

YoDaMa reviewed Oct 7, 2021

View reviewed changes

redboltz added 2 commits October 8, 2021 08:24

Removed invalid fallback code.

7b3b7fc

Stored CONNACK packet instead of sessionPresent.

6493c4f

YoDaMa merged commit 7466819 into mqttjs:master Oct 11, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix resubscribe message id allocate twice #1337

Fix resubscribe message id allocate twice #1337

redboltz commented Oct 6, 2021

redboltz commented Oct 7, 2021

redboltz commented Oct 7, 2021

YoDaMa Oct 7, 2021

redboltz Oct 7, 2021

YoDaMa Oct 7, 2021

redboltz Oct 7, 2021

YoDaMa Oct 7, 2021

YoDaMa Oct 7, 2021

redboltz Oct 7, 2021 •

edited

	MqttClient.prototype._onConnect = function (packet) {
	if (this.disconnected) {
	this.emit('connect', packet)
	return
	}

	var that = this

	this.messageIdProvider.clear()
	this._setupPingTimer()
	this._resubscribe(packet)

	this.connected = true

	if (!this.connected) {
	debug('_sendPacket :: client not connected. Storing packet offline.')
	this._storePacket(packet, cb, cbStorePut)
	return
	}

	MqttClient.prototype._handleAck = function (packet) {
	/* eslint no-fallthrough: "off" */
	var messageId = packet.messageId
	var type = packet.cmd
	var response = null
	var cb = this.outgoing[messageId] ? this.outgoing[messageId].cb : null
	var that = this
	var err

	if (!cb) {
	debug('_handleAck :: Server sent an ack in error. Ignoring.')
	// Server sent an ack in error, ignore it.
	return
	}

	if (packet.messageId && packet.messageId !== 0) {
	if (!that.messageIdProvider.register(packet.messageId)) {
	packet.messageeId = that.messageIdProvider.allocate()

		@@ -1714,9 +1718,9 @@ MqttClient.prototype._onConnect = function (packet) {

		var that = this

		this.sessionPresent = packet.sessionPresent

Fix resubscribe message id allocate twice #1337

Fix resubscribe message id allocate twice #1337

Conversation

redboltz commented Oct 6, 2021

redboltz commented Oct 7, 2021

redboltz commented Oct 7, 2021

YoDaMa Oct 7, 2021

Choose a reason for hiding this comment

redboltz Oct 7, 2021

Choose a reason for hiding this comment

YoDaMa Oct 7, 2021

Choose a reason for hiding this comment

redboltz Oct 7, 2021

Choose a reason for hiding this comment

Call resubscribe

resubscribe

subscribe

subscribeProc

_sendPacket

_storePacket

_handleAck

YoDaMa Oct 7, 2021

Choose a reason for hiding this comment

YoDaMa Oct 7, 2021

Choose a reason for hiding this comment

redboltz Oct 7, 2021 • edited

Choose a reason for hiding this comment

redboltz Oct 7, 2021 •

edited