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