From 7d0c5bce50c66cc5d68caccde58a142c1dca89d6 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 23 Apr 2026 21:05:57 -0700 Subject: [PATCH] enhance(mqtt): improve QoS handling and retry logic in MQTTBridge Refactor the MQTTBridge to implement enhanced QoS handling for publish operations. Introduce retry mechanisms for QoS0 packets, allowing for transient failures to be retried with a delay. Update publish methods to return success status, improving error handling and logging. Additionally, adjust the packet queue processing to accommodate new retry logic and ensure better message delivery control. --- .../src/PsychicMqttClient.cpp | 6 +- src/helpers/bridges/MQTTBridge.cpp | 149 +++++++++++++++--- src/helpers/bridges/MQTTBridge.h | 6 +- variants/heltec_v3/platformio.ini | 39 +++++ 4 files changed, 174 insertions(+), 26 deletions(-) diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp index 1fd6fe633..b68e5352f 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp @@ -471,7 +471,11 @@ int PsychicMqttClient::publish(const char *topic, int qos, bool retain, const ch if (async) { ESP_LOGV(TAG, "Enqueuing message to topic %s with QoS %d", topic, qos); - return esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, true); + // QoS0 telemetry is high-rate and does not need durable outbox storage. + // Avoiding store=true reduces small allocation churn in reconnect storms. + // Keep store=true for QoS1/2 so status/control semantics remain durable. + bool store_in_outbox = (qos > 0); + return esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, store_in_outbox); } else { diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 62831cdde..8c9251d7d 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -1531,11 +1531,15 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload // values (-1 generic failure, -2 outbox full) are the only actual failures. int result = slot.client->publish(topic, qos, retained, payload, strlen(payload), true); if (result < 0) { - static unsigned long last_fail_log = 0; - unsigned long now = millis(); - if (now - last_fail_log > 60000) { - MQTT_DEBUG_PRINTLN("MQTT%d publish failed (result=%d)", index + 1, result); - last_fail_log = now; + // QoS0 packet/raw publishes are best-effort and may be retried from the + // bridge queue; avoid logging transient first-attempt failures here. + if (qos > 0) { + static unsigned long last_fail_log = 0; + unsigned long now = millis(); + if (now - last_fail_log > 60000) { + MQTT_DEBUG_PRINTLN("MQTT%d publish failed (result=%d qos=%u)", index + 1, result, (unsigned)qos); + last_fail_log = now; + } } return false; } @@ -2138,6 +2142,12 @@ void MQTTBridge::processPacketQueue() { int max_per_loop = (_queue_count > 5) ? 5 : 1; unsigned long loop_start_time = millis(); const unsigned long MAX_PROCESSING_TIME_MS = (_queue_count > 5) ? 100 : 30; + static const uint8_t MAX_QOS0_RETRY_ATTEMPTS = 3; + static const unsigned long RETRY_DELAY_BASE_MS = 300UL; + static const unsigned long RETRY_DELAY_JITTER_MS = 200UL; + #ifdef MQTT_DIAG_VERBOSE + static unsigned long last_retry_schedule_log = 0; + #endif while (processed < max_per_loop) { unsigned long elapsed = millis() - loop_start_time; @@ -2151,6 +2161,13 @@ void MQTTBridge::processPacketQueue() { break; // No more packets } + unsigned long now_ms = millis(); + if (queued.next_retry_ms != 0 && now_ms < queued.next_retry_ms) { + // Not ready yet; put it back and stop draining this cycle. + xQueueSend(_packet_queue_handle, &queued, 0); + break; + } + // Update Core 0-owned last-raw-data for publishStatus() — no mutex needed since // _last_raw_data is now written only here (Core 0) and read only by publishStatus() (Core 0). if (!queued.is_tx && queued.has_raw_data && _last_raw_data) { @@ -2161,18 +2178,52 @@ void MQTTBridge::processPacketQueue() { _last_raw_timestamp = millis(); } - publishPacket(&queued.packet_copy, queued.is_tx, - queued.has_raw_data ? queued.raw_data : nullptr, - queued.has_raw_data ? queued.raw_len : 0, - queued.snr, queued.rssi); + bool packet_published = publishPacket(&queued.packet_copy, queued.is_tx, + queued.has_raw_data ? queued.raw_data : nullptr, + queued.has_raw_data ? queued.raw_len : 0, + queued.snr, queued.rssi); taskYIELD(); // allow higher-priority tasks to run between packet publishes // Publish raw if enabled + bool raw_published = false; if (_raw_enabled) { - publishRaw(&queued.packet_copy); + raw_published = publishRaw(&queued.packet_copy); } - _queue_count--; + bool any_published = packet_published || raw_published; + if (!any_published && queued.retry_attempts < MAX_QOS0_RETRY_ATTEMPTS) { + queued.retry_attempts++; + unsigned long retry_delay_ms = RETRY_DELAY_BASE_MS + (now_ms % RETRY_DELAY_JITTER_MS); + queued.next_retry_ms = now_ms + retry_delay_ms; + #ifdef MQTT_DIAG_VERBOSE + if (now_ms - last_retry_schedule_log > 5000UL) { + unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + MQTT_DEBUG_PRINTLN("Retry scheduled: attempt=%u/%u delay=%lu age=%lu q=%u pkt_type=%u packet_ok=%d raw_ok=%d", + (unsigned)queued.retry_attempts, (unsigned)MAX_QOS0_RETRY_ATTEMPTS, + retry_delay_ms, age_ms, (unsigned)uxQueueMessagesWaiting(_packet_queue_handle), + (unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0); + last_retry_schedule_log = now_ms; + } + #endif + if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) { + MQTT_DEBUG_PRINTLN("Retry requeue failed, dropping packet (attempt=%u)", queued.retry_attempts); + } + } else if (!any_published) { + // Intentional: QoS0 best-effort packets are dropped silently in normal + // builds; detailed exhaustion logs are only emitted in verbose mode. + #ifdef MQTT_DIAG_VERBOSE + static unsigned long last_retry_drop_log = 0; + if (now_ms - last_retry_drop_log > 60000UL) { + unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + MQTT_DEBUG_PRINTLN("Packet dropped after retry exhaustion (attempts=%u age=%lu pkt_type=%u packet_ok=%d raw_ok=%d)", + queued.retry_attempts, age_ms, (unsigned)queued.packet_copy.getPayloadType(), + packet_published ? 1 : 0, raw_published ? 1 : 0); + last_retry_drop_log = now_ms; + } + #endif + } + + _queue_count = uxQueueMessagesWaiting(_packet_queue_handle); processed++; } #else @@ -2201,6 +2252,12 @@ void MQTTBridge::processPacketQueue() { int max_per_loop = (_queue_count > 5) ? 5 : 1; unsigned long loop_start_time = millis(); const unsigned long MAX_PROCESSING_TIME_MS = (_queue_count > 5) ? 100 : 30; + static const uint8_t MAX_QOS0_RETRY_ATTEMPTS = 3; + static const unsigned long RETRY_DELAY_BASE_MS = 300UL; + static const unsigned long RETRY_DELAY_JITTER_MS = 200UL; + #ifdef MQTT_DIAG_VERBOSE + static unsigned long last_retry_schedule_log = 0; + #endif while (_queue_count > 0 && processed < max_per_loop) { unsigned long elapsed = millis() - loop_start_time; @@ -2209,6 +2266,10 @@ void MQTTBridge::processPacketQueue() { } QueuedPacket& queued = _packet_queue[_queue_head]; + unsigned long now_ms = millis(); + if (queued.next_retry_ms != 0 && now_ms < queued.next_retry_ms) { + break; + } if (!queued.is_tx && queued.has_raw_data && _last_raw_data) { memcpy(_last_raw_data, queued.raw_data, queued.raw_len); @@ -2218,14 +2279,46 @@ void MQTTBridge::processPacketQueue() { _last_raw_timestamp = millis(); } - publishPacket(&queued.packet_copy, queued.is_tx, - queued.has_raw_data ? queued.raw_data : nullptr, - queued.has_raw_data ? queued.raw_len : 0, - queued.snr, queued.rssi); + bool packet_published = publishPacket(&queued.packet_copy, queued.is_tx, + queued.has_raw_data ? queued.raw_data : nullptr, + queued.has_raw_data ? queued.raw_len : 0, + queued.snr, queued.rssi); // No taskYIELD() on non-ESP32 platforms (non-FreeRTOS, cooperative scheduling not needed) + bool raw_published = false; if (_raw_enabled) { - publishRaw(&queued.packet_copy); + raw_published = publishRaw(&queued.packet_copy); + } + + bool any_published = packet_published || raw_published; + if (!any_published && queued.retry_attempts < MAX_QOS0_RETRY_ATTEMPTS) { + queued.retry_attempts++; + unsigned long retry_delay_ms = RETRY_DELAY_BASE_MS + (now_ms % RETRY_DELAY_JITTER_MS); + queued.next_retry_ms = now_ms + retry_delay_ms; + #ifdef MQTT_DIAG_VERBOSE + if (now_ms - last_retry_schedule_log > 5000UL) { + unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + MQTT_DEBUG_PRINTLN("Retry scheduled: attempt=%u/%u delay=%lu age=%lu q=%d pkt_type=%u packet_ok=%d raw_ok=%d", + (unsigned)queued.retry_attempts, (unsigned)MAX_QOS0_RETRY_ATTEMPTS, + retry_delay_ms, age_ms, _queue_count, + (unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0); + last_retry_schedule_log = now_ms; + } + #endif + break; // keep packet at head for delayed retry + } else if (!any_published) { + // Intentional: QoS0 best-effort packets are dropped silently in normal + // builds; detailed exhaustion logs are only emitted in verbose mode. + #ifdef MQTT_DIAG_VERBOSE + static unsigned long last_retry_drop_log = 0; + if (now_ms - last_retry_drop_log > 60000UL) { + unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + MQTT_DEBUG_PRINTLN("Packet dropped after retry exhaustion (attempts=%u age=%lu pkt_type=%u packet_ok=%d raw_ok=%d)", + queued.retry_attempts, age_ms, (unsigned)queued.packet_copy.getPayloadType(), + packet_published ? 1 : 0, raw_published ? 1 : 0); + last_retry_drop_log = now_ms; + } + #endif } dequeuePacket(); @@ -2326,10 +2419,10 @@ bool MQTTBridge::publishStatus() { return false; } -void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, +bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, const uint8_t* raw_data, int raw_len, float snr, float rssi) { - if (!packet) return; + if (!packet) return false; // Memory pressure check: Skip publishes when there's not enough contiguous // heap for the publish itself (JSON buffer + esp-mqtt outbox frame + WiFi TX @@ -2355,7 +2448,7 @@ void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, max_alloc, (int)PUBLISH_SKIP_MAX_ALLOC_THRESHOLD, _skipped_publishes); last_skip_log = now; } - return; + return false; } _last_memory_check = now; } @@ -2412,24 +2505,29 @@ void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, } if (len > 0) { + bool published = false; char topic[128]; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { if (_slots[i].enabled && _slots[i].client && _slots[i].connected) { if (buildTopicForSlot(i, MSG_PACKETS, topic, sizeof(topic))) { - publishToSlot(i, topic, active_buffer, false); + if (publishToSlot(i, topic, active_buffer, false)) { + published = true; + } } } } + return published; } else { uint8_t packet_type = packet->getPayloadType(); if (packet_type == 4 || packet_type == 9) { MQTT_DEBUG_PRINTLN("Failed to build packet JSON for type=%d (len=%d), packet not published", packet_type, len); } } + return false; } -void MQTTBridge::publishRaw(mesh::Packet* packet) { - if (!packet) return; +bool MQTTBridge::publishRaw(mesh::Packet* packet) { + if (!packet) return false; // Use pre-allocated buffer; fallback to single stack buffer if not available char json_buffer_stack[PUBLISH_JSON_BUFFER_SIZE]; @@ -2452,15 +2550,20 @@ void MQTTBridge::publishRaw(mesh::Packet* packet) { ); if (len > 0) { + bool published = false; char topic[128]; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { if (_slots[i].enabled && _slots[i].client && _slots[i].connected) { if (buildTopicForSlot(i, MSG_RAW, topic, sizeof(topic))) { - publishToSlot(i, topic, active_buffer, false); + if (publishToSlot(i, topic, active_buffer, false)) { + published = true; + } } } } + return published; } + return false; } // --------------------------------------------------------------------------- diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 3dac71223..785812bae 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -125,6 +125,8 @@ private: struct QueuedPacket { mesh::Packet packet_copy; // ~258 bytes, full value copy unsigned long timestamp; + unsigned long next_retry_ms; // Earliest millis() when a failed send may be retried + uint8_t retry_attempts; // Bounded resend attempts for transient QoS0 publish failures bool is_tx; float snr; float rssi; @@ -311,10 +313,10 @@ private: void mqttTaskLoop(); // Main loop for MQTT task void initializeWiFiInTask(); // WiFi initialization moved to task #endif - void publishPacket(mesh::Packet* packet, bool is_tx, + bool publishPacket(mesh::Packet* packet, bool is_tx, const uint8_t* raw_data = nullptr, int raw_len = 0, float snr = 0.0f, float rssi = 0.0f); - void publishRaw(mesh::Packet* packet); + bool publishRaw(mesh::Packet* packet); void queuePacket(mesh::Packet* packet, bool is_tx); void dequeuePacket(); bool isAnySlotConnected(); diff --git a/variants/heltec_v3/platformio.ini b/variants/heltec_v3/platformio.ini index 106a7a18d..82cb619fb 100644 --- a/variants/heltec_v3/platformio.ini +++ b/variants/heltec_v3/platformio.ini @@ -465,6 +465,45 @@ lib_deps = ${Heltec_lora32_v3.lib_deps} ${esp32_ota.lib_deps} +[env:Heltec_WSL3_repeater_observer_mqtt] +extends = Heltec_lora32_v3 +build_flags = + ${Heltec_lora32_v3.build_flags} + -D ADVERT_NAME='"MQTT Observer"' + -D ADVERT_LAT=0.0 + -D ADVERT_LON=0.0 + -D ADMIN_PASSWORD='"password"' + -D MAX_NEIGHBOURS=50 + -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 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"' +# -D WIFI_PWD='"password"' +# -D MQTT_SERVER='"your-mqtt-broker.com"' +# -D MQTT_PORT=1883 +# -D MQTT_USERNAME='"your-username"' +# -D MQTT_PASSWORD='"your-password"' +build_src_filter = ${Heltec_lora32_v3.build_src_filter} + + + + + + + +<../examples/simple_repeater> +lib_deps = + ${Heltec_lora32_v3.lib_deps} + ${esp32_ota.lib_deps} + elims/PsychicMqttClient@^0.2.4 + bblanchon/ArduinoJson + arduino-libraries/NTPClient + JChristensen/Timezone + paulstoffregen/Time@1.6.1 + [env:Heltec_v3_kiss_modem] extends = Heltec_lora32_v3 build_flags =