From a1a8c19459144e8ab05e5aca07a8fcc49df042fe Mon Sep 17 00:00:00 2001 From: agessaman Date: Wed, 9 Sep 2026 17:46:20 -0700 Subject: [PATCH] fix(mqtt): decide slot teardown by SDK lifecycle state, not connectivity (F04) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `client->connected()` answers "is the network up", and it 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 `teardownSlot()` skipped it: selecting `none` cleared the bridge flags while the SDK task kept running, and when the in-flight handshake completed the callback marked the disabled slot connected and scheduled its status. Each slot now carries the client's SDK lifecycle state (Absent, Configured, Starting, Connected, Disconnected, Stopped, Quarantined) and a generation counter for the logs. The bridge task owns that state, which makes it the authority the callbacks consult: - a CONNECTED event for a slot that is disabled, never started, or already stopped is logged and dropped instead of marking the slot connected; - teardown stops any *live* client, so a mid-handshake client can no longer outlive its configuration. Teardown also gains a reason, because "stop the task" and "close the transport" are different needs: - `Disable` (preset `none`, shutdown) stops the client — the stop is the point; - `Reconfigure` closes the transport with `softDisconnect()` and keeps the esp-mqtt task. Stopping it there would return its 6 KiB stack into the hole the two 16 KiB mbedTLS record buffers just vacated, which is the fork's documented internal-heap fragmentation driver — and per the soak campaign's own conclusion, feeding `softDisconnect()` into the reconfigure path is the fix for it, not serialisation. The campaign closed 2026-08-20, so the reconfigure churn is no longer anyone's measurement lever. A stop that does not complete now quarantines that client: its SDK task was never joined, so it is never reused, never destroyed, and the token buffer its config still points at is never freed. `get mqttN.diag` reports `quarantined`. Two knock-ons this required: - `setupSlot()` starts or reconnects according to the SDK state. `esp_mqtt_client_start()` fails on an already-started client, and now that its result is honoured, a reconfigure that kept the task would otherwise leave the slot permanently unactivated. - the active-slot cap counts resource holders as well as configured ones. It keyed off `initial_connect_done`, which teardown clears, so a started or quarantined client stopped counting against the cap and a board could oversubscribe past the concurrent-TLS limit the cap exists to enforce. --- src/helpers/bridges/MQTTBridge.cpp | 172 ++++++++++++++++++++++++++--- src/helpers/bridges/MQTTBridge.h | 41 ++++++- 2 files changed, 194 insertions(+), 19 deletions(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 97ecd867..3e3cdf76 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -463,6 +463,19 @@ 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; } @@ -561,6 +574,11 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) // is configured but missing a token/IATA/credential, so it was never set up and // has no client yet. Previously reported "disc", which read as a network fault. state = "wait"; + } else if (slot.client && slot.client_state == ClientState::Quarantined) { + // Its stop did not complete, so the SDK task was never joined: the slot is + // out of service for the rest of the boot and its resources are retained + // on purpose. + state = "quarantined"; } else if (!slot.client) { // Ready to connect but the client object could not be allocated. state = "no client"; @@ -1723,10 +1741,24 @@ bool MQTTBridge::ensureSlotClient(int index) { MQTT_DEBUG_PRINTLN("MQTT%d: out of memory allocating client", index + 1); return false; } + slot.client_state = ClientState::Configured; slot.client->setAutoReconnect(false); // we handle reconnect with our own backoff slot.client->onConnect([this, index](bool sessionPresent) { + // A CONNECT started before this slot was disabled or reconfigured can still + // complete afterwards. Accepting it marked a slot connected that the + // operator had switched off, scheduled its status publish, and published + // through the old session (F04). The bridge task owns client_state, so it + // is the authority on whether this event was asked for. + const ClientState st = _slots[index].client_state; + if (!_slots[index].enabled || !(st == ClientState::Starting || st == ClientState::Disconnected)) { + MQTT_DEBUG_PRINTLN("MQTT%d ignoring late CONNECTED (state=%s, gen=%lu, enabled=%d)", + index + 1, clientStateName(st), + (unsigned long)_slots[index].generation, (int)_slots[index].enabled); + return; + } MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); + _slots[index].client_state = ClientState::Connected; _slots[index].connected = true; _slot_force_jwt_mint[index] = false; // NOTE: reconnect_backoff / max_backoff_failures are NOT reset here. @@ -1759,6 +1791,11 @@ bool MQTTBridge::ensureSlotClient(int index) { }); slot.client->onDisconnect([this, index](bool sessionPresent) { MQTT_DEBUG_PRINTLN("MQTT%d disconnected", index + 1); + // Only a live client's disconnect is news. One arriving for a client we + // already stopped (or quarantined) must not resurrect its state. + if (clientStateIsLive(_slots[index].client_state)) { + _slots[index].client_state = ClientState::Disconnected; + } _slots[index].disconnect_count++; if (_slots[index].first_disconnect_time == 0) { _slots[index].first_disconnect_time = millis(); @@ -1841,20 +1878,42 @@ void MQTTBridge::releaseSlotAuthToken(int index) { void MQTTBridge::destroySlotClients() { for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { MQTTSlot& slot = _slots[i]; - if (slot.client != nullptr) { - if (slot.client->connected()) { - slot.client->disconnect(); + if (slot.client == nullptr) { + releaseSlotAuthToken(i); + continue; + } + + if (slot.client_state == ClientState::Quarantined) { + // A previous stop failed, so this client's SDK task was never joined. + // Deleting it now is the use-after-free F01 is about, and freeing its + // token would pull the buffer out from under a config the task may still + // read. Leak both, deliberately, until the node reboots. + MQTT_DEBUG_PRINTLN("MQTT%d client quarantined - not destroyed, token retained", i + 1); + continue; + } + + 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; + continue; } + slot.client_state = ClientState::Stopped; #ifdef ESP_PLATFORM vTaskDelay(pdMS_TO_TICKS(50)); #else delay(50); #endif - delete slot.client; - slot.client = nullptr; } - // Unconditional: only now is the token unreachable from the client's stored - // config, and a token without a client would otherwise leak. + + delete slot.client; + slot.client = nullptr; + slot.client_state = ClientState::Absent; + slot.generation++; + // Only now is the token unreachable from the client's stored config, and a + // token without a client would otherwise leak. releaseSlotAuthToken(i); } } @@ -1862,7 +1921,18 @@ void MQTTBridge::destroySlotClients() { int MQTTBridge::activatedSlotCount() const { int n = 0; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (_slots[i].enabled && _slots[i].initial_connect_done) n++; + const MQTTSlot& s = _slots[i]; + // Two ways to hold a position, and the cap must respect both. The original + // one is the configuration view: enabled and through a successful setup. + // The second is the resource view: a client that has been started still + // owns a task, a socket and an mbedTLS context — which is what the cap + // actually protects — and a quarantined one owns them for the rest of the + // boot. `initial_connect_done` alone missed those, because teardown clears + // it, so a board could oversubscribe past _max_active_slots. + const bool holds_config_position = s.enabled && s.initial_connect_done; + const bool holds_resources = s.client != nullptr && + (clientStateIsLive(s.client_state) || s.client_state == ClientState::Quarantined); + if (holds_config_position || holds_resources) n++; } return n; } @@ -1909,8 +1979,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) { - if (slot.client->connected()) { - slot.client->disconnect(); + // 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. + 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; + slot.generation++; } // Clear TLS verification fields so a stale CA-bundle attach or cert // pointer from a prior preset doesn't override the new one. @@ -2097,13 +2176,20 @@ bool MQTTBridge::setupSlot(int index) { // 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(); + // 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) + // has to reconnect instead — and with connect()'s result now honoured, using + // the wrong one would leave the slot permanently unactivated. + const esp_err_t connect_result = reconnectSlotClient(index); 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.client_state = ClientState::Starting; + slot.generation++; slot.initial_connect_done = true; return true; } @@ -2112,12 +2198,43 @@ bool MQTTBridge::setupSlot(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(). -void MQTTBridge::teardownSlot(int index) { +void MQTTBridge::teardownSlot(int index, TeardownReason reason) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[index]; - if (slot.client && slot.client->connected()) { - slot.client->disconnect(); + // Gated on the SDK lifecycle state, not on connectivity: a client resolving + // DNS, negotiating TLS or waiting after a failed CONNECT reports + // not-connected, and the old `connected()` gate left exactly those running — + // free to complete their handshake against an endpoint the operator had + // 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; + } 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; + } + } + slot.generation++; #ifdef ESP_PLATFORM vTaskDelay(pdMS_TO_TICKS(50)); #else @@ -2154,11 +2271,24 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { MQTTSlot& slot = _slots[index]; if (slot.client == nullptr) return ESP_ERR_INVALID_STATE; + if (slot.client_state == ClientState::Quarantined) { + // Its SDK task was never joined; touching it again is exactly what F01 + // forbids. + return ESP_ERR_INVALID_STATE; + } + + esp_err_t r; if (!slot.client->isStarted()) { MQTT_DEBUG_PRINTLN("MQTT%d start (client was stopped)", index + 1); - return slot.client->connect(); + r = slot.client->connect(); + } else { + r = slot.client->reconnect(); } - return slot.client->reconnect(); + if (r == ESP_OK) { + slot.client_state = ClientState::Starting; + slot.generation++; + } + return r; } @@ -2841,9 +2971,15 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[slot_index]; - teardownSlot(slot_index); + const bool disabling = (strcmp(preset_name, MQTT_PRESET_NONE) == 0 || preset_name[0] == '\0'); + // Selecting `none` must actually stop the client: clearing the bridge flags + // used to leave its task and transport running until full shutdown, and an + // in-flight handshake could still complete and mark the disabled slot + // connected (F04). Every other case keeps the task and only closes the + // transport. + teardownSlot(slot_index, disabling ? TeardownReason::Disable : TeardownReason::Reconfigure); - if (strcmp(preset_name, MQTT_PRESET_NONE) == 0 || preset_name[0] == '\0') { + if (disabling) { slot.enabled = false; slot.preset = nullptr; return; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 4450a4f3..0e9eeb06 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -98,9 +98,38 @@ 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; + } + // Connection slot - each slot holds one MQTT connection struct MQTTSlot { PsychicMqttClient* client; + ClientState client_state; + // Bumped on every start/stop. Only used for diagnostics and log lines: the + // accept/reject decision for a late callback is made on client_state, which + // the bridge task owns. + uint32_t generation; const MQTTPresetDef* preset; // Points to MQTT_PRESETS[] entry, nullptr for custom/none bool enabled; // true when preset is not "none" bool connected; // Updated in callbacks @@ -487,7 +516,17 @@ private: int activatedSlotCount() const; bool canActivateSlot(int index) const; // force as in destroySlotClients(): skip the unbounded wait, dirty-stop path only. - void teardownSlot(int index); // Disconnect the slot's client (keeps the object alive) + // Why a slot is being torn down. The distinction is not cosmetic: + // - Reconfigure: the slot is about to connect somewhere else, so the + // transport must close (or an in-flight handshake could complete against + // the OLD endpoint) but the esp-mqtt task should stay. Stopping it returns + // its 6 KiB stack into the hole the two 16 KiB mbedTLS record buffers just + // vacated, which is the fork's documented internal-heap fragmentation + // driver; softDisconnect() avoids exactly that. + // - Disable: the slot is going away, so the task and its transport must go + // with it. Here the stop IS the point. + enum class TeardownReason : uint8_t { Reconfigure, Disable }; + void teardownSlot(int index, TeardownReason reason = TeardownReason::Disable); // 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