From abe838bf58b2bba24e76f826234ea079195970ec Mon Sep 17 00:00:00 2001 From: agessaman Date: Wed, 9 Sep 2026 17:40:45 -0700 Subject: [PATCH] fix(mqtt): typed results for client operations, and stop advancing state on failure (F06) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `connect()`, `reconnect()`, `disconnect()`, `softDisconnect()` and `forceStop()` all returned void, so every caller in the bridge treated "asked" as "done": - `setupSlot()` marked a slot activated after `connect()` whatever happened. A failed `esp_mqtt_client_start()` therefore consumed one of the scarce active-slot positions, handed the slot to a reconnect ladder that is gated on activation, and was never retried by the deferred-setup path. - `reconnect()` explicitly proceeded after `esp_mqtt_set_config()` failed, reconnecting on the previous configuration — the renewed token in the buffer, the old one on the wire. - the renewal path recorded the new expiry before the bounce succeeded, so a failed bounce left the live session on the old credential with the next renewal not due for a whole token lifetime. - `softDisconnect()` logged its timeout and told its caller nothing. - `disconnect()` waited for the DISCONNECTED event with no bound, on the very task whose stop acknowledgement the shutdown waits for. Now every one of them returns `esp_err_t`, a failed configuration transaction aborts rather than starting or reconnecting on a half-updated config, and `disconnect()`'s wait is bounded (it still stops the client, and reports ESP_ERR_TIMEOUT when the event never arrived). Bridge consequences: - a failed start leaves the slot unactivated, so the existing deferred-setup retry revisits it and it holds no active-slot position; - a failed renewal bounce re-arms the renewal instead of recording it, so the next maintenance pass retries; - a reconnect that fails *locally* rolls back the backoff advance made for it. The ladder and the breaker bound broker and network faults; an uninitialised client or an uncommitted config transaction is neither, and inflating the ladder for it was how a local fault could trip a breaker meant for a broker. --- .../src/PsychicMqttClient.cpp | 133 +++++++++++------- lib/PsychicMqttClient/src/PsychicMqttClient.h | 47 +++++-- src/helpers/bridges/MQTTBridge.cpp | 78 ++++++++-- src/helpers/bridges/MQTTBridge.h | 5 +- 4 files changed, 194 insertions(+), 69 deletions(-) diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp index ba1dc5f1..cec11177 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp @@ -371,20 +371,20 @@ bool PsychicMqttClient::connected() return _connected; } -void PsychicMqttClient::connect() +esp_err_t PsychicMqttClient::connect() { #if ESP_IDF_VERSION_MAJOR == 5 if (_mqtt_cfg.broker.address.uri == nullptr) { ESP_LOGE(TAG, "MQTT URI not set."); - return; + return ESP_ERR_INVALID_STATE; } int desired_buffer = _mqtt_cfg.buffer.size > 0 ? _mqtt_cfg.buffer.size : 1024; #else if (_mqtt_cfg.uri == nullptr) { ESP_LOGE(TAG, "MQTT URI not set."); - return; + return ESP_ERR_INVALID_STATE; } int desired_buffer = _mqtt_cfg.buffer_size > 0 ? _mqtt_cfg.buffer_size : 1024; #endif @@ -400,41 +400,44 @@ void PsychicMqttClient::connect() { ESP_LOGE(TAG, "Failed to allocate reassembly buffer (%u bytes)", (unsigned)_buffer_capacity); _buffer_capacity = 0; + return ESP_ERR_NO_MEM; } } if (_client == nullptr) { _client = esp_mqtt_client_init(&_mqtt_cfg); + if (_client == nullptr) + { + ESP_LOGE(TAG, "esp_mqtt_client_init failed"); + return ESP_ERR_NO_MEM; + } // Register event handler only once when client is first created // to avoid memory leak from repeated registrations esp_mqtt_client_register_event(_client, MQTT_EVENT_ANY, _onMqttEventStatic, this); _config_dirty = false; } + else if (_config_dirty) + { + // A failed config update must not be followed by a start: the client + // would come up on a partly-updated configuration, which is how a slot + // connects with the previous broker's credentials. Leave _config_dirty + // set so the next attempt retries the whole transaction. + esp_err_t cfg_result = esp_mqtt_set_config(_client, &_mqtt_cfg); + if (cfg_result != ESP_OK) + { + ESP_LOGE(TAG, "connect(): failed to apply mqtt config: %s", esp_err_to_name(cfg_result)); + return cfg_result; + } + _config_dirty = false; + ESP_LOGD(TAG, "connect(): applied mqtt config update"); + } else { - if (_config_dirty) - { - esp_err_t cfg_result = esp_mqtt_set_config(_client, &_mqtt_cfg); - ESP_ERROR_CHECK_WITHOUT_ABORT(cfg_result); - if (cfg_result == ESP_OK) - { - _config_dirty = false; - ESP_LOGD(TAG, "connect(): applied mqtt config update"); - } - else - { - ESP_LOGW(TAG, "connect(): failed to apply mqtt config, will retry"); - } - } - else - { - ESP_LOGD(TAG, "connect(): mqtt config unchanged, skipping set_config"); - } + ESP_LOGD(TAG, "connect(): mqtt config unchanged, skipping set_config"); } esp_err_t start_result = esp_mqtt_client_start(_client); - ESP_ERROR_CHECK_WITHOUT_ABORT(start_result); if (start_result == ESP_OK) { _started = true; @@ -445,76 +448,103 @@ void PsychicMqttClient::connect() // Reporting success here hides the one state reconnect() cannot recover from. ESP_LOGE(TAG, "MQTT client failed to start: %s", esp_err_to_name(start_result)); } + return start_result; } -void PsychicMqttClient::reconnect() +esp_err_t PsychicMqttClient::reconnect() { if (_client == nullptr) { ESP_LOGW(TAG, "MQTT client not initialized, cannot reconnect."); - return; + return ESP_ERR_INVALID_STATE; } if (_config_dirty) { - // Apply config only when mutating setters changed _mqtt_cfg. + // Apply config only when mutating setters changed _mqtt_cfg. A failure + // aborts the reconnect: reconnecting on the previous config was how a + // renewed token silently failed to reach the connection. esp_err_t cfg_result = esp_mqtt_set_config(_client, &_mqtt_cfg); - ESP_ERROR_CHECK_WITHOUT_ABORT(cfg_result); - if (cfg_result == ESP_OK) + if (cfg_result != ESP_OK) { - _config_dirty = false; - ESP_LOGD(TAG, "reconnect(): applied mqtt config update"); - } - else - { - ESP_LOGW(TAG, "reconnect(): failed to apply mqtt config, reconnecting with previous config"); + ESP_LOGE(TAG, "reconnect(): failed to apply mqtt config: %s", esp_err_to_name(cfg_result)); + return cfg_result; } + _config_dirty = false; + ESP_LOGD(TAG, "reconnect(): applied mqtt config update"); } else { ESP_LOGD(TAG, "reconnect(): mqtt config unchanged, skipping set_config"); } - ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_reconnect(_client)); + esp_err_t r = esp_mqtt_client_reconnect(_client); + if (r != ESP_OK) + { + ESP_LOGE(TAG, "MQTT reconnect request failed: %s", esp_err_to_name(r)); + return r; + } ESP_LOGI(TAG, "MQTT client reconnect requested."); + return ESP_OK; } -void PsychicMqttClient::disconnect() +esp_err_t PsychicMqttClient::disconnect(unsigned long timeout_ms) { if (_client == nullptr) { ESP_LOGW(TAG, "MQTT client not started."); - return; + return ESP_ERR_INVALID_STATE; } + bool clean = true; if (_connected) { ESP_LOGI(TAG, "Disconnecting MQTT client."); _stopMqttClient = false; esp_mqtt_client_disconnect(_client); - // Wait for all disconnect events to be processed - while (!_stopMqttClient) + // Bounded, unlike the original: this runs on the MQTT task whose stop + // acknowledgement the bridge's shutdown waits for, so a lost + // DISCONNECTED event used to wedge the whole teardown. + unsigned long waited = 0; + while (!_stopMqttClient && waited < timeout_ms) { vTaskDelay(10 / portTICK_PERIOD_MS); + waited += 10; + } + if (!_stopMqttClient) + { + ESP_LOGW(TAG, "disconnect: no DISCONNECTED event in %lums; stopping anyway", timeout_ms); + clean = false; } } - esp_mqtt_client_stop(_client); - _started = false; - ESP_LOGI(TAG, "MQTT client stopped."); + esp_err_t stop_result = esp_mqtt_client_stop(_client); + if (stop_result == ESP_OK) + { + _started = false; + ESP_LOGI(TAG, "MQTT client stopped."); + } + else + { + // The SDK task was not joined. Saying otherwise is what let the caller + // destroy the client from under a live task. + ESP_LOGE(TAG, "esp_mqtt_client_stop failed: %s", esp_err_to_name(stop_result)); + return stop_result; + } + return clean ? ESP_OK : ESP_ERR_TIMEOUT; } -void PsychicMqttClient::softDisconnect(unsigned long timeout_ms) +esp_err_t PsychicMqttClient::softDisconnect(unsigned long timeout_ms) { if (_client == nullptr) { ESP_LOGW(TAG, "MQTT client not started."); - return; + return ESP_ERR_INVALID_STATE; } if (!_connected) { // Nothing to close; leaving the task alone is the whole point. - return; + return ESP_OK; } ESP_LOGI(TAG, "Disconnecting MQTT transport (client task retained)."); @@ -530,25 +560,34 @@ void PsychicMqttClient::softDisconnect(unsigned long timeout_ms) if (!_stopMqttClient) { ESP_LOGW(TAG, "softDisconnect: no DISCONNECTED event in %lums", timeout_ms); + return ESP_ERR_TIMEOUT; } + return ESP_OK; } -void PsychicMqttClient::forceStop() +esp_err_t PsychicMqttClient::forceStop() { if (_client == nullptr) { ESP_LOGW(TAG, "MQTT client not started."); - return; + return ESP_ERR_INVALID_STATE; } if (_connected) { ESP_LOGI(TAG, "Forced stop MQTT client."); } - ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_stop(_client)); + esp_err_t r = esp_mqtt_client_stop(_client); + if (r != ESP_OK) + { + // The SDK task was not joined: the caller must not destroy this client. + ESP_LOGE(TAG, "forceStop: esp_mqtt_client_stop failed: %s", esp_err_to_name(r)); + return r; + } _connected = false; _started = false; ESP_LOGI(TAG, "MQTT client forcefully stopped."); + return ESP_OK; } int PsychicMqttClient::subscribe(const char *topic, int qos) diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.h b/lib/PsychicMqttClient/src/PsychicMqttClient.h index 28e53fc3..5d214679 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.h +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.h @@ -349,8 +349,17 @@ public: * @brief Connects the MQTT client to the server. * * @note All parameters must be set before calling this method. + * + * @return ESP_OK when esp_mqtt_client_start() accepted the client. + * ESP_ERR_INVALID_STATE if no URI is set, ESP_ERR_NO_MEM if the + * client or its reassembly buffer could not be allocated, or the + * error from esp_mqtt_set_config()/esp_mqtt_client_start(). + * + * A failed configuration update does NOT start the client: starting with a + * half-updated config is how a slot ends up connected with the previous + * broker's credentials. */ - void connect(); + esp_err_t connect(); /** * @brief Reconnects a previously started MQTT client. @@ -359,14 +368,28 @@ public: * re-initiating a connection on an already-started client, especially * when auto-reconnect is disabled. Updates config before reconnecting * so credential changes (e.g., refreshed JWT tokens) take effect. + * + * @return ESP_OK when the reconnect was requested, ESP_ERR_INVALID_STATE on + * an uninitialised client, or the error from esp_mqtt_set_config(). + * A failed config update aborts the reconnect rather than + * reconnecting with the previous configuration. */ - void reconnect(); + esp_err_t reconnect(); /** - * @brief Disconnects the MQTT client from the server. - * This call might be blocking until the client is stopped cleanly + * @brief Disconnects the MQTT client and stops its task. + * + * Blocks until the DISCONNECTED event arrives or timeout_ms elapses, then + * stops the client either way. The wait used to be unbounded, which turned + * a lost event into a wedged caller — and that caller is the MQTT task + * whose acknowledgement the bridge's shutdown waits on. + * + * @return ESP_OK when the client stopped after a clean disconnect, + * ESP_ERR_TIMEOUT when the DISCONNECTED event never arrived (the + * stop still ran), ESP_ERR_INVALID_STATE on an uninitialised + * client, or the error from esp_mqtt_client_stop(). */ - void disconnect(); + esp_err_t disconnect(unsigned long timeout_ms = 10000); /** * @brief Closes the transport but leaves the client task running. @@ -380,8 +403,12 @@ public: * @param timeout_ms how long to wait for the DISCONNECTED event before * giving up. Bounded on purpose: disconnect()'s wait is * unbounded and a lost event would wedge the caller. + * + * @return ESP_OK when the DISCONNECTED event arrived, ESP_ERR_TIMEOUT if it + * did not (the transport may still be open), ESP_ERR_INVALID_STATE + * on an uninitialised client, ESP_OK when already disconnected. */ - void softDisconnect(unsigned long timeout_ms = 5000); + esp_err_t softDisconnect(unsigned long timeout_ms = 5000); /** * @brief True once esp_mqtt_client_start() has succeeded and no stop has run. @@ -392,10 +419,14 @@ public: bool isStarted() const { return _started; } /** - * @brief Forcefully stops the MQTT client and disconnects from the server. + * @brief Stops the MQTT client without waiting for a DISCONNECTED event. * This does not trigger the onDisconnect callbacks. + * + * @return the result of esp_mqtt_client_stop(), or ESP_ERR_INVALID_STATE on + * an uninitialised client. A non-OK result means the SDK task was + * NOT joined: the caller must not destroy the client after it. */ - void forceStop(); + esp_err_t forceStop(); /** * @brief Subscribes to a topic. Server must be connected diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 8c0ee058..97ecd867 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -2091,7 +2091,19 @@ bool MQTTBridge::setupSlot(int index) { } } - slot.client->connect(); + // Activation is now conditional on the client actually starting. A failed + // esp_mqtt_set_config()/esp_mqtt_client_start() used to be invisible: the slot + // was marked activated, so it consumed one of the scarce active-slot + // positions, the reconnect ladder (which is gated on activation) governed it, + // and nothing retried the setup. Leaving it unactivated hands it to the + // existing deferred-setup retry in maintainSlotConnections() instead. + const esp_err_t connect_result = slot.client->connect(); + if (connect_result != ESP_OK) { + MQTT_DEBUG_PRINTLN("MQTT%d start failed (%s) - will retry", index + 1, + esp_err_to_name(connect_result)); + slot.last_reconnect_attempt = millis(); + return false; + } slot.initial_connect_done = true; return true; } @@ -2137,17 +2149,16 @@ void MQTTBridge::teardownSlot(int index) { // which only stops slots still marked connected: a publishing slot's socket fails first, so // the guard skips it — measured across a 62 s deauth, five slots, zero stops. An idle slot // with no traffic to fail on is the one case that could still reach here that way. -void MQTTBridge::reconnectSlotClient(int index) { - if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; +esp_err_t MQTTBridge::reconnectSlotClient(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return ESP_ERR_INVALID_ARG; MQTTSlot& slot = _slots[index]; - if (slot.client == nullptr) return; + if (slot.client == nullptr) return ESP_ERR_INVALID_STATE; if (!slot.client->isStarted()) { MQTT_DEBUG_PRINTLN("MQTT%d start (client was stopped)", index + 1); - slot.client->connect(); - return; + return slot.client->connect(); } - slot.client->reconnect(); + return slot.client->reconnect(); } @@ -2304,21 +2315,40 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // to avoid internal heap leak/fragmentation from destroy/create cycles MQTT_DEBUG_PRINTLN("MQTT%d token renewal: reconnecting with fresh credentials", index + 1); MQTT_TRACE_HEAP("renewal:before-bounce", index); + esp_err_t bounce_result; if (slot.client->isStarted()) { // Keep the esp-mqtt task alive across the handshake. disconnect() // would stop it, returning its 6 KiB stack into the hole the two // 16 KiB mbedTLS record buffers just vacated — which is what walks // the largest free block down 16 KiB at a time on non-PSRAM boards. - slot.client->softDisconnect(); + const esp_err_t soft = slot.client->softDisconnect(); + if (soft != ESP_OK) { + MQTT_DEBUG_PRINTLN("MQTT%d renewal: soft disconnect did not complete (%s)", + index + 1, esp_err_to_name(soft)); + } MQTT_TRACE_HEAP("renewal:after-disconnect", index); slot.client->setCredentials(_jwt_username, slot.auth_token); MQTT_TRACE_HEAP("renewal:after-credentials", index); - slot.client->reconnect(); + bounce_result = slot.client->reconnect(); } else { // Client was stopped (teardown/reconfigure). reconnect() is a no-op // on a stopped client, so this path must start it. slot.client->setCredentials(_jwt_username, slot.auth_token); - slot.client->connect(); + bounce_result = slot.client->connect(); + } + if (bounce_result != ESP_OK) { + // The fresh token is in the buffer but did not reach the + // connection: the config transaction failed, or the client would + // not start. Recording this as a completed renewal would leave the + // live session on the old credential until the broker enforced exp, + // with the next renewal not due for a whole token lifetime. Re-arm + // the renewal instead so the next maintenance pass retries the + // bounce, and leave the reconnect ladder alone — this is a local + // failure, not a broker fault. + MQTT_DEBUG_PRINTLN("MQTT%d renewal bounce failed (%s) - retrying next pass", + index + 1, esp_err_to_name(bounce_result)); + slot.last_token_renewal = 0; + return; } MQTT_TRACE_HEAP("renewal:after-reconnect", index); reconnect_attempted = true; @@ -2439,6 +2469,9 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns static_cast(now_millis), static_cast(slot.last_reconnect_attempt), slot.reconnect_backoff, static_cast(index))) { slot.last_reconnect_attempt = now_millis; + const uint8_t backoff_before = slot.reconnect_backoff; + const uint8_t failures_before = slot.max_backoff_failures; + const bool breaker_before = slot.circuit_breaker_tripped; MQTTConnectionPolicy::BackoffAdvance advance = MQTTConnectionPolicy::advanceBackoff( slot.reconnect_backoff, slot.max_backoff_failures); slot.reconnect_backoff = advance.reconnect_backoff; @@ -2463,7 +2496,18 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns } // Via the helper: reconnect() is a no-op on a client whose start failed, // which would back off forever without ever starting it. - reconnectSlotClient(index); + const esp_err_t r = reconnectSlotClient(index); + if (r != ESP_OK) { + // A local failure — uninitialised client, or a config transaction that + // did not commit — is not a broker fault. The ladder and the breaker + // exist to bound broker/network faults, so roll back the advance made + // above; the retry interval still paces the next attempt. + MQTT_DEBUG_PRINTLN("MQTT%d reconnect not accepted locally (%s) - backoff unchanged", + index + 1, esp_err_to_name(r)); + slot.reconnect_backoff = backoff_before; + slot.max_backoff_failures = failures_before; + slot.circuit_breaker_tripped = breaker_before; + } } } } @@ -4367,8 +4411,16 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) { reconnectSlotClient(i); } else if (action == MQTTConnectionPolicy::StaleTokenAction::Bounce) { MQTT_DEBUG_PRINTLN("MQTT%d bouncing for the corrected-clock token", i + 1); - _slots[i].client->softDisconnect(); - _slots[i].client->reconnect(); + const esp_err_t soft = _slots[i].client->softDisconnect(); + if (soft != ESP_OK) { + MQTT_DEBUG_PRINTLN("MQTT%d soft disconnect did not complete (%s)", i + 1, + esp_err_to_name(soft)); + } + const esp_err_t rc = _slots[i].client->reconnect(); + if (rc != ESP_OK) { + MQTT_DEBUG_PRINTLN("MQTT%d corrected-clock reconnect failed (%s)", i + 1, + esp_err_to_name(rc)); + } } else { MQTT_DEBUG_PRINTLN("MQTT%d token re-created, no bounce (broker does not enforce exp)", i + 1); diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 0ec28ccc..4450a4f3 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -490,7 +490,10 @@ private: void teardownSlot(int index); // Disconnect the slot's client (keeps the object alive) // Reconnect a slot, starting it instead when the client is stopped (reconnect() is a // no-op on a stopped client). See the definition. - void reconnectSlotClient(int index); + // ESP_OK when the reconnect/start was accepted by the SDK. A local failure + // (uninitialised client, failed config transaction) is not a broker fault and + // must not advance this slot's backoff ladder — see maintainSlotConnection(). + esp_err_t reconnectSlotClient(int index); void maintainSlotConnections(); // Maintain all slot connections (token renewal, reconnect) void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted); bool createSlotAuthToken(int index); // Create/renew JWT token for a slot