Merge remote-tracking branch 'review/observer-firmware' into keymindCascade

# Conflicts:
#	MQTT_IMPLEMENTATION.md
#	examples/simple_repeater/MyMesh.cpp
#	examples/simple_room_server/MyMesh.cpp
#	examples/simple_sensor/SensorMesh.cpp
#	src/helpers/CommonCLI.h
#	src/helpers/MQTTPresets.h
#	src/helpers/bridges/MQTTBridge.cpp
#	src/helpers/bridges/MQTTBridge.h
#	src/helpers/radiolib/CustomLLCC68Wrapper.h
#	src/helpers/ui/SH1106Display.cpp
This commit is contained in:
mikecarper committed 2026-08-19 13:45:47 -07:00
commit 4ebbdc3d9b
17 files changed
+836 -253

No files matched your search

+14 -13
View File
@@ -100,6 +100,7 @@ int MQTTMessageBuilder::buildPacketMessage(
}
int MQTTMessageBuilder::buildRawMessage(
JsonDocument& doc,
const char* origin,
const char* origin_id,
const char* timestamp,
@@ -108,7 +109,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(
@@ -188,10 +189,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);
@@ -266,9 +266,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
@@ -306,6 +306,7 @@ int MQTTMessageBuilder::buildPacketJSONFromRaw(
}
int MQTTMessageBuilder::buildRawJSON(
JsonDocument& doc,
mesh::Packet* packet,
const char* origin,
const char* origin_id,
@@ -322,11 +323,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) {
@@ -364,9 +364,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
@@ -5,6 +5,7 @@
#include "MeshCore.h"
#include <ArduinoJson.h>
#include "MQTTPayloadBuilder.h"
#include "MQTTWireScratch.h"
#include <Mesh.h>
#include <Timezone.h>
@@ -23,6 +24,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
@@ -148,6 +158,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,
@@ -234,6 +245,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
@@ -167,6 +167,7 @@ int MQTTPayloadBuilder::buildPacketMessage(
}
int MQTTPayloadBuilder::buildRawMessage(
JsonDocument& doc,
const char* origin,
const char* origin_id,
const char* timestamp,
@@ -174,7 +175,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,
+2 -1
View File
@@ -79,7 +79,7 @@ static inline bool mqttPresetNeedsSlotCredentials(const MQTTPresetDef* preset) {
}
// Number of built-in presets
static const int MQTT_PRESET_COUNT = 34;
static const int MQTT_PRESET_COUNT = 35;
// Keep the certificate and preset tables in one translation unit. Defining
// these as header-local constants created a complete flash copy in every MQTT
@@ -189,6 +189,7 @@ extern const MQTTPresetDef MQTT_PRESETS[MQTT_PRESET_COUNT] = {
// JWT token auth; LE Gen-Y ECDSA chain (YE2 -> Root YE -> X2) still anchors at ISRG Root X1.
{ "gomesh", "wss://mqtt.gomesh.dev:443", "mqtt.gomesh.dev", ISRG_ROOT_X1, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55, nullptr, nullptr },
{ "idahomesh", "wss://mqtt.idahomesh.org:443/mqtt", "mqtt.idahomesh.org", ISRG_ROOT_X1, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55, nullptr, nullptr },
{ "ntxmesh", "wss://ntxmesh.dhovin.me:8883", "ntxmesh.dhovin.me", ISRG_ROOT_X1, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55, nullptr, nullptr },
};
#else
+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
+382 -207
View File
@@ -14,6 +14,7 @@
#include <time.h>
#include <sys/time.h>
#include <math.h>
#include <new>
#include <strings.h>
#ifdef WITH_SNMP
@@ -198,6 +199,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;
@@ -497,7 +529,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";
@@ -638,7 +676,7 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs,
#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),
@@ -654,7 +692,7 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs,
_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
@@ -683,7 +721,7 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs,
_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;
@@ -741,8 +779,8 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs,
#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() {
@@ -752,14 +790,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)
@@ -779,12 +814,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(
@@ -806,21 +844,13 @@ void MQTTBridge::releaseRuntimeBuffers() {
// ---------------------------------------------------------------------------
void MQTTBridge::begin() {
MQTT_DEBUG_PRINTLN("Initializing MQTT Bridge...");
if (_initialized) return;
// end() releases the PSRAM-backed scratch buffers. Recreate them when the
// same bridge object is started again after a settings or OTA restart.
#if defined(BOARD_HAS_PSRAM)
if (_last_raw_data == nullptr) {
_last_raw_data = static_cast<uint8_t*>(psram_malloc(LAST_RAW_DATA_SIZE));
// Idempotent start: a second begin() would re-run allocation and re-create
// the task, leaking the previous queue and task.
if (_initialized) {
MQTT_DEBUG_PRINTLN("MQTT Bridge already running - begin() ignored");
return;
}
if (_publish_json_buffer == nullptr) {
_publish_json_buffer = static_cast<char*>(psram_malloc(PUBLISH_JSON_BUFFER_SIZE));
}
if (_status_json_buffer == nullptr) {
_status_json_buffer = static_cast<char*>(psram_malloc(STATUS_JSON_BUFFER_SIZE));
}
#endif
// A restarted bridge may be using a different preset or custom endpoint.
// Clear the old slot description before rebuilding it from _obs below.
@@ -844,14 +874,6 @@ void MQTTBridge::begin() {
_ntp_estimate_ok = false;
_ntp_estimate_epoch = 0;
// Idempotent start (Phase 5): a second begin() on an already-running bridge
// would re-run allocation and re-create the task, leaking the previous
// queue/task. Guard here instead of relying on caller discipline.
if (_initialized) {
MQTT_DEBUG_PRINTLN("MQTT Bridge already running - begin() ignored");
return;
}
// PSRAM diagnostic - helps debug memory fragmentation on boards with external RAM
#ifdef BOARD_HAS_PSRAM
{
@@ -1027,9 +1049,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
@@ -1048,8 +1069,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)
@@ -1071,10 +1090,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
@@ -1149,7 +1169,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()
_ntp_estimate_requested = false;
@@ -1207,10 +1227,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) {
@@ -1452,11 +1468,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;
}
@@ -1465,8 +1480,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) {
@@ -1631,119 +1648,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
@@ -1770,7 +1867,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;
@@ -1800,12 +1897,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
@@ -1920,11 +2021,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);
@@ -1933,6 +2036,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
@@ -1952,7 +2056,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';
@@ -1998,8 +2104,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) ||
@@ -2008,6 +2118,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);
}
}
@@ -2020,8 +2161,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
// so a link that can't survive a single keepalive period never resets the
// 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.
// hammering full TLS handshakes at the 10 s rung - see the onConnect
// 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),
@@ -2120,8 +2261,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);
@@ -2221,6 +2362,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];
@@ -2260,7 +2405,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) {
@@ -2288,7 +2433,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.
@@ -2305,11 +2450,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;
}
}
@@ -2371,15 +2516,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];
@@ -2433,7 +2579,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,
@@ -2448,7 +2594,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);
}
@@ -2523,6 +2669,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;
@@ -2544,6 +2697,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);
}
}
@@ -2742,10 +2902,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;
}
@@ -2753,7 +2912,6 @@ void MQTTBridge::loop() {
continue;
}
setupSlot(i);
active_count++;
}
}
}
@@ -3199,7 +3357,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];
@@ -3228,7 +3389,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;
@@ -3250,12 +3414,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];
@@ -3308,7 +3472,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,
@@ -3326,7 +3490,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;
}
}
@@ -3400,15 +3564,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];
@@ -3426,33 +3590,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
);
}
@@ -3468,7 +3636,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;
}
}
@@ -3500,15 +3668,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];
@@ -3517,6 +3685,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
);
@@ -3527,7 +3696,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;
}
}
@@ -3577,7 +3746,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;
}
}
@@ -3849,7 +4018,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
@@ -104,9 +104,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;
@@ -194,9 +201,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;
@@ -309,17 +313,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)
@@ -342,10 +351,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
@@ -380,6 +405,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
@@ -407,25 +436,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();