diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index 3ec13277..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( @@ -182,10 +183,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 +258,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 @@ -298,6 +298,7 @@ int MQTTMessageBuilder::buildPacketJSONFromRaw( } int MQTTMessageBuilder::buildRawJSON( + JsonDocument& doc, mesh::Packet* packet, const char* origin, const char* origin_id, @@ -314,11 +315,10 @@ 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); + + return buildRawMessage(doc, origin, origin_id, timestamp, raw_hex, buffer, buffer_size); } const char* MQTTMessageBuilder::getRouteTypeString(int route_type) { @@ -356,9 +356,10 @@ 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]; + if (!canSerializePacket(packet, 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..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,6 +22,15 @@ */ class MQTTMessageBuilder { public: + // 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 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). * Always UTC with an explicit "+00:00" offset, ISO-8601 @@ -146,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, @@ -232,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..d5c79336 --- /dev/null +++ b/src/helpers/MQTTWireScratch.h @@ -0,0 +1,47 @@ +#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 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; +} + +} // namespace MQTTWireScratch diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 9f1ecab3..735999e9 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 @@ -177,6 +178,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; @@ -474,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"; @@ -609,7 +647,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 +663,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 @@ -654,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; @@ -712,8 +750,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 +761,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 +785,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 +999,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 +1019,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) @@ -1005,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 @@ -1083,7 +1119,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 +1173,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) { @@ -1367,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; } @@ -1380,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) { @@ -1546,119 +1579,199 @@ 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 _status_json_doc/_status_json_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; +} + +// 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. +// +// 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]; + 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; + } + // 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; +} + +// 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, psram_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); } } -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; } - // 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); - return; + // 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; } // Reconfigure path: if we're re-applying (e.g. after a preset change), stop @@ -1685,7 +1798,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; @@ -1715,12 +1828,16 @@ 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[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); + slot.last_reconnect_attempt = millis(); + 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 @@ -1829,11 +1946,13 @@ 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) - createSlotAuthToken(index); - if (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); + slot.last_reconnect_attempt = millis(); + 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); @@ -1842,6 +1961,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 @@ -1861,7 +1981,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'; @@ -1907,8 +2029,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) || @@ -1917,6 +2043,37 @@ 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)) { + // 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; + } + if (!_slots[i].client) continue; + maintainSlotConnection(i, now_millis, current_time, time_synced, reconnect_attempted_this_cycle, teardown_attempted_this_cycle); } } @@ -1930,7 +2087,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), @@ -2029,8 +2186,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); @@ -2130,6 +2287,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]; @@ -2169,7 +2330,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 +2358,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 +2375,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 +2441,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 +2501,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 +2516,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); } @@ -2429,6 +2591,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; @@ -2450,6 +2619,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); } } @@ -2636,10 +2812,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; } @@ -2647,7 +2822,6 @@ void MQTTBridge::loop() { continue; } setupSlot(i); - active_count++; } } } @@ -3093,7 +3267,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]; @@ -3122,7 +3299,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; @@ -3144,12 +3324,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 +3379,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 +3397,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 +3471,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 +3497,37 @@ 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. 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 (MQTTMessageBuilder::canSerializePacket(packet, 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 +3543,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 +3575,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]; @@ -3408,6 +3592,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 ); @@ -3418,7 +3603,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 +3653,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; } } @@ -3740,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) { diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index aa2660cf..43af42b2 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -86,9 +86,16 @@ 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; 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 + // 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; @@ -176,9 +183,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 +289,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 +327,26 @@ 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 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; + 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 @@ -356,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 @@ -383,25 +412,37 @@ 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 + 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); 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(); 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..d8258aea --- /dev/null +++ b/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp @@ -0,0 +1,132 @@ +// 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)); +} + +// 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; + 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)); +} + +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(); +}