fix(mqtt): bound esp-mqtt outbox for QoS0 publishes
QoS0 packet/raw publishes are forced into the esp-mqtt outbox (store=true, async) so packet topics keep flowing, but the outbox has no size bound of its own — esp-mqtt frees entries only on send-ack or ~30s expiry. On a stalled or slow uplink (socket still "connected") QoS0 frames accumulate on internal heap without limit, driving the heap exhaustion/fragmentation seen in the field. Cap the outbox at the application level: PsychicMqttClient::setOutboxLimit() records a per-client byte cap, and publish() drops a QoS0 message (returns -2) when esp_mqtt_client_get_outbox_size() is already at/over the cap, before enqueuing. The bridge's existing processPacketQueue retry/drop path handles the -2 as backpressure. Caps: 16 KiB PSRAM / 8 KiB non-PSRAM (outbox lives on internal heap, so non-PSRAM is the fragmentation-sensitive case). Portable across IDF 4.4 and 5 via esp_mqtt_client_get_outbox_size(); esp-mqtt's own outbox.limit config is not used (its enqueue path does not reliably enforce it for QoS0, and the app-level guard fires before enqueue regardless). Adds getOutboxSize()/getOutboxLimit()/getOutboxDrops() and surfaces per-slot outbox size/cap/drops via a throttled logMemoryStatus() in the MQTT task loop (MQTT_DEBUG-gated) to confirm the bound on-target.
This commit is contained in:
1 parent
ba1dbefb59
commit
b7c145929f
3 files changed
+146
-9
No files matched your search
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
Reference in new issue
Block a user