diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp index 2ae395c8..abe49c88 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp @@ -48,6 +48,17 @@ PsychicMqttClient &PsychicMqttClient::setKeepAlive(int keepAlive) return *this; } +PsychicMqttClient &PsychicMqttClient::setNetworkTimeout(int timeoutMs) +{ +#if ESP_IDF_VERSION_MAJOR == 5 + _mqtt_cfg.network.timeout_ms = timeoutMs; +#else + _mqtt_cfg.network_timeout_ms = timeoutMs; +#endif + _config_dirty = true; + return *this; +} + PsychicMqttClient &PsychicMqttClient::setAutoReconnect(bool reconnect) { #if ESP_IDF_VERSION_MAJOR == 5 @@ -529,17 +540,44 @@ int PsychicMqttClient::publish(const char *topic, int qos, bool retain, const ch if (async) { + // QoS0 async publishes are stored in the esp-mqtt outbox (store=true) so that + // packet topics keep flowing — enqueue(store=false) produced false-failure + // semantics that stalled the packet path. The outbox has no size bound of its + // own (esp-mqtt frees entries only on send-ack or time-based expiry), so on a + // stalled uplink QoS0 frames pile up on internal heap without limit. Cap it + // here: if the outbox is already at/over _outbox_limit, drop this QoS0 message + // and report -2 ("outbox full") so the caller can retry/drop. QoS1/2 (durable, + // low-rate) are never gated. + if (qos == 0 && _outbox_limit > 0 && _client != nullptr && + esp_mqtt_client_get_outbox_size(_client) >= _outbox_limit) + { + _outbox_drops++; + static unsigned long last_full_log = 0; + unsigned long now = millis(); + if (now - last_full_log > 5000) + { + ESP_LOGW(TAG, "Outbox at cap (%u bytes); dropping QoS0 message to topic %s", + (unsigned)_outbox_limit, topic); + last_full_log = now; + } + return -2; + } + ESP_LOGV(TAG, "Enqueuing message to topic %s with QoS %d", topic, qos); - // Hotfix: restore legacy outbox behavior for QoS0 async publishes. - // This avoids false-failure semantics from enqueue(store=false) on some - // connected paths where packet topics stop flowing. bool store_in_outbox = true; - return esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, store_in_outbox); + int result = esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, store_in_outbox); + if (result < 0) _publish_err++; else _publish_ok++; + return result; } else { ESP_LOGV(TAG, "Publishing message to topic %s with QoS %d", topic, qos); - return esp_mqtt_client_publish(_client, topic, payload, length, qos, retain); + // Synchronous write (used for QoS0 packet publishes). A negative result is a real + // send failure (socket error / network_timeout on a stalled link), tracked here so + // callers can surface delivery health without per-message logging. + int result = esp_mqtt_client_publish(_client, topic, payload, length, qos, retain); + if (result < 0) _publish_err++; else _publish_ok++; + return result; } } @@ -557,6 +595,40 @@ esp_mqtt_client_config_t *PsychicMqttClient::getMqttConfig() return &_mqtt_cfg; } +PsychicMqttClient &PsychicMqttClient::setOutboxLimit(size_t bytes) +{ + _outbox_limit = bytes; + return *this; +} + +size_t PsychicMqttClient::getOutboxSize() +{ + if (_client == nullptr) + return 0; + int size = esp_mqtt_client_get_outbox_size(_client); + return size > 0 ? (size_t)size : 0; +} + +size_t PsychicMqttClient::getOutboxLimit() +{ + return _outbox_limit; +} + +unsigned long PsychicMqttClient::getOutboxDrops() +{ + return _outbox_drops; +} + +unsigned long PsychicMqttClient::getPublishOk() +{ + return _publish_ok; +} + +unsigned long PsychicMqttClient::getPublishErr() +{ + return _publish_err; +} + void PsychicMqttClient::_onMqttEventStatic(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) { // Since this is a static function, we need to cast the first argument (void*) back to the class instance type diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.h b/lib/PsychicMqttClient/src/PsychicMqttClient.h index c45cf272..42cf8d5c 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.h +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.h @@ -126,6 +126,18 @@ public: */ PsychicMqttClient &setAutoReconnect(bool reconnect = true); + /** + * @brief Sets the network operation timeout in milliseconds. esp-mqtt aborts a + * network read/write (including a synchronous publish's socket write) if it does + * not complete within this window. A lower value bounds how long a synchronous + * QoS0 publish can block on a stalled/half-open socket before failing and letting + * the slot flip to disconnected. esp-mqtt's default is 10000. + * + * @param timeoutMs Network timeout in milliseconds. + * @return A reference to the PsychicMqttClient instance. + */ + PsychicMqttClient &setNetworkTimeout(int timeoutMs); + /** * @brief Sets the retransmit timeout for unacknowledged QoS 1/2 messages. * esp-mqtt resends an unacked PUBLISH (DUP flag set) every time this @@ -410,6 +422,55 @@ public: */ esp_mqtt_client_config_t *getMqttConfig(); + /** + * @brief Caps the size of the esp-mqtt outbox for async QoS 0 publishes. + * + * QoS 0 async publishes are forced into the esp-mqtt outbox (store=true) so + * packet topics keep flowing, but the outbox has no size bound of its own — + * it only frees entries on send-ack or time-based expiry. On a stalled uplink + * the entries accumulate on internal heap without limit. When the current + * outbox size is at/over this cap, publish() drops the new QoS 0 message + * (returns -2) instead of enqueuing it, applying backpressure. 0 disables the + * cap. QoS 1/2 publishes are never gated by this. + * + * @param bytes Maximum outbox size in bytes, or 0 to disable. + * @return A reference to the PsychicMqttClient instance. + */ + PsychicMqttClient &setOutboxLimit(size_t bytes); + + /** + * @brief Returns the current esp-mqtt outbox size in bytes (0 if the client + * is not initialized). Useful for diagnostics/backpressure monitoring. + * + * @return Current outbox size in bytes. + */ + size_t getOutboxSize(); + + /** + * @brief Returns the configured outbox cap in bytes (0 = disabled). Lets + * callers confirm the cap is actually applied to this client. + */ + size_t getOutboxLimit(); + + /** + * @brief Returns the cumulative count of QoS0 messages dropped because the + * outbox was at/over the cap. Monotonic; useful for backpressure diagnostics. + */ + unsigned long getOutboxDrops(); + + /** + * @brief Cumulative count of publishes that were accepted (enqueued or written + * successfully). Monotonic. Pair with getPublishErr() for delivery-health stats. + */ + unsigned long getPublishOk(); + + /** + * @brief Cumulative count of publishes that failed (negative return from the + * synchronous write or async enqueue — socket error / network timeout). Monotonic. + * A rising value indicates the uplink is dropping publishes. + */ + unsigned long getPublishErr(); + private: esp_mqtt_client_handle_t _client = nullptr; esp_mqtt_client_config_t _mqtt_cfg; @@ -418,6 +479,15 @@ private: bool _stopMqttClient = false; bool _config_dirty = true; + // Runtime cap on the esp-mqtt outbox for QoS 0 async publishes (bytes). + // 0 = disabled. Enforced in publish(); not an esp-mqtt config field. + size_t _outbox_limit = 0; + // Cumulative count of QoS0 publishes dropped because the outbox hit the cap. + unsigned long _outbox_drops = 0; + // Cumulative publish accept/fail counts (any QoS, sync or async path). + unsigned long _publish_ok = 0; + unsigned long _publish_err = 0; + // Multipart message reassembly. _buffer is lazily allocated at connect() time // to match the configured buffer size, then reused for the client's lifetime. // _topic is inline storage, never heap-allocated. diff --git a/src/helpers/CommonCLI_Observer.cpp b/src/helpers/CommonCLI_Observer.cpp index 3b4219f3..695aacef 100644 --- a/src/helpers/CommonCLI_Observer.cpp +++ b/src/helpers/CommonCLI_Observer.cpp @@ -695,6 +695,8 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf start = (int)_atoi(start_arg); } formatMQTTPresetListReply(reply, 160, start); + } else if (memcmp(config, "mqtt.stats", 10) == 0) { + MQTTBridge::formatMqttStatsReply(reply, 160); } else if (memcmp(config, "mqtt.status", 11) == 0) { MQTTBridge::formatMqttStatusReply(reply, 160, &_mqtt_prefs); } else if (memcmp(config, "mqtt.packets", 12) == 0) { diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index aa737d98..15104a65 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -250,6 +250,45 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref snprintf(buf + pos, bufsize - pos, ", q:%d", q); } +// On-demand publish-health + heap snapshot for the `get mqtt.stats` CLI command. +// Same data as the (MQTT_MEMORY_DEBUG-only) periodic logMemoryStatus() line, but +// returned as a reply instead of logged. Per-slot "sN=ok/err": ok = cumulative +// accepted publishes, err = cumulative failures (socket error / network timeout). +// Outbox should read ~0 (QoS0 publishes synchronously); a rising err isolates a +// broker whose uplink is dropping writes. +void MQTTBridge::formatMqttStatsReply(char* buf, size_t bufsize) { + if (buf == nullptr || bufsize == 0) return; + if (s_mqtt_bridge_instance == nullptr || !s_mqtt_bridge_instance->_initialized) { + snprintf(buf, bufsize, "> (bridge not running)"); + return; + } + MQTTBridge* b = s_mqtt_bridge_instance; + + int q = 0; +#ifdef ESP_PLATFORM + if (b->_packet_queue_handle != nullptr) { + q = (int)uxQueueMessagesWaiting(b->_packet_queue_handle); + } +#else + q = b->_queue_count; +#endif + + size_t outbox_total = 0; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (b->_slots[i].client) outbox_total += b->_slots[i].client->getOutboxSize(); + } + + int pos = snprintf(buf, bufsize, "> Free=%d Max=%d q:%d/%d Outbox=%u |", + (int)ESP.getFreeHeap(), (int)ESP.getMaxAllocHeap(), + q, MAX_QUEUE_SIZE, (unsigned)outbox_total); + for (int i = 0; i < RUNTIME_MQTT_SLOTS && pos < (int)bufsize - 1; i++) { + if (!b->_slots[i].enabled || !b->_slots[i].client) continue; + pos += snprintf(buf + pos, bufsize - pos, " s%d=%lu/%lu", i + 1, + b->_slots[i].client->getPublishOk(), + b->_slots[i].client->getPublishErr()); + } +} + uint8_t MQTTBridge::getLastWifiDisconnectReason() { return s_wifi_disconnect_reason; } unsigned long MQTTBridge::getLastWifiDisconnectTime() { return s_wifi_disconnect_time; } @@ -898,6 +937,19 @@ void MQTTBridge::mqttTaskLoop() { #endif unsigned long now = millis(); + + // Periodic heap + publish-health snapshot. Gated behind MQTT_MEMORY_DEBUG (a + // dedicated diagnostics flag, NOT enabled on production or plain MQTT_DEBUG builds) + // so it stays off by default — the same data is available on demand via the + // `get mqtt.stats` CLI command (formatMqttStatsReply / logMemoryStatus()). + #ifdef MQTT_MEMORY_DEBUG + static unsigned long last_mem_log = 0; + if (now - last_mem_log >= 30000) { + last_mem_log = now; + logMemoryStatus(); + } + #endif + bool wifi_just_connected = handleWiFiConnection(now); if (wifi_just_connected) { // WiFi recovered — reset last_reconnect_attempt for disconnected slots so they @@ -1742,14 +1794,23 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload return false; } - // QoS 0 for the high-rate packet/raw publish paths: no PUBACK, no outbox store, - // no per-message heap alloc — critical for non-PSRAM fragmentation. QoS 1 is used - // only for low-rate retained status messages where delivery matters. + // Publish path by QoS: + // - QoS 0 (high-rate packets/raw): SYNCHRONOUS (async=false → esp_mqtt_client_publish), + // which writes straight to the socket. The async/outbox path drains only one queued + // item per esp-mqtt task loop (~1 msg/s/conn, gated by the 1s poll_read), so under + // even light packet load the outbox pins at its cap and drops ~20-30%. A synchronous + // write bypasses that drain ceiling entirely and does not store in the outbox. It can + // block the (Core-0, prio-1) MQTT task on a stalled socket, but only up to + // network_timeout_ms (lowered in optimizeMqttClientConfig); mesh RX (Core 1) and the + // WiFi/TCP stack (higher-prio system tasks) are unaffected, and a failed write flips + // the slot to disconnected so subsequent packets skip it. + // - QoS 1 (low-rate retained status): async, so it keeps the durable outbox + retransmit. // - // esp_mqtt_client_enqueue return convention: QoS 0 returns msg_id == 0 on success - // (no tracking since there's no PUBACK); QoS 1/2 return a positive msg_id. Negative - // values (-1 generic failure, -2 outbox full) are the only actual failures. - int result = slot.client->publish(topic, qos, retained, payload, strlen(payload), true); + // Return convention: QoS 0 sync publish returns msg_id == 0 on success (no PUBACK + // tracking). Negative values (-1 write/failure) are the only actual failures; the queue + // retry/drop path below handles them. + bool async = (qos > 0); + int result = slot.client->publish(topic, qos, retained, payload, strlen(payload), async); if (result < 0) { // QoS0 packet/raw publishes are best-effort and may be retried from the // bridge queue; avoid logging transient first-attempt failures here. @@ -3408,6 +3469,23 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_ client->setBufferSize(MQTT_CLIENT_BUFFER_SIZE); + // Bound how long a synchronous QoS0 publish (see publishToSlot) can block the MQTT + // task on a stalled/half-open socket before esp-mqtt aborts the write. Default is 10s; + // 2.5s lets a first stall resolve fast (write fails → slot flips to disconnected → + // subsequent packets skip it) without holding up publishing to the other slots. Mesh + // RX (Core 1) and the WiFi/TCP stack are unaffected by this block regardless. + client->setNetworkTimeout(2500); + + // Dormant safety net: cap the esp-mqtt outbox for any residual async QoS0 path. QoS0 + // packets now publish synchronously (store=false, no outbox), so this normally never + // engages, but it bounds internal-heap growth if a QoS0 message ever takes the async + // path. Non-PSRAM (outbox on internal heap) gets the tighter cap. +#if defined(BOARD_HAS_PSRAM) + client->setOutboxLimit(16384); +#else + client->setOutboxLimit(8192); +#endif + // Access ESP-IDF config to optimize additional settings esp_mqtt_client_config_t* config = client->getMqttConfig(); if (config) { @@ -3425,8 +3503,31 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_ } void MQTTBridge::logMemoryStatus() { - MQTT_DEBUG_PRINTLN("Memory: Free=%d, Max=%d, Queue=%d/%d", - ESP.getFreeHeap(), ESP.getMaxAllocHeap(), _queue_count, MAX_QUEUE_SIZE); + // QoS0 packets now publish synchronously, so the outbox stays ~0 and is only a sanity + // check (a non-zero total would mean the QoS1 status path is backing up or the dormant + // async cap engaged). The live signal is per-slot publish health: ok = cumulative + // accepted writes, err = cumulative failures (socket error / network_timeout on a + // stalled link). A rising err on a slot means that broker's uplink is dropping packets; + // ok climbing with err flat is healthy delivery. + char pub_detail[200]; + size_t pos = 0; + size_t outbox_total = 0; + pub_detail[0] = '\0'; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].client) { + outbox_total += _slots[i].client->getOutboxSize(); + if (_slots[i].enabled) { + pos += snprintf(pub_detail + pos, sizeof(pub_detail) - pos, "%ss%d=%lu/%lu", + pos ? " " : "", i + 1, + _slots[i].client->getPublishOk(), + _slots[i].client->getPublishErr()); + if (pos >= sizeof(pub_detail)) break; + } + } + } + MQTT_DEBUG_PRINTLN("Memory: Free=%d, Max=%d, Queue=%d/%d, Outbox=%u | pub(ok/err) %s", + ESP.getFreeHeap(), ESP.getMaxAllocHeap(), _queue_count, MAX_QUEUE_SIZE, + (unsigned)outbox_total, pub_detail); } // --------------------------------------------------------------------------- diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 377ea51c..69782bd4 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -468,6 +468,9 @@ public: * (for LoRa). Returns false if the bridge is not running. */ bool ntpDiag(char* reply, size_t reply_size, bool verbose); static void formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPrefs* obs); + /** On-demand publish-health + heap snapshot for `get mqtt.stats` (per-slot ok/err, + * outbox size, free/max heap, queue depth). */ + static void formatMqttStatsReply(char* buf, size_t bufsize); /** True when WiFi is set and at least one MQTT slot can run (preset + custom host if needed). */ static bool isConfigValid(const MQTTPrefs* obs); static void formatSlotDiagReply(char* buf, size_t bufsize, int slot_index); diff --git a/variants/heltec_v3/platformio.ini b/variants/heltec_v3/platformio.ini index 4bd3819a..2d13237b 100644 --- a/variants/heltec_v3/platformio.ini +++ b/variants/heltec_v3/platformio.ini @@ -127,7 +127,8 @@ build_flags = -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 - -D MQTT_MEMORY_DEBUG=1 +; Periodic 30s heap/pub-stats serial log — enable only for debugging (use `get mqtt.stats` on demand instead). +; -D MQTT_MEMORY_DEBUG=1 ; Keep default observer profile less verbose to reduce runtime contention. ; -D MESH_PACKET_LOGGING=1 ; -D MESH_DEBUG=1 @@ -486,7 +487,8 @@ build_flags = -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 - -D MQTT_MEMORY_DEBUG=1 +; Periodic 30s heap/pub-stats serial log — enable only for debugging (use `get mqtt.stats` on demand instead). +; -D MQTT_MEMORY_DEBUG=1 ; Keep default observer profile less verbose to reduce runtime contention. ; -D MESH_PACKET_LOGGING=1 ; -D MESH_DEBUG=1