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.

This commit is contained in:
agessaman committed 2026-03-23 20:21:47 -07:00
1 parent 95874f0c24
commit 4a60f16683
10 files changed
+102 -35

No files matched your search

+6 -4
View File
@@ -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
+6 -4
View File
@@ -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
+3 -3
View File
@@ -22,15 +22,15 @@
#if MESH_DEBUG && ARDUINO
#include <Arduino.h>
#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
+1 -1
View File
@@ -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]));
+1 -1
View File
@@ -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;
}
+9 -6
View File
@@ -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
+68 -12
View File
@@ -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;
}
}
+3 -1
View File
@@ -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);
+2
View File
@@ -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}
+3 -3
View File
@@ -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"'