diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp index 2ae395c8..69cb35b1 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp @@ -529,10 +529,30 @@ 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); } @@ -557,6 +577,30 @@ 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; +} + 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..a9e273ba 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.h +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.h @@ -410,6 +410,42 @@ 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(); + private: esp_mqtt_client_handle_t _client = nullptr; esp_mqtt_client_config_t _mqtt_cfg; @@ -418,6 +454,12 @@ 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; + // 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/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index aa737d98..1e9b218c 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -898,6 +898,16 @@ void MQTTBridge::mqttTaskLoop() { #endif unsigned long now = millis(); + + // Periodic heap + outbox snapshot (compiles to nothing unless MQTT_DEBUG is set). + // Outbox= is the value to watch: it should sit near 0 on a healthy uplink and + // plateau at the configured cap (not climb) during a stall. + static unsigned long last_mem_log = 0; + if (now - last_mem_log >= 30000) { + last_mem_log = now; + logMemoryStatus(); + } + bool wifi_just_connected = handleWiFiConnection(now); if (wifi_just_connected) { // WiFi recovered — reset last_reconnect_attempt for disconnected slots so they @@ -1742,13 +1752,16 @@ 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. + // QoS 0 for the high-rate packet/raw publish paths: no PUBACK, so delivery is + // best-effort. These still enter the esp-mqtt outbox (store=true, needed so packet + // topics keep flowing), but the client bounds that outbox via setOutboxLimit() to + // stop QoS0 frames accumulating on internal heap during a stalled uplink — over the + // cap, publish() returns -2 and the queue retry/drop path below handles it. QoS 1 is + // used only for low-rate retained status messages where delivery matters. // // 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. + // values (-1 generic failure, -2 outbox full / over cap) are the only actual failures. int result = slot.client->publish(topic, qos, retained, payload, strlen(payload), true); if (result < 0) { // QoS0 packet/raw publishes are best-effort and may be retried from the @@ -3408,6 +3421,22 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_ client->setBufferSize(MQTT_CLIENT_BUFFER_SIZE); + // Bound the esp-mqtt outbox for high-rate QoS0 packet/raw publishes. Those go in + // with store=true (so packet topics keep flowing) but the outbox has no size limit + // of its own — it frees entries only on send-ack or ~30s expiry, so a stalled uplink + // lets QoS0 frames accumulate on internal heap without bound. Above this cap the + // client drops the new QoS0 message (returns -2), and processPacketQueue's existing + // retry/drop path applies backpressure. Non-PSRAM is the fragmentation-sensitive case + // (outbox lives on internal heap), so it gets the tighter cap. Under healthy operation + // the outbox drains to ~0 in ms, so this only bites during a stall. Note: esp-mqtt's + // own outbox.limit config is not used — its enqueue path does not reliably enforce it + // for QoS0, and this app-level guard fires before enqueue regardless of IDF version. +#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 +3454,30 @@ 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); + // The esp-mqtt outbox is the structure that grew unbounded before the cap. The cap + // (setOutboxLimit) is enforced PER CLIENT, so log it per slot — a summed total hides + // whether any single slot has blown past its cap. Each entry is size/limit; drops is + // the cumulative count of QoS0 messages rejected at the cap (backpressure firing). + // Under a healthy uplink size sits near 0; under a stall/saturation it should plateau + // at limit (not climb) and drops should increase. + char outbox_detail[160]; + size_t pos = 0; + size_t outbox_total = 0; + outbox_detail[0] = '\0'; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].client) { + size_t sz = _slots[i].client->getOutboxSize(); + outbox_total += sz; + pos += snprintf(outbox_detail + pos, sizeof(outbox_detail) - pos, "%ss%d=%u/%u(d%lu)", + pos ? " " : "", i, (unsigned)sz, + (unsigned)_slots[i].client->getOutboxLimit(), + _slots[i].client->getOutboxDrops()); + if (pos >= sizeof(outbox_detail)) break; + } + } + MQTT_DEBUG_PRINTLN("Memory: Free=%d, Max=%d, Queue=%d/%d, Outbox=%u [%s]", + ESP.getFreeHeap(), ESP.getMaxAllocHeap(), _queue_count, MAX_QUEUE_SIZE, + (unsigned)outbox_total, outbox_detail); } // ---------------------------------------------------------------------------