MQTT tolerate delayed PINGRESP (#25067)

* MQTT tolerate delayed PINGRESP

* Add default MQTT_MAX_PING_OUTSTANDING

CHANGELOG.md

@@ -26,6 +26,7 @@ All notable changes to this project will be documented in this file.
 - BLE MI32 display icons instead of data lines. disable by removing `#define USE_SENSOR_ICON`
 - MiELHVAC accepts `fan_only` as an alias for fan mode in `HVACSetMode` / `HVACSetHAMode` (Home Assistant) (#24992)
 - Command `SetOption46 201..255` init wait 1 to 55 seconds instead of 2010 to 2550 msec (#25035)
+- MQTT connection closed before TCP retransmission could deliver a delayed PINGRESP, now tolerates 2 unanswered pings
 
 ### Fixed
 - MiELHVAC Modbus and CC1101 GPIO names shown as `MbsRelay ...` due to a duplicate entry in the GPIO name table (#24992)

lib/default/TasmotaPubSub/src/PubSubClient_311.cpp

@@ -311,7 +311,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
             if (len == 4 && (buffer[0] == MQTTCONNACK) && (buffer[1] == 2)) {
                 if (buffer[3] == 0) {
                     lastInActivity = millis();
-                    pingOutstanding = false;
+                    pingOutstanding = 0;
                     _state = MQTT_CONNECTED;
                     return true;
                 } else {
@@ -480,7 +480,11 @@ boolean PubSubClient::loop() {
 // the next one tore the connection down with MQTT_CONNECTION_TIMEOUT.
         if (this->keepAlive &&
             ((t - lastInActivity > this->keepAlive*1000UL) || (t - lastOutActivity > this->keepAlive*1000UL))) {
-            if (pingOutstanding) {
+// Up to maxPingOutstanding PINGREQ may stay unanswered before giving up. Sending a further
+// PINGREQ (allowed at any time by MQTT 3.1.2.10) keeps the broker's 1.5 x keepAlive timer
+// fed while TCP retransmission recovers a delayed PINGRESP.
+// See https://github.com/arendst/Tasmota/issues/24985
+            if (pingOutstanding >= this->maxPingOutstanding) {
                 this->_state = MQTT_CONNECTION_TIMEOUT;
                 _client->stop();
                 return false;
@@ -493,7 +497,7 @@ boolean PubSubClient::loop() {
                   lastInActivity = t;
                 }
 
-                pingOutstanding = true;
+                pingOutstanding++;
             }
         }
         if (_client->available()) {
@@ -566,7 +570,7 @@ boolean PubSubClient::loop() {
                     this->buffer[1] = 0;
                     _client->write(this->buffer,2);
                 } else if (type == MQTTPINGRESP) {
-                    pingOutstanding = false;
+                    pingOutstanding = 0;
                 }
             } else if (!connected()) {
                 // readPacket has closed the connection
@@ -871,7 +875,7 @@ void PubSubClient::disconnect(bool disconnect_package) {
     }
     _state = MQTT_DISCONNECTED;
     // A stale outstanding ping would otherwise time out the next connection immediately
-    pingOutstanding = false;
+    pingOutstanding = 0;
 
     lastInActivity = lastOutActivity = millis();
 }
@@ -911,7 +915,7 @@ boolean PubSubClient::connected() {
                 _client->flush();
                 _client->stop();
 
-                pingOutstanding = false;
+                pingOutstanding = 0;
             }
         } else {
             return this->_state == MQTT_CONNECTED;

lib/default/TasmotaPubSub/src/PubSubClient_311.h

@@ -37,6 +37,15 @@
 #define MQTT_SOCKET_TIMEOUT 15
 #endif
 
+// MQTT_MAX_PING_OUTSTANDING : number of unanswered PINGREQ tolerated before the connection
+//  is closed with MQTT_CONNECTION_TIMEOUT. Each one is sent one keepalive interval after the
+//  previous, so a PINGRESP may take up to N x keepAlive to arrive. 1 = original PubSubClient
+//  behavior. Override with setMaxPingOutstanding() (clamped to 1..4).
+//  See https://github.com/arendst/Tasmota/issues/24985
+#ifndef MQTT_MAX_PING_OUTSTANDING
+#define MQTT_MAX_PING_OUTSTANDING 2
+#endif
+
 // MQTT_MAX_TRANSFER_SIZE : limit how much data is passed to the network client
 //  in each write call. Needed for the Arduino Wifi Shield. Leave undefined to
 //  pass the entire MQTT packet in each write call.
@@ -99,7 +108,8 @@ private:
    uint16_t nextMsgId = 0;
    unsigned long lastOutActivity = 0;
    unsigned long lastInActivity = 0;
-   bool pingOutstanding = false;
+   uint8_t pingOutstanding = 0;       // number of PINGREQ sent without a PINGRESP yet
+   uint8_t maxPingOutstanding = MQTT_MAX_PING_OUTSTANDING;
    MQTT_CALLBACK_SIGNATURE;
    uint32_t readPacket(uint8_t*);
    boolean readByte(uint8_t * result);
@@ -154,6 +164,9 @@ public:
    PubSubClient& setStream(Stream& stream);
    PubSubClient& setKeepAlive(uint16_t keepAlive);
    PubSubClient& setSocketTimeout(uint16_t timeout);
+   // Number of unanswered PINGREQ tolerated before closing (clamped to 1..4)
+   PubSubClient& setMaxPingOutstanding(uint8_t n) { this->maxPingOutstanding = (n < 1) ? 1 : ((n > 4) ? 4 : n); return *this; }
+   uint8_t getMaxPingOutstanding() const { return this->maxPingOutstanding; }
 
    boolean setBufferSize(uint16_t size);
    uint16_t getBufferSize() const;

lib/default/TasmotaPubSub/src/PubSubClient_5.cpp

@@ -1400,7 +1400,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
                     }
 
                     lastInActivity = millis();
-                    pingOutstanding = false;
+                    pingOutstanding = 0;
                     _state = MQTT_CONNECTED;
 // Task 18.9: retransmission is part of connection establishment. Keep the prior
 // identity intact while replay verifies each entry belongs to this persistent session;
@@ -1423,7 +1423,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
                 if (len == 4 && (buffer[0] == MQTTCONNACK) && (buffer[1] == 2)) {
                     if (buffer[3] == 0) {
                         lastInActivity = millis();
-                        pingOutstanding = false;
+                        pingOutstanding = 0;
                         _state = MQTT_CONNECTED;
                         return true;
                     } else {
@@ -1736,7 +1736,12 @@ boolean PubSubClient::loop() {
 // the next one tore the connection down with MQTT_CONNECTION_TIMEOUT.
         if (this->keepAlive &&
             ((t - lastInActivity > this->keepAlive*1000UL) || (t - lastOutActivity > this->keepAlive*1000UL))) {
-            if (pingOutstanding) {
+// Up to maxPingOutstanding PINGREQ may stay unanswered before giving up. Sending a further
+// PINGREQ (allowed at any time by MQTT 3.1.2.10) keeps the broker's 1.5 x keepAlive timer
+// fed while TCP retransmission recovers a delayed PINGRESP. The window follows the
+// effective keepAlive, including a broker-imposed Server Keep Alive.
+// See https://github.com/arendst/Tasmota/issues/24985
+            if (pingOutstanding >= this->maxPingOutstanding) {
                 this->_state = MQTT_CONNECTION_TIMEOUT;
                 _client->stop();
                 return false;
@@ -1750,7 +1755,7 @@ boolean PubSubClient::loop() {
                 }
                 lastOutActivity = t;
                 lastInActivity = t;
-                pingOutstanding = true;
+                pingOutstanding++;
             }
         }
         if (_client->available()) {
@@ -2086,7 +2091,7 @@ boolean PubSubClient::loop() {
                     this->buffer[1] = 0;
                     transportWrite(this->buffer,2);
                 } else if (type == MQTTPINGRESP) {
-                    pingOutstanding = false;
+                    pingOutstanding = 0;
                 }
             } else if (!connected()) {
                 // readPacket has closed the connection
@@ -4337,7 +4342,7 @@ void PubSubClient::resetConnectionState() {
     publishBytesExpected = 0;
     publishBytesWritten = 0;
     // A stale outstanding ping would otherwise time out the next connection immediately
-    pingOutstanding = false;
+    pingOutstanding = 0;
 
     lastInActivity = lastOutActivity = millis();
 
@@ -4475,7 +4480,7 @@ boolean PubSubClient::connected() {
                 _client->flush();
                 _client->stop();
 
-                pingOutstanding = false;
+                pingOutstanding = 0;
                 if (MQTT_RUNTIME_IS_5() && (!this->sessionIdentity.valid ||
                     (this->sessionIdentity.sessionExpiryInterval == 0))) {
 // Clean Start, explicit zero expiry, or a failed identity capture leaves no session that can

lib/default/TasmotaPubSub/src/PubSubClient_5.h

@@ -38,6 +38,15 @@
 #define MQTT_SOCKET_TIMEOUT 15
 #endif
 
+// MQTT_MAX_PING_OUTSTANDING : number of unanswered PINGREQ tolerated before the connection
+//  is closed with MQTT_CONNECTION_TIMEOUT. Each one is sent one keepalive interval after the
+//  previous, so a PINGRESP may take up to N x keepAlive to arrive. 1 = original PubSubClient
+//  behavior. Override with setMaxPingOutstanding() (clamped to 1..4).
+//  See https://github.com/arendst/Tasmota/issues/24985
+#ifndef MQTT_MAX_PING_OUTSTANDING
+#define MQTT_MAX_PING_OUTSTANDING 2
+#endif
+
 // USE_MQTT_QOS : opt-in to OUTBOUND QoS 1/2 acknowledged delivery.
 //
 // Tasmota publishes at QoS 0 in almost every real deployment, so this MQTT5-capable client
@@ -822,7 +831,8 @@ private:
    uint16_t nextMsgId = 0;
    unsigned long lastOutActivity = 0;
    unsigned long lastInActivity = 0;
-   bool pingOutstanding = false;
+   uint8_t pingOutstanding = 0;       // number of PINGREQ sent without a PINGRESP yet
+   uint8_t maxPingOutstanding = MQTT_MAX_PING_OUTSTANDING;
 
    // State for the beginPublish()/write()/endPublish() streaming contract. A
    // mismatch leaves a truncated PUBLISH on the wire, so it is terminal.
@@ -1412,6 +1422,9 @@ public:
    PubSubClient& setStream(Stream& stream);
    PubSubClient& setKeepAlive(uint16_t keepAlive);
    PubSubClient& setSocketTimeout(uint16_t timeout);
+   // Number of unanswered PINGREQ tolerated before closing (clamped to 1..4)
+   PubSubClient& setMaxPingOutstanding(uint8_t n) { this->maxPingOutstanding = (n < 1) ? 1 : ((n > 4) ? 4 : n); return *this; }
+   uint8_t getMaxPingOutstanding() const { return this->maxPingOutstanding; }
    // Session Expiry Interval used by a legacy `cleanSession == false` CONNECT. The
    // default is 0xFFFFFFFF (indefinite persistence); zero is preserved as an explicit
    // Clean Start 0 / immediate-expiry request rather than replaced by the default.

lib/default/TasmotaPubSub/tests/src/keepalive_test.cpp

@@ -15,16 +15,18 @@
     - A scripted PINGRESP clears the outstanding-ping state, so the connection
       stays alive and a later idle interval issues a further PINGREQ instead of
       declaring a timeout (Requirement 12.2).
-    - Advancing beyond the interval a second time with no PINGRESP reports a lost
-      connection (Requirement 12.3).
+    - Up to MQTT_MAX_PING_OUTSTANDING (setMaxPingOutstanding()) PINGREQ may stay
+      unanswered, one per interval; the next interval with no PINGRESP reports a
+      lost connection (Requirement 12.3, issue #24985).
 
   Properties (folded in as single deterministic, data-driven cases over a curated
   interval table {15 s default, 2 s short} - NO randomized generators):
     - Property 8: exactly one PINGREQ is issued per idle interval, and after a
       matching PINGRESP another PINGREQ is issued on the next interval rather
       than a timeout being declared (Requirements 12.1, 12.2, 18.9).
-    - Property 9: advancing beyond two consecutive intervals with no PINGRESP
-      reports a lost/timed-out connection (Requirements 12.3, 18.8, 18.9).
+    - Property 9: with a budget of N tolerated pings, advancing beyond N+1
+      consecutive intervals with no PINGRESP reports a lost/timed-out connection
+      (Requirements 12.3, 18.8, 18.9).
 
   Every assertion goes through the public API and decoded MockClient.outbound()
   wire bytes - never private members - so the baseline stays durable across a
@@ -72,6 +74,19 @@ bool isSinglePingreq(const std::vector<uint8_t>& out) {
            out.size() == 2;
 }
 
+// Advance `count` idle intervals without delivering any PINGRESP, checking that
+// each one emits exactly one PINGREQ and leaves the connection up.
+void sendUnansweredPings(MockClient& client, PubSubClient& psc, uint16_t keepAliveSecs, uint8_t count) {
+    for (uint8_t i = 0; i < count; i++) {
+        CAPTURE(i);
+        client.clearOutbound();
+        advanceBeyondInterval(keepAliveSecs);
+        REQUIRE(psc.loop());
+        CHECK(isSinglePingreq(client.outbound()));
+        CHECK(psc.connected());
+    }
+}
+
 }  // namespace
 
 TEST_SUITE("baseline") {
@@ -130,14 +145,14 @@ TEST_SUITE("baseline") {
         MockClient client;
         PubSubClient psc(client);
         connectAndClear(client, psc, MQTT_KEEPALIVE);
+        REQUIRE(psc.getMaxPingOutstanding() == MQTT_MAX_PING_OUTSTANDING);
 
-        // First interval: PINGREQ sent, ping now outstanding.
-        advanceBeyondInterval(MQTT_KEEPALIVE);
-        REQUIRE(psc.loop());
-        REQUIRE(isSinglePingreq(client.outbound()));
+        // One PINGREQ per interval while fewer than MQTT_MAX_PING_OUTSTANDING are
+        // unanswered; the connection stays up.
+        sendUnansweredPings(client, psc, MQTT_KEEPALIVE, MQTT_MAX_PING_OUTSTANDING);
 
-        // Second interval with no PINGRESP delivered: the library detects the
-        // unanswered ping and reports a lost/timed-out connection.
+        // Next interval with still no PINGRESP delivered: the library detects the
+        // unanswered pings and reports a lost/timed-out connection.
         advanceBeyondInterval(MQTT_KEEPALIVE);
         CHECK_FALSE(psc.loop());
         CHECK_FALSE(psc.connected());
@@ -145,6 +160,61 @@ TEST_SUITE("baseline") {
         CHECK(client.stopCalled());
     }
 
+    // --- Tolerated unanswered PINGREQ (issue #24985) -------------------------
+
+    TEST_CASE("setMaxPingOutstanding(1) restores the original single-ping timeout") {
+        TestClock::instance().reset();
+        MockClient client;
+        PubSubClient psc(client);
+        psc.setMaxPingOutstanding(1);
+        connectAndClear(client, psc, MQTT_KEEPALIVE);
+
+        advanceBeyondInterval(MQTT_KEEPALIVE);
+        REQUIRE(psc.loop());
+        REQUIRE(isSinglePingreq(client.outbound()));
+
+        advanceBeyondInterval(MQTT_KEEPALIVE);
+        CHECK_FALSE(psc.loop());
+        CHECK(psc.state() == MQTT_CONNECTION_TIMEOUT);
+        CHECK(client.stopCalled());
+    }
+
+    TEST_CASE("setMaxPingOutstanding clamps to 1..4") {
+        MockClient client;
+        PubSubClient psc(client);
+        CHECK(psc.setMaxPingOutstanding(0).getMaxPingOutstanding() == 1);
+        CHECK(psc.setMaxPingOutstanding(3).getMaxPingOutstanding() == 3);
+        CHECK(psc.setMaxPingOutstanding(200).getMaxPingOutstanding() == 4);
+    }
+
+    TEST_CASE("a late PINGRESP after a second PINGREQ keeps the connection alive") {
+        TestClock::instance().reset();
+        MockClient client;
+        PubSubClient psc(client);
+        psc.setMaxPingOutstanding(2);
+        connectAndClear(client, psc, MQTT_KEEPALIVE);
+
+        // Two intervals without answer: two PINGREQ, still connected.
+        sendUnansweredPings(client, psc, MQTT_KEEPALIVE, 2);
+
+        // Both PINGRESP arrive late (TCP retransmission recovered); the counter is
+        // cleared and the surplus PINGRESP is harmless.
+        client.clearOutbound();
+        client.pushPacket(MqttPacket::pingresp());
+        client.pushPacket(MqttPacket::pingresp());
+        REQUIRE(psc.loop());
+        REQUIRE(psc.loop());
+        CHECK(client.outbound().empty());
+        CHECK(psc.connected());
+
+        // A full new budget is available again: two more unanswered intervals do
+        // not time out, the third does.
+        sendUnansweredPings(client, psc, MQTT_KEEPALIVE, 2);
+        advanceBeyondInterval(MQTT_KEEPALIVE);
+        CHECK_FALSE(psc.loop());
+        CHECK(psc.state() == MQTT_CONNECTION_TIMEOUT);
+    }
+
     // --- Property 8: exactly one PINGREQ per idle interval ------------------
 
     // Feature: tasmota-pubsub-tests, Property 8: for all keepalive intervals,
@@ -194,34 +264,38 @@ TEST_SUITE("baseline") {
 
     // --- Property 9: missing PINGRESP produces a lost connection ------------
 
-    // Feature: tasmota-pubsub-tests, Property 9: for all keepalive intervals,
-    // advancing the virtual clock beyond two consecutive intervals without
-    // delivering a PINGRESP causes the library to report a lost/timed-out
-    // connection. Validated deterministically over a curated interval table
-    // (default 15 s and a short 2 s), no randomized generators.
-    TEST_CASE("Property 9: missing PINGRESP over two intervals reports a lost connection") {
+    // Feature: tasmota-pubsub-tests, Property 9: for all keepalive intervals and
+    // all tolerated-ping budgets N, advancing the virtual clock beyond N
+    // consecutive intervals without a PINGRESP sends exactly one PINGREQ per
+    // interval and keeps the connection; the (N+1)th interval reports a
+    // lost/timed-out connection. Validated deterministically over a curated
+    // interval table (default 15 s and a short 2 s) and N in {1, 2, 4}, no
+    // randomized generators.
+    TEST_CASE("Property 9: missing PINGRESP over N+1 intervals reports a lost connection") {
         const uint16_t intervals[] = {MQTT_KEEPALIVE, 2};
+        const uint8_t budgets[] = {1, 2, 4};
 
         for (uint16_t keepAliveSecs : intervals) {
-            CAPTURE(keepAliveSecs);
-            TestClock::instance().reset();
-            MockClient client;
-            PubSubClient psc(client);
-            connectAndClear(client, psc, keepAliveSecs);
+            for (uint8_t budget : budgets) {
+                CAPTURE(keepAliveSecs);
+                CAPTURE(budget);
+                TestClock::instance().reset();
+                MockClient client;
+                PubSubClient psc(client);
+                psc.setMaxPingOutstanding(budget);
+                connectAndClear(client, psc, keepAliveSecs);
 
-            // First interval: PINGREQ sent, connection still alive.
-            advanceBeyondInterval(keepAliveSecs);
-            REQUIRE(psc.loop());
-            CHECK(isSinglePingreq(client.outbound()));
-            CHECK(psc.connected());
+                // N intervals: one PINGREQ each, connection still alive.
+                sendUnansweredPings(client, psc, keepAliveSecs, budget);
 
-            // Second interval with no PINGRESP: the unanswered ping trips the
-            // keepalive timeout and the connection is reported lost.
-            advanceBeyondInterval(keepAliveSecs);
-            CHECK_FALSE(psc.loop());
-            CHECK_FALSE(psc.connected());
-            CHECK(psc.state() == MQTT_CONNECTION_TIMEOUT);
-            CHECK(client.stopCalled());
+                // Next interval with no PINGRESP: the unanswered pings trip the
+                // keepalive timeout and the connection is reported lost.
+                advanceBeyondInterval(keepAliveSecs);
+                CHECK_FALSE(psc.loop());
+                CHECK_FALSE(psc.connected());
+                CHECK(psc.state() == MQTT_CONNECTION_TIMEOUT);
+                CHECK(client.stopCalled());
+            }
         }
     }
 }

tasmota/my_user_config.h

@@ -117,6 +117,7 @@
 // -- PubSubClient defaults -----------------------
 #define MQTT_KEEPALIVE         30                // [MqttKeepAlive] Number of seconds between KeepAlive messages
 #define MQTT_SOCKET_TIMEOUT    4                 // [MqttTimeout] Number of seconds before Mqtt connection timeout
+#define MQTT_MAX_PING_OUTSTANDING 2              // Number of unanswered PINGREQ (one per KeepAlive) before closing the connection (1..4, 1 = legacy)
 #define MQTT_MAX_PACKET_SIZE   1200              // Working buffer and advertised maximum incoming packet size
 #define MQTT_PACKET_TIMEOUT    MQTT_SOCKET_TIMEOUT // Seconds allowed to receive a complete MQTT packet