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"'