From c28548aecf7cdb63d0d9ef5981604e729704a9e0 Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 3 Aug 2026 22:15:12 -0700 Subject: [PATCH 1/9] perf(mqtt): tranche 1 memory reductions in the bridge Five independent, behavior-preserving reductions. Measured on Heltec_v3_repeater_observer_mqtt (non-PSRAM) and ThinkNode_M7_repeater_observer_mqtt (PSRAM), 267/267 native tests green. Drop the unused static-task bookkeeping. StaticTask_t _mqtt_task_tcb (344 B) and StackType_t* _mqtt_task_stack were never used: there is no xTaskCreateStatic call, the pointer was assigned nullptr immediately before xTaskCreatePinnedToCore, and the two psram_free() calls on it were dead. Size the wire-format scratch buffers from the protocol maximum. raw_hex[1024] at three sites becomes 2*MAX_TRANS_UNIT+1, and raw_buf/reconstructed[512] become MAX_TRANS_UNIT. Both writeTo() sites now check getRawLength() first: writeTo() does not bounds-check and returns uint8_t, so the old 512-byte buffers were the only thing absorbing a malformed payload_len, and the post-hoc "raw_len > sizeof(buf)" test ran after the overrun. Pass the already-known serialized length into publishToSlot() instead of re-running strlen() per destination slot (up to 2 KB per packet per slot, and NEIGHBORS_JSON_BUFFER_SIZE per neighbor snapshot). Same for the direct publish in publishStatusToSlot(). Share one JSON buffer and one document across packet, raw, and status. All publish paths serialize on the bridge task, so the separate status buffer and document were never concurrent. Status keeps STATUS_JSON_BUFFER_SIZE as its serialization ceiling, so which oversized status documents get dropped is unchanged. Route the document's pools through a PSRAM-preferring allocator. Under ArduinoJson 7 StaticJsonDocument is a deprecated empty subclass of JsonDocument whose template argument only feeds capacity(); the object is 64 B and each pool block (4096 B here) came from plain malloc(), i.e. the internal DRAM the mbedTLS working set needs. Mirrors NeighborsDocAllocator. The old comment claiming an inline pool has been corrected. Measured: sizeof(MQTTBridge) non-PSRAM 13208 -> 12032 B (-1176, internal heap) PSRAM 10492 -> 10080 B (-412, plus one fewer 768 B PSRAM allocation) stack, non-PSRAM buildPacketJSON[FromRaw] 1264 -> 736 B buildRawJSON 1136 -> 608 B publishPacket 688 -> ~432 B packetToHex 560 -> ~304 B deepest publish chain ~2.6 -> ~1.8 KB of 8 KB flash 1592513 -> 1592573 B (+60) static RAM 74656 B unchanged -- MQTTBridge is heap-allocated, so these savings are internal heap, not the linker figure Deferred from the review: the QueuedPacket wire-only redesign (reward is 1.56 KB on non-PSRAM only, and Packet::readFrom() rejects payload_len == 0), demand-driven slot clients/JWT tokens, and pool retention across publishes -- JsonDocument::to() always calls clear(), which destroys pools, so "retain pools by clearing the root object" needs a string-pool lifetime analysis first. --- src/helpers/MQTTMessageBuilder.cpp | 23 ++--- src/helpers/MQTTMessageBuilder.h | 8 ++ src/helpers/bridges/MQTTBridge.cpp | 148 +++++++++++++++++------------ src/helpers/bridges/MQTTBridge.h | 50 ++++++---- 4 files changed, 142 insertions(+), 87 deletions(-) diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index 3ec13277..ebe05120 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -182,10 +182,9 @@ int MQTTMessageBuilder::buildPacketJSON( } // Convert packet to hex - // MAX_TRANS_UNIT is 255 bytes, hex = 510 chars, but allow for larger with headers - char raw_hex[1024]; + char raw_hex[WIRE_HEX_SCRATCH_SIZE]; packetToHex(packet, raw_hex, sizeof(raw_hex)); - + // Get packet characteristics int packet_type = packet->getPayloadType(); const char* route_str = getRouteTypeString(packet->isRouteDirect() ? 1 : 0); @@ -258,9 +257,9 @@ int MQTTMessageBuilder::buildPacketJSONFromRaw( strcpy(date_str, "01/01/2024"); } - // Convert raw radio data to hex (this includes radio headers) - // MAX_TRANS_UNIT is 255 bytes, hex = 510 chars, but allow for larger with headers - char raw_hex[1024]; + // Convert raw radio data to hex (this includes radio headers). bytesToHex() emits + // an empty string rather than truncating if raw_len exceeds the protocol maximum. + char raw_hex[WIRE_HEX_SCRATCH_SIZE]; bytesToHex(raw_data, raw_len, raw_hex, sizeof(raw_hex)); // Get packet characteristics from the parsed packet @@ -314,10 +313,9 @@ int MQTTMessageBuilder::buildRawJSON( formatIsoTimestampForMqtt(now_tv.tv_sec, now_tv.tv_usec, timezone, timestamp, sizeof(timestamp)); // Convert packet to hex - // MAX_TRANS_UNIT is 255, so max hex size is 510 chars + null = 511 bytes - char raw_hex[1024]; + char raw_hex[WIRE_HEX_SCRATCH_SIZE]; packetToHex(packet, raw_hex, sizeof(raw_hex)); - + return buildRawMessage(origin, origin_id, timestamp, raw_hex, buffer, buffer_size); } @@ -356,9 +354,12 @@ void MQTTMessageBuilder::packetToHex(mesh::Packet* packet, char* hex, size_t hex hex[0] = '\0'; // Serialize full on-air/wire format using Packet::writeTo() // This includes header, transport codes (if present), path_len, path, and payload - uint8_t raw_buf[512]; + uint8_t raw_buf[WIRE_SCRATCH_SIZE]; + // writeTo() neither bounds-checks nor can report an overrun (it returns uint8_t), + // so reject an over-long packet on getRawLength() before writing anything. + if (packet->getRawLength() > (int)sizeof(raw_buf)) return; uint8_t raw_len = packet->writeTo(raw_buf); - if (raw_len == 0 || raw_len > sizeof(raw_buf)) return; + if (raw_len == 0) return; // Check if hex buffer is large enough (2 hex chars per byte + null terminator) if (hex_size < (size_t)raw_len * 2 + 1) return; diff --git a/src/helpers/MQTTMessageBuilder.h b/src/helpers/MQTTMessageBuilder.h index 2d1fc805..7ef35a1a 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -21,6 +21,14 @@ */ class MQTTMessageBuilder { public: + // Wire-format scratch sizing, from the protocol maximum: Packet::writeTo() returns + // uint8_t, so MAX_TRANS_UNIT is the hard ceiling, and hex is 2 chars/byte + NUL. + static const size_t WIRE_SCRATCH_SIZE = MAX_TRANS_UNIT; + static const size_t WIRE_HEX_SCRATCH_SIZE = 2 * MAX_TRANS_UNIT + 1; + + static_assert(1 + 4 + 1 + MAX_PATH_SIZE + MAX_PACKET_PAYLOAD <= MAX_TRANS_UNIT, + "serialized packet no longer fits MAX_TRANS_UNIT — resize the wire scratch buffers"); + /** * Format the MQTT JSON `timestamp` field (same rule for status, packet, raw). * Always UTC with an explicit "+00:00" offset, ISO-8601 diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 9f1ecab3..eafa2321 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -177,6 +177,37 @@ static void psram_free(void* ptr) { #endif } +static void* psram_realloc(void* ptr, size_t new_size) { + if (new_size == 0) { + psram_free(ptr); + return nullptr; + } +#if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) + void* p = heap_caps_realloc(ptr, new_size, MALLOC_CAP_SPIRAM); + if (p != nullptr) return p; + // A block that fell back to internal DRAM on allocation (PSRAM exhausted) cannot + // be grown in PSRAM; retry there rather than reporting failure. + return heap_caps_realloc(ptr, new_size, MALLOC_CAP_INTERNAL); +#else + return realloc(ptr, new_size); +#endif +} + +// Shared JSON document pools follow the same PSRAM-first policy as the bridge's +// text buffers. ArduinoJson calls reallocate() when shrinking its pool list and +// asserts the result is non-null for a shrink, which both branches above satisfy. +void* MQTTBridge::JsonScratchAllocator::allocate(size_t size) { + return psram_malloc(size); +} + +void MQTTBridge::JsonScratchAllocator::deallocate(void* ptr) { + psram_free(ptr); +} + +void* MQTTBridge::JsonScratchAllocator::reallocate(void* ptr, size_t new_size) { + return psram_realloc(ptr, new_size); +} + // Time (millis()) when WiFi was last seen connected; 0 when disconnected. Used for get wifi.status uptime. static unsigned long s_wifi_connected_at = 0; @@ -609,7 +640,7 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg #endif _last_raw_len(0), _last_snr(0), _last_rssi(0), _last_raw_timestamp(0), #if defined(BOARD_HAS_PSRAM) - _publish_json_buffer(nullptr), _status_json_buffer(nullptr), + _json_scratch_buffer(nullptr), #endif _identity(identity), _cached_has_connected_slots(false), @@ -625,7 +656,7 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _last_slot_reconnect_ms(0) #ifdef ESP_PLATFORM , _packet_queue_handle(nullptr), _mqtt_task_handle(nullptr), - _mqtt_task_stack(nullptr), _packet_queue_storage(nullptr) + _packet_queue_storage(nullptr) #else , _queue_head(0), _queue_tail(0) #endif @@ -712,8 +743,8 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg #if !defined(BOARD_HAS_PSRAM) memset(_last_raw_data, 0, sizeof(_last_raw_data)); #endif - // JSON document scratch space is now a StaticJsonDocument inline class member — - // no heap allocation needed; reused via doc.clear() on every publish. + // The shared JSON document needs no setup here: its pools are allocated lazily on + // the first publish through _json_allocator and released by releaseRuntimeBuffers(). } void MQTTBridge::allocateRuntimeBuffers() { @@ -723,14 +754,11 @@ void MQTTBridge::allocateRuntimeBuffers() { // next begin() will retry only the missing buffer. _last_raw_data = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( _last_raw_data, LAST_RAW_DATA_SIZE, psram_malloc)); - _publish_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( - _publish_json_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc)); - _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( - _status_json_buffer, STATUS_JSON_BUFFER_SIZE, psram_malloc)); - MQTT_DEBUG_PRINTLN("Runtime buffers: raw=%s publish=%s status=%s", + _json_scratch_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( + _json_scratch_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc)); + MQTT_DEBUG_PRINTLN("Runtime buffers: raw=%s json=%s", _last_raw_data ? "PSRAM" : "unavailable", - _publish_json_buffer ? "PSRAM" : "stack fallback", - _status_json_buffer ? "PSRAM" : "stack fallback"); + _json_scratch_buffer ? "PSRAM" : "stack fallback"); #endif #if defined(WITH_MQTT_NEIGHBORS) @@ -750,12 +778,15 @@ void MQTTBridge::releaseRuntimeBuffers() { #if defined(BOARD_HAS_PSRAM) _last_raw_data = static_cast(MQTTRuntimeBufferLifecycle::release( _last_raw_data, psram_free)); - _publish_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( - _publish_json_buffer, psram_free)); - _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( - _status_json_buffer, psram_free)); + _json_scratch_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( + _json_scratch_buffer, psram_free)); #endif + // Drop the shared document's pools with the buffers. clear() destroys every pool + // and resets the list to its inline array; the next publish reallocates. Holding + // 4 KB of pool across a stopped bridge is pure overhead. + _json_scratch_doc.clear(); + #if defined(WITH_MQTT_NEIGHBORS) // Paired with the unconditional allocation in allocateRuntimeBuffers(). _neighbors_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( @@ -961,9 +992,8 @@ void MQTTBridge::begin() { #define MQTT_TASK_PRIORITY 1 #endif - // Task stack: use dynamic allocation (internal RAM). PSRAM stack was disabled because it - // causes resets on some boards (e.g. Heltec V4) when the task runs from PSRAM stack. - _mqtt_task_stack = nullptr; + // Task stack: dynamic allocation (internal RAM). A PSRAM-backed stack was tried and + // reverted — it resets some boards (e.g. Heltec V4) when the task runs from PSRAM. _mqtt_task_handle = nullptr; // Clear the cooperative-stop handshake before the new task starts reading it. // deliverStop() leaves _stop_requested latched true after a stop cycle, so a @@ -982,8 +1012,6 @@ void MQTTBridge::begin() { if (create_result != pdPASS) _mqtt_task_handle = nullptr; if (_mqtt_task_handle == nullptr) { MQTT_DEBUG_PRINTLN("Failed to create MQTT task!"); - psram_free(_mqtt_task_stack); - _mqtt_task_stack = nullptr; vQueueDelete(_packet_queue_handle); _packet_queue_handle = nullptr; #if defined(BOARD_HAS_PSRAM) @@ -1083,7 +1111,7 @@ void MQTTBridge::end() { #endif // Timezone is inline class storage (_timezone_storage) — nothing to delete. - // JSON documents are StaticJsonDocument inline members — no heap to free. + // The shared JSON document's pools were freed by releaseRuntimeBuffers() above. _initialized = false; _slots_setup_done = false; // Reset so deferred setup runs again on next begin() MQTT_DEBUG_PRINTLN("MQTT Bridge stopped (%s)", @@ -1137,10 +1165,6 @@ void MQTTBridge::LifecycleOps::releaseResources() { // dynamically-allocated stack/TCB in the idle task. b->_mqtt_task_handle = nullptr; - // Free the PSRAM task stack (nullptr for dynamic tasks — no-op). - psram_free(b->_mqtt_task_stack); - b->_mqtt_task_stack = nullptr; - // Drain and delete the FreeRTOS packet queue (value-copied packets, no // external pointers to clean up). Safe on Core 1: not a TLS resource. if (b->_packet_queue_handle != nullptr) { @@ -1585,7 +1609,7 @@ void MQTTBridge::initSlotClients() { updateCachedConnectionStatus(); // bool store — safe from this (esp-mqtt) task // This callback runs on the client's esp-mqtt event task, not the bridge // task. Do NOT build/publish status here: publishStatusToSlot() writes the - // shared _status_json_doc/_status_json_buffer/_origin that the periodic + // shared _json_scratch_doc/_json_scratch_buffer/_origin that the periodic // publishStatus() uses on the bridge task, and two slots' callbacks could // race each other over them. Marshal the publish onto the bridge task via a // per-slot flag (see mqttTaskLoop consumer / A2). @@ -2169,7 +2193,7 @@ bool MQTTBridge::createSlotAuthToken(int index) { return false; } -bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload, bool retained, uint8_t qos) { +bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload, size_t payload_len, bool retained, uint8_t qos) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; if (!slot.client || !slot.connected) { @@ -2197,7 +2221,7 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload // 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); + int result = slot.client->publish(topic, qos, retained, payload, (int)payload_len, 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. @@ -2214,11 +2238,11 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload return true; } -bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, bool retained, uint8_t qos) { +bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, size_t payload_len, bool retained, uint8_t qos) { bool published = false; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { if (_slots[i].enabled && _slots[i].client && _slots[i].connected) { - if (publishToSlot(i, topic, payload, retained, qos)) { + if (publishToSlot(i, topic, payload, payload_len, retained, qos)) { published = true; } } @@ -2280,15 +2304,16 @@ void MQTTBridge::publishStatusToSlot(int index) { } // Reuse pre-allocated buffer to avoid heap alloc/free churn under memory pressure. - // _status_json_doc/_status_json_buffer/_origin are shared with publishStatus(); - // both callers run only on the bridge task (this function is reached solely via - // the _status_publish_pending consumer in mqttTaskLoop, never from the onConnect - // callback thread — see A2), so the accesses are serialized and need no mutex. + // _json_scratch_doc/_json_scratch_buffer/_origin are shared with publishStatus() and + // with the packet/raw paths; every one of them runs only on the bridge task (this + // function is reached solely via the _status_publish_pending consumer in + // mqttTaskLoop, never from the onConnect callback thread — see A2), so the accesses + // are serialized and need no mutex. #if defined(BOARD_HAS_PSRAM) char fallback_status_buffer[STATUS_JSON_BUFFER_SIZE]; - char* json_buffer = (_status_json_buffer != nullptr) ? _status_json_buffer : fallback_status_buffer; + char* json_buffer = (_json_scratch_buffer != nullptr) ? _json_scratch_buffer : fallback_status_buffer; #else - char* json_buffer = _status_json_buffer; + char* json_buffer = _json_scratch_buffer; #endif char origin_id[65]; @@ -2339,7 +2364,7 @@ void MQTTBridge::publishStatusToSlot(int index) { int internal_heap_free = (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL); int len = MQTTMessageBuilder::buildStatusMessage( - _status_json_doc, + _json_scratch_doc, _origin, origin_id, _board_model, _firmware_version, radio_info, client_version, "online", timestamp, json_buffer, STATUS_JSON_BUFFER_SIZE, battery_mv, uptime_secs, errors, _queue_count, noise_floor, @@ -2354,7 +2379,7 @@ void MQTTBridge::publishStatusToSlot(int index) { // on-connect status must not force retain=true. Custom slots default to // non-retained here too, keeping both status paths consistent. bool use_retain = slot.preset ? slot.preset->allow_retain : false; - int result = slot.client->publish(status_topic, 1, use_retain, json_buffer, strlen(json_buffer)); + int result = slot.client->publish(status_topic, 1, use_retain, json_buffer, len); if (result <= 0) { MQTT_DEBUG_PRINTLN("MQTT%d status publish failed", index + 1); } @@ -3144,12 +3169,12 @@ bool MQTTBridge::publishStatus() { refreshOriginFromPrefs(); // Reuse pre-allocated buffer to avoid heap alloc/free churn under memory pressure. - // _status_json_buffer and _last_raw_data are both Core 0-owned; no mutex needed. + // _json_scratch_buffer and _last_raw_data are both Core 0-owned; no mutex needed. #if defined(BOARD_HAS_PSRAM) char fallback_status_buffer[STATUS_JSON_BUFFER_SIZE]; - char* json_buffer = (_status_json_buffer != nullptr) ? _status_json_buffer : fallback_status_buffer; + char* json_buffer = (_json_scratch_buffer != nullptr) ? _json_scratch_buffer : fallback_status_buffer; #else - char* json_buffer = _status_json_buffer; + char* json_buffer = _json_scratch_buffer; #endif char origin_id[65]; char timestamp[40]; @@ -3199,7 +3224,7 @@ bool MQTTBridge::publishStatus() { int internal_heap_free = (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL); int len = MQTTMessageBuilder::buildStatusMessage( - _status_json_doc, + _json_scratch_doc, _origin, origin_id, _board_model, _firmware_version, radio_info, client_version, "online", timestamp, json_buffer, STATUS_JSON_BUFFER_SIZE, battery_mv, uptime_secs, errors, _queue_count, noise_floor, @@ -3217,7 +3242,7 @@ bool MQTTBridge::publishStatus() { if (buildTopicForSlot(i, MSG_STATUS, topic, sizeof(topic))) { any_slot_wants_status = true; bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false; - if (publishToSlot(i, topic, json_buffer, use_retain, 1)) { + if (publishToSlot(i, topic, json_buffer, (size_t)len, use_retain, 1)) { published = true; } } @@ -3291,15 +3316,15 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, char json_buffer_stack[PUBLISH_JSON_BUFFER_SIZE]; char* active_buffer; size_t active_buffer_size; - if (_publish_json_buffer != nullptr) { - active_buffer = _publish_json_buffer; + if (_json_scratch_buffer != nullptr) { + active_buffer = _json_scratch_buffer; active_buffer_size = PUBLISH_JSON_BUFFER_SIZE; } else { active_buffer = json_buffer_stack; active_buffer_size = PUBLISH_JSON_BUFFER_SIZE; } #else - char* active_buffer = _publish_json_buffer; + char* active_buffer = _json_scratch_buffer; const size_t active_buffer_size = PUBLISH_JSON_BUFFER_SIZE; #endif char origin_id[65]; @@ -3317,33 +3342,36 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, if (raw_data && raw_len > 0) { float score = (_radio && !is_tx) ? _radio->packetScore(snr, raw_len) : NAN; len = MQTTMessageBuilder::buildPacketJSONFromRaw( - _packet_json_doc, + _json_scratch_doc, raw_data, raw_len, packet, is_tx, _origin, origin_id, snr, rssi, score, _timezone, active_buffer, active_buffer_size ); } else if (!is_tx && _last_raw_data && _last_raw_len > 0 && (millis() - _last_raw_timestamp) < 1000) { float score = _radio ? _radio->packetScore(_last_snr, _last_raw_len) : NAN; len = MQTTMessageBuilder::buildPacketJSONFromRaw( - _packet_json_doc, + _json_scratch_doc, _last_raw_data, _last_raw_len, packet, is_tx, _origin, origin_id, _last_snr, _last_rssi, score, _timezone, active_buffer, active_buffer_size ); } else { // Reconstruct wire-format bytes from packet (same as MQTTMessageBuilder::packetToHex). - // This path is used on non-PSRAM boards where raw_data is not stored in the queue, - // and ensures the "raw" hex field and SNR/RSSI are accurate in the JSON output. - uint8_t reconstructed[512]; - uint8_t rlen = packet->writeTo(reconstructed); + // Reached when the queued item carried no captured raw frame, so the "raw" hex field + // is re-serialized rather than dropped. Length-guarded for the same reason. + uint8_t reconstructed[MQTTMessageBuilder::WIRE_SCRATCH_SIZE]; + uint8_t rlen = 0; + if (packet->getRawLength() <= (int)sizeof(reconstructed)) { + rlen = packet->writeTo(reconstructed); + } if (rlen > 0) { float score = (_radio && !is_tx) ? _radio->packetScore(snr, rlen) : NAN; len = MQTTMessageBuilder::buildPacketJSONFromRaw( - _packet_json_doc, + _json_scratch_doc, reconstructed, rlen, packet, is_tx, _origin, origin_id, snr, rssi, score, _timezone, active_buffer, active_buffer_size ); } else { len = MQTTMessageBuilder::buildPacketJSON( - _packet_json_doc, + _json_scratch_doc, packet, is_tx, _origin, origin_id, _timezone, active_buffer, active_buffer_size ); } @@ -3359,7 +3387,7 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, if ((eligible_slots & static_cast(1u << i)) != 0 && _slots[i].enabled && _slots[i].client && _slots[i].connected) { if (buildTopicForSlot(i, MSG_PACKETS, topic, sizeof(topic))) { - if (publishToSlot(i, topic, active_buffer, false)) { + if (publishToSlot(i, topic, active_buffer, (size_t)len, false)) { published = true; } } @@ -3391,15 +3419,15 @@ bool MQTTBridge::publishRaw(mesh::Packet* packet, bool& has_eligible_target) { char json_buffer_stack[PUBLISH_JSON_BUFFER_SIZE]; char* active_buffer; size_t active_buffer_size; - if (_publish_json_buffer != nullptr) { - active_buffer = _publish_json_buffer; + if (_json_scratch_buffer != nullptr) { + active_buffer = _json_scratch_buffer; active_buffer_size = PUBLISH_JSON_BUFFER_SIZE; } else { active_buffer = json_buffer_stack; active_buffer_size = PUBLISH_JSON_BUFFER_SIZE; } #else - char* active_buffer = _publish_json_buffer; + char* active_buffer = _json_scratch_buffer; const size_t active_buffer_size = PUBLISH_JSON_BUFFER_SIZE; #endif char origin_id[65]; @@ -3418,7 +3446,7 @@ bool MQTTBridge::publishRaw(mesh::Packet* packet, bool& has_eligible_target) { if ((eligible_slots & static_cast(1u << i)) != 0 && _slots[i].enabled && _slots[i].client && _slots[i].connected) { if (buildTopicForSlot(i, MSG_RAW, topic, sizeof(topic))) { - if (publishToSlot(i, topic, active_buffer, false)) { + if (publishToSlot(i, topic, active_buffer, (size_t)len, false)) { published = true; } } @@ -3468,7 +3496,7 @@ bool MQTTBridge::publishNeighbors() { // Neighbor snapshots are periodically refreshed. Publish synchronously // at QoS 0 to avoid the QoS 1 outbox, retaining where the broker allows. bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false; - if (publishToSlot(i, topic, _neighbors_json_buffer, use_retain, 0)) { + if (publishToSlot(i, topic, _neighbors_json_buffer, _neighbors_publish_len, use_retain, 0)) { published = true; } } diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 143aae36..d1211fc6 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -176,9 +176,6 @@ private: #ifdef ESP_PLATFORM QueueHandle_t _packet_queue_handle; TaskHandle_t _mqtt_task_handle; - // PSRAM-backed task stack; TCB kept in internal RAM - StackType_t* _mqtt_task_stack; // nullptr if using dynamic task creation - StaticTask_t _mqtt_task_tcb; // Packet queue storage: PSRAM heap on PSRAM boards, inline array on non-PSRAM boards. // Using xQueueCreateStatic with inline storage eliminates a separate heap allocation. uint8_t* _packet_queue_storage; @@ -285,17 +282,22 @@ private: float _last_rssi; unsigned long _last_raw_timestamp; - // JSON publish/status serialization buffers — reused for every publish (no alloc/free churn). - // On PSRAM boards: heap pointer into PSRAM to save internal heap. On non-PSRAM: inline in - // class object so these allocations don't interleave with large TLS buffers at startup. + // One JSON serialization buffer shared by every publish path — packet, raw, and + // status all serialize on the bridge task (Core 0), so they are never in flight at + // the same time and a second buffer bought nothing. Reused rather than reallocated + // per publish (no alloc/free churn). On PSRAM boards: heap pointer into PSRAM to save + // internal heap. On non-PSRAM: inline in the class object so the allocation doesn't + // interleave with large TLS buffers at startup. static const size_t PUBLISH_JSON_BUFFER_SIZE = 2048; + // Status keeps its own smaller ceiling: raising it would change which oversized + // status documents get published instead of dropped. static const size_t STATUS_JSON_BUFFER_SIZE = 768; + static_assert(STATUS_JSON_BUFFER_SIZE <= PUBLISH_JSON_BUFFER_SIZE, + "status payloads serialize into the shared publish buffer"); #if defined(BOARD_HAS_PSRAM) - char* _publish_json_buffer; - char* _status_json_buffer; + char* _json_scratch_buffer; #else - char _publish_json_buffer[PUBLISH_JSON_BUFFER_SIZE]; - char _status_json_buffer[STATUS_JSON_BUFFER_SIZE]; + char _json_scratch_buffer[PUBLISH_JSON_BUFFER_SIZE]; #endif #if defined(WITH_MQTT_NEIGHBORS) @@ -318,10 +320,23 @@ private: std::atomic _neighbors_secs_until_next; #endif - // JSON document scratch space — inline StaticJsonDocument keeps the pool off the MQTT - // task stack and eliminates two separate heap allocations (fragmentation reduction). - StaticJsonDocument _packet_json_doc; - StaticJsonDocument _status_json_doc; + // Routes the shared document's pools to PSRAM where the board has it, matching the + // neighbors document's allocator in MyMesh.cpp. ArduinoJson's default allocator is + // plain malloc(), which puts every per-publish pool block (4096 bytes here) in + // internal DRAM next to the mbedTLS working set. + struct JsonScratchAllocator : ArduinoJson::Allocator { + void* allocate(size_t size) override; + void deallocate(void* ptr) override; + void* reallocate(void* ptr, size_t new_size) override; + }; + JsonScratchAllocator _json_allocator; + + // Shared by the packet/raw/status builders, like _json_scratch_buffer above. + // Declared after _json_allocator so the allocator is constructed first. + // This was a StaticJsonDocument described as an inline pool; under ArduinoJson 7 + // that is a deprecated empty subclass of JsonDocument whose template argument only + // feeds capacity(), so the object is 64 bytes and every pool comes from the allocator. + JsonDocument _json_scratch_doc{&_json_allocator}; // Memory pressure monitoring (per-publish skip; see publishPacket()). // The broader fragmentation-recovery machinery was removed in Phase 4 of @@ -400,8 +415,11 @@ private: void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted); bool createSlotAuthToken(int index); // Create/renew JWT token for a slot unsigned long slotTokenLifetime(int index) const; // effective JWT lifetime (preset/default minus slot stagger), seconds - bool publishToSlot(int index, const char* topic, const char* payload, bool retained = false, uint8_t qos = 0); - bool publishToAllSlots(const char* topic, const char* payload, bool retained = false, uint8_t qos = 0); + // payload_len is the serialized length the builder already returned. Every caller + // knows it, and passing it avoids re-scanning up to 2 KB of JSON per destination + // slot (and up to NEIGHBORS_JSON_BUFFER_SIZE per neighbor snapshot). + bool publishToSlot(int index, const char* topic, const char* payload, size_t payload_len, bool retained = false, uint8_t qos = 0); + bool publishToAllSlots(const char* topic, const char* payload, size_t payload_len, bool retained = false, uint8_t qos = 0); void publishStatusToSlot(int index); void updateCachedConnectionStatus(); From c833da10e98778d556d32df4bc35bf11f894a422 Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 09:16:47 -0700 Subject: [PATCH 2/9] perf(mqtt): allocate slot clients on demand initSlotClients() created a PsychicMqttClient for every one of RUNTIME_MQTT_SLOTS at begin(), without consulting slot.enabled or the active cap -- even though presets are applied and _max_active_slots is computed earlier in the same function. RUNTIME_MQTT_SLOTS is deliberately cap+1 (6/5 with PSRAM, 3/2 without), so at least one client was always unusable, and a lightly-configured board wasted several. Replaced with ensureSlotClient(index), called from setupSlot() -- i.e. only for a slot that is enabled, inside the cap, and ready to connect. A client that never reaches setupSlot() is completely inert: the reconnect ladder is gated on initial_connect_done, which only setupSlot() sets. Retained for the bridge lifetime as before, so reconfigure/reconnect still reuse one mbedTLS context; that context is built by connect(), not by the constructor, so deferring costs nothing but the 1284-byte object. Internal DRAM saved, by configured slot count: non-PSRAM (3 runtime / 2 cap) 1 configured 2568 B 2-3 1284 B PSRAM (6 runtime / 5 cap) 1 configured 6420 B 2 5136 B 5-6 1284 B A BOARD_HAS_PSRAM board whose PSRAM fails to init gets cap 2 against 6 runtime slots, so it saves at least 5136 B -- on the board that just lost its PSRAM. Two gates used "client != nullptr" as a proxy for "configured". Under eager allocation that conjunct was always true and therefore harmless; with lazy allocation it would have made shouldQueuePacketType() drop every packet received before the post-NTP-sync slot setup, which is precisely the window the queue exists to cover. Both now key on slot.enabled, matching the documented intent above eligiblePacketSlots() that a configured-but- disconnected broker is still a target. formatSlotDiagReply() gains the !isSlotReady() -> "wait" branch that get mqtt.status and getSlotStatusSnapshot() already have; that state previously fell through to "disc" and read as a network fault when it is really a missing token/IATA/credential. "no client" now means only what its name says: the slot is ready but the client could not be allocated. That state is newly reachable, so the allocation uses new (std::nothrow) -- this framework enables C++ exceptions, and a throwing new on exhaustion would panic the node instead of degrading one slot. flash non-PSRAM 1592573 -> 1592697 B (+124) PSRAM 1555001 -> 1555109 B (+108) 267/267 native tests pass; both observer envs build clean. --- src/helpers/bridges/MQTTBridge.cpp | 204 ++++++++++++++++------------- src/helpers/bridges/MQTTBridge.h | 11 +- 2 files changed, 120 insertions(+), 95 deletions(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index eafa2321..556936a1 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #ifdef WITH_SNMP @@ -505,7 +506,13 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) return; } else if (!slot.enabled) { state = "inactive"; + } else if (!b->isSlotReady(slot_index)) { + // Same classification as `get mqtt.status` and getSlotStatusSnapshot(): the slot + // is configured but missing a token/IATA/credential, so it was never set up and + // has no client yet. Previously reported "disc", which read as a network fault. + state = "wait"; } else if (!slot.client) { + // Ready to connect but the client object could not be allocated. state = "no client"; } else if (slot.connected) { state = "ok"; @@ -1033,10 +1040,11 @@ void MQTTBridge::begin() { // NOTE: Slot setup deferred until after NTP sync in loop() #endif - // Allocate persistent MQTT client objects once. They live for the bridge's - // lifetime so reconfigure/reconnect paths reuse the same mbedTLS context - // instead of churning ~40 KB of internal heap per cycle. - initSlotClients(); + // MQTT client objects are NOT allocated here. setupSlot() creates one on a slot's + // first setup, so unconfigured and capped-off slots never cost their ~1.3 KB of + // internal DRAM. Once created a client lives for the bridge's lifetime, so the + // reconfigure/reconnect paths still reuse the same mbedTLS context instead of + // churning ~40 KB of internal heap per cycle. // Sync the lifecycle Coordinator to Running now that all resources exist and // the task is created. Driven only on the success path: the failure rollbacks @@ -1570,85 +1578,96 @@ void MQTTBridge::mqttTaskLoop() { // Slot management // --------------------------------------------------------------------------- -// Allocate one PsychicMqttClient per slot and register its persistent callbacks. -// Called exactly once per bridge lifetime from begin(); the objects live until -// destroySlotClients(). Reconfiguring a slot (preset change, JWT renewal, -// reconnect) reuses the same client — no delete/new cycles, so the mbedTLS -// context and its ~40 KB of internal-heap buffers are allocated once instead -// of every reconfigure. -void MQTTBridge::initSlotClients() { - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - MQTTSlot& slot = _slots[i]; - if (slot.client != nullptr) continue; +// Allocate this slot's PsychicMqttClient and register its persistent callbacks. +// Called from setupSlot(), i.e. only for a slot that is enabled, within the active +// cap, and ready to connect — a client is ~1.3 KB of internal DRAM and does nothing +// at all until setupSlot() runs (the reconnect ladder is gated on +// initial_connect_done), so slots that are unconfigured or capped off never get one. +// +// Once created the object lives until destroySlotClients(): reconfiguring a slot +// (preset change, JWT renewal, reconnect) reuses it, so the mbedTLS context and its +// ~40 KB of internal-heap buffers are allocated once instead of every reconfigure. +// That context is created by connect(), not by this constructor, so deferring the +// allocation to first use costs nothing beyond the object itself. +bool MQTTBridge::ensureSlotClient(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; + MQTTSlot& slot = _slots[index]; + if (slot.client != nullptr) return true; - slot.client = new PsychicMqttClient(); - slot.client->setAutoReconnect(false); // we handle reconnect with our own backoff - - const int index = i; // capture a fresh copy so lambdas refer to the right slot - slot.client->onConnect([this, index](bool sessionPresent) { - MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); - _slots[index].connected = true; - // NOTE: reconnect_backoff / max_backoff_failures are NOT reset here. - // A CONNACK alone doesn't prove the link is healthy — a broker that - // accepts and then drops within seconds would reset the ladder every - // cycle and retry at the 10 s rung forever, and each retry is a full - // TLS session alloc/free (~40 KB of internal-heap churn, a known - // fragmentation driver). The ladder is instead cleared by - // maintainSlotConnection() once the connection has stayed up for - // BACKOFF_STABLE_RESET_MS, so flapping endpoints keep their earned - // backoff level. The breaker itself does clear now: while connected - // the diag/status must not claim the slot gave up, and the next - // disconnect should be governed by the (still-elevated) ladder. - _slots[index].connected_at_ms = millis(); - _slots[index].circuit_breaker_tripped = false; - _slots[index].last_tls_err = 0; - _slots[index].last_tls_stack_err = 0; - _slots[index].last_sock_errno = 0; - _slots[index].last_error_time = 0; - _slots[index].current_outage_started_ms = 0; // clear current-outage timer for AlertReporter - updateCachedConnectionStatus(); // bool store — safe from this (esp-mqtt) task - // This callback runs on the client's esp-mqtt event task, not the bridge - // task. Do NOT build/publish status here: publishStatusToSlot() writes the - // shared _json_scratch_doc/_json_scratch_buffer/_origin that the periodic - // publishStatus() uses on the bridge task, and two slots' callbacks could - // race each other over them. Marshal the publish onto the bridge task via a - // per-slot flag (see mqttTaskLoop consumer / A2). - _status_publish_pending[index] = true; - }); - slot.client->onDisconnect([this, index](bool sessionPresent) { - MQTT_DEBUG_PRINTLN("MQTT%d disconnected", index + 1); - _slots[index].disconnect_count++; - if (_slots[index].first_disconnect_time == 0) { - _slots[index].first_disconnect_time = millis(); - } - if (_slots[index].current_outage_started_ms == 0) { - _slots[index].current_outage_started_ms = millis(); - } - _slots[index].connected = false; - _slots[index].connected_at_ms = 0; // stability clock only runs while connected - updateCachedConnectionStatus(); - }); - slot.client->onError([this, index](esp_mqtt_error_codes error) { - _slots[index].last_tls_err = error.esp_tls_last_esp_err; - _slots[index].last_tls_stack_err = error.esp_tls_stack_err; - _slots[index].last_sock_errno = error.esp_transport_sock_errno; - _slots[index].last_error_time = millis(); - if (error.error_type == MQTT_ERROR_TYPE_CONNECTION_REFUSED) { - // Broker rejected the MQTT CONNECT itself — not a transport failure. - // return code: 1=protocol, 2=client-id rejected, 3=server unavailable, - // 4=bad username/password, 5=not authorized. Codes 3/4/5 point at a - // server-side lockout or auth problem rather than the network. - MQTT_DEBUG_PRINTLN("MQTT%d connection refused by broker (return code=%d)", - index + 1, (int)error.connect_return_code); - } else if (error.esp_tls_last_esp_err != 0 || error.esp_tls_stack_err != 0 || error.esp_transport_sock_errno != 0) { - MQTT_DEBUG_PRINTLN("MQTT%d error: tls=%d, tls_stack=%d, sock=%d, type=%d", - index + 1, error.esp_tls_last_esp_err, error.esp_tls_stack_err, - error.esp_transport_sock_errno, error.error_type); - } else { - MQTT_DEBUG_PRINTLN("MQTT%d error: type=%d", index + 1, error.error_type); - } - }); + // nothrow: this framework builds with C++ exceptions enabled, so a plain new would + // throw on exhaustion and panic the node. A slot that cannot get a client should + // degrade to the "no client" diag state instead. + slot.client = new (std::nothrow) PsychicMqttClient(); + if (slot.client == nullptr) { + MQTT_DEBUG_PRINTLN("MQTT%d: out of memory allocating client", index + 1); + return false; } + slot.client->setAutoReconnect(false); // we handle reconnect with our own backoff + + slot.client->onConnect([this, index](bool sessionPresent) { + MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); + _slots[index].connected = true; + // NOTE: reconnect_backoff / max_backoff_failures are NOT reset here. + // A CONNACK alone doesn't prove the link is healthy — a broker that + // accepts and then drops within seconds would reset the ladder every + // cycle and retry at the 10 s rung forever, and each retry is a full + // TLS session alloc/free (~40 KB of internal-heap churn, a known + // fragmentation driver). The ladder is instead cleared by + // maintainSlotConnection() once the connection has stayed up for + // BACKOFF_STABLE_RESET_MS, so flapping endpoints keep their earned + // backoff level. The breaker itself does clear now: while connected + // the diag/status must not claim the slot gave up, and the next + // disconnect should be governed by the (still-elevated) ladder. + _slots[index].connected_at_ms = millis(); + _slots[index].circuit_breaker_tripped = false; + _slots[index].last_tls_err = 0; + _slots[index].last_tls_stack_err = 0; + _slots[index].last_sock_errno = 0; + _slots[index].last_error_time = 0; + _slots[index].current_outage_started_ms = 0; // clear current-outage timer for AlertReporter + updateCachedConnectionStatus(); // bool store — safe from this (esp-mqtt) task + // This callback runs on the client's esp-mqtt event task, not the bridge + // task. Do NOT build/publish status here: publishStatusToSlot() writes the + // shared _json_scratch_doc/_json_scratch_buffer/_origin that the periodic + // publishStatus() uses on the bridge task, and two slots' callbacks could + // race each other over them. Marshal the publish onto the bridge task via a + // per-slot flag (see mqttTaskLoop consumer / A2). + _status_publish_pending[index] = true; + }); + slot.client->onDisconnect([this, index](bool sessionPresent) { + MQTT_DEBUG_PRINTLN("MQTT%d disconnected", index + 1); + _slots[index].disconnect_count++; + if (_slots[index].first_disconnect_time == 0) { + _slots[index].first_disconnect_time = millis(); + } + if (_slots[index].current_outage_started_ms == 0) { + _slots[index].current_outage_started_ms = millis(); + } + _slots[index].connected = false; + _slots[index].connected_at_ms = 0; // stability clock only runs while connected + updateCachedConnectionStatus(); + }); + slot.client->onError([this, index](esp_mqtt_error_codes error) { + _slots[index].last_tls_err = error.esp_tls_last_esp_err; + _slots[index].last_tls_stack_err = error.esp_tls_stack_err; + _slots[index].last_sock_errno = error.esp_transport_sock_errno; + _slots[index].last_error_time = millis(); + if (error.error_type == MQTT_ERROR_TYPE_CONNECTION_REFUSED) { + // Broker rejected the MQTT CONNECT itself — not a transport failure. + // return code: 1=protocol, 2=client-id rejected, 3=server unavailable, + // 4=bad username/password, 5=not authorized. Codes 3/4/5 point at a + // server-side lockout or auth problem rather than the network. + MQTT_DEBUG_PRINTLN("MQTT%d connection refused by broker (return code=%d)", + index + 1, (int)error.connect_return_code); + } else if (error.esp_tls_last_esp_err != 0 || error.esp_tls_stack_err != 0 || error.esp_transport_sock_errno != 0) { + MQTT_DEBUG_PRINTLN("MQTT%d error: tls=%d, tls_stack=%d, sock=%d, type=%d", + index + 1, error.esp_tls_last_esp_err, error.esp_tls_stack_err, + error.esp_transport_sock_errno, error.error_type); + } else { + MQTT_DEBUG_PRINTLN("MQTT%d error: type=%d", index + 1, error.error_type); + } + }); + return true; } void MQTTBridge::destroySlotClients() { @@ -1678,10 +1697,9 @@ void MQTTBridge::setupSlot(int index) { return; } - // Persistent client is expected to have been allocated by initSlotClients(). - // If it hasn't, we can't proceed — bail loudly rather than silently leaking. - if (slot.client == nullptr) { - MQTT_DEBUG_PRINTLN("MQTT%d: setupSlot before initSlotClients() - skipping", index + 1); + // First setup for this slot allocates its persistent client; later ones reuse it. + if (!ensureSlotClient(index)) { + MQTT_DEBUG_PRINTLN("MQTT%d: client allocation failed - skipping", index + 1); return; } @@ -1954,7 +1972,7 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // ladder. Flapping endpoints therefore stay at their earned backoff rung // (worst case the 300 s rung / 30-minute breaker probes) instead of // hammering full TLS handshakes at the 10 s rung — see the onConnect - // handler in initSlotClients() for why this doesn't happen on CONNACK. + // handler in ensureSlotClient() for why this doesn't happen on CONNACK. if (slot.connected && (slot.reconnect_backoff != 0 || slot.max_backoff_failures != 0) && MQTTConnectionPolicy::stableConnection(static_cast(now_millis), @@ -2053,8 +2071,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns (_radio && _radio->getLastRecvMillis() > 0) ? (_ms->getMillis() - _radio->getLastRecvMillis()) : 0); if (slot_uses_jwt) { // Regenerate or refresh token, then reconnect the persistent client. - // The client object and its mbedTLS context are always live post - // initSlotClients(), so no full setup is ever needed here. + // Reaching the ladder at all means setupSlot() ran, so the client object + // and its mbedTLS context are live and no full setup is needed here. if (createSlotAuthToken(index)) { slot.client->setCredentials(_jwt_username, slot.auth_token); MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (fresh token)", index + 1); @@ -3118,7 +3136,10 @@ uint8_t MQTTBridge::eligiblePacketSlots(uint8_t packet_type, MQTTMessageType typ uint8_t eligible_slots = 0; char topic[128]; for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { - const bool slot_enabled = _slots[i].enabled && _slots[i].client != nullptr; + // Configuration is the gate, not client allocation: a slot whose client has not + // been created yet (setupSlot() runs only after NTP sync) is still a target, so + // the packet stays queued for the bounded retry per the note above. + const bool slot_enabled = _slots[i].enabled; // Load once so a live CLI/WebConfig update cannot split this packet's // decision across two different masks. const uint16_t filter_mask = _obs->mqtt_slot_packet_filter[i]; @@ -3147,7 +3168,10 @@ bool MQTTBridge::shouldQueuePacketType(uint8_t packet_type, bool& filtered) { bool any_enabled = false; for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { masks[i] = _obs->mqtt_slot_packet_filter[i]; - enabled[i] = _slots[i].enabled && _slots[i].client != nullptr; + // Configured, not allocated — see eligiblePacketSlots(). Gating on the client + // here would silently drop every packet received before the post-NTP-sync slot + // setup, which is exactly the window the queue exists to cover. + enabled[i] = _slots[i].enabled; any_enabled = any_enabled || enabled[i]; } if (!any_enabled) return false; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index d1211fc6..850794e8 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -398,16 +398,17 @@ private: // Internal methods - slot management // Lifetime model (Phase 1 of MQTT memory-defrag): - // - initSlotClients() allocates one PsychicMqttClient per slot and registers - // its persistent callbacks. Runs once per bridge lifetime in begin(). + // - ensureSlotClient() allocates this slot's PsychicMqttClient and registers its + // persistent callbacks. Called from setupSlot() on a slot's first setup, so an + // unconfigured or capped-off slot never pays for a client it cannot use. // - destroySlotClients() disconnects and deletes each client. Runs once in end(). - // - setupSlot() configures an already-allocated client (server, credentials, - // CA) and calls connect(). Safe to call multiple times to reconfigure. + // - setupSlot() ensures the client exists, then configures it (server, + // credentials, CA) and calls connect(). Safe to call again to reconfigure. // - teardownSlot() only disconnects — it never deletes the client. Leaves // the mbedTLS/transport state ready for a subsequent setupSlot(). // This avoids delete/new cycles that shed ~40 KB of mbedTLS buffers per // reconfigure and fragment the internal heap on non-PSRAM boards. - void initSlotClients(); // Allocate persistent clients + register callbacks (once) + bool ensureSlotClient(int index); // Allocate this slot's persistent client + callbacks on first use void destroySlotClients(); // Delete all persistent clients (shutdown only) void setupSlot(int index); // Configure and connect the slot's existing client void teardownSlot(int index); // Disconnect the slot's client (keeps the object alive) From 9141ad854fce703627eea781ab09e3e154dbd4ad Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 09:45:30 -0700 Subject: [PATCH 3/9] perf(mqtt): allocate slot JWT token buffers on demand MQTTSlot carried an inline char auth_token[768] -- 768 of its 1192 bytes -- for every runtime slot, whether the slot was JWT, username/password, disabled, or above the active-slot cap. Now a char* allocated by ensureSlotAuthToken() from createSlotAuthToken(), the sole writer, which runs only once a slot is confirmed to have a JWT audience. Because _slots[] is a fixed member array, this shrinks the bridge object unconditionally at construction and re-adds only what is used: MQTTSlot 1192 -> 428 B (-764) MQTTBridge non-PSRAM 12032 -> 9740 B (-2292) PSRAM 10080 -> 5496 B (-4584) The object is allocated as one contiguous block at boot, so a smaller boot-time chunk is easier to satisfy and leaves a larger contiguous remainder; JWT slots then take 768 B each as separate blocks. Net steady-state saving is 768 B per slot that never creates a token -- 768 B on a fully configured board, up to 3840 B on a PSRAM board with one JWT slot. 22 of the 29 presets are MQTT_AUTH_JWT, so this is mostly reclaiming unconfigured slots rather than non-JWT ones. Lifetime rules, which are the whole risk here: - setCredentials() stores this pointer in _mqtt_cfg rather than copying, and esp-mqtt re-reads it whenever a later connect() re-applies a dirtied config. So the buffer is freed only alongside the client, in destroySlotClients(), after the delete. MQTTSlot::broker_uri already carries an "avoids dangling pointer" comment from this same hazard with setServer(). - teardownSlot() still clears the token to an empty string but keeps the buffer: the client survives teardown and does not clear cfg->password (only setupSlot()'s reconfigure branch does). Freeing there would dangle. - Never freed per reconnect, preserving the churn-avoidance the inline buffer was there for. Allocation goes through MQTTRuntimeBufferLifecycle (already host-tested) using plain malloc, so the buffer stays in internal DRAM exactly where it was when inline. Failure propagates through createSlotAuthToken()'s existing bool. The two call sites that create a token and then test it (preset JWT in setupSlot, and the custom-slot audience path) now null-check first; the five other readers are all inside if (createSlotAuthToken(...)) success branches and need no change. destroySlotClients() releases the token unconditionally rather than after its client null-check, so a token can never outlive its slot. flash non-PSRAM 1592697 -> 1592801 B (+104) PSRAM 1555109 -> 1555213 B (+104) 267/267 native tests pass; both observer envs build clean. --- src/helpers/bridges/MQTTBridge.cpp | 71 +++++++++++++++++++++++------- src/helpers/bridges/MQTTBridge.h | 14 ++++-- 2 files changed, 65 insertions(+), 20 deletions(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 556936a1..5d42f00f 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -692,7 +692,7 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _slots[i].enabled = false; _slots[i].client = nullptr; _slots[i].preset = nullptr; - // auth_token[0] == '\0' after memset above — no valid token + // auth_token == nullptr after memset above — allocated on first token creation _slots[i].connected = false; _slots[i].initial_connect_done = false; _slots[i].token_expires_at = 0; @@ -1670,21 +1670,52 @@ bool MQTTBridge::ensureSlotClient(int index) { return true; } +// Allocate this slot's JWT token buffer. Called only from createSlotAuthToken(), the +// sole writer, so a slot on a non-JWT preset (or no preset at all) never allocates. +// Internal DRAM, matching where the buffer lived when it was inline in MQTTSlot. +bool MQTTBridge::ensureSlotAuthToken(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; + MQTTSlot& slot = _slots[index]; + slot.auth_token = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( + slot.auth_token, AUTH_TOKEN_SIZE, malloc)); + if (slot.auth_token == nullptr) { + MQTT_DEBUG_PRINTLN("MQTT%d: out of memory allocating auth token", index + 1); + return false; + } + slot.auth_token[0] = '\0'; + return true; +} + +// Safe only once this slot's client is gone: setCredentials() gave the client this +// pointer, and esp-mqtt re-reads it from _mqtt_cfg on any later connect() that +// re-applies a dirtied config. See the MQTTSlot::auth_token comment. +void MQTTBridge::releaseSlotAuthToken(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; + MQTTSlot& slot = _slots[index]; + slot.auth_token = static_cast( + MQTTRuntimeBufferLifecycle::release(slot.auth_token, free)); + slot.token_expires_at = 0; + slot.last_token_renewal = 0; +} + void MQTTBridge::destroySlotClients() { for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { MQTTSlot& slot = _slots[i]; - if (slot.client == nullptr) continue; - - if (slot.client->connected()) { - slot.client->disconnect(); + if (slot.client != nullptr) { + if (slot.client->connected()) { + slot.client->disconnect(); + } + #ifdef ESP_PLATFORM + vTaskDelay(pdMS_TO_TICKS(50)); + #else + delay(50); + #endif + delete slot.client; + slot.client = nullptr; } - #ifdef ESP_PLATFORM - vTaskDelay(pdMS_TO_TICKS(50)); - #else - delay(50); - #endif - delete slot.client; - slot.client = nullptr; + // Unconditional: only now is the token unreachable from the client's stored + // config, and a token without a client would otherwise leak. + releaseSlotAuthToken(i); } } @@ -1727,7 +1758,7 @@ void MQTTBridge::setupSlot(int index) { cfg->username = nullptr; cfg->password = nullptr; #endif - slot.auth_token[0] = '\0'; + if (slot.auth_token) slot.auth_token[0] = '\0'; slot.connected = false; slot.token_expires_at = 0; slot.last_token_renewal = 0; @@ -1760,7 +1791,7 @@ void MQTTBridge::setupSlot(int index) { // Try to create token and connect (will succeed only if NTP synced) if (slot.preset->auth_type == MQTT_AUTH_JWT) { createSlotAuthToken(index); - if (slot.auth_token[0] != '\0') { + if (slot.auth_token && slot.auth_token[0] != '\0') { slot.client->setCredentials(_jwt_username, slot.auth_token); } } else if (slot.preset->auth_type == MQTT_AUTH_USERPASS) { @@ -1871,9 +1902,9 @@ void MQTTBridge::setupSlot(int index) { // Custom slot authentication: JWT if audience is set, else username/password if (slot.audience[0] != '\0') { - // JWT auth for custom slot — create initial token (buffer is always inline) + // JWT auth for custom slot — create initial token (allocates the buffer) createSlotAuthToken(index); - if (slot.auth_token[0] != '\0') { + if (slot.auth_token && slot.auth_token[0] != '\0') { slot.client->setCredentials(_jwt_username, slot.auth_token); } MQTT_DEBUG_PRINTLN("MQTT%d custom broker using JWT auth (audience: %s)", index + 1, slot.audience); @@ -1903,7 +1934,9 @@ void MQTTBridge::teardownSlot(int index) { #endif } - slot.auth_token[0] = '\0'; + // Invalidate the token but keep the buffer: the client survives teardown and still + // holds this pointer in its config (see MQTTSlot::auth_token). + if (slot.auth_token) slot.auth_token[0] = '\0'; slot.connected = false; slot.initial_connect_done = false; slot.broker_uri[0] = '\0'; @@ -2172,6 +2205,10 @@ bool MQTTBridge::createSlotAuthToken(int index) { } if (!audience || audience[0] == '\0') return false; + // This slot is confirmed JWT, so it needs the token buffer. Allocated on first use + // and kept thereafter; every caller already treats false as "no usable token". + if (!ensureSlotAuthToken(index)) return false; + // Ensure JWT username is set if (_jwt_username[0] == '\0') { char public_key_hex[65]; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 850794e8..d1c72e3f 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -86,9 +86,15 @@ private: bool connected; // Updated in callbacks bool initial_connect_done; // True after first connect() call - // JWT auth state (used by preset JWT slots and custom slots with audience set) - // Inline buffer avoids per-reconnect heap alloc/free churn (fragmentation source). - char auth_token[AUTH_TOKEN_SIZE]; // empty string = no valid token + // JWT auth state (used by preset JWT slots and custom slots with audience set). + // nullptr until this slot first creates a token, so a slot that is unconfigured, + // capped off, or on a non-JWT preset never pays for AUTH_TOKEN_SIZE. Allocated by + // ensureSlotAuthToken() and then held for the client's lifetime -- never freed per + // reconnect (alloc/free churn is a fragmentation source) and never freed on + // teardown, because setCredentials() hands this exact pointer to the client and + // esp-mqtt re-reads it whenever a later connect() re-applies a dirtied config. + // Freed only alongside the client in destroySlotClients(). + char* auth_token; // nullptr or empty string = no valid token unsigned long token_expires_at; unsigned long last_token_renewal; @@ -409,6 +415,8 @@ private: // This avoids delete/new cycles that shed ~40 KB of mbedTLS buffers per // reconfigure and fragment the internal heap on non-PSRAM boards. bool ensureSlotClient(int index); // Allocate this slot's persistent client + callbacks on first use + bool ensureSlotAuthToken(int index); // Allocate this slot's JWT token buffer on first token creation + void releaseSlotAuthToken(int index);// Free the token buffer (only with the client — see MQTTSlot) void destroySlotClients(); // Delete all persistent clients (shutdown only) void setupSlot(int index); // Configure and connect the slot's existing client void teardownSlot(int index); // Disconnect the slot's client (keeps the object alive) From b93b6b1560a08e5e707caf09d59ddd38e37a8593 Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 09:47:06 -0700 Subject: [PATCH 4/9] perf(mqtt): keep slot JWT tokens in PSRAM where available ensureSlotAuthToken()/releaseSlotAuthToken() now use the bridge's existing psram_malloc/psram_free rather than malloc/free, so a JWT slot's 768-byte token comes out of PSRAM instead of internal DRAM. On PSRAM boards this is the larger half of the auth_token work: the previous commit reclaims 768 B per slot that never creates a token, while this reclaims 768 B per slot that does -- up to 3840 B of internal DRAM on a fully configured five-slot board, which is where the mbedTLS working set is competing for space. No effect on non-PSRAM boards: psram_malloc falls back to internal DRAM, so the buffer stays exactly where the previous commit left it. Same for a BOARD_HAS_PSRAM board whose PSRAM fails to initialise. Safe to move because every access is a CPU copy on the bridge task, never DMA, an ISR, or a cache-disabled window: JWTHelper memcpy's the token into this buffer, and esp-mqtt copies it out of _mqtt_cfg into its own internal-DRAM storage when connect() applies the config. This is unlike the PSRAM-backed MQTT task stack that was tried and reverted for resetting Heltec V4 boards, where the fault was PSRAM execution context rather than a plain buffer read. Split from the previous commit so it can be reverted alone if hardware soak shows any PSRAM-related instability on the JWT path. flash non-PSRAM 1592801 -> 1592865 B (+64) PSRAM 1555213 -> 1555237 B (+24) 267/267 native tests pass; both observer envs and an nRF52 repeater build clean. --- src/helpers/bridges/MQTTBridge.cpp | 12 +++++++++--- src/helpers/bridges/MQTTBridge.h | 3 ++- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 5d42f00f..ebb63887 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -1672,12 +1672,18 @@ bool MQTTBridge::ensureSlotClient(int index) { // Allocate this slot's JWT token buffer. Called only from createSlotAuthToken(), the // sole writer, so a slot on a non-JWT preset (or no preset at all) never allocates. -// Internal DRAM, matching where the buffer lived when it was inline in MQTTSlot. +// +// PSRAM where the board has it (psram_malloc falls back to internal DRAM otherwise), +// which is what moves the token off internal heap for slots that DO use JWT. Safe +// because the only readers are CPU copies on the bridge task: JWTHelper memcpy's the +// token in here, and esp-mqtt copies it out of _mqtt_cfg into its own internal-DRAM +// storage when connect() applies the config. No DMA, no ISR, and no cache-disabled +// window -- unlike the PSRAM task stack that reset Heltec V4 boards. bool MQTTBridge::ensureSlotAuthToken(int index) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; slot.auth_token = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( - slot.auth_token, AUTH_TOKEN_SIZE, malloc)); + slot.auth_token, AUTH_TOKEN_SIZE, psram_malloc)); if (slot.auth_token == nullptr) { MQTT_DEBUG_PRINTLN("MQTT%d: out of memory allocating auth token", index + 1); return false; @@ -1693,7 +1699,7 @@ void MQTTBridge::releaseSlotAuthToken(int index) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[index]; slot.auth_token = static_cast( - MQTTRuntimeBufferLifecycle::release(slot.auth_token, free)); + MQTTRuntimeBufferLifecycle::release(slot.auth_token, psram_free)); slot.token_expires_at = 0; slot.last_token_renewal = 0; } diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index d1c72e3f..4c7436c2 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -88,7 +88,8 @@ private: // JWT auth state (used by preset JWT slots and custom slots with audience set). // nullptr until this slot first creates a token, so a slot that is unconfigured, - // capped off, or on a non-JWT preset never pays for AUTH_TOKEN_SIZE. Allocated by + // capped off, or on a non-JWT preset never pays for AUTH_TOKEN_SIZE; slots that do + // use JWT keep their buffer in PSRAM where the board has it. Allocated by // ensureSlotAuthToken() and then held for the client's lifetime -- never freed per // reconnect (alloc/free churn is a fragmentation source) and never freed on // teardown, because setCredentials() hands this exact pointer to the client and From e242cc6fd303096e6f64ff3e4342f411d5788ff2 Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 10:13:04 -0700 Subject: [PATCH 5/9] fix(mqtt): address review of the demand-driven slot work [P1] A failed setup no longer strands the slot. setupSlot() returns bool and the startup loops count only successful activations, so a slot that fails on a client allocation neither consumes an active-slot position (starving a later healthy broker on capped hardware) nor sits dead forever: maintainSlotConnections() previously skipped clientless slots and the reconnect ladder is gated on initial_connect_done, so nothing retried it. It now retries an enabled but unactivated slot on a 60 s timer, one per cycle, gated on the same _slots_setup_done ordering so the NTP-deferred setup sequence is preserved. [P2] JWT setup no longer proceeds without a usable token. Both the preset and custom-audience paths returned after ignoring createSlotAuthToken()'s result, then called connect() and latched initial_connect_done -- so the token-allocation failure introduced by the previous commit produced an unauthenticated attempt exactly when memory was exhausted. They now return false and let the retry path handle it. [P2] ensureSlotAuthToken() no longer clears an existing token. It cleared unconditionally, so every renewal wiped the current token before JWTHelper ran; a renewal that then failed left an empty password where the inline buffer used to preserve working credentials (JWTHelper writes only on success). Only freshly allocated buffers are initialised now. [P2] Raw publications reuse the shared document. buildRawJSON() reached MQTTPayloadBuilder::buildRawMessage(), which constructed its own default JsonDocument and therefore malloc'd and freed an internal-heap variant pool per message -- on the highest-rate topic. The document is threaded through both builders and the bridge passes _json_scratch_doc. [P3] The writeTo() guard validates the source fields, not just the destination. A corrupt payload_len of MAX_PACKET_PAYLOAD + 1 still leaves getRawLength() inside MAX_TRANS_UNIT, so writeTo() read past packet->payload. Sizing and validation moved to a pure MQTTWireScratch header with host tests covering the accept/reject edges, matching the MQTTPacketFilter/MQTTConnectionPolicy pattern. Two findings fell out: MAX_PATH_SIZE one-byte hops is not encodable (the hop count is 6 bits, so 64 & 63 == 0; 32 two-byte hops is the widest real path), and a zero-payload packet serializes but does not survive readFrom() -- pinned as a test because it constrains any future wire-only queue. [P3] Corrected the pool-size comment: these targets are 32-bit, so ARDUINOJSON_SLOT_ID_SIZE is 2 and a pool block is 128 slots / 1024 bytes, not 4096. The 4096 figure came from a pre-existing comment near NEIGHBORS_DOC_POOL_BUDGET, which is left alone -- its byte measurements are empirical and still stand, only the block-size attribution is wrong. Activation is now centralized in activatedSlotCount()/canActivateSlot(), used by both startup loops, the retry path, and applySlotPreset(). That closes the pre-existing divergence where a live preset change called setupSlot() without consulting _max_active_slots, letting a non-PSRAM board reach three concurrent TLS sessions against a cap of two. BEHAVIOUR CHANGE: a reconfigure that would exceed the cap now logs and leaves the slot inactive instead of connecting. Reconfiguring an already-active slot still works, because teardownSlot() releases its position first. 272/272 native tests pass (5 new); both observer envs and an nRF52 repeater build clean. Flash 1593249 B non-PSRAM, 1555625 B PSRAM. --- src/helpers/MQTTMessageBuilder.cpp | 10 +- src/helpers/MQTTMessageBuilder.h | 16 ++- src/helpers/MQTTPayloadBuilder.cpp | 3 +- src/helpers/MQTTPayloadBuilder.h | 1 + src/helpers/MQTTWireScratch.h | 39 ++++++ src/helpers/bridges/MQTTBridge.cpp | 119 ++++++++++++++---- src/helpers/bridges/MQTTBridge.h | 19 ++- .../test_mqtt_payload_builder.cpp | 6 +- .../test_mqtt_wire_scratch.cpp | 101 +++++++++++++++ 9 files changed, 272 insertions(+), 42 deletions(-) create mode 100644 src/helpers/MQTTWireScratch.h create mode 100644 test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index ebe05120..062f52ed 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -96,6 +96,7 @@ int MQTTMessageBuilder::buildPacketMessage( } int MQTTMessageBuilder::buildRawMessage( + JsonDocument& doc, const char* origin, const char* origin_id, const char* timestamp, @@ -104,7 +105,7 @@ int MQTTMessageBuilder::buildRawMessage( size_t buffer_size ) { return MQTTPayloadBuilder::buildRawMessage( - origin, origin_id, timestamp, raw, buffer, buffer_size); + doc, origin, origin_id, timestamp, raw, buffer, buffer_size); } int MQTTMessageBuilder::buildNeighborsMessage( @@ -297,6 +298,7 @@ int MQTTMessageBuilder::buildPacketJSONFromRaw( } int MQTTMessageBuilder::buildRawJSON( + JsonDocument& doc, mesh::Packet* packet, const char* origin, const char* origin_id, @@ -316,7 +318,7 @@ int MQTTMessageBuilder::buildRawJSON( char raw_hex[WIRE_HEX_SCRATCH_SIZE]; packetToHex(packet, raw_hex, sizeof(raw_hex)); - return buildRawMessage(origin, origin_id, timestamp, raw_hex, buffer, buffer_size); + return buildRawMessage(doc, origin, origin_id, timestamp, raw_hex, buffer, buffer_size); } const char* MQTTMessageBuilder::getRouteTypeString(int route_type) { @@ -355,9 +357,7 @@ void MQTTMessageBuilder::packetToHex(mesh::Packet* packet, char* hex, size_t hex // Serialize full on-air/wire format using Packet::writeTo() // This includes header, transport codes (if present), path_len, path, and payload uint8_t raw_buf[WIRE_SCRATCH_SIZE]; - // writeTo() neither bounds-checks nor can report an overrun (it returns uint8_t), - // so reject an over-long packet on getRawLength() before writing anything. - if (packet->getRawLength() > (int)sizeof(raw_buf)) return; + if (!canSerializePacket(packet, sizeof(raw_buf))) return; uint8_t raw_len = packet->writeTo(raw_buf); if (raw_len == 0) return; diff --git a/src/helpers/MQTTMessageBuilder.h b/src/helpers/MQTTMessageBuilder.h index 7ef35a1a..91a47b6e 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -3,6 +3,7 @@ #include "MeshCore.h" #include #include "MQTTPayloadBuilder.h" +#include "MQTTWireScratch.h" #include #include @@ -21,13 +22,14 @@ */ class MQTTMessageBuilder { public: - // Wire-format scratch sizing, from the protocol maximum: Packet::writeTo() returns - // uint8_t, so MAX_TRANS_UNIT is the hard ceiling, and hex is 2 chars/byte + NUL. - static const size_t WIRE_SCRATCH_SIZE = MAX_TRANS_UNIT; - static const size_t WIRE_HEX_SCRATCH_SIZE = 2 * MAX_TRANS_UNIT + 1; + // Wire-format scratch sizing and validation live in the pure, host-tested + // MQTTWireScratch; these are the firmware-facing aliases. + static const size_t WIRE_SCRATCH_SIZE = MQTTWireScratch::kWireBytes; + static const size_t WIRE_HEX_SCRATCH_SIZE = MQTTWireScratch::kWireHexChars; - static_assert(1 + 4 + 1 + MAX_PATH_SIZE + MAX_PACKET_PAYLOAD <= MAX_TRANS_UNIT, - "serialized packet no longer fits MAX_TRANS_UNIT — resize the wire scratch buffers"); + static bool canSerializePacket(const mesh::Packet* packet, size_t dest_size) { + return packet != nullptr && MQTTWireScratch::canSerialize(*packet, dest_size); + } /** * Format the MQTT JSON `timestamp` field (same rule for status, packet, raw). @@ -154,6 +156,7 @@ public: * @return Length of JSON string, or 0 on error */ static int buildRawMessage( + JsonDocument& doc, const char* origin, const char* origin_id, const char* timestamp, @@ -240,6 +243,7 @@ public: * @return Length of JSON string, or 0 on error */ static int buildRawJSON( + JsonDocument& doc, mesh::Packet* packet, const char* origin, const char* origin_id, diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp index f1b49223..265969ae 100644 --- a/src/helpers/MQTTPayloadBuilder.cpp +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -166,6 +166,7 @@ int MQTTPayloadBuilder::buildPacketMessage( } int MQTTPayloadBuilder::buildRawMessage( + JsonDocument& doc, const char* origin, const char* origin_id, const char* timestamp, @@ -173,7 +174,7 @@ int MQTTPayloadBuilder::buildRawMessage( char* buffer, size_t buffer_size ) { - JsonDocument doc; + doc.clear(); JsonObject root = doc.to(); root["origin"] = origin; diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h index 24e56124..7bca170d 100644 --- a/src/helpers/MQTTPayloadBuilder.h +++ b/src/helpers/MQTTPayloadBuilder.h @@ -61,6 +61,7 @@ public: ); static int buildRawMessage( + JsonDocument& doc, const char* origin, const char* origin_id, const char* timestamp, diff --git a/src/helpers/MQTTWireScratch.h b/src/helpers/MQTTWireScratch.h new file mode 100644 index 00000000..c631cdf1 --- /dev/null +++ b/src/helpers/MQTTWireScratch.h @@ -0,0 +1,39 @@ +#pragma once + +#include +#include + +// Sizing and validation for the scratch buffers that hold a serialized packet. +// Pure so the boundary conditions can be tested on the host: the firmware-side +// callers are MQTTMessageBuilder::packetToHex() and MQTTBridge::publishPacket(). +namespace MQTTWireScratch { + +// A serialized packet is header(1) + transport codes(0|4) + path_len(1) + +// path(<= MAX_PATH_SIZE) + payload(<= MAX_PACKET_PAYLOAD). Packet::writeTo() +// returns uint8_t, so MAX_TRANS_UNIT is the hard ceiling. +static const size_t kWireBytes = MAX_TRANS_UNIT; +// Two uppercase hex chars per byte, plus the NUL. +static const size_t kWireHexChars = 2 * MAX_TRANS_UNIT + 1; + +static_assert(1 + 4 + 1 + MAX_PATH_SIZE + MAX_PACKET_PAYLOAD <= MAX_TRANS_UNIT, + "serialized packet no longer fits MAX_TRANS_UNIT — resize the wire scratch buffers"); + +// True when Packet::writeTo() can safely serialize `packet` into `dest_size` bytes. +// +// writeTo() trusts the packet's own length fields and cannot report an overrun (its +// return type is uint8_t), so the source fields must be checked as well as the +// destination: +// - payload_len drives an unchecked memcpy out of a MAX_PACKET_PAYLOAD array, and a +// corrupt value can still leave getRawLength() inside MAX_TRANS_UNIT. +// - path_len is written into a single wire byte, so anything above 255 is silently +// truncated and would disagree with getPathByteLen(). +// The path bytes themselves need no check here: writePath() already refuses a +// getPathByteLen() above MAX_PATH_SIZE. +inline bool canSerialize(const mesh::Packet& packet, size_t dest_size) { + if (packet.payload_len > MAX_PACKET_PAYLOAD) return false; + if (packet.path_len > 0xFF) return false; + const int raw_len = packet.getRawLength(); + return raw_len > 0 && (size_t)raw_len <= dest_size; +} + +} // namespace MQTTWireScratch diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index ebb63887..6fc26693 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -1399,11 +1399,10 @@ void MQTTBridge::mqttTaskLoop() { #endif MQTT_DEBUG_PRINTLN("NTP synced, setting up MQTT slots (max %d active)...", _max_active_slots); - int active_count = 0; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { if (_slots[i].enabled) { - if (active_count >= _max_active_slots) { - MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached (no PSRAM)", i + 1, _max_active_slots); + if (!canActivateSlot(i)) { + MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", i + 1, _max_active_slots); _slots[i].enabled = false; // Disable so other loops skip it continue; } @@ -1412,8 +1411,10 @@ void MQTTBridge::mqttTaskLoop() { MQTT_DEBUG_PRINTLN("MQTT%d not ready - run '%s' to connect", i + 1, reason); continue; } - setupSlot(i); - active_count++; + // A slot that fails to activate consumes no position and stays enabled, so + // maintainSlotConnections() retries it and a later healthy broker is not + // starved by it on a capped board. + if (!setupSlot(i)) continue; // Stagger connections: 5s between slots to avoid simultaneous TLS handshakes // which compete for ~40KB internal heap each if (i < RUNTIME_MQTT_SLOTS - 1) { @@ -1682,13 +1683,18 @@ bool MQTTBridge::ensureSlotClient(int index) { bool MQTTBridge::ensureSlotAuthToken(int index) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; + const bool fresh = (slot.auth_token == nullptr); slot.auth_token = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( slot.auth_token, AUTH_TOKEN_SIZE, psram_malloc)); if (slot.auth_token == nullptr) { MQTT_DEBUG_PRINTLN("MQTT%d: out of memory allocating auth token", index + 1); return false; } - slot.auth_token[0] = '\0'; + // Initialise only a newly allocated buffer. Clearing on every call would discard a + // valid token at the start of each renewal, so a renewal that then failed inside + // JWTHelper would leave the slot with an empty password where it previously kept + // working credentials (JWTHelper writes the token only on success). + if (fresh) slot.auth_token[0] = '\0'; return true; } @@ -1725,19 +1731,38 @@ void MQTTBridge::destroySlotClients() { } } -void MQTTBridge::setupSlot(int index) { - if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; +int MQTTBridge::activatedSlotCount() const { + int n = 0; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].enabled && _slots[i].initial_connect_done) n++; + } + return n; +} + +bool MQTTBridge::canActivateSlot(int index) const { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; + // Already holding a position (a reconfigure of a live slot) — no new position needed. + if (_slots[index].enabled && _slots[index].initial_connect_done) return true; + return activatedSlotCount() < _max_active_slots; +} + +// Returns true only when the slot reached connect(). A false result leaves the slot +// enabled but not activated, so it holds no active-slot position and +// maintainSlotConnections() will retry it — the allocation failures below are transient +// memory conditions, not permanent misconfiguration. +bool MQTTBridge::setupSlot(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; if (!slot.enabled) { teardownSlot(index); - return; + return false; } // First setup for this slot allocates its persistent client; later ones reuse it. if (!ensureSlotClient(index)) { - MQTT_DEBUG_PRINTLN("MQTT%d: client allocation failed - skipping", index + 1); - return; + MQTT_DEBUG_PRINTLN("MQTT%d: client allocation failed - will retry", index + 1); + return false; } // Reconfigure path: if we're re-applying (e.g. after a preset change), stop @@ -1794,12 +1819,15 @@ void MQTTBridge::setupSlot(int index) { slot.client->setCACert(slot.preset->ca_cert); } - // Try to create token and connect (will succeed only if NTP synced) + // A JWT slot with no usable token would connect unauthenticated and be rejected. + // Stay unactivated instead, so the retry path tries again — the failure is either + // a transient token-buffer allocation or a JWTHelper error, not a config problem. if (slot.preset->auth_type == MQTT_AUTH_JWT) { - createSlotAuthToken(index); - if (slot.auth_token && slot.auth_token[0] != '\0') { - slot.client->setCredentials(_jwt_username, slot.auth_token); + if (!createSlotAuthToken(index) || !slot.auth_token || slot.auth_token[0] == '\0') { + MQTT_DEBUG_PRINTLN("MQTT%d: no usable JWT token - will retry", index + 1); + return false; } + slot.client->setCredentials(_jwt_username, slot.auth_token); } else if (slot.preset->auth_type == MQTT_AUTH_USERPASS) { const char* user = nullptr; const char* pass = slot.preset->userpass_password @@ -1908,11 +1936,12 @@ void MQTTBridge::setupSlot(int index) { // Custom slot authentication: JWT if audience is set, else username/password if (slot.audience[0] != '\0') { - // JWT auth for custom slot — create initial token (allocates the buffer) - createSlotAuthToken(index); - if (slot.auth_token && slot.auth_token[0] != '\0') { - slot.client->setCredentials(_jwt_username, slot.auth_token); + // JWT auth for custom slot — same rule as the preset JWT path above. + if (!createSlotAuthToken(index) || !slot.auth_token || slot.auth_token[0] == '\0') { + MQTT_DEBUG_PRINTLN("MQTT%d: no usable JWT token - will retry", index + 1); + return false; } + slot.client->setCredentials(_jwt_username, slot.auth_token); MQTT_DEBUG_PRINTLN("MQTT%d custom broker using JWT auth (audience: %s)", index + 1, slot.audience); } else if (strlen(slot.username) > 0) { slot.client->setCredentials(slot.username, slot.password); @@ -1921,6 +1950,7 @@ void MQTTBridge::setupSlot(int index) { slot.client->connect(); slot.initial_connect_done = true; + return true; } // Disconnect the slot's MQTT client and clear per-connection state, but leave @@ -1988,8 +2018,12 @@ void MQTTBridge::maintainSlotConnections() { // when multiple slots fail simultaneously bool teardown_attempted_this_cycle = false; + // At most one deferred setup retry per cycle: a successful one ends in connect(), so + // this shares the "no simultaneous TLS handshakes" rule the reconnect guard enforces. + bool setup_retry_this_cycle = false; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (!_slots[i].enabled || !_slots[i].client) continue; + if (!_slots[i].enabled) continue; // JWT slots need time sync before we can manage tokens bool slot_jwt = (_slots[i].preset && _slots[i].preset->auth_type == MQTT_AUTH_JWT) || @@ -1998,6 +2032,27 @@ void MQTTBridge::maintainSlotConnections() { continue; } + // Enabled but never activated: setupSlot() failed on a client or token allocation, + // or on token creation. The ladder below is gated on initial_connect_done and would + // never revisit it, and maintenance used to skip clientless slots entirely, so + // without this the slot stayed dead until a reconfigure or reboot. Only retried + // after the initial pass has run, so the NTP-deferred setup order is preserved. + if (!_slots[i].initial_connect_done) { + if (_slots_setup_done && !setup_retry_this_cycle && !reconnect_attempted_this_cycle && + isSlotReady(i) && canActivateSlot(i) && + MQTTConnectionPolicy::elapsedMs(static_cast(now_millis), + static_cast(_slots[i].last_reconnect_attempt)) + >= SLOT_SETUP_RETRY_INTERVAL) { + _slots[i].last_reconnect_attempt = now_millis; + setup_retry_this_cycle = true; + MQTT_DEBUG_PRINTLN("MQTT%d retrying deferred setup (int_heap=%d)", i + 1, + (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL)); + if (setupSlot(i)) _last_slot_reconnect_ms = now_millis; + } + continue; + } + if (!_slots[i].client) continue; + maintainSlotConnection(i, now_millis, current_time, time_synced, reconnect_attempted_this_cycle, teardown_attempted_this_cycle); } } @@ -2515,6 +2570,13 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { slot.audience[sizeof(slot.audience) - 1] = '\0'; slot.enabled = (slot.host[0] != '\0'); if (_initialized && slot.enabled && customEndpointComplete(slot.host, slot.port)) { + // Same cap startup applies. teardownSlot() above already released this slot's own + // position, so reconfiguring a live slot still passes. + if (!canActivateSlot(slot_index)) { + MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", slot_index + 1, _max_active_slots); + slot.enabled = false; + return; + } setupSlot(slot_index); } return; @@ -2536,6 +2598,13 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { MQTT_DEBUG_PRINTLN("MQTT%d (%s) not ready - run '%s' to connect", slot_index + 1, preset_name, reason); return; } + // Same cap startup applies. Without this a live reconfigure could raise a + // non-PSRAM board to three concurrent TLS sessions against a cap of two. + if (!canActivateSlot(slot_index)) { + MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", slot_index + 1, _max_active_slots); + slot.enabled = false; + return; + } setupSlot(slot_index); } } @@ -2722,10 +2791,9 @@ void MQTTBridge::loop() { // Deferred slot setup after NTP sync (non-ESP32 path) if (_ntp_synced && !_slots_setup_done) { _slots_setup_done = true; - int active_count = 0; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { if (_slots[i].enabled) { - if (active_count >= _max_active_slots) { + if (!canActivateSlot(i)) { _slots[i].enabled = false; continue; } @@ -2733,7 +2801,6 @@ void MQTTBridge::loop() { continue; } setupSlot(i); - active_count++; } } } @@ -3423,10 +3490,11 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, } else { // Reconstruct wire-format bytes from packet (same as MQTTMessageBuilder::packetToHex). // Reached when the queued item carried no captured raw frame, so the "raw" hex field - // is re-serialized rather than dropped. Length-guarded for the same reason. + // is re-serialized rather than dropped. Guarded on the packet's own length fields + // as well as the destination, for the reasons in canSerializePacket(). uint8_t reconstructed[MQTTMessageBuilder::WIRE_SCRATCH_SIZE]; uint8_t rlen = 0; - if (packet->getRawLength() <= (int)sizeof(reconstructed)) { + if (MQTTMessageBuilder::canSerializePacket(packet, sizeof(reconstructed))) { rlen = packet->writeTo(reconstructed); } if (rlen > 0) { @@ -3503,6 +3571,7 @@ bool MQTTBridge::publishRaw(mesh::Packet* packet, bool& has_eligible_target) { origin_id[sizeof(origin_id) - 1] = '\0'; int len = MQTTMessageBuilder::buildRawJSON( + _json_scratch_doc, packet, _origin, origin_id, _timezone, active_buffer, active_buffer_size ); diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 4c7436c2..eb7ab943 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -329,8 +329,11 @@ private: // Routes the shared document's pools to PSRAM where the board has it, matching the // neighbors document's allocator in MyMesh.cpp. ArduinoJson's default allocator is - // plain malloc(), which puts every per-publish pool block (4096 bytes here) in - // internal DRAM next to the mbedTLS working set. + // plain malloc(), which puts every per-publish pool block in internal DRAM next to + // the mbedTLS working set. A block is ARDUINOJSON_POOL_CAPACITY slots: these targets + // are 32-bit, so ARDUINOJSON_SLOT_ID_SIZE is 2 and that resolves to 128 slots = + // 1024 bytes per block, not the 4096 quoted near NEIGHBORS_DOC_POOL_BUDGET below + // (which describes a 64-bit configuration; its own byte measurements still stand). struct JsonScratchAllocator : ArduinoJson::Allocator { void* allocate(size_t size) override; void deallocate(void* ptr) override; @@ -378,6 +381,10 @@ private: unsigned long _last_no_broker_log; static const unsigned long NO_BROKER_LOG_INTERVAL = 30000; // Log every 30 seconds max static const unsigned long SLOT_LOG_INTERVAL = 30000; // Log every 30 seconds max + // Retry cadence for a slot whose setup failed on an allocation. Deliberately slower + // than the first backoff rung: the failure means internal heap is exhausted, and a + // retry that succeeds immediately launches a TLS handshake. + static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000; unsigned long _last_config_warning; // Throttle configuration mismatch warnings static const unsigned long CONFIG_WARNING_INTERVAL = 300000; // Log every 5 minutes max @@ -419,7 +426,13 @@ private: bool ensureSlotAuthToken(int index); // Allocate this slot's JWT token buffer on first token creation void releaseSlotAuthToken(int index);// Free the token buffer (only with the client — see MQTTSlot) void destroySlotClients(); // Delete all persistent clients (shutdown only) - void setupSlot(int index); // Configure and connect the slot's existing client + bool setupSlot(int index); // Configure and connect the slot; false = not activated + // Single definition of "this slot holds one of the _max_active_slots positions": + // it is enabled and has been through a successful setupSlot(). Startup, the + // setup-retry path, and live reconfigure all gate on these so the cap cannot be + // exceeded by one route while another enforces it. + int activatedSlotCount() const; + bool canActivateSlot(int index) const; void teardownSlot(int index); // Disconnect the slot's client (keeps the object alive) void maintainSlotConnections(); // Maintain all slot connections (token renewal, reconnect) void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted); diff --git a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp index a3b0866f..37486fd2 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -171,8 +171,9 @@ TEST(MQTTPayloadBuilder, RxPacketOmitsUnknownNanScore) { TEST(MQTTPayloadBuilder, RawMessageHasExactContractAndEscapesData) { char buffer[512]; + JsonDocument doc; int len = MQTTPayloadBuilder::buildRawMessage( - "node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer)); + doc, "node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer)); ASSERT_GT(len, 0); EXPECT_EQ(static_cast(len), strlen(buffer)); @@ -224,8 +225,9 @@ TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) { EXPECT_STREQ("3c3d3e3f", parsed_path[15].as()); char raw_buffer[1024]; + JsonDocument raw_doc; int raw_len = MQTTPayloadBuilder::buildRawMessage( - "node", "0123456789ABCDEF", kTimestamp, raw.c_str(), + raw_doc, "node", "0123456789ABCDEF", kTimestamp, raw.c_str(), raw_buffer, sizeof(raw_buffer)); ASSERT_GT(raw_len, 0); JsonDocument parsed_raw; diff --git a/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp b/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp new file mode 100644 index 00000000..b78e23bd --- /dev/null +++ b/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp @@ -0,0 +1,101 @@ +// Boundary tests for the wire-format scratch sizing used by the MQTT raw/packet +// publish paths. Packet::writeTo() cannot report an overrun (uint8_t return) and +// trusts the packet's own length fields, so canSerialize() is what keeps it in +// bounds — these cases pin the exact accept/reject edges. +#include + +#include "helpers/MQTTWireScratch.h" + +namespace { + +// A packet that serializes to the largest legal wire form: transport codes present, +// a full path, and a full payload. +mesh::Packet maxPacket() { + mesh::Packet p; + p.header = ROUTE_TYPE_TRANSPORT_DIRECT | (PAYLOAD_TYPE_TXT_MSG << PH_TYPE_SHIFT); + p.transport_codes[0] = 0x1234; + p.transport_codes[1] = 0x5678; + // The hop count field is 6 bits, so MAX_PATH_SIZE one-byte hops is NOT encodable + // (64 & 63 == 0). 32 hops of 2 bytes is the widest path that reaches MAX_PATH_SIZE. + p.setPathHashSizeAndCount(2, 32); + EXPECT_EQ(MAX_PATH_SIZE, p.getPathByteLen()); + p.payload_len = MAX_PACKET_PAYLOAD; + memset(p.payload, 0xAB, sizeof(p.payload)); + return p; +} + +} // namespace + +TEST(MQTTWireScratch, MaxLegalPacketFitsTheScratchBuffer) { + mesh::Packet p = maxPacket(); + // 1 header + 4 transport + 1 path_len + 64 path + 184 payload = 254. + EXPECT_EQ(254, p.getRawLength()); + EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); + + uint8_t buf[MQTTWireScratch::kWireBytes]; + const uint8_t written = p.writeTo(buf); + EXPECT_EQ(254, (int)written); + EXPECT_LE((size_t)written, sizeof(buf)); + // The hex buffer must hold two chars per byte plus the NUL. + EXPECT_GE(MQTTWireScratch::kWireHexChars, (size_t)written * 2 + 1); +} + +TEST(MQTTWireScratch, RejectsPayloadLenPastTheArrayEvenWhenEncodedLengthFits) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.setPathHashSizeAndCount(1, 0); + // getRawLength() == 2 + 0 + 185 == 187, comfortably inside MAX_TRANS_UNIT, but + // writeTo() would memcpy 185 bytes out of a 184-byte array. + p.payload_len = MAX_PACKET_PAYLOAD + 1; + EXPECT_LE(p.getRawLength(), (int)MQTTWireScratch::kWireBytes); + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); + + p.payload_len = MAX_PACKET_PAYLOAD; + EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} + +TEST(MQTTWireScratch, RejectsPathLenThatWouldTruncateIntoOneWireByte) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.payload_len = 4; + p.path_len = 0x100; // writeTo() stores this in a single byte + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); + + p.path_len = 0xFF; + // Still rejected, but now on the destination check rather than truncation: + // 0xFF encodes 63 hops of 4 bytes. + EXPECT_GT(p.getRawLength(), (int)MQTTWireScratch::kWireBytes); + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} + +TEST(MQTTWireScratch, DestinationEdgeIsInclusive) { + mesh::Packet p = maxPacket(); + const size_t exact = (size_t)p.getRawLength(); + EXPECT_TRUE(MQTTWireScratch::canSerialize(p, exact)); + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, exact - 1)); +} + +// A zero-payload packet serializes fine but does not survive readFrom(), which +// requires at least one payload byte. canSerialize() deliberately does not reject it: +// the raw/packet publish paths only ever write the bytes out. Any future change that +// reconstructs a Packet from queued wire bytes has to handle this asymmetry. +TEST(MQTTWireScratch, ZeroPayloadPacketSerializesButDoesNotRoundTrip) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.setPathHashSizeAndCount(1, 0); + p.payload_len = 0; + EXPECT_EQ(2, p.getRawLength()); + EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); + + uint8_t buf[MQTTWireScratch::kWireBytes]; + const uint8_t written = p.writeTo(buf); + EXPECT_EQ(2, (int)written); + + mesh::Packet restored; + EXPECT_FALSE(restored.readFrom(buf, written)); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From 6be84eb4eb4bb23acb4770b916499c5cc5bd6856 Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 14:03:21 -0700 Subject: [PATCH 6/9] fix(mqtt): let a retried setup consume the reconnect allowance reconnect_attempted_this_cycle is computed once before the maintenance loop and only maintainSlotConnection() was setting it. The deferred-setup retry armed the 15 s cross-slot guard via _last_slot_reconnect_ms but left the local flag false, so a disconnected slot later in the same pass still saw "no reconnect yet" and started a second TLS handshake. Two concurrent ~40 KB sessions is exactly the contention the guard prevents, and it landed in the one situation where internal heap is already known to be short -- a failed allocation is why the retry runs. Set on success only: setupSlot() returns true only after client->connect(), so success means a handshake was launched. A failed retry launches nothing and continues to spend only setup_retry_this_cycle, which rate-limits the allocation attempts without consuming the handshake allowance. --- src/helpers/bridges/MQTTBridge.cpp | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 6fc26693..90ffbf71 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -2047,7 +2047,17 @@ void MQTTBridge::maintainSlotConnections() { setup_retry_this_cycle = true; MQTT_DEBUG_PRINTLN("MQTT%d retrying deferred setup (int_heap=%d)", i + 1, (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL)); - if (setupSlot(i)) _last_slot_reconnect_ms = now_millis; + if (setupSlot(i)) { + // A successful setup ends in connect(), so it spends this cycle's single + // handshake allowance as well as arming the 15 s cross-slot guard. Without + // the local flag, a disconnected slot later in this same pass would start a + // second concurrent TLS handshake — the contention the guard exists to + // prevent, and most damaging here because a failed allocation is why we are + // retrying at all. A failed setup launches nothing and so spends only + // setup_retry_this_cycle. + _last_slot_reconnect_ms = now_millis; + reconnect_attempted_this_cycle = true; + } } continue; } From 53bae5684e79ad425e4e05bcf7d4fd18b16437ac Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 14:04:08 -0700 Subject: [PATCH 7/9] fix(mqtt): measure the setup-retry interval from the failure last_reconnect_attempt starts at zero and teardownSlot() re-zeroes it, so the retry gate in maintainSlotConnections() reduced to "uptime >= SLOT_SETUP_RETRY_INTERVAL". Past 60 s of uptime a failed setup was therefore retried on the next maintenance pass rather than 60 s later -- in the same task iteration for a live reconfigure, since reconfigure processing runs before maintenance in the loop. setupSlot() now stamps last_reconnect_attempt on each failure that represents a real attempt (client allocation, and both JWT token paths), so all three callers get the interval measured from the failure. The bounds check and the !enabled early return are not attempts and stay unstamped. The reconnect ladder reads this field only for slots with initial_connect_done set, which a failed setup never sets, so reconnect timing is unaffected. The retry path's own pre-call stamp is kept as a backstop for any future false-returning path that does not stamp itself. --- src/helpers/bridges/MQTTBridge.cpp | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 90ffbf71..89373341 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -1759,9 +1759,18 @@ bool MQTTBridge::setupSlot(int index) { return false; } + // Every failure below is a real attempt, so stamp it: the retry interval in + // maintainSlotConnections() measures from last_reconnect_attempt, which starts at 0 + // and is re-zeroed by teardownSlot(). Left unstamped, the gate degenerates to + // "uptime >= SLOT_SETUP_RETRY_INTERVAL" and a failure past that point is retried on + // the very next maintenance pass — the same task iteration, for a live reconfigure. + // The reconnect ladder never reads this field for an unactivated slot (it is gated + // on initial_connect_done), so stamping here cannot perturb reconnect timing. + // First setup for this slot allocates its persistent client; later ones reuse it. if (!ensureSlotClient(index)) { MQTT_DEBUG_PRINTLN("MQTT%d: client allocation failed - will retry", index + 1); + slot.last_reconnect_attempt = millis(); return false; } @@ -1825,6 +1834,7 @@ bool MQTTBridge::setupSlot(int index) { if (slot.preset->auth_type == MQTT_AUTH_JWT) { if (!createSlotAuthToken(index) || !slot.auth_token || slot.auth_token[0] == '\0') { MQTT_DEBUG_PRINTLN("MQTT%d: no usable JWT token - will retry", index + 1); + slot.last_reconnect_attempt = millis(); return false; } slot.client->setCredentials(_jwt_username, slot.auth_token); @@ -1939,6 +1949,7 @@ bool MQTTBridge::setupSlot(int index) { // JWT auth for custom slot — same rule as the preset JWT path above. if (!createSlotAuthToken(index) || !slot.auth_token || slot.auth_token[0] == '\0') { MQTT_DEBUG_PRINTLN("MQTT%d: no usable JWT token - will retry", index + 1); + slot.last_reconnect_attempt = millis(); return false; } slot.client->setCredentials(_jwt_username, slot.auth_token); From dbdf2d732f298ca97d8bdb131e879a122f31126c Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 14:04:39 -0700 Subject: [PATCH 8/9] fix(mqtt): scan every slot for a stale JWT after a clock correction The post-NTP-correction refresh looped to _max_active_slots, which is a count of activation positions and never an index bound. The indices holding those positions are not contiguous: a slot passed over by isSlotReady() -- or, since the demand-driven work, by a failed setup -- leaves a higher index activated. On a two-position board that meant slots 1 and 2 could be live while only indices 0 and 1 were scanned, so slot 3 kept a JWT issued against the pre-correction clock until its own expiry or a reconnect regenerated it. Now bounded by RUNTIME_MQTT_SLOTS. The existing guard already skips disabled, non-JWT, and clientless slots, so widening the range cannot touch a slot that was never set up. Pre-existing (the loop predates the demand-driven work) and kept as its own commit so it can be picked separately. Audited the other _max_active_slots uses: all are count comparisons or log arguments, so this was the only misuse. --- src/helpers/bridges/MQTTBridge.cpp | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 89373341..735999e9 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -3925,7 +3925,13 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) { // and re-setup all JWT-authenticated slots so they get fresh tokens. if (_slots_setup_done && was_ntp_synced) { unsigned long current_time = (unsigned long)time(nullptr); - for (int i = 0; i < _max_active_slots; i++) { + // Every slot, not _max_active_slots: that is a count of positions, never an + // index bound. Which indices hold those positions is not contiguous — a slot can + // fail isSlotReady() or its setup and be passed over, leaving a higher index + // activated — so bounding by the cap silently skipped an activated slot and left + // it holding a JWT issued against the pre-correction clock. The guard below + // already excludes disabled, non-JWT, and clientless slots. + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { bool slot_jwt = (_slots[i].preset && _slots[i].preset->auth_type == MQTT_AUTH_JWT) || (!_slots[i].preset && _slots[i].audience[0] != '\0'); if (_slots[i].enabled && slot_jwt && _slots[i].client) { From 8d1a0eb3334a2120d362d0ffa8be77c8ad29b64d Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 4 Aug 2026 14:06:32 -0700 Subject: [PATCH 9/9] fix(mqtt): reject invalid path encodings before serializing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit canSerialize() validated payload_len and the destination size but not whether the path encoding is one writePath() will actually emit. writePath() self-guards against overrunning the path array, but it does so by writing nothing and returning 0 — a correctness problem, not a safety one, because getRawLength() still counts the path. An over-long or reserved encoding therefore passed the size check and then serialized to a truncated frame that was published as the packet. Worst case is path_len 0xFF with no payload: 63 hops of 4 bytes counts as 254 bytes, inside the 255-byte buffer, while writeTo() emits just the 2-byte header. The `raw` field would carry 4 hex chars presented as the frame. Reserved 4-byte hash encodings passed too, producing frames Packet::readFrom() rejects. Now gated on Packet::isValidPathLen(), which rejects the reserved 4-byte hash size and any count * size above MAX_PATH_SIZE in one predicate. It is the same check readFrom() applies to every received packet, and TX packets are built via setPathHashSizeAndCount() with real hash sizes, so no decodable packet is turned away. Two tests added for the cases a destination-size check cannot reach. The existing truncation test passed for the wrong reason -- its payload_len of 4 pushed getRawLength() to 258 and tripped the size check, masking the hole -- so it is split into the >0xFF truncation case and the counted-length-fits case, with the 254/2-byte asymmetry asserted explicitly so it cannot be masked again. 274/274 native tests; both observer envs and an nRF52 repeater build clean. --- src/helpers/MQTTWireScratch.h | 12 +++++- .../test_mqtt_wire_scratch.cpp | 37 +++++++++++++++++-- 2 files changed, 44 insertions(+), 5 deletions(-) diff --git a/src/helpers/MQTTWireScratch.h b/src/helpers/MQTTWireScratch.h index c631cdf1..d5c79336 100644 --- a/src/helpers/MQTTWireScratch.h +++ b/src/helpers/MQTTWireScratch.h @@ -27,11 +27,19 @@ static_assert(1 + 4 + 1 + MAX_PATH_SIZE + MAX_PACKET_PAYLOAD <= MAX_TRANS_UNIT, // corrupt value can still leave getRawLength() inside MAX_TRANS_UNIT. // - path_len is written into a single wire byte, so anything above 255 is silently // truncated and would disagree with getPathByteLen(). -// The path bytes themselves need no check here: writePath() already refuses a -// getPathByteLen() above MAX_PATH_SIZE. +// - the path encoding must be one writePath() will actually emit. It self-guards +// against overrunning the path array, but by writing nothing and returning 0, which +// is a correctness problem rather than a safety one: getRawLength() still counts the +// path, so an over-long or reserved encoding passes a destination-size check and +// then serializes to a truncated frame that gets published as the packet. The worst +// case is path_len 0xFF with no payload — 254 counted bytes, 2 bytes emitted. +// isValidPathLen() rejects both the reserved 4-byte hash size and any +// count * size above MAX_PATH_SIZE, and is the same predicate Packet::readFrom() +// applies to every received packet, so no decodable packet is turned away. inline bool canSerialize(const mesh::Packet& packet, size_t dest_size) { if (packet.payload_len > MAX_PACKET_PAYLOAD) return false; if (packet.path_len > 0xFF) return false; + if (!mesh::Packet::isValidPathLen((uint8_t)packet.path_len)) return false; const int raw_len = packet.getRawLength(); return raw_len > 0 && (size_t)raw_len <= dest_size; } diff --git a/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp b/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp index b78e23bd..d8258aea 100644 --- a/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp +++ b/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp @@ -60,11 +60,42 @@ TEST(MQTTWireScratch, RejectsPathLenThatWouldTruncateIntoOneWireByte) { p.payload_len = 4; p.path_len = 0x100; // writeTo() stores this in a single byte EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} +// The case a destination-size check cannot catch, and which an earlier version of +// these tests masked by using a payload big enough to push getRawLength() over the +// limit: 0xFF encodes 63 hops of 4 bytes, so with no payload the counted length is +// 254 — inside the buffer — while writePath() refuses the 252-byte path and writeTo() +// emits only the 2-byte header. Publishing that would put 4 hex chars in the `raw` +// field and call them the packet. +TEST(MQTTWireScratch, RejectsOverlongPathEvenWhenTheCountedLengthFits) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; p.path_len = 0xFF; - // Still rejected, but now on the destination check rather than truncation: - // 0xFF encodes 63 hops of 4 bytes. - EXPECT_GT(p.getRawLength(), (int)MQTTWireScratch::kWireBytes); + p.payload_len = 0; + + ASSERT_EQ(254, p.getRawLength()); + ASSERT_LE((size_t)p.getRawLength(), MQTTWireScratch::kWireBytes); + uint8_t buf[MQTTWireScratch::kWireBytes]; + ASSERT_EQ(2, (int)p.writeTo(buf)); + + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} + +// hash_size 4 is reserved: isValidPathLen() and therefore Packet::readFrom() reject +// it, so serializing one produces a frame no receiver can parse back — even though the +// hop bytes fit and writePath() copies them happily. +TEST(MQTTWireScratch, RejectsReservedFourByteHashEncoding) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.setPathHashSizeAndCount(4, 2); + p.payload_len = 4; + + ASSERT_EQ(4, p.getPathHashSize()); + ASSERT_EQ(8, p.getPathByteLen()); // fits the path array + ASSERT_LE(p.getRawLength(), (int)MQTTWireScratch::kWireBytes); + ASSERT_FALSE(mesh::Packet::isValidPathLen((uint8_t)p.path_len)); + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); }