From e2673f79e488342dbe3130a9176881a0c80cc9f5 Mon Sep 17 00:00:00 2001 From: agessaman Date: Wed, 9 Sep 2026 20:03:13 -0700 Subject: [PATCH] fix(mqtt): close the review's P1 and P2 findings on the F01/F04/F06 work MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the branch found one merge-blocking lifecycle hole and three correctness gaps where the implementation stopped short of contracts the design had already written down. All four are real; each was confirmed against the source (two of them against hardware) before anything changed. **P1 — a quarantined client could still produce a clean bridge stop.** The cooperative teardown set `_teardown_complete` unconditionally, so a slot whose `esp_mqtt_client_stop()` had not completed — deliberately skipped by `destroySlotClients()` and marked Quarantined — still let the trampoline publish the acknowledgement. The owner then freed the queue and buffers and allowed a restart while that SDK task might still be running: exactly the ownership ambiguity StopUnproven exists to remove. The ack is now withheld unless EVERY client is proven stopped, so one unproven client leaves the whole bridge unproven. The rule lives in MQTTClientState.h (`mqttStopMayBeAcknowledged`) with host tests, alongside the state predicates moved out of the bridge. **P2 — the F04 protection did not cover a client that was still connecting.** `softDisconnect()` returns ESP_OK immediately when the client is not connected, so for a slot mid-DNS/TLS/CONNECT it cancelled nothing: the attempt ran on and its CONNECTED event arrived after the new configuration was applied, and with callbacks registered once per client and esp-mqtt events carrying no generation, nothing could tell it from the new attempt's. A reconfigure that lands on a `Starting` client now stops it, joining its SDK task, before applying the new configuration. A Connected client still takes the cheap softDisconnect path, which is where the fragmentation argument applies. One helper (`closeLiveClientForReconfigure`) so the two call sites cannot drift. **P2 — a failed renewal bounce still advanced the effective expiry.** Minting updates `token_expires_at` immediately and the renewal decision read it, so a bounce that failed looked complete: the next pass saw a fresh future expiry and never retried, and clearing `last_token_renewal` re-armed nothing. Slots now carry `applied_token_expires_at` — the expiry of the credential the CONNECTION is using — which only advances when a connect or reconnect has carried it. A failed bounce leaves it on the old credential, so the renewal stays due. **P2 — config-committed and start-accepted were conflated.** `connect()` returned one result for both, so a start that failed after the configuration had committed left `applied_config` describing the previous configuration, and the next recreate-or-reuse decision could reuse a client whose trust policy was not the one it believed. `applyConfig()` is now its own wrapper operation; `applied_config` records the commit, activation records the start. The reconnect ladder resets there too rather than in `teardownSlot()` — the old endpoint's history still applies until a replacement configuration actually commits. Two of my own bugs surfaced on hardware while testing this, both fixed here: - `recreateSlotClient()` called the full `teardownSlot()`, which cleared `broker_uri`, the just-minted token and both expiries out from under a configuration that had already been decided, so a recreate handed the SDK an empty URI and an empty token. It now stops the client and swaps the object, touching nothing else, and the apply step refuses to configure a URI that changed under it rather than passing it on. - `stopSlotClient()` quarantined on any non-OK result, but `ESP_FAIL` from `esp_mqtt_client_stop()` means "client is in invalid state", i.e. not started: there was no task to join, the safest state there is. It was observed quarantining healthy clients on hardware. The case that genuinely cannot be proven is a stop that never RETURNS, which cannot surface here at all — it hangs the task, which is what the bridge-level timeout contains. Hardware (Heltec V4, 5 live slots): a reconfigure landing on a connecting client logs `reconfigure during connect - stopping to cancel the attempt`; a broker holding the CONNACK sees the client close the socket and the disabled slot never connects; `wss`→`mqtt`→`wss` recreate cycles reconnect each way; a blackholed endpoint recovers. 499/499 native tests, four envs clean. --- .../src/PsychicMqttClient.cpp | 110 ++++---- lib/PsychicMqttClient/src/PsychicMqttClient.h | 16 ++ src/helpers/MQTTClientState.h | 82 ++++++ src/helpers/bridges/MQTTBridge.cpp | 263 +++++++++++++----- src/helpers/bridges/MQTTBridge.h | 42 ++- .../test_mqtt_client_state.cpp | 89 ++++++ 6 files changed, 458 insertions(+), 144 deletions(-) create mode 100644 src/helpers/MQTTClientState.h create mode 100644 test/test_mqtt_client_state/test_mqtt_client_state.cpp diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp index cec11177a..ccd11341d 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp @@ -371,21 +371,59 @@ bool PsychicMqttClient::connected() return _connected; } -esp_err_t PsychicMqttClient::connect() +esp_err_t PsychicMqttClient::applyConfig() { #if ESP_IDF_VERSION_MAJOR == 5 if (_mqtt_cfg.broker.address.uri == nullptr) - { - ESP_LOGE(TAG, "MQTT URI not set."); - return ESP_ERR_INVALID_STATE; - } - int desired_buffer = _mqtt_cfg.buffer.size > 0 ? _mqtt_cfg.buffer.size : 1024; #else if (_mqtt_cfg.uri == nullptr) +#endif { ESP_LOGE(TAG, "MQTT URI not set."); return ESP_ERR_INVALID_STATE; } + + if (_client == nullptr) + { + // esp_mqtt_client_init() takes the whole configuration, including the + // buffer sizes it allocates once and never resizes. + _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; + return ESP_OK; + } + + if (!_config_dirty) + { + ESP_LOGD(TAG, "applyConfig(): mqtt config unchanged, skipping set_config"); + return ESP_OK; + } + + esp_err_t cfg_result = esp_mqtt_set_config(_client, &_mqtt_cfg); + if (cfg_result != ESP_OK) + { + // Leave _config_dirty set so the next attempt retries the whole + // transaction rather than starting on a partly-updated configuration. + ESP_LOGE(TAG, "applyConfig(): failed to apply mqtt config: %s", esp_err_to_name(cfg_result)); + return cfg_result; + } + _config_dirty = false; + ESP_LOGD(TAG, "applyConfig(): applied mqtt config update"); + return ESP_OK; +} + +esp_err_t PsychicMqttClient::connect() +{ +#if ESP_IDF_VERSION_MAJOR == 5 + int desired_buffer = _mqtt_cfg.buffer.size > 0 ? _mqtt_cfg.buffer.size : 1024; +#else int desired_buffer = _mqtt_cfg.buffer_size > 0 ? _mqtt_cfg.buffer_size : 1024; #endif @@ -404,38 +442,11 @@ esp_err_t PsychicMqttClient::connect() } } - 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 - { - ESP_LOGD(TAG, "connect(): mqtt config unchanged, skipping set_config"); - } + // A failed config transaction 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. + esp_err_t cfg_result = applyConfig(); + if (cfg_result != ESP_OK) return cfg_result; esp_err_t start_result = esp_mqtt_client_start(_client); if (start_result == ESP_OK) @@ -458,24 +469,11 @@ esp_err_t PsychicMqttClient::reconnect() ESP_LOGW(TAG, "MQTT client not initialized, cannot reconnect."); return ESP_ERR_INVALID_STATE; } - if (_config_dirty) - { - // 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); - if (cfg_result != ESP_OK) - { - 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"); - } + // A failed config update aborts the reconnect: reconnecting on the previous + // config was how a renewed token silently failed to reach the connection. + esp_err_t cfg_result = applyConfig(); + if (cfg_result != ESP_OK) return cfg_result; + esp_err_t r = esp_mqtt_client_reconnect(_client); if (r != ESP_OK) { diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.h b/lib/PsychicMqttClient/src/PsychicMqttClient.h index 5d214679e..93b85a437 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.h +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.h @@ -345,6 +345,22 @@ public: */ bool connected(); + /** + * @brief Commits the pending configuration to the SDK client, creating it + * if it does not exist yet, WITHOUT starting it. + * + * Exists so a caller can tell "the configuration is now what I asked for" + * from "the client started": esp_mqtt_client_start() can fail after the + * configuration has been committed, and a caller that cannot distinguish + * those records a stale view of what the SDK actually holds. + * + * @return ESP_OK when the SDK client holds this configuration. + * ESP_ERR_INVALID_STATE if no URI is set, ESP_ERR_NO_MEM if the + * client could not be allocated, or the error from + * esp_mqtt_set_config() (which leaves the update pending). + */ + esp_err_t applyConfig(); + /** * @brief Connects the MQTT client to the server. * diff --git a/src/helpers/MQTTClientState.h b/src/helpers/MQTTClientState.h new file mode 100644 index 000000000..dd1fee44d --- /dev/null +++ b/src/helpers/MQTTClientState.h @@ -0,0 +1,82 @@ +#pragma once + +#include + +// What a slot's MQTT client is doing, as opposed to whether its network is up, +// and the two decisions that hang off it. +// +// Pure logic: no Arduino, esp-mqtt or bridge dependency, so the shutdown +// contract below is a host test rather than a hardware run. +// +// `client->connected()` answers "is the network up" and was being used for "is +// there anything to stop": a client resolving DNS, negotiating TLS or waiting +// after a failed CONNECT reports not-connected, so teardown skipped it and left +// its task running (F04). + +enum class MqttClientState : uint8_t { + Absent = 0, // no client object + Configured, // client allocated, never started + Starting, // start/reconnect requested, awaiting CONNECTED + Connected, // CONNECTED received + Disconnected, // started, no session (our reconnect ladder governs it) + Stopped, // stop completed; the SDK task is joined and gone + Quarantined, // stop failed: the SDK task was NOT joined. Never destroy, + // never reuse, never free anything it still points at. +}; + +static inline const char* mqttClientStateName(MqttClientState s) { + switch (s) { + case MqttClientState::Absent: return "absent"; + case MqttClientState::Configured: return "configured"; + case MqttClientState::Starting: return "starting"; + case MqttClientState::Connected: return "connected"; + case MqttClientState::Disconnected: return "disconnected"; + case MqttClientState::Stopped: return "stopped"; + case MqttClientState::Quarantined: return "quarantined"; + } + return "?"; +} + +// The client has been started and not proven stopped: it may still own a task, +// a socket and a TLS context. Quarantined is deliberately NOT live — it is +// worse: it may own them and we can never find out. +static inline bool mqttClientStateIsLive(MqttClientState s) { + return s == MqttClientState::Starting || s == MqttClientState::Connected || + s == MqttClientState::Disconnected; +} + +// A connection attempt is in flight, so an event for it can still arrive. +// Closing the transport is not enough here: PsychicMqttClient::softDisconnect() +// returns immediately when the client is not yet connected, which means an +// in-progress DNS/TLS/CONNECT is left to complete on its own. Cancelling one +// requires a real stop. +static inline bool mqttClientStateHasAttemptInFlight(MqttClientState s) { + return s == MqttClientState::Starting; +} + +// The SDK task is known to be gone (or never existed), so the object may be +// destroyed and anything it pointed at may be freed. +static inline bool mqttClientStateIsProvenStopped(MqttClientState s) { + return s == MqttClientState::Absent || s == MqttClientState::Configured || + s == MqttClientState::Stopped; +} + +// The bridge's cooperative stop may only be acknowledged when EVERY client is +// proven stopped. +// +// This is the invariant the whole shutdown contract rests on: the owner treats +// the acknowledgement as proof that the task destroyed its clients, and only +// then frees the queue and buffers and allows a restart. One client whose +// esp_mqtt_client_stop() never completed makes that false — its SDK task may +// still be running — so the acknowledgement must be withheld and the bridge +// left in its unproven state, however many other slots stopped cleanly. +static inline bool mqttStopMayBeAcknowledged(const MqttClientState* states, int count) { + // An empty set is trivially proven (no slots, nothing to stop); a null array + // with a nonzero count is a caller bug and must not read as proof. + if (count <= 0) return true; + if (states == nullptr) return false; + for (int i = 0; i < count; i++) { + if (!mqttClientStateIsProvenStopped(states[i])) return false; + } + return true; +} diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index da0fa3022..9e176c4b4 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -463,19 +463,6 @@ void MQTTBridge::applyWifiPowerSave() { #endif } -const char* MQTTBridge::clientStateName(ClientState st) { - switch (st) { - case ClientState::Absent: return "absent"; - case ClientState::Configured: return "configured"; - case ClientState::Starting: return "starting"; - case ClientState::Connected: return "connected"; - case ClientState::Disconnected: return "disconnected"; - case ClientState::Stopped: return "stopped"; - case ClientState::Quarantined: return "quarantined"; - } - return "?"; -} - bool MQTTBridge::stopUnprovenLatched() { return s_stop_unproven; } uint8_t MQTTBridge::getLastWifiDisconnectReason() { return s_wifi_disconnect_reason; } @@ -1444,6 +1431,29 @@ void MQTTBridge::mqttTaskLoop() { teardownSlot(i); } destroySlotClients(); + + // The acknowledgement means "every client is destroyed, nothing I own is + // still running". One client whose esp_mqtt_client_stop() never completed + // makes that false — destroySlotClients() deliberately skipped it and its + // SDK task may still be executing — so the ack must be WITHHELD, whatever + // the other slots did. Publishing it anyway would tell the owner to free + // the queue and buffers and allow a restart while that task lives, which + // is the ownership ambiguity this whole contract exists to remove. + MqttClientState states[RUNTIME_MQTT_SLOTS]; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) states[i] = _slots[i].client_state; + if (!mqttStopMayBeAcknowledged(states, RUNTIME_MQTT_SLOTS)) { + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (!mqttClientStateIsProvenStopped(states[i])) { + MQTT_DEBUG_PRINTLN("MQTT%d stop unproven (%s) - withholding the bridge stop ack", + i + 1, mqttClientStateName(states[i])); + } + } + // Leave _teardown_complete false: the trampoline publishes no ack, the + // owner's stop times out into StopUnproven, and nothing is released or + // restarted for the rest of this boot. + return; + } + // Record that the ordered teardown finished, but do NOT publish the ack // here: the owner treats the ack as permission to free everything this // task can reach, and between this point and vTaskDelete(nullptr) the task @@ -1881,6 +1891,7 @@ void MQTTBridge::releaseSlotAuthToken(int index) { slot.auth_token = static_cast( MQTTRuntimeBufferLifecycle::release(slot.auth_token, psram_free)); slot.token_expires_at = 0; + slot.applied_token_expires_at = 0; slot.last_token_renewal = 0; } @@ -1902,14 +1913,11 @@ void MQTTBridge::destroySlotClients() { } if (clientStateIsLive(slot.client_state)) { - const esp_err_t r = slot.client->disconnect(); - if (r != ESP_OK && r != ESP_ERR_TIMEOUT) { - MQTT_DEBUG_PRINTLN("MQTT%d stop FAILED during shutdown (%s) - not destroying", i + 1, - esp_err_to_name(r)); - slot.client_state = ClientState::Quarantined; + stopSlotClient(i); + if (slot.client_state == ClientState::Quarantined) { + MQTT_DEBUG_PRINTLN("MQTT%d not destroyed: stop unproven during shutdown", i + 1); continue; } - slot.client_state = ClientState::Stopped; #ifdef ESP_PLATFORM vTaskDelay(pdMS_TO_TICKS(50)); #else @@ -1996,17 +2004,17 @@ bool MQTTBridge::setupSlot(int index) { // them. setCredentials / setServer below overwrite the config fields in place // before connect() restarts the ESP-IDF client. if (slot.initial_connect_done) { - // Close the transport (keeping the task) if this client is live, for the - // same reason applySlotPreset() does: a handshake in flight against the - // previous endpoint must not complete after the new config is applied. + // Same rule as applySlotPreset(): a handshake in flight against the previous + // endpoint must not survive the new configuration, and only a stop can + // cancel one. One helper so the two paths cannot drift apart. if (clientStateIsLive(slot.client_state)) { - const esp_err_t r = slot.client->softDisconnect(); - if (r != ESP_OK) { - MQTT_DEBUG_PRINTLN("MQTT%d re-apply: transport did not close cleanly (%s)", - index + 1, esp_err_to_name(r)); - } - slot.client_state = ClientState::Disconnected; + closeLiveClientForReconfigure(index); slot.generation++; + if (slot.client_state == ClientState::Quarantined) { + MQTT_DEBUG_PRINTLN("MQTT%d: cannot re-apply, stop unproven", index + 1); + slot.last_reconnect_attempt = millis(); + return false; + } } // The config fields are NOT cleared here any more. Two reasons: this block // only ran when initial_connect_done was set, which teardownSlot() clears @@ -2018,11 +2026,10 @@ bool MQTTBridge::setupSlot(int index) { if (slot.auth_token) slot.auth_token[0] = '\0'; slot.connected = false; slot.token_expires_at = 0; + slot.applied_token_expires_at = 0; slot.last_token_renewal = 0; - slot.reconnect_backoff = 0; - slot.max_backoff_failures = 0; - slot.circuit_breaker_tripped = false; slot.last_reconnect_attempt = 0; + // Ladder reset happens after the new configuration commits, below. // The refusal that set this belonged to the credentials being cleared here. _slot_force_jwt_mint[index] = false; } @@ -2243,6 +2250,18 @@ bool MQTTBridge::setupSlot(int index) { // credentials are written as "" rather than left alone: NULL means "leave // unchanged" to esp_mqtt_set_config(), while an empty string clears the // stored value and leaves the CONNECT's username flag clear. + // cfg_uri points at storage the slot owns (slot.broker_uri) or at flash + // (preset->server_url), and esp-mqtt keeps the pointer, so it must still hold + // what `desired` was built from. This caught a real bug: an earlier version of + // recreateSlotClient() ran the full slot teardown, which cleared broker_uri + // out from under the configuration that had already been decided, and the SDK + // was then handed an empty URI. + if (cfg_uri == nullptr || strcmp(cfg_uri, desired.uri) != 0) { + MQTT_DEBUG_PRINTLN("MQTT%d: broker URI changed under the config apply - not connecting", + index + 1); + slot.last_reconnect_attempt = millis(); + return false; + } slot.client->setServer(cfg_uri); if (desired.trust == MqttTrust::PemCert) { slot.client->setCACert(desired.pem); @@ -2255,6 +2274,29 @@ bool MQTTBridge::setupSlot(int index) { slot.client->setCredentials(mqttFieldOrEmpty(desired.username), mqttFieldOrEmpty(desired.password)); + // Commit the configuration as its own transaction, and record it as applied + // the moment it commits — not after the client starts. The SDK really does + // hold this configuration once set_config/init returns OK, so a start that + // fails afterwards must not leave the bridge believing the PREVIOUS config is + // applied: the next recreate-or-reuse decision would compare against it and + // could reuse a client whose trust policy is not the one it thinks. + const esp_err_t config_result = slot.client->applyConfig(); + if (config_result != ESP_OK) { + MQTT_DEBUG_PRINTLN("MQTT%d config transaction failed (%s) - will retry", index + 1, + esp_err_to_name(config_result)); + slot.last_reconnect_attempt = millis(); + return false; + } + slot.applied_config = desired; + + // The new endpoint's history starts here, now that the configuration it + // belongs to is committed. Resetting the ladder in teardownSlot() cleared it + // before there was any replacement config — so a reconfigure that then failed + // to commit lost the backoff state that still applied to the old endpoint. + slot.reconnect_backoff = 0; + slot.max_backoff_failures = 0; + slot.circuit_breaker_tripped = false; + // Start or reconnect according to what the SDK client actually is, not what // the network is doing. esp_mqtt_client_start() fails on an already-started // client, so a reconfigure that kept the task (TeardownReason::Reconfigure) @@ -2268,9 +2310,8 @@ bool MQTTBridge::setupSlot(int index) { return false; } // reconnectSlotClient() moved the state to Starting and bumped the generation. - // Record what the client is now configured with, so the next apply can tell - // an in-place change from a structural one. - slot.applied_config = desired; + // applied_config was recorded at the commit above; activation records that the + // client also started. slot.initial_connect_done = true; return true; } @@ -2288,11 +2329,25 @@ bool MQTTBridge::recreateSlotClient(int index) { MQTTSlot& slot = _slots[index]; if (slot.client == nullptr) return ensureSlotClient(index); - teardownSlot(index, TeardownReason::Disable); // stop, do not just close the transport + // Stop the client only. NOT teardownSlot(): that is the "this slot is going + // away" path, and it clears the very state the in-flight setup is holding — + // broker_uri (which cfg_uri points at), the token just minted for this + // configuration, and both expiries. Recreation swaps the client object under + // a configuration that has already been decided; it must not disturb it. + if (clientStateIsLive(slot.client_state)) { + stopSlotClient(index); + slot.generation++; + #ifdef ESP_PLATFORM + vTaskDelay(pdMS_TO_TICKS(50)); + #else + delay(50); + #endif + } if (slot.client_state == ClientState::Quarantined) { MQTT_DEBUG_PRINTLN("MQTT%d cannot recreate: previous stop unproven", index + 1); return false; } + slot.connected = false; delete slot.client; slot.client = nullptr; @@ -2308,6 +2363,80 @@ bool MQTTBridge::recreateSlotClient(int index) { // the client object alive so a subsequent setupSlot() can reuse its mbedTLS // context. This is called both on reconfigure (preset change) and at shutdown; // destruction of the underlying client happens once in destroySlotClients(). +// Close a live client for a reconfigure: cheap where that is safe, a real stop +// where it is not. +// +// softDisconnect() cannot cancel a connection that has not completed. +// esp_mqtt_client_disconnect() acts on a live session, and the wrapper returns +// ESP_OK immediately when the client is not connected — so for a client +// mid-DNS/TLS/CONNECT nothing is closed and nothing is cancelled. That attempt +// would run to completion against the OLD endpoint and deliver its CONNECTED +// event after the new configuration was applied, indistinguishable from the new +// attempt's (F04): esp-mqtt gives events no generation of their own, and the +// callbacks are registered once per client, so there is nothing in the event to +// tell them apart. The only way to separate them is to make sure the old +// attempt is dead first, which means stopping the client and joining its task. +// +// That costs a stop/start cycle, but only when an operator reconfigures a slot +// *while it is connecting*. A Connected client still takes the cheap route, +// which is where the fragmentation argument actually applies. +void MQTTBridge::closeLiveClientForReconfigure(int index) { + MQTTSlot& slot = _slots[index]; + if (mqttClientStateHasAttemptInFlight(slot.client_state)) { + MQTT_DEBUG_PRINTLN("MQTT%d reconfigure during connect - stopping to cancel the attempt", + index + 1); + stopSlotClient(index); + return; + } + const esp_err_t r = slot.client->softDisconnect(); + if (r != ESP_OK) { + MQTT_DEBUG_PRINTLN("MQTT%d reconfigure: transport did not close cleanly (%s)", + index + 1, esp_err_to_name(r)); + } + slot.client_state = ClientState::Disconnected; +} + +// Stop a live client and record whether the stop was proven. ESP_ERR_TIMEOUT +// means no DISCONNECTED event arrived but esp_mqtt_client_stop() itself +// returned OK, so the SDK task is joined and the object is safe to reuse; any +// other error means it was NOT joined, and that client is out of service for +// the rest of the boot. +void MQTTBridge::stopSlotClient(int index) { + MQTTSlot& slot = _slots[index]; + const esp_err_t r = slot.client->disconnect(); + + // What the SDK's results actually mean here (mqtt_client.h documents + // esp_mqtt_client_stop as "ESP_OK on success, ESP_ERR_INVALID_ARG on wrong + // initialization, ESP_FAIL if client is in invalid state"): + // + // ESP_OK the SDK task was joined. + // ESP_ERR_TIMEOUT our own wrapper's code for "no DISCONNECTED event, but + // the stop itself returned OK" — the task is still joined. + // ESP_FAIL the client was not started. There was no task to join, + // which is the safest state of all. Treating this as a + // failure quarantined perfectly healthy clients — observed + // on hardware when a reconfigure landed on a slot whose + // connect attempt had already failed. + // + // The case that genuinely cannot be proven is a stop that never RETURNS: the + // SDK waits on its API mutex and the task's stopped event without a bound. + // That does not surface here at all — it hangs this task, which is exactly + // what the bridge-level stop timeout (StopUnproven) exists to contain. So + // with IDF 4.4's contract the branch below should be unreachable; it stays + // because acting on an unexpected result by not touching the client again is + // the only safe response if that ever changes. + if (r == ESP_OK || r == ESP_ERR_TIMEOUT || r == ESP_FAIL) { + if (r == ESP_FAIL) { + MQTT_DEBUG_PRINTLN("MQTT%d stop: client was not started (nothing to join)", index + 1); + } + slot.client_state = ClientState::Stopped; + return; + } + MQTT_DEBUG_PRINTLN("MQTT%d stop returned %s - client quarantined for this boot", + index + 1, esp_err_to_name(r)); + slot.client_state = ClientState::Quarantined; +} + void MQTTBridge::teardownSlot(int index, TeardownReason reason) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[index]; @@ -2319,30 +2448,9 @@ void MQTTBridge::teardownSlot(int index, TeardownReason reason) { // already replaced or switched off (F04). if (slot.client && clientStateIsLive(slot.client_state)) { if (reason == TeardownReason::Reconfigure) { - // Close the transport, keep the task. The in-flight handshake cannot - // complete against the old endpoint any more, and the esp-mqtt task's - // 6 KiB stack does not get returned into the hole the mbedTLS record - // buffers just vacated (the documented fragmentation driver). - const esp_err_t r = slot.client->softDisconnect(); - if (r != ESP_OK) { - MQTT_DEBUG_PRINTLN("MQTT%d reconfigure: transport did not close cleanly (%s)", - index + 1, esp_err_to_name(r)); - } - slot.client_state = ClientState::Disconnected; + closeLiveClientForReconfigure(index); } else { - const esp_err_t r = slot.client->disconnect(); - if (r == ESP_OK || r == ESP_ERR_TIMEOUT) { - // ESP_ERR_TIMEOUT: no DISCONNECTED event, but the stop itself returned - // OK, so the SDK task is joined and the object is safe to reuse. - slot.client_state = ClientState::Stopped; - } else { - // The stop did not complete: the SDK task was not joined. Never touch - // this client again — not to reuse it, not to destroy it, and do not - // free the token buffer its config still points at. - MQTT_DEBUG_PRINTLN("MQTT%d stop FAILED (%s) - client quarantined for this boot", - index + 1, esp_err_to_name(r)); - slot.client_state = ClientState::Quarantined; - } + stopSlotClient(index); } slot.generation++; #ifdef ESP_PLATFORM @@ -2359,10 +2467,13 @@ void MQTTBridge::teardownSlot(int index, TeardownReason reason) { slot.initial_connect_done = false; slot.broker_uri[0] = '\0'; slot.token_expires_at = 0; + slot.applied_token_expires_at = 0; slot.last_token_renewal = 0; - slot.reconnect_backoff = 0; - slot.max_backoff_failures = 0; - slot.circuit_breaker_tripped = false; + // The reconnect ladder is deliberately NOT cleared here. It describes the + // endpoint this slot has been failing against, and that history still applies + // until a replacement configuration actually commits — setupSlot() clears it + // there. Clearing it on teardown meant a reconfigure that never committed + // (bad URI, failed config transaction) also forgave a tripped breaker. slot.last_reconnect_attempt = 0; slot.last_log_time = 0; slot.last_deferred_log_ms = 0; @@ -2397,6 +2508,9 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { if (r == ESP_OK) { slot.client_state = ClientState::Starting; slot.generation++; + // The attempt carries whatever credential is configured right now, so this + // is the point at which a freshly minted token becomes the one in use. + slot.applied_token_expires_at = slot.token_expires_at; } return r; } @@ -2513,9 +2627,15 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // exp on live sessions (waev's 55-minute tokens). const unsigned long renewal_buffer = MQTTConnectionPolicy::renewalBufferSecs( static_cast(slotTokenLifetime(index))); + // Against the APPLIED expiry — the credential the live connection is using — + // not the one sitting in the token buffer. Minting updates the buffer's + // expiry immediately, so reading that here meant a renewal whose bounce + // failed looked complete: the next pass saw a fresh future expiry, decided + // no renewal was due, and never retried the bounce. The connection stayed on + // the old credential until the broker enforced exp (F06). bool token_needs_renewal = MQTTConnectionPolicy::tokenNeedsRenewal( time_synced, static_cast(current_time), - static_cast(slot.token_expires_at), + static_cast(slot.applied_token_expires_at), static_cast(renewal_buffer)); // Throttle renewal attempts to once per minute @@ -2525,7 +2645,7 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns if (token_needs_renewal && can_attempt_renewal) { slot.last_token_renewal = now_millis; - unsigned long old_token_expires_at = slot.token_expires_at; + unsigned long old_token_expires_at = slot.applied_token_expires_at; if (createSlotAuthToken(index)) { MQTT_DEBUG_PRINTLN("MQTT%d token renewed", index + 1); @@ -2549,6 +2669,11 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns if (!exp_forces_bounce && old_token_expired_or_imminent && slot.client->connected()) { MQTT_DEBUG_PRINTLN("MQTT%d token renewed, no bounce (broker does not enforce exp)", index + 1); + // This broker does not act on exp, so the live session's older + // credential is not a problem and the renewal is complete. Recording + // it stops the renewal from staying due and re-attempting every + // minute for the rest of the token's life. + slot.applied_token_expires_at = slot.token_expires_at; } if (exp_forces_bounce || !slot.client->connected()) { // Disconnect + reconnect with fresh credentials, reusing existing client @@ -2585,9 +2710,13 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // 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", + // Leave applied_token_expires_at pointing at the OLD credential: + // that is still what the connection is using, so the renewal stays + // due and the next pass retries the bounce (paced by the one-a- + // minute renewal throttle). The freshly minted token stays in the + // buffer and will be used by that retry, or by the next reconnect. + MQTT_DEBUG_PRINTLN("MQTT%d renewal bounce failed (%s) - still due, retrying", index + 1, esp_err_to_name(bounce_result)); - slot.last_token_renewal = 0; return; } MQTT_TRACE_HEAP("renewal:after-reconnect", index); @@ -2601,10 +2730,14 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns } else { // Token renewed but old one still valid — just update credentials for next reconnect slot.client->setCredentials(_jwt_username, slot.auth_token); + // Staged for the next reconnect, and the credential in use is still + // valid, so the renewal decision is settled for this token. + slot.applied_token_expires_at = slot.token_expires_at; } } else { MQTT_DEBUG_PRINTLN("MQTT%d token renewal failed", index + 1); slot.token_expires_at = 0; + slot.applied_token_expires_at = 0; } return; // Token renewal handled connect; skip backoff logic below } diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 6c81cfc2b..d919211de 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -12,6 +12,7 @@ #include "helpers/MQTTLifecycle.h" #include "helpers/AlertFaultPolicy.h" #include "helpers/MQTTEffectiveConfig.h" +#include "helpers/MQTTClientState.h" #include #ifdef WITH_SNMP @@ -105,29 +106,12 @@ private: static const uint32_t kNtpMinValidEpoch = 1767225600UL; // 2026-01-01 UTC static const uint32_t kNtpMaxValidEpoch = 4102444800UL; // 2100-01-01 UTC - // What the SDK client is doing, as opposed to whether the network is up. - // `client->connected()` answers the second question and was being used for - // the first: a client resolving DNS, negotiating TLS or waiting after a - // failed CONNECT reports not-connected, so teardown skipped it and left its - // task running (F04). Absent is 0 so a memset-initialised slot is correct. - enum class ClientState : uint8_t { - Absent = 0, // no client object - Configured, // client allocated, never started - Starting, // start/reconnect requested, awaiting CONNECTED - Connected, // CONNECTED received - Disconnected, // started, no session (our reconnect ladder governs it) - Stopped, // stop completed; the SDK task is joined and gone - Quarantined, // stop failed: the SDK task was NOT joined. Never destroy, - // never reuse, never free anything it still points at. - }; - - static const char* clientStateName(ClientState s); - // True while the SDK client has been started and not proven stopped, i.e. - // while it may still own a task, a socket and a TLS context. - static bool clientStateIsLive(ClientState s) { - return s == ClientState::Starting || s == ClientState::Connected || - s == ClientState::Disconnected; - } + // Per-slot SDK client state and the shutdown contract live in + // MQTTClientState.h so both are host-testable; ClientState is an alias so the + // bridge code reads naturally. + using ClientState = MqttClientState; + static const char* clientStateName(ClientState s) { return mqttClientStateName(s); } + static bool clientStateIsLive(ClientState s) { return mqttClientStateIsLive(s); } // Connection slot - each slot holds one MQTT connection struct MQTTSlot { @@ -158,7 +142,14 @@ private: // esp-mqtt re-reads it whenever a later connect() re-applies a dirtied config. // Freed only alongside the client in destroySlotClients(). char* auth_token; // nullptr or empty string = no valid token + // Two expiries, deliberately. token_expires_at describes the token in the + // buffer (minting updates it immediately); applied_token_expires_at + // describes the credential the CONNECTION is actually using, and only + // advances when a connect/reconnect has carried it. The renewal decision + // reads the applied one, so a renewal whose bounce failed stays due and is + // retried instead of looking complete (F06). unsigned long token_expires_at; + unsigned long applied_token_expires_at; unsigned long last_token_renewal; // Custom broker settings (only used when preset_name is "custom") @@ -547,6 +538,11 @@ private: // with it. Here the stop IS the point. enum class TeardownReason : uint8_t { Reconfigure, Disable }; void teardownSlot(int index, TeardownReason reason = TeardownReason::Disable); + // Close a live client for a reconfigure: softDisconnect where that is enough, + // a real stop where an in-flight connection attempt has to be cancelled. + void closeLiveClientForReconfigure(int index); + // Stop a live client, recording Stopped (proven) or Quarantined (not joined). + void stopSlotClient(int index); // Reconnect a slot, starting it instead when the client is stopped (reconnect() is a // no-op on a stopped client). See the definition. // ESP_OK when the reconnect/start was accepted by the SDK. A local failure diff --git a/test/test_mqtt_client_state/test_mqtt_client_state.cpp b/test/test_mqtt_client_state/test_mqtt_client_state.cpp new file mode 100644 index 000000000..9b0f7c36c --- /dev/null +++ b/test/test_mqtt_client_state/test_mqtt_client_state.cpp @@ -0,0 +1,89 @@ +#include "helpers/MQTTClientState.h" + +#include + +#include +#include + +namespace { + +const MqttClientState kAllStates[] = { + MqttClientState::Absent, MqttClientState::Configured, + MqttClientState::Starting, MqttClientState::Connected, + MqttClientState::Disconnected, MqttClientState::Stopped, + MqttClientState::Quarantined, +}; + +bool ack(std::vector states) { + return mqttStopMayBeAcknowledged(states.data(), (int)states.size()); +} + +} // namespace + +TEST(MqttClientState, LiveMeansStartedAndNotProvenStopped) { + EXPECT_TRUE(mqttClientStateIsLive(MqttClientState::Starting)); + EXPECT_TRUE(mqttClientStateIsLive(MqttClientState::Connected)); + EXPECT_TRUE(mqttClientStateIsLive(MqttClientState::Disconnected)); + + EXPECT_FALSE(mqttClientStateIsLive(MqttClientState::Absent)); + EXPECT_FALSE(mqttClientStateIsLive(MqttClientState::Configured)); + EXPECT_FALSE(mqttClientStateIsLive(MqttClientState::Stopped)); + // Quarantined is not "live": it is worse. It may own a task and we can never + // find out, which is why it is not simply lumped in with the live states. + EXPECT_FALSE(mqttClientStateIsLive(MqttClientState::Quarantined)); +} + +TEST(MqttClientState, OnlyStartingHasAnAttemptInFlight) { + for (MqttClientState s : kAllStates) { + EXPECT_EQ(s == MqttClientState::Starting, mqttClientStateHasAttemptInFlight(s)) + << mqttClientStateName(s); + } +} + +TEST(MqttClientState, ProvenStoppedIsTheComplementOfLiveAndQuarantined) { + for (MqttClientState s : kAllStates) { + const bool proven = mqttClientStateIsProvenStopped(s); + EXPECT_EQ(!mqttClientStateIsLive(s) && s != MqttClientState::Quarantined, proven) + << mqttClientStateName(s); + } +} + +// The invariant the shutdown contract rests on: the acknowledgement means the +// task destroyed its clients, so it may not be published while any client's +// stop is unproven. +TEST(MqttClientState, StopIsAcknowledgedOnlyWhenEveryClientIsProvenStopped) { + EXPECT_TRUE(ack({})); // no slots configured at all + EXPECT_TRUE(ack({MqttClientState::Absent, MqttClientState::Stopped, + MqttClientState::Configured})); + + // One quarantined client withholds the acknowledgement, whatever the others did. + EXPECT_FALSE(ack({MqttClientState::Quarantined})); + EXPECT_FALSE(ack({MqttClientState::Stopped, MqttClientState::Stopped, + MqttClientState::Quarantined, MqttClientState::Absent})); + EXPECT_FALSE(ack({MqttClientState::Quarantined, MqttClientState::Absent, + MqttClientState::Absent, MqttClientState::Absent, + MqttClientState::Absent, MqttClientState::Absent})); + + // So does a client that is merely still live: teardown did not finish it. + EXPECT_FALSE(ack({MqttClientState::Stopped, MqttClientState::Connected})); + EXPECT_FALSE(ack({MqttClientState::Starting})); + EXPECT_FALSE(ack({MqttClientState::Disconnected})); + + // An empty set is trivially proven; a null array with a nonzero count is a + // caller bug and must never read as proof. + EXPECT_TRUE(mqttStopMayBeAcknowledged(nullptr, 0)); + EXPECT_FALSE(mqttStopMayBeAcknowledged(nullptr, 3)); +} + +TEST(MqttClientState, EveryStateHasAName) { + for (MqttClientState s : kAllStates) { + ASSERT_NE(nullptr, mqttClientStateName(s)); + EXPECT_GT(strlen(mqttClientStateName(s)), 0u); + } + EXPECT_STREQ("quarantined", mqttClientStateName(MqttClientState::Quarantined)); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}