merge: perf/memory-savings into observer-firmware-dev

Nine commits reducing the MQTT bridge's internal-DRAM footprint, plus four
fixes that rode with them (invalid path encodings, stale-JWT scan after a
clock correction, setup-retry interval measured from the failure, retried
setup consuming the reconnect allowance).

Touches no prefs surface -- nothing in NodePrefs, MQTTPrefs, or
ConfigSerializer -- so /prefs.json layout is unaffected and there is no
fleet config risk.

Soak evidence: every soak branch already contained this work in full. Device 1
has run it 69.6 h with 325,852 publishes, 0 errors and 0 reboots. The caveat
worth carrying: that long-duration evidence is all on the reduced-TLS
framework (OUT_CONTENT_LEN 4096). Device 3 is now soaking it on the stock
framework, which is where the allocation-ordering interaction with the full
16 KiB record buffers actually gets exercised.
This commit is contained in:
agessaman
2026-08-09 09:33:20 -07:00
9 changed files with 654 additions and 226 deletions
+14 -13
View File
@@ -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;
+12
View File
@@ -3,6 +3,7 @@
#include "MeshCore.h"
#include <ArduinoJson.h>
#include "MQTTPayloadBuilder.h"
#include "MQTTWireScratch.h"
#include <Mesh.h>
#include <Timezone.h>
@@ -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,
+2 -1
View File
@@ -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<JsonObject>();
root["origin"] = origin;
+1
View File
@@ -61,6 +61,7 @@ public:
);
static int buildRawMessage(
JsonDocument& doc,
const char* origin,
const char* origin_id,
const char* timestamp,
+47
View File
@@ -0,0 +1,47 @@
#pragma once
#include <Packet.h>
#include <stddef.h>
// 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
+376 -185
View File
@@ -12,6 +12,7 @@
#include <time.h>
#include <sys/time.h>
#include <math.h>
#include <new>
#include <strings.h>
#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<uint8_t*>(MQTTRuntimeBufferLifecycle::allocateIfMissing(
_last_raw_data, LAST_RAW_DATA_SIZE, psram_malloc));
_publish_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::allocateIfMissing(
_publish_json_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc));
_status_json_buffer = static_cast<char*>(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<char*>(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<uint8_t*>(MQTTRuntimeBufferLifecycle::release(
_last_raw_data, psram_free));
_publish_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::release(
_publish_json_buffer, psram_free));
_status_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::release(
_status_json_buffer, psram_free));
_json_scratch_buffer = static_cast<char*>(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<char*>(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<char*>(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<char*>(
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<uint32_t>(now_millis),
static_cast<uint32_t>(_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<uint32_t>(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<uint8_t>(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<uint8_t>(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) {
+66 -25
View File
@@ -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<uint32_t> _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<PUBLISH_JSON_BUFFER_SIZE> _packet_json_doc;
StaticJsonDocument<STATUS_JSON_BUFFER_SIZE> _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<N> 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();
@@ -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<size_t>(len), strlen(buffer));
@@ -224,8 +225,9 @@ TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) {
EXPECT_STREQ("3c3d3e3f", parsed_path[15].as<const char*>());
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;
@@ -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 <gtest/gtest.h>
#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();
}