Merge pull request #26 from agessaman/fix/mqtt-outbox-bound

fix(mqtt): bound QoS0 outbox and publish synchronously to stop heap exhaustion + packet drops

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

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.

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) {

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);
 }
 
 // ---------------------------------------------------------------------------

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);

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