From 51233375c636eb9e2a71ef2655c7a7eca2d0873d Mon Sep 17 00:00:00 2001 From: Jared Dohrman Date: Tue, 28 Apr 2026 10:31:07 +1000 Subject: [PATCH 1/2] fix: preflight mqtt websocket brokers before connect --- src/helpers/mqtt/MQTTUplink.cpp | 271 +++++++++++++++++++++++++++++--- src/helpers/mqtt/MQTTUplink.h | 7 + 2 files changed, 255 insertions(+), 23 deletions(-) diff --git a/src/helpers/mqtt/MQTTUplink.cpp b/src/helpers/mqtt/MQTTUplink.cpp index 8e6eff92..f7a37ae4 100644 --- a/src/helpers/mqtt/MQTTUplink.cpp +++ b/src/helpers/mqtt/MQTTUplink.cpp @@ -11,9 +11,15 @@ #include #include #include +#if defined(CONFIG_MBEDTLS_CERTIFICATE_BUNDLE) && !(defined(ARDUINO) && ESP_IDF_VERSION_MAJOR < 5) +#include +#define MQTT_USE_CRT_BUNDLE 1 +#define MQTT_CRT_BUNDLE_ATTACH esp_crt_bundle_attach +#endif #include #include #include +#include #include #ifndef FIRMWARE_VERSION @@ -43,8 +49,14 @@ namespace { constexpr unsigned long kBrokerRetryBaseMillis = 10000; constexpr unsigned long kBrokerRetryMaxMillis = 300000; +constexpr unsigned long kHeapHeadroomRetryMillis = 30000; +constexpr unsigned long kWsPreflightTimeoutMillis = 5000; constexpr unsigned long kTokenRefreshEventWindowMs = 2UL * 60UL * 1000UL; constexpr size_t kBrokerTokenSize = 640; +constexpr size_t kWsTransportBufferSize = 1024; +constexpr size_t kWsPreflightResponseSize = kWsTransportBufferSize; +constexpr size_t kDualBrokerMinFreeHeap = 80U * 1024U; +constexpr size_t kDualBrokerMinLargestHeap = 32U * 1024U; constexpr time_t kTokenLifetimeSecs = 6UL * 60UL * 60UL; constexpr time_t kTokenRefreshSlackSecs = 300; constexpr time_t kMinSaneEpoch = 1735689600; // 2025-01-01T00:00:00Z @@ -75,6 +87,19 @@ void freeScratchBuffer(void* ptr) { } } +bool containsIgnoreCase(const char* haystack, const char* needle) { + if (haystack == nullptr || needle == nullptr || needle[0] == 0) { + return false; + } + const size_t needle_len = strlen(needle); + for (const char* scan = haystack; *scan != 0; ++scan) { + if (strncasecmp(scan, needle, needle_len) == 0) { + return true; + } + } + return false; +} + #if MQTT_DEBUG void logMqttMemorySnapshot(const char* phase, const char* broker_label = nullptr) { MQTT_LOG("mem phase=%s broker=%s uptime_ms=%lu heap_free=%u heap_min=%u heap_max=%u psram_free=%u psram_min=%u " @@ -134,10 +159,37 @@ bool MQTTUplink::hasEnabledBroker() const { return (_prefs.enabled_mask & 0x07) != 0; } +uint8_t MQTTUplink::countEnabledBrokers() const { + uint8_t count = 0; + for (const BrokerState& broker : _brokers) { + if (broker.spec != nullptr && (_prefs.enabled_mask & broker.spec->bit) != 0) { + ++count; + } + } + return count; +} + +uint8_t MQTTUplink::countConnectedBrokers() const { + uint8_t count = 0; + for (const BrokerState& broker : _brokers) { + if (broker.spec != nullptr && broker.connected) { + ++count; + } + } + return count; +} + bool MQTTUplink::isUnsetIataValue(const char* iata) { return iata == nullptr || iata[0] == 0 || strcmp(iata, MQTT_UNSET_IATA) == 0; } +const char* MQTTUplink::brokerCaCert(const BrokerSpec& spec) { + if (spec.bit == kEastmeshBit) { + return mqtt_ca_certs::kEastmeshIsrgRootX1Pem; + } + return mqtt_ca_certs::kLetsmeshWe1Pem; +} + uint8_t MQTTUplink::normalizeEnabledMask(uint8_t mask) { uint8_t normalized = 0; uint8_t count = 0; @@ -153,6 +205,125 @@ uint8_t MQTTUplink::normalizeEnabledMask(uint8_t mask) { return normalized; } +bool MQTTUplink::hasConnectHeadroom(const BrokerState& broker) const { + const uint8_t enabled_count = countEnabledBrokers(); + const uint8_t connected_count = countConnectedBrokers(); + const bool dual_broker_attempt = enabled_count > 1 && connected_count > 0; + if (!dual_broker_attempt) { + return true; + } + + const size_t free_heap = heap_caps_get_free_size(MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT); + const size_t largest_heap = heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT); + + if (free_heap >= kDualBrokerMinFreeHeap && largest_heap >= kDualBrokerMinLargestHeap) { + return true; + } + + MQTT_LOG("%s connect deferred: heap headroom low heap_free=%u heap_max=%u required_heap_free=%u " + "required_heap_max=%u enabled=%u connected=%u", + broker.spec != nullptr ? broker.spec->label : "-", + static_cast(free_heap), + static_cast(largest_heap), + static_cast(kDualBrokerMinFreeHeap), + static_cast(kDualBrokerMinLargestHeap), + static_cast(enabled_count), + static_cast(connected_count)); + return false; +} + +bool MQTTUplink::preflightBroker(BrokerState& broker) const { + if (broker.spec == nullptr) { + return false; + } + + logMqttMemorySnapshot("preflight-pre", broker.spec->label); + WiFiClientSecure client; + client.setCACert(brokerCaCert(*broker.spec)); + client.setTimeout(kWsPreflightTimeoutMillis); + if (!client.connect(broker.spec->host, 443)) { + MQTT_LOG("%s preflight failed: tcp/tls connect", broker.spec->label); + client.stop(); + logMqttMemorySnapshot("preflight-failed", broker.spec->label); + return false; + } + + client.print("GET /mqtt HTTP/1.1\r\n" + "Connection: Upgrade\r\n" + "Upgrade: websocket\r\n" + "Sec-WebSocket-Version: 13\r\n" + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + "User-Agent: MeshCore-EastMesh\r\n" + "Host: "); + client.print(broker.spec->host); + client.print(":443\r\n\r\n"); + + char response[kWsPreflightResponseSize]; + size_t used = 0; + bool header_complete = false; + unsigned long deadline_ms = millis() + kWsPreflightTimeoutMillis; + response[0] = 0; + + while (static_cast(millis() - deadline_ms) < 0) { + while (client.available() > 0) { + int c = client.read(); + if (c < 0) { + break; + } + if (used + 1 >= sizeof(response)) { + response[used] = 0; + MQTT_LOG("%s preflight failed: response header too large bytes=%u", broker.spec->label, + static_cast(used)); + client.stop(); + logMqttMemorySnapshot("preflight-failed", broker.spec->label); + return false; + } + response[used++] = static_cast(c); + response[used] = 0; + if (strstr(response, "\r\n\r\n") != nullptr) { + header_complete = true; + break; + } + } + if (header_complete) { + break; + } + if (!client.connected() && client.available() <= 0) { + break; + } + delay(10); + } + + client.stop(); + if (!header_complete) { + MQTT_LOG("%s preflight failed: incomplete response bytes=%u", broker.spec->label, static_cast(used)); + logMqttMemorySnapshot("preflight-failed", broker.spec->label); + return false; + } + + char status_line[48]; + size_t status_len = 0; + while (status_len + 1 < sizeof(status_line) && response[status_len] != 0 && response[status_len] != '\r' && + response[status_len] != '\n') { + status_line[status_len] = response[status_len]; + ++status_len; + } + status_line[status_len] = 0; + + bool ok = strncmp(response, "HTTP/", 5) == 0 && strstr(status_line, " 101") != nullptr && + containsIgnoreCase(response, "\r\nSec-WebSocket-Accept:"); + if (!ok) { + MQTT_LOG("%s preflight failed: status=\"%s\" accept=%d", broker.spec->label, status_line, + containsIgnoreCase(response, "\r\nSec-WebSocket-Accept:") ? 1 : 0); + logMqttMemorySnapshot("preflight-failed", broker.spec->label); + return false; + } + + MQTT_LOG("%s preflight ok: status=\"%s\"", broker.spec->label, status_line); + logMqttMemorySnapshot("preflight-ok", broker.spec->label); + return true; +} + void MQTTUplink::formatTopic(char* dst, size_t dst_size, const char* leaf) const { if (dst == nullptr || dst_size == 0) { return; @@ -340,7 +511,7 @@ bool MQTTUplink::refreshToken(BrokerState& broker) { } void MQTTUplink::destroyBroker(BrokerState& broker, bool reset_retry_state) { - bool had_runtime_state = broker.client != nullptr || broker.token != nullptr || broker.connected || + bool had_runtime_state = broker.client != nullptr || broker.token != nullptr || broker.connected || broker.started || broker.connect_announced || broker.reconnect_pending || broker.next_connect_attempt != 0 || broker.last_connect_attempt != 0 || broker.reconnect_failures != 0 || broker.token_expires_at != 0; @@ -349,14 +520,33 @@ void MQTTUplink::destroyBroker(BrokerState& broker, bool reset_retry_state) { } logMqttMemorySnapshot("destroy-pre", broker.spec != nullptr ? broker.spec->label : nullptr); if (broker.client != nullptr) { - MQTT_LOG("%s destroy broker client", broker.spec->label); - esp_mqtt_client_stop(broker.client); - esp_mqtt_client_destroy(broker.client); + if (broker.started) { + MQTT_LOG("%s stop broker client", broker.spec->label); + esp_err_t stop_rc = esp_mqtt_client_stop(broker.client); + MQTT_LOG("%s stop broker client rc=0x%x", broker.spec->label, stop_rc); + if (heap_caps_check_integrity_all(true)) { + MQTT_LOG("%s destroy broker client", broker.spec->label); + esp_err_t destroy_rc = esp_mqtt_client_destroy(broker.client); + MQTT_LOG("%s destroy broker client rc=0x%x", broker.spec->label, destroy_rc); + } else { + // Avoid freeing through esp-mqtt after the IDF 4.4 WSS transport has already poisoned the heap. + MQTT_LOG("%s abandon stopped broker client: heap corrupt", broker.spec->label); + } + } else { + if (heap_caps_check_integrity_all(true)) { + MQTT_LOG("%s destroy broker client", broker.spec->label); + esp_err_t destroy_rc = esp_mqtt_client_destroy(broker.client); + MQTT_LOG("%s destroy broker client rc=0x%x", broker.spec->label, destroy_rc); + } else { + MQTT_LOG("%s abandon broker client: heap corrupt", broker.spec->label); + } + } broker.client = nullptr; } freeScratchBuffer(broker.token); broker.token = nullptr; broker.connected = false; + broker.started = false; broker.connect_announced = false; broker.connected_since_ms = 0; broker.token_expires_at = 0; @@ -537,6 +727,17 @@ void MQTTUplink::publishStatus(bool online) { logMqttMemorySnapshot(online ? "status-post" : "status-offline-post"); } +void MQTTUplink::scheduleBrokerRetry(BrokerState& broker, unsigned long now_ms, bool count_failure) { + if (count_failure && broker.reconnect_failures < 10) { + broker.reconnect_failures++; + } + broker.reconnect_pending = true; + broker.next_connect_attempt = now_ms + getBrokerRetryDelayMillis(broker.reconnect_failures); + MQTT_LOG("%s reconnect in %lu ms (failures=%u)", broker.spec != nullptr ? broker.spec->label : "-", + getBrokerRetryDelayMillis(broker.reconnect_failures), + static_cast(broker.reconnect_failures)); +} + void MQTTUplink::handleMqttEvent(void* handler_args, esp_event_base_t, int32_t event_id, void* event_data) { auto* broker = static_cast(handler_args); if (broker == nullptr) { @@ -563,14 +764,7 @@ void MQTTUplink::handleMqttEvent(void* handler_args, esp_event_base_t, int32_t e static_cast(WiFi.status()), WiFi.RSSI(), connected_for_ms); logMqttMemorySnapshot("disconnected", broker->spec->label); if (!broker->reconnect_pending) { - if (broker->reconnect_failures < 10) { - broker->reconnect_failures++; - } - broker->reconnect_pending = true; - broker->next_connect_attempt = now_ms + getBrokerRetryDelayMillis(broker->reconnect_failures); - MQTT_LOG("%s reconnect in %lu ms (failures=%u)", broker->spec->label, - getBrokerRetryDelayMillis(broker->reconnect_failures), - static_cast(broker->reconnect_failures)); + scheduleBrokerRetry(*broker, now_ms, true); } break; case MQTT_EVENT_ERROR: @@ -589,14 +783,7 @@ void MQTTUplink::handleMqttEvent(void* handler_args, esp_event_base_t, int32_t e MQTT_LOG("%s wifi_status=%d rssi=%d", broker->spec->label, static_cast(WiFi.status()), WiFi.RSSI()); logMqttMemorySnapshot("error", broker->spec->label); if (!broker->reconnect_pending) { - if (broker->reconnect_failures < 10) { - broker->reconnect_failures++; - } - broker->reconnect_pending = true; - broker->next_connect_attempt = now_ms + getBrokerRetryDelayMillis(broker->reconnect_failures); - MQTT_LOG("%s reconnect in %lu ms (failures=%u)", broker->spec->label, - getBrokerRetryDelayMillis(broker->reconnect_failures), - static_cast(broker->reconnect_failures)); + scheduleBrokerRetry(*broker, now_ms, true); } break; case MQTT_EVENT_BEFORE_CONNECT: @@ -644,7 +831,26 @@ void MQTTUplink::ensureBroker(BrokerState& broker, bool allow_new_connect) { if (!broker.reconnect_pending || now_ms < broker.next_connect_attempt) { return; } - destroyBroker(broker, false); + if (!allow_new_connect) { + return; + } + broker.last_connect_attempt = now_ms; + broker.reconnect_pending = false; + if (!preflightBroker(broker)) { + scheduleBrokerRetry(broker, now_ms, true); + return; + } + MQTT_LOG("%s mqtt reconnect requested", broker.spec->label); + logMqttMemorySnapshot("reconnect-pre", broker.spec->label); + esp_err_t reconnect_rc = esp_mqtt_client_reconnect(broker.client); + if (reconnect_rc != ESP_OK) { + MQTT_LOG("%s mqtt reconnect failed rc=0x%x", broker.spec->label, reconnect_rc); + logMqttMemorySnapshot("reconnect-failed", broker.spec->label); + scheduleBrokerRetry(broker, now_ms, true); + } else { + logMqttMemorySnapshot("reconnect-requested", broker.spec->label); + } + return; } if (broker.next_connect_attempt != 0 && now_ms < broker.next_connect_attempt) { @@ -653,9 +859,19 @@ void MQTTUplink::ensureBroker(BrokerState& broker, bool allow_new_connect) { if (!allow_new_connect) { return; } + if (!hasConnectHeadroom(broker)) { + broker.reconnect_pending = true; + broker.next_connect_attempt = now_ms + kHeapHeadroomRetryMillis; + return; + } broker.last_connect_attempt = now_ms; broker.reconnect_pending = false; + if (!preflightBroker(broker)) { + scheduleBrokerRetry(broker, now_ms, true); + return; + } + if (!refreshToken(broker)) { broker.reconnect_pending = true; broker.next_connect_attempt = now_ms + kBrokerRetryBaseMillis; @@ -672,7 +888,11 @@ void MQTTUplink::ensureBroker(BrokerState& broker, bool allow_new_connect) { cfg.broker.address.port = 443; cfg.broker.address.transport = MQTT_TRANSPORT_OVER_WSS; cfg.broker.address.path = "/mqtt"; - cfg.broker.verification.certificate = mqtt_ca_certs::kCombinedPem; +#if defined(MQTT_USE_CRT_BUNDLE) + cfg.broker.verification.crt_bundle_attach = MQTT_CRT_BUNDLE_ATTACH; +#else + cfg.broker.verification.certificate = brokerCaCert(*broker.spec); +#endif cfg.credentials.username = broker.username; cfg.credentials.client_id = broker.client_id; cfg.credentials.authentication.password = broker.token; @@ -699,7 +919,11 @@ void MQTTUplink::ensureBroker(BrokerState& broker, bool allow_new_connect) { cfg.network_timeout_ms = 10000; cfg.disable_auto_reconnect = true; cfg.transport = MQTT_TRANSPORT_OVER_WSS; - cfg.cert_pem = mqtt_ca_certs::kCombinedPem; +#if defined(MQTT_USE_CRT_BUNDLE) + cfg.crt_bundle_attach = MQTT_CRT_BUNDLE_ATTACH; +#else + cfg.cert_pem = brokerCaCert(*broker.spec); +#endif cfg.lwt_topic = broker.status_topic; cfg.lwt_msg = broker.offline_payload; cfg.lwt_qos = 1; @@ -723,6 +947,7 @@ void MQTTUplink::ensureBroker(BrokerState& broker, bool allow_new_connect) { broker.next_connect_attempt = now_ms + kBrokerRetryBaseMillis; destroyBroker(broker, false); } else { + broker.started = true; MQTT_LOG("%s mqtt start requested", broker.spec->label); logMqttMemorySnapshot("start-requested", broker.spec->label); } diff --git a/src/helpers/mqtt/MQTTUplink.h b/src/helpers/mqtt/MQTTUplink.h index a4391109..619f7907 100644 --- a/src/helpers/mqtt/MQTTUplink.h +++ b/src/helpers/mqtt/MQTTUplink.h @@ -82,6 +82,7 @@ private: const BrokerSpec* spec; esp_mqtt_client_handle_t client; bool connected; + bool started; bool connect_announced; bool reconnect_pending; unsigned long last_connect_attempt; @@ -117,12 +118,18 @@ private: static constexpr uint8_t kMaxEnabledBrokers = 2; static const BrokerSpec kBrokerSpecs[3]; static bool isUnsetIataValue(const char* iata); + static const char* brokerCaCert(const BrokerSpec& spec); BrokerState _brokers[3]; static void handleMqttEvent(void* handler_args, esp_event_base_t base, int32_t event_id, void* event_data); + static void scheduleBrokerRetry(BrokerState& broker, unsigned long now_ms, bool count_failure); bool hasEnabledBroker() const; + uint8_t countEnabledBrokers() const; + uint8_t countConnectedBrokers() const; static uint8_t normalizeEnabledMask(uint8_t mask); + bool hasConnectHeadroom(const BrokerState& broker) const; + bool preflightBroker(BrokerState& broker) const; void formatTopic(char* dst, size_t dst_size, const char* leaf) const; void refreshIdentityStrings(); void refreshBrokerIdentity(BrokerState& broker); From 68772d6a89eebf5a54b893220caa7fcd5f876853 Mon Sep 17 00:00:00 2001 From: Jared Dohrman Date: Tue, 28 Apr 2026 10:34:26 +1000 Subject: [PATCH 2/2] docs: add repeater-mqtt-eastmesh-v1.3.13 release notes --- release-notes.yml | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/release-notes.yml b/release-notes.yml index beb97fe8..64ad8772 100644 --- a/release-notes.yml +++ b/release-notes.yml @@ -500,3 +500,30 @@ releases: area: web-panel text: "Updated the web panel and custom CLI docs for the planned MQTT token-refresh event and the new web-panel region shortcut." breaking_changes: [] + + - track: repeater-mqtt + version: "1.3.13" + tag: "repeater-mqtt-eastmesh-v1.3.13" + date: "2026-04-28" + previous_version: "1.3.12" + summary: "Adds a safe MQTT WebSocket preflight for broker failover, restores clean MQTT client destruction when the heap is healthy, and clarifies that EastMesh MQTT repeaters feed Eastern Australia Core telemetry." + changes: + - type: fixed + area: mqtt + text: "Added a bounded WebSocket upgrade preflight before MQTT connects and reconnects so down or misrouted brokers that return non-upgrade HTTP responses are backed off before the ESP32 IDF 4.4 WebSocket transport can corrupt heap." + - type: fixed + area: mqtt + text: "Restored normal MQTT client destruction when heap integrity is clean, while keeping a guarded abandon path if heap poisoning is already detected before `esp_mqtt_client_destroy()`." + - type: changed + area: mqtt + text: "Reduced dual-broker connection pressure by checking both free internal heap and largest internal heap block before starting a second broker while another broker is already connected." + - type: changed + area: mqtt + text: "Switched retrying disconnected MQTT clients to `esp_mqtt_client_reconnect()` instead of destroying and recreating the client on every reconnect attempt." + - type: changed + area: mqtt + text: "Scoped MQTT TLS CA selection per broker instead of passing the combined CA bundle on Arduino-ESP32 2.x builds, while keeping certificate-bundle support for newer ESP-IDF stacks." + - type: docs + area: readme + text: "Updated the README to describe EastMesh as an Eastern Australia community mesh network, introduce CoreScope telemetry, and warn that MQTT repeater builds are intended for Eastern Australia repeaters only." + breaking_changes: []