Calling reconnect() causes endless connect/disconnect loop

[ferm10n]

I tested this with mqtt@4.2.1 and node 12.16.3.

Something as simple as this:

const mqtt = require('mqtt');

const client = mqtt.connect('mqtt://localhost');
client.on('connect', () => console.log('connected', new Date()));
client.on('close', () => console.log('disconnected', new Date()));
client.on('error', err => console.error('error', err));

setTimeout(() => client.reconnect(), 5000);

Gives this output:

connected 2020-08-28T22:27:42.598Z
disconnected 2020-08-28T22:27:47.606Z
connected 2020-08-28T22:27:47.606Z
disconnected 2020-08-28T22:27:48.611Z
connected 2020-08-28T22:27:48.614Z
disconnected 2020-08-28T22:27:49.616Z
connected 2020-08-28T22:27:49.619Z
disconnected 2020-08-28T22:27:50.621Z
connected 2020-08-28T22:27:50.624Z
disconnected 2020-08-28T22:27:51.625Z
connected 2020-08-28T22:27:51.629Z

I'm going to poke around a little and see if I can't figure out why, but in the meantime I saw that:

• the inner stream emits a close event because the Socket was closed
• Socket gets closed because a null is written to the stream, which looks like that signals the end of the stream. The null is written because some node internals gets an arrayBuffer of undefined inside onStreamRead... but idk how to debug that any deeper because of async stack traces :/

[ferm10n]

I noticed that if I call .end() before .reconnect(), then this issue doesn't happen. That's okay as a workaround I guess. Maybe the resolution to this could be to add a call to end() before _reconnect()

[Odame]

For reference, something like this worked for me.

const reconnect = () => {
	return new Promise((resolve, reject) => {
		mqttClient.end(true, {}, () => {
			mqttClient.reconnect();
			resolve();
		});
	});
}

[YoDaMa]

so the issue is that it gets trapped in a reconnect loop? But by calling end and then reconnect, there's no loop, it just reconnects?

[ferm10n]

@YoDaMa Yes that's what I saw. You should be able to reproduce it with the snippet at the very top.

[YoDaMa]

Posting logs with debugs enabled from the repro:

  mqttjs connecting to an MQTT broker... +0ms
  mqttjs:client MqttClient :: options.protocol mqtt +0ms
  mqttjs:client MqttClient :: options.protocolVersion 4 +2ms
  mqttjs:client MqttClient :: options.username undefined +1ms
  mqttjs:client MqttClient :: options.keepalive 60 +0ms
  mqttjs:client MqttClient :: options.reconnectPeriod 1000 +1ms
  mqttjs:client MqttClient :: options.rejectUnauthorized undefined +1ms
  mqttjs:client MqttClient :: clientId mqttjs_f6caf460 +1ms
  mqttjs:client MqttClient :: setting up stream +1ms
  mqttjs:client _setupStream :: calling method to clear reconnect +2ms
  mqttjs:client _clearReconnect : clearing reconnect timer +0ms
  mqttjs:client _setupStream :: using streamBuilder provided to client to create stream +1ms
  mqttjs calling streambuilder for mqtt +18ms
  mqttjs:tcp port 1883 and host localhost +0ms
  mqttjs:client _setupStream :: pipe stream to writable stream +19ms
  mqttjs:client _setupStream: sending packet `connect` +2ms
  mqttjs:client sendPacket :: packet: { cmd: 'connect' } +1ms
  mqttjs:client sendPacket :: emitting `packetsend` +5ms
  mqttjs:client sendPacket :: writing to stream +1ms
  mqttjs:client sendPacket :: writeToStream result true +29ms
connecting client
  mqttjs:client writable stream :: parsing buffer +11ms
  mqttjs:client parser :: on packet push to packets array. +4ms
  mqttjs:client work :: getting next packet in queue +1ms
  mqttjs:client work :: packet pulled from queue +2ms
  mqttjs:client _handlePacket :: emitting packetreceive +1ms
  mqttjs:client _handleConnack +1ms
  mqttjs:client _setupPingTimer :: keepalive 60 (seconds) +2ms
  mqttjs:client _resubscribe +2ms
  mqttjs:client connect :: sending queued packets +3ms
  mqttjs:client deliver :: entry undefined +1ms
connected 2020-09-21T22:14:29.468Z
  mqttjs:client client reconnect +5s
  mqttjs:client _reconnect: emitting reconnect to client +3ms
  mqttjs:client _reconnect: calling _setupStream +1ms
  mqttjs:client _setupStream :: calling method to clear reconnect +1ms
  mqttjs:client _clearReconnect : clearing reconnect timer +1ms
  mqttjs:client _setupStream :: using streamBuilder provided to client to create stream +1ms
  mqttjs calling streambuilder for mqtt +5s
  mqttjs:tcp port 1883 and host localhost +5s
  mqttjs:client _setupStream :: pipe stream to writable stream +4ms
  mqttjs:client _setupStream: sending packet `connect` +1ms
  mqttjs:client sendPacket :: packet: { cmd: 'connect' } +1ms
  mqttjs:client sendPacket :: emitting `packetsend` +1ms
  mqttjs:client sendPacket :: writing to stream +1ms
  mqttjs:client sendPacket :: writeToStream result true +2ms
  mqttjs:client (mqttjs_f6caf460)stream :: on close +4ms
  mqttjs:client flushVolatile :: deleting volatile messages from the queue and setting their callbacks as error function +1ms
  mqttjs:client stream: emit close to MqttClient +1ms
  mqttjs:client close :: connected set to `false` +1ms
  mqttjs:client close :: clearing connackTimer +1ms
  mqttjs:client close :: clearing ping timer +1ms
  mqttjs:client close :: calling _setupReconnect +1ms
  mqttjs:client _setupReconnect :: emit `offline` state +1ms
  mqttjs:client _setupReconnect :: set `reconnecting` to `true` +1ms
  mqttjs:client _setupReconnect :: setting reconnectTimer for 1000 ms +1ms
disconnected 2020-09-21T22:14:34.488Z
  mqttjs:client writable stream :: parsing buffer +3ms
  mqttjs:client parser :: on packet push to packets array. +2ms
  mqttjs:client work :: getting next packet in queue +2ms
  mqttjs:client work :: packet pulled from queue +1ms
  mqttjs:client _handlePacket :: emitting packetreceive +1ms
  mqttjs:client _handleConnack +1ms
  mqttjs:client _setupPingTimer :: keepalive 60 (seconds) +0ms
  mqttjs:client _resubscribe +1ms
  mqttjs:client connect :: sending queued packets +2ms
  mqttjs:client deliver :: entry undefined +1ms
connected 2020-09-21T22:14:34.503Z
  mqttjs:client reconnectTimer :: reconnect triggered! +1s
  mqttjs:client _reconnect: emitting reconnect to client +3ms
  mqttjs:client _reconnect: calling _setupStream +2ms
  mqttjs:client _setupStream :: calling method to clear reconnect +1ms
  mqttjs:client _clearReconnect : clearing reconnect timer +1ms
  mqttjs:client _setupStream :: using streamBuilder provided to client to create stream +1ms
  mqttjs calling streambuilder for mqtt +1s
  mqttjs:tcp port 1883 and host localhost +1s
  mqttjs:client _setupStream :: pipe stream to writable stream +3ms
  mqttjs:client _setupStream: sending packet `connect` +1ms
  mqttjs:client sendPacket :: packet: { cmd: 'connect' } +1ms
  mqttjs:client sendPacket :: emitting `packetsend` +1ms
  mqttjs:client sendPacket :: writing to stream +1ms
  mqttjs:client sendPacket :: writeToStream result true +2ms
  mqttjs:client (mqttjs_f6caf460)stream :: on close +3ms
  mqttjs:client flushVolatile :: deleting volatile messages from the queue and setting their callbacks as error function +2ms
  mqttjs:client stream: emit close to MqttClient +1ms
  mqttjs:client close :: connected set to `false` +1ms
  mqttjs:client close :: clearing connackTimer +1ms
  mqttjs:client close :: clearing ping timer +0ms
  mqttjs:client close :: calling _setupReconnect +1ms
  mqttjs:client _setupReconnect :: emit `offline` state +1ms
  mqttjs:client _setupReconnect :: set `reconnecting` to `true` +1ms
  mqttjs:client _setupReconnect :: setting reconnectTimer for 1000 ms +1ms
disconnected 2020-09-21T22:14:35.536Z

Looks like the problem is that we do not check if the client is connected or not when reconnecting, and then call _setupStream internally, which will reset the stream and emit a close to the client, triggering the infinite reconnect loop. Adding a check if things are connected and calling end if they are seems to fix the problem:

  mqttjs connecting to an MQTT broker... +0ms
  mqttjs:client MqttClient :: options.protocol mqtt +0ms
  mqttjs:client MqttClient :: options.protocolVersion 4 +2ms
  mqttjs:client MqttClient :: options.username undefined +1ms
  mqttjs:client MqttClient :: options.keepalive 60 +0ms
  mqttjs:client MqttClient :: options.reconnectPeriod 1000 +1ms
  mqttjs:client MqttClient :: options.rejectUnauthorized undefined +1ms
  mqttjs:client MqttClient :: clientId mqttjs_a09b79d3 +1ms
  mqttjs:client MqttClient :: setting up stream +1ms
  mqttjs:client _setupStream :: calling method to clear reconnect +2ms
  mqttjs:client _clearReconnect : clearing reconnect timer +1ms
  mqttjs:client _setupStream :: using streamBuilder provided to client to create stream +0ms
  mqttjs calling streambuilder for mqtt +20ms
  mqttjs:tcp port 1883 and host localhost +0ms
  mqttjs:client _setupStream :: pipe stream to writable stream +21ms
  mqttjs:client _setupStream: sending packet `connect` +2ms
  mqttjs:client sendPacket :: packet: { cmd: 'connect' } +2ms
  mqttjs:client sendPacket :: emitting `packetsend` +5ms
  mqttjs:client sendPacket :: writing to stream +1ms
  mqttjs:client sendPacket :: writeToStream result true +36ms
connecting client
  mqttjs:client writable stream :: parsing buffer +9ms
  mqttjs:client parser :: on packet push to packets array. +2ms
  mqttjs:client work :: getting next packet in queue +1ms
  mqttjs:client work :: packet pulled from queue +1ms
  mqttjs:client _handlePacket :: emitting packetreceive +1ms
  mqttjs:client _handleConnack +3ms
  mqttjs:client _setupPingTimer :: keepalive 60 (seconds) +1ms
  mqttjs:client _resubscribe +2ms
  mqttjs:client connect :: sending queued packets +3ms
  mqttjs:client deliver :: entry undefined +1ms
connected 2020-09-21T22:27:06.549Z
  mqttjs:client client reconnect +5s
  mqttjs:client _reconnect: emitting reconnect to client +9ms
  mqttjs:client end :: (mqttjs_a09b79d3) +6ms
  mqttjs:client end :: cb? true +6ms
  mqttjs:client _clearReconnect : clearing reconnect timer +4ms
  mqttjs:client end :: (mqttjs_a09b79d3) :: immediately calling finish +3ms
  mqttjs:client end :: (mqttjs_a09b79d3) :: finish :: calling _cleanUp with force false +3ms
  mqttjs:client _cleanUp :: done callback provided for on stream close +3ms
  mqttjs:client _cleanUp :: forced? false +2ms
  mqttjs:client _cleanUp :: (mqttjs_a09b79d3) :: call _sendPacket with disconnect packet +1ms
  mqttjs:client _sendPacket :: (mqttjs_a09b79d3) ::  start +1ms
  mqttjs:client sendPacket :: packet: { cmd: 'disconnect' } +1ms
  mqttjs:client sendPacket :: emitting `packetsend` +1ms
  mqttjs:client sendPacket :: writing to stream +1ms
  mqttjs:client sendPacket :: writeToStream result true +1ms
  mqttjs:client sendPacket :: invoking cb +1ms
  mqttjs:client _cleanUp :: clearing pingTimer +1ms
  mqttjs:client client already connected. disconnecting first. +1ms
  mqttjs:client (mqttjs_a09b79d3)stream :: on close +3ms
  mqttjs:client flushVolatile :: deleting volatile messages from the queue and setting their callbacks as error function +2ms
  mqttjs:client stream: emit close to MqttClient +1ms
  mqttjs:client close :: connected set to `false` +1ms
  mqttjs:client close :: clearing connackTimer +1ms
  mqttjs:client close :: clearing ping timer +1ms
  mqttjs:client close :: calling _setupReconnect +1ms
  mqttjs:client _setupReconnect :: doing nothing... +1ms
disconnected 2020-09-21T22:27:11.591Z
  mqttjs:client end :: finish :: calling process.nextTick on closeStores +2ms
  mqttjs:client end :: closeStores: closing incoming and outgoing stores +1ms
  mqttjs:client end :: closeStores: emitting end +1ms
  mqttjs:client end :: closeStores: invoking callback with args +1ms
  mqttjs:client _setupStream :: calling method to clear reconnect +2ms
  mqttjs:client _clearReconnect : clearing reconnect timer +1ms
  mqttjs:client _setupStream :: using streamBuilder provided to client to create stream +2ms
  mqttjs calling streambuilder for mqtt +5s
  mqttjs:tcp port 1883 and host localhost +5s
  mqttjs:client _setupStream :: pipe stream to writable stream +4ms
  mqttjs:client _setupStream: sending packet `connect` +1ms
  mqttjs:client sendPacket :: packet: { cmd: 'connect' } +1ms
  mqttjs:client sendPacket :: emitting `packetsend` +2ms
  mqttjs:client sendPacket :: writing to stream +1ms
  mqttjs:client sendPacket :: writeToStream result true +1ms
  mqttjs:client writable stream :: parsing buffer +4ms
  mqttjs:client parser :: on packet push to packets array. +2ms
  mqttjs:client work :: getting next packet in queue +1ms
  mqttjs:client work :: packet pulled from queue +1ms
  mqttjs:client _handlePacket :: emitting packetreceive +1ms
  mqttjs:client _handleConnack +1ms
  mqttjs:client connect :: sending queued packets +2ms
  mqttjs:client deliver :: entry undefined +1ms
connected 2020-09-21T22:27:11.624Z
  mqttjs:client (mqttjs_a09b79d3)stream :: on close +1m
  mqttjs:client flushVolatile :: deleting volatile messages from the queue and setting their callbacks as error function +3ms
  mqttjs:client stream: emit close to MqttClient +1ms
  mqttjs:client close :: connected set to `false` +1ms
  mqttjs:client close :: clearing connackTimer +1ms
  mqttjs:client close :: clearing ping timer +1ms
  mqttjs:client close :: calling _setupReconnect +2ms
  mqttjs:client _setupReconnect :: doing nothing... +2ms
disconnected 2020-09-21T22:28:41.631Z