From 4a60f166835bd1a97989b471602ff88585dd65a3 Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 23 Mar 2026 20:21:47 -0700 Subject: [PATCH] Refactor logging and error handling in MyMesh and MQTTBridge. Implement checks for Serial availability before printing logs to prevent potential blocking. Update MQTTBridge to include circuit breaker logic for managing reconnect attempts and enhance slot management with new state tracking. Expand MQTTPresetDef to include a new preset with configurable token lifetime and retain options. --- examples/simple_repeater/MyMesh.cpp | 10 ++-- examples/simple_room_server/MyMesh.cpp | 10 ++-- src/MeshCore.h | 6 +- src/helpers/CommonCLI.cpp | 2 +- src/helpers/JWTHelper.cpp | 2 +- src/helpers/MQTTPresets.h | 15 +++-- src/helpers/bridges/MQTTBridge.cpp | 80 ++++++++++++++++++++++---- src/helpers/bridges/MQTTBridge.h | 4 +- variants/heltec_v4/platformio.ini | 2 + variants/lilygo_t3s3/platformio.ini | 6 +- 10 files changed, 102 insertions(+), 35 deletions(-) diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index 2b2f9a87..48adfb04 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -487,10 +487,12 @@ const char *MyMesh::getLogDateTime() { void MyMesh::logRxRaw(float snr, float rssi, const uint8_t raw[], int len) { #if MESH_PACKET_LOGGING - Serial.print(getLogDateTime()); - Serial.print(" RAW: "); - mesh::Utils::printHex(Serial, raw, len); - Serial.println(); + if (Serial.availableForWrite() > 0) { + Serial.print(getLogDateTime()); + Serial.print(" RAW: "); + mesh::Utils::printHex(Serial, raw, len); + Serial.println(); + } #endif #ifdef WITH_BRIDGE diff --git a/examples/simple_room_server/MyMesh.cpp b/examples/simple_room_server/MyMesh.cpp index 5e98cd61..1e27e1a1 100644 --- a/examples/simple_room_server/MyMesh.cpp +++ b/examples/simple_room_server/MyMesh.cpp @@ -201,10 +201,12 @@ int MyMesh::handleRequest(ClientInfo *sender, uint32_t sender_timestamp, uint8_t void MyMesh::logRxRaw(float snr, float rssi, const uint8_t raw[], int len) { #if MESH_PACKET_LOGGING - Serial.print(getLogDateTime()); - Serial.print(" RAW: "); - mesh::Utils::printHex(Serial, raw, len); - Serial.println(); + if (Serial.availableForWrite() > 0) { + Serial.print(getLogDateTime()); + Serial.print(" RAW: "); + mesh::Utils::printHex(Serial, raw, len); + Serial.println(); + } #endif #ifdef WITH_MQTT_BRIDGE diff --git a/src/MeshCore.h b/src/MeshCore.h index 70cd0f06..c73dcf4c 100644 --- a/src/MeshCore.h +++ b/src/MeshCore.h @@ -22,15 +22,15 @@ #if MESH_DEBUG && ARDUINO #include - #define MESH_DEBUG_PRINT(F, ...) Serial.printf("DEBUG: " F, ##__VA_ARGS__) - #define MESH_DEBUG_PRINTLN(F, ...) Serial.printf("DEBUG: " F "\n", ##__VA_ARGS__) + #define MESH_DEBUG_PRINT(F, ...) do { if (Serial.availableForWrite() > 0) { Serial.printf("DEBUG: " F, ##__VA_ARGS__); } } while(0) + #define MESH_DEBUG_PRINTLN(F, ...) do { if (Serial.availableForWrite() > 0) { Serial.printf("DEBUG: " F "\n", ##__VA_ARGS__); } } while(0) #else #define MESH_DEBUG_PRINT(...) {} #define MESH_DEBUG_PRINTLN(...) {} #endif #if BRIDGE_DEBUG && ARDUINO -#define BRIDGE_DEBUG_PRINTLN(F, ...) Serial.printf("%s BRIDGE: " F, getLogDateTime(), ##__VA_ARGS__) +#define BRIDGE_DEBUG_PRINTLN(F, ...) do { if (Serial.availableForWrite() > 0) { Serial.printf("%s BRIDGE: " F, getLogDateTime(), ##__VA_ARGS__); } } while(0) #else #define BRIDGE_DEBUG_PRINTLN(...) {} #endif diff --git a/src/helpers/CommonCLI.cpp b/src/helpers/CommonCLI.cpp index d6d5e0e0..8e9a5cca 100644 --- a/src/helpers/CommonCLI.cpp +++ b/src/helpers/CommonCLI.cpp @@ -1222,7 +1222,7 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch sprintf(reply, "OK - slot %d preset: %s", slot + 1, preset_name); } } else { - strcpy(reply, "Error: valid presets are: analyzer-us, analyzer-eu, meshmapper, meshrank, custom, none"); + strcpy(reply, "Error: valid presets are: analyzer-us, analyzer-eu, meshmapper, meshrank, waev, custom, none"); } } else if (memcmp(subcmd, "server ", 7) == 0) { StrHelper::strncpy(_prefs->mqtt_slot_host[slot], &subcmd[7], sizeof(_prefs->mqtt_slot_host[slot])); diff --git a/src/helpers/JWTHelper.cpp b/src/helpers/JWTHelper.cpp index dc4154ba..0b73d87e 100644 --- a/src/helpers/JWTHelper.cpp +++ b/src/helpers/JWTHelper.cpp @@ -79,7 +79,7 @@ bool JWTHelper::createAuthToken( // Verify the signature locally int verify_result = ed25519_verify(signature, (const unsigned char*)signingInput, signingInputLen, public_key); if (verify_result != 1) { - Serial.println("JWTHelper: Signature verification failed!"); + if (Serial.availableForWrite() > 0) Serial.println("JWTHelper: Signature verification failed!"); return false; } diff --git a/src/helpers/MQTTPresets.h b/src/helpers/MQTTPresets.h index a538a58b..e74bf255 100644 --- a/src/helpers/MQTTPresets.h +++ b/src/helpers/MQTTPresets.h @@ -14,12 +14,14 @@ enum MQTTTopicStyle : uint8_t { }; struct MQTTPresetDef { - const char* name; // Preset identifier: "analyzer-us", "analyzer-eu", "meshmapper", "meshrank" + const char* name; // Preset identifier: "analyzer-us", "analyzer-eu", "meshmapper", "meshrank", "waev" const char* server_url; // Full URL including scheme: "wss://host:port/path" or "mqtts://host:port" const char* jwt_audience; // JWT audience field (only for MQTT_AUTH_JWT, nullptr otherwise) const char* ca_cert; // PEM CA certificate (nullptr to skip cert pinning) MQTTAuthType auth_type; MQTTTopicStyle topic_style; + unsigned long token_lifetime; // JWT token lifetime in seconds (0 = use default 86400) + bool allow_retain; // Whether the broker allows the MQTT retain flag }; // Google Trust Services - GTS Root R4 (used by LetsMesh Analyzer) @@ -81,14 +83,15 @@ static const char ISRG_ROOT_X1[] PROGMEM = "-----END CERTIFICATE-----\n"; // Number of built-in presets -static const int MQTT_PRESET_COUNT = 4; +static const int MQTT_PRESET_COUNT = 5; // Built-in preset definitions (stored in flash) static const MQTTPresetDef MQTT_PRESETS[MQTT_PRESET_COUNT] = { - { "analyzer-us", "wss://mqtt-us-v1.letsmesh.net:443/mqtt", "mqtt-us-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE }, - { "analyzer-eu", "wss://mqtt-eu-v1.letsmesh.net:443/mqtt", "mqtt-eu-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE }, - { "meshmapper", "wss://mqtt.meshmapper.cc:443/mqtt", "mqtt.meshmapper.cc", ISRG_ROOT_X1, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE }, - { "meshrank", "mqtts://meshrank.net:8883", nullptr, ISRG_ROOT_X1, MQTT_AUTH_NONE, MQTT_TOPIC_MESHRANK }, + { "analyzer-us", "wss://mqtt-us-v1.letsmesh.net:443/mqtt", "mqtt-us-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true }, + { "analyzer-eu", "wss://mqtt-eu-v1.letsmesh.net:443/mqtt", "mqtt-eu-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true }, + { "meshmapper", "wss://mqtt.meshmapper.cc:443/mqtt", "mqtt.meshmapper.cc", ISRG_ROOT_X1, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true }, + { "meshrank", "mqtts://meshrank.net:8883", nullptr, ISRG_ROOT_X1, MQTT_AUTH_NONE, MQTT_TOPIC_MESHRANK, 0, false }, + { "waev", "wss://mqtt.waev.app:443/mqtt", "mqtt.waev.app", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 3300, false }, }; // Find a preset by name, returns nullptr if not found diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 1f21d20e..98d44dd6 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -128,13 +128,15 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const NodePref const char* name = slot.preset ? slot.preset->name : "custom"; snprintf(slot_info[i], sizeof(slot_info[i]), "slot%d: %s (not ready)", i + 1, name); } else if (slot.preset) { + const char* state = slot.connected ? "connected" : + slot.circuit_breaker_tripped ? "failed" : "disconnected"; snprintf(slot_info[i], sizeof(slot_info[i]), "slot%d: %s (%s)", i + 1, - slot.preset->name, - slot.connected ? "connected" : "disconnected"); + slot.preset->name, state); } else { // Custom broker - snprintf(slot_info[i], sizeof(slot_info[i]), "slot%d: custom (%s)", i + 1, - slot.connected ? "connected" : "disconnected"); + const char* state = slot.connected ? "connected" : + slot.circuit_breaker_tripped ? "failed" : "disconnected"; + snprintf(slot_info[i], sizeof(slot_info[i]), "slot%d: custom (%s)", i + 1, state); } } @@ -198,6 +200,8 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCCloc _slots[i].token_expires_at = 0; _slots[i].last_token_renewal = 0; _slots[i].reconnect_backoff = 0; + _slots[i].max_backoff_failures = 0; + _slots[i].circuit_breaker_tripped = false; _slots[i].last_reconnect_attempt = 0; _slots[i].last_log_time = 0; _slots[i].port = 1883; @@ -767,6 +771,7 @@ void MQTTBridge::setupSlot(int index) { if (slot.client) return; slot.client = new PsychicMqttClient(); + slot.client->setAutoReconnect(false); // We handle reconnect with our own backoff logic optimizeMqttClientConfig(slot.client, slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT); // Callbacks (capture index by value) @@ -774,6 +779,8 @@ void MQTTBridge::setupSlot(int index) { MQTT_DEBUG_PRINTLN("Slot %d connected", index); _slots[index].connected = true; _slots[index].reconnect_backoff = 0; + _slots[index].max_backoff_failures = 0; + _slots[index].circuit_breaker_tripped = false; updateCachedConnectionStatus(); publishStatusToSlot(index); }); @@ -848,6 +855,8 @@ void MQTTBridge::teardownSlot(int index) { slot.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; slot.last_log_time = 0; } @@ -867,6 +876,16 @@ void MQTTBridge::maintainSlotConnections() { bool clock_looks_set = (clock_sec >= 1735689600); // 2025-01-01 00:00:00 UTC bool can_do_jwt = _ntp_synced || clock_looks_set; + // Count connected slots to inform reconnect decisions + int connected_count = 0; + for (int i = 0; i < MAX_MQTT_SLOTS; i++) { + if (_slots[i].enabled && _slots[i].connected) connected_count++; + } + + // Only allow one reconnect attempt per maintenance cycle to avoid + // multiple simultaneous TLS handshakes blocking the network stack + bool reconnect_attempted_this_cycle = false; + for (int i = 0; i < MAX_MQTT_SLOTS; i++) { if (!_slots[i].enabled || !_slots[i].client) continue; @@ -875,15 +894,16 @@ void MQTTBridge::maintainSlotConnections() { continue; } - maintainSlotConnection(i, now_millis, current_time, time_synced); + maintainSlotConnection(i, now_millis, current_time, time_synced, reconnect_attempted_this_cycle); } } -void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced) { +void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted) { MQTTSlot& slot = _slots[index]; if (slot.connected) { slot.reconnect_backoff = 0; + slot.max_backoff_failures = 0; } // JWT token renewal (only for preset slots with JWT auth) @@ -938,10 +958,29 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns } } + // Periodic probe for circuit-breaker-tripped slots (recovery from transient outages) + // Attempts a single reconnect every 30 minutes to see if the server has come back + if (slot.circuit_breaker_tripped && !reconnect_attempted) { + static const unsigned long CIRCUIT_BREAKER_PROBE_INTERVAL_MS = 1800000UL; // 30 minutes + unsigned long probe_elapsed = (now_millis >= slot.last_reconnect_attempt) ? + (now_millis - slot.last_reconnect_attempt) : + (ULONG_MAX - slot.last_reconnect_attempt + now_millis + 1); + if (probe_elapsed >= CIRCUIT_BREAKER_PROBE_INTERVAL_MS) { + slot.last_reconnect_attempt = now_millis; + reconnect_attempted = true; + MQTT_DEBUG_PRINTLN("Slot %d circuit breaker probe (attempting single reconnect after %lu ms)", index, probe_elapsed); + slot.client->connect(); + // If the connect callback fires and sets slot.connected = true, + // the next maintenance cycle will clear the circuit breaker via the + // existing slot.reconnect_backoff = 0 / circuit_breaker reset logic + } + } + // Reconnect with exponential backoff (for disconnected slots that already have valid config) - // Stagger reconnections: each slot offsets by 3s * slot_index to avoid simultaneous TLS handshakes - if (!slot.connected && slot.initial_connect_done) { + // Only one reconnect per maintenance cycle to prevent TLS handshakes from blocking other slots + if (!slot.connected && slot.initial_connect_done && !slot.circuit_breaker_tripped && !reconnect_attempted) { static const unsigned long SLOT_BACKOFF_MS[] = { 10000, 30000, 60000, 120000, 300000 }; + static const uint8_t MAX_FAILURES_AT_MAX_BACKOFF = 3; // ~15 min at max backoff before giving up unsigned long reconnect_elapsed = (now_millis >= slot.last_reconnect_attempt) ? (now_millis - slot.last_reconnect_attempt) : (ULONG_MAX - slot.last_reconnect_attempt + now_millis + 1); @@ -951,8 +990,16 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns slot.last_reconnect_attempt = now_millis; if (slot.reconnect_backoff < 5) { slot.reconnect_backoff++; + } else { + slot.max_backoff_failures++; + if (slot.max_backoff_failures >= MAX_FAILURES_AT_MAX_BACKOFF) { + slot.circuit_breaker_tripped = true; + MQTT_DEBUG_PRINTLN("Slot %d circuit breaker tripped after %d failures at max backoff - stopping reconnect attempts. Reconfigure slot to retry.", index, slot.max_backoff_failures); + return; + } } - MQTT_DEBUG_PRINTLN("Slot %d reconnecting (backoff level %d)", index, slot.reconnect_backoff); + MQTT_DEBUG_PRINTLN("Slot %d reconnecting (backoff level %d, failures at max: %d)", index, slot.reconnect_backoff, slot.max_backoff_failures); + reconnect_attempted = true; slot.client->connect(); } } @@ -989,8 +1036,13 @@ bool MQTTBridge::createSlotAuthToken(int index) { const char* email = (_prefs->mqtt_email[0] != '\0') ? _prefs->mqtt_email : nullptr; unsigned long current_time = time(nullptr); + // Use preset-specific lifetime if set, otherwise default to 24h // Stagger token expiry per slot to avoid simultaneous renewal/reconnect - unsigned long expires_in = 86400 - (index * 300); // slot 0: 24h, slot 1: 23h55m, slot 2: 23h50m + unsigned long base_lifetime = (slot.preset->token_lifetime > 0) ? slot.preset->token_lifetime : 86400; + // Stagger token expiry per slot to avoid simultaneous renewal/reconnect + // Use 5% of lifetime per slot, capped at 300s, so short-lived tokens aren't over-reduced + unsigned long stagger = index * min((unsigned long)300, base_lifetime / 20); + unsigned long expires_in = base_lifetime - stagger; bool time_synced = (current_time >= 1000000000); if (JWTHelper::createAuthToken( @@ -1017,7 +1069,10 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload return false; } - int result = slot.client->publish(topic, 1, retained, payload, strlen(payload)); + // Use async publish (enqueue) to avoid blocking the MQTT task loop. + // Synchronous publish with QoS 1 blocks waiting for PUBACK, which can stall + // all slots when one slot's network connection is degraded. + int result = slot.client->publish(topic, 1, retained, payload, strlen(payload), true); if (result <= 0) { static unsigned long last_fail_log = 0; unsigned long now = millis(); @@ -1775,7 +1830,8 @@ bool MQTTBridge::publishStatus() { if (_slots[i].enabled && _slots[i].client && _slots[i].connected) { if (buildTopicForSlot(i, MSG_STATUS, topic, sizeof(topic))) { any_slot_wants_status = true; - if (publishToSlot(i, topic, json_buffer, true)) { + bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false; + if (publishToSlot(i, topic, json_buffer, use_retain)) { published = true; } } diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 7693e438..8088481b 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -74,6 +74,8 @@ private: // Reconnect backoff uint8_t reconnect_backoff; // 0..4 index into backoff table + uint8_t max_backoff_failures; // consecutive failures at max backoff level + bool circuit_breaker_tripped; // true = stop reconnecting until reconfigured unsigned long last_reconnect_attempt; unsigned long last_log_time; // Throttle disconnect log messages }; @@ -202,7 +204,7 @@ private: void setupSlot(int index); // Create/destroy client for a slot based on its preset void teardownSlot(int index); // Disconnect and free slot resources void maintainSlotConnections(); // Maintain all slot connections (token renewal, reconnect) - void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced); + void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted); bool createSlotAuthToken(int index); // Create/renew JWT token for a slot bool publishToSlot(int index, const char* topic, const char* payload, bool retained = false); bool publishToAllSlots(const char* topic, const char* payload, bool retained = false); diff --git a/variants/heltec_v4/platformio.ini b/variants/heltec_v4/platformio.ini index df0ed6e8..f5bb4ccc 100644 --- a/variants/heltec_v4/platformio.ini +++ b/variants/heltec_v4/platformio.ini @@ -142,6 +142,7 @@ build_flags = -D MESH_PACKET_LOGGING=1 -D MESH_DEBUG=1 -D CONFIG_MBEDTLS_CERTIFICATE_BUNDLE=y + -D ESP32_CPU_FREQ=160 -D MQTT_WIFI_TX_POWER=WIFI_POWER_19_5dBm -D MQTT_WIFI_POWER_SAVE_DEFAULT=1 # -D WIFI_SSID='"ssid"' @@ -182,6 +183,7 @@ build_flags = ; -D MESH_PACKET_LOGGING=1 ; -D MESH_DEBUG=1 -D CONFIG_MBEDTLS_CERTIFICATE_BUNDLE=y + -D ESP32_CPU_FREQ=160 -D MQTT_WIFI_TX_POWER=WIFI_POWER_19_5dBm -D MQTT_WIFI_POWER_SAVE_DEFAULT=1 build_src_filter = ${heltec_v4_oled.build_src_filter} diff --git a/variants/lilygo_t3s3/platformio.ini b/variants/lilygo_t3s3/platformio.ini index f207653e..88519f49 100644 --- a/variants/lilygo_t3s3/platformio.ini +++ b/variants/lilygo_t3s3/platformio.ini @@ -112,9 +112,9 @@ build_flags = -D WITH_MQTT_BRIDGE=1 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 - -D MQTT_DEBUG=1 - -D MESH_PACKET_LOGGING=1 - -D MESH_DEBUG=1 +; -D MQTT_DEBUG=1 +; -D MESH_PACKET_LOGGING=1 +; -D MESH_DEBUG=1 -D CONFIG_MBEDTLS_CERTIFICATE_BUNDLE=y -D ESP32_CPU_FREQ=160 # -D WIFI_SSID='"ssid"'