diff --git a/.github/workflows/build-observer-firmwares-beta.yml b/.github/workflows/build-observer-firmwares-beta.yml index 3c5e7d19..be54ee72 100644 --- a/.github/workflows/build-observer-firmwares-beta.yml +++ b/.github/workflows/build-observer-firmwares-beta.yml @@ -56,7 +56,7 @@ env: # OTA comparison treats a different base version as "always an update", so a # distinct base here would make every beta node think it is permanently behind. # Channels are separated by manifest URL, not by base version. - FIRMWARE_VERSION: v1.17.0 + FIRMWARE_VERSION: v1.17.1 # Beta-only rolling release. A separate tag is required, not cosmetic: the # publish step prunes all but the KEEP_BUILDS most recent build hashes within diff --git a/.github/workflows/build-observer-firmwares.yml b/.github/workflows/build-observer-firmwares.yml index 03b2d38c..8d553875 100644 --- a/.github/workflows/build-observer-firmwares.yml +++ b/.github/workflows/build-observer-firmwares.yml @@ -41,7 +41,7 @@ concurrency: env: # Version embedded in firmware filenames; must match the version key in the # flasher's config.json. Bump here when the observer version changes. - FIRMWARE_VERSION: v1.17.0 + FIRMWARE_VERSION: v1.17.1 # Rolling release tag that hosts the latest observer binaries. RELEASE_TAG: observer-mqtt-latest # Download host serving RELEASE_TAG's assets (cloudflare-worker in the flasher diff --git a/MQTT_IMPLEMENTATION.md b/MQTT_IMPLEMENTATION.md index 0fd636a8..58ccbd5e 100644 --- a/MQTT_IMPLEMENTATION.md +++ b/MQTT_IMPLEMENTATION.md @@ -217,6 +217,7 @@ below documents the current build. | `atvirastinklas` | `wss://mqtt-mc.atvirastinklas.lt:443` | JWT | -- | | `gomesh` | `wss://mqtt.gomesh.dev:443` | JWT | -- | | `idahomesh` | `wss://mqtt.idahomesh.org:443/mqtt` | JWT | -- | +| `ntxmesh` | `wss://ntxmesh.dhovin.me:8883` | JWT | -- | | `custom` | your own broker | User/pass, or JWT when `mqttN.audience` is set | `set mqttN.server` (see [custom broker setup](#custom-brokers)) | | `none` | (slot disabled) | -- | -- | diff --git a/docs/mbedtls-tls-footprint.md b/docs/mbedtls-tls-footprint.md new file mode 100644 index 00000000..be528e9d --- /dev/null +++ b/docs/mbedtls-tls-footprint.md @@ -0,0 +1,153 @@ +# Shrinking the per-connection TLS footprint on non-PSRAM observers + +## Why + +On a non-PSRAM Heltec V3 running two WSS/JWT broker slots, the largest allocatable block in +internal DRAM walks down in ~16 KiB steps at every TLS reconnect while total free heap stays +flat. Measured on hardware over 50 reconnect cycles: 62,452 -> 16,372 bytes, permanently. + +The step size is not a coincidence. `framework-arduinoespressif32 3.20017` (Arduino 2.0.17, +IDF 4.4) builds mbedTLS with the **symmetric** buffer configuration: + +``` +CONFIG_MBEDTLS_SSL_MAX_CONTENT_LEN 16384 # sizes BOTH the in and out record buffers +CONFIG_MBEDTLS_SSL_KEEP_PEER_CERTIFICATE 1 +# CONFIG_MBEDTLS_ASYMMETRIC_CONTENT_LEN -- not defined +# CONFIG_MBEDTLS_DYNAMIC_BUFFER -- not defined +# CONFIG_MBEDTLS_SSL_VARIABLE_BUFFER_LENGTH -- not defined +``` + +Read from `packages/framework-arduinoespressif32/tools/sdk/esp32s3/qio_qspi/include/sdkconfig.h`. +Note the separate `framework-arduinoespressif32-libs` package in `~/.platformio` belongs to +the esp32c6 env (pioarduino, IDF 5.3) and is **not** what this env links against -- though it +happens to carry identical mbedTLS values. + +So each broker slot costs **2 x 16 KiB = 32 KiB** of contiguous internal DRAM in record +buffers alone, and two slots cost 64 KiB on a board with roughly 80 KiB free. Every reconnect +frees and re-allocates those buffers, and anything that lands in the vacated hole in between +prevents them from going back, which is the ratchet. + +Confirmed by two independent observations: losing a whole TLS session returned exactly 16,384 +bytes of contiguity on one device and exactly 32,768 on another, and per-connection teardown +frees ~41.6-44.7 KB total. + +## What the firmware could already do, and its limit + +`softDisconnect()` (branch `perf/mqtt-renewal-no-stop`, commit `6c51e468`) stops the JWT +renewal bounce from destroying and recreating the esp-mqtt task, keeping its 6 KiB stack out +of the hole. Measured: the staircase arrests after 2 steps at 36,852 through cycle 16, where +the baseline took 4 steps and settled at 16,372 by cycle 11 -- about 20 KB better. + +That is as far as the application layer reaches. MQTT 3.1.1 has no re-authentication packet, +so presenting a fresh JWT *requires* a transport reconnect; mbedTLS's internal allocation +order during the handshake is not controllable from the application. The remaining cost is +the record buffers themselves. + +## The changes + +All three are compile-time in mbedTLS, and the Arduino framework ships precompiled `.a` +archives (`tools/sdk/esp32s3/lib/libmbedtls.a`), so a project-level `-D` cannot change them. +A custom framework build is required. + +| Setting | From | To | Saving per connection | +|---|---|---|---| +| `CONFIG_MBEDTLS_ASYMMETRIC_CONTENT_LEN` | unset | `y` | enables the two below | +| `CONFIG_MBEDTLS_SSL_IN_CONTENT_LEN` | 16384 (implied) | 16384 | none -- keep it | +| `CONFIG_MBEDTLS_SSL_OUT_CONTENT_LEN` | 16384 (implied) | 4096 | **~12 KiB** | +| `CONFIG_MBEDTLS_SSL_KEEP_PEER_CERTIFICATE` | `1` | `n` | ~4 KiB | + +Roughly 16 KiB per connection, 32 KiB across two slots. + +### Why inbound stays at 16 KiB + +A peer may legitimately send a 16 KiB TLS record. Lowering the inbound limit only works if +both brokers negotiate the max-fragment-length extension or provably never send larger, and +getting it wrong produces invalid-record and handshake failures rather than a clean error. +Outbound is ours to choose: this firmware's MQTT and WebSocket frames are far below 4 KiB +(`MAX_TRANS_UNIT`-bounded packets plus small JSON), so 4 KiB is comfortable. + +### Risk on the peer certificate + +Dropping `KEEP_PEER_CERTIFICATE` means `mbedtls_ssl_get_peer_cert()` returns NULL after the +handshake. Chain validation still happens -- only retention of the parsed leaf changes. This +firmware verifies against a CA (`GTS_ROOT_R4` / the bundle) and never inspects the peer +certificate or a fingerprint, so it should be safe. It does change `mbedtls_ssl_session` +layout, which is exactly why the whole framework must be rebuilt together rather than +swapping a single archive in. + +## Build procedure + +### Do not use esp32-arduino-lib-builder for this + +`release/v4.4` is the branch matching Arduino 2.0.x, but its `update-components.sh` clones +every dependency at **master**, so it no longer resolves. Five successive failures, each a +different repo: the arduino branch name it passes to `-A` does not exist; `jq` is absent +from the IDF image and its absence makes `build.sh`'s target loop a **silent no-op that +still exits 0**; `esp_littlefs` and `esp32-camera` master require IDF >=5.0/>=5.1; +`esp32-camera` later needs an `esp_jpeg` version the 4.4 registry cannot satisfy; and +tinyusb's source layout no longer matches `arduino_tinyusb/CMakeLists.txt`. Pinning each +one in turn just surfaces the next. + +### Rebuild only the mbedTLS archives + +More rigorous anyway, because it reuses the shipped `sdkconfig` verbatim -- so the archives +differ from stock *only* by the intended change, with no arduino-version or +`DYNAMIC_BUFFER` drift. + +This is ABI-safe for the content-length change specifically: `ssl.h` declares `in_buf` and +`out_buf` as `unsigned char *`, allocated in `ssl_setup()`, and no public struct embeds a +CONTENT_LEN-sized array. The other precompiled archives (esp-tls, esp_http_client, +esp-mqtt) therefore remain compatible. **It is not safe for +`CONFIG_MBEDTLS_SSL_KEEP_PEER_CERTIFICATE`**, which changes `mbedtls_ssl_session` layout -- +that one needs everything rebuilt together, so it is excluded here. + +1. Minimal IDF project whose only component requirement is `mbedtls`. +2. `sdkconfig.defaults` = the shipped + `packages/framework-arduinoespressif32/tools/sdk/esp32s3/sdkconfig`, with + `CONFIG_MBEDTLS_SSL_MAX_CONTENT_LEN=16384` replaced by the three asymmetric lines. + Verify the diff is exactly 5 lines before building. +3. `docker run --rm -v $PWD:/project -w /project espressif/idf:v4.4.7 idf.py -DIDF_TARGET=esp32s3 build` +4. Confirm `build/config/sdkconfig.h` carries `OUT_CONTENT_LEN 4096`. +5. Stage the four archives under the framework's names -- note the rename: + + | built | framework name | stock size | rebuilt | + |---|---|---:|---:| + | `esp-idf/mbedtls/libmbedtls.a` | `libmbedtls.a` | 113,914 | 113,338 | + | `esp-idf/mbedtls/mbedtls/library/libmbedtls.a` | **`libmbedtls_2.a`** | 1,252,150 | 1,245,718 | + | `.../libmbedcrypto.a` | `libmbedcrypto.a` | 4,302,698 | 4,259,458 | + | `.../libmbedx509.a` | `libmbedx509.a` | 676,590 | 673,318 | + + All within ~1% of stock, which is a good check that only the config differs. + +### Wire it in with -L, never platform_packages + +``` +PLATFORMIO_BUILD_FLAGS="-L/path/to/staged/archives" pio run -e Heltec_v3_repeater_observer_mqtt +``` + +Verify with `grep -oE "[^ ]*libmbed[a-z0-9_]*\.a" .pio/build//firmware.map | sort -u` -- +every path must be the staged directory. + +**Do not** point `platform_packages` at a `file://` copy of the framework. PlatformIO +installs it *over* the shared `~/.platformio/packages/framework-arduinoespressif32`, +silently changing mbedTLS for every other ESP32 env and project on the machine. It does this +even when the copy's `package.json` version differs -- verified twice here, and both times the +fix was `rm -rf` the package and `pio pkg install` to re-download stock. A prepended library +search path keeps the change scoped to one env, because the linker takes each archive member +from the first archive that satisfies an undefined symbol. + +## How to verify it worked + +1. `strings`/`grep` the new `sdkconfig.h` for the four settings. +2. Build and check the RAM figure; static usage should be unchanged (these are heap buffers). +3. On hardware, `get mqtt.stats` at boot with two slots connected: the largest free block + should start roughly 24-32 KiB higher than the current 62-67 KiB. +4. Soak across reconnect cycles and compare the floor against the two recorded runs: + baseline settled 16,372 (cycle 11); `softDisconnect` holds 36,852 (cycle 16). + +## Prior art in this investigation + +`.scratch/mqtt-non-psram-heap-staircase-analysis-2026-08-05.md` (untracked -- `.scratch/` is +globally gitignored) holds the full allocation inventory. `~/mqtt-soak/STATE.md` holds the +soak evidence, including two retracted hypotheses worth not repeating: the perf commits were +not the cause, and waev does not cap connections per IP. diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index 584c15b3..eead4f46 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -3120,7 +3120,7 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc _prefs.rx_boosted_gain = mesh::radio::configuredRxBoostedGainDefault(); #endif _prefs.radio_fem_rxgain = 1; // LoRa FEM RX gain on by default (FEM boards) - _prefs.radio_fem_txgain = 0; + _prefs.radio_fem_txgain = 0; // LoRa FEM TX gain off by default (FEM boards) pending_discover_tag = 0; pending_discover_until = 0; diff --git a/firmware-notes.html b/firmware-notes.html index eeda7b07..a6d5a786 100644 --- a/firmware-notes.html +++ b/firmware-notes.html @@ -1 +1 @@ -MQTT Observer v1.16.0 (experimental)

See setup guide for configuration instructions.

+MQTT Observer v1.17.1

See setup guide for configuration instructions.

diff --git a/platformio.local.ini.hold b/platformio.local.ini.hold new file mode 100644 index 00000000..2655f2aa --- /dev/null +++ b/platformio.local.ini.hold @@ -0,0 +1,15 @@ +; Local-only override (gitignored) pointing the Heltec V3 observer env at a custom +; framework whose mbedTLS archives were rebuilt with an asymmetric TLS record buffer: +; CONFIG_MBEDTLS_ASYMMETRIC_CONTENT_LEN=y +; CONFIG_MBEDTLS_SSL_IN_CONTENT_LEN=16384 (unchanged -- a peer may send a 16 KiB record) +; CONFIG_MBEDTLS_SSL_OUT_CONTENT_LEN=4096 (was 16384) +; +; Expected: ~12 KiB less internal DRAM per TLS connection, ~24 KiB across two broker slots. +; Built from the shipped sdkconfig verbatim plus those three lines, so the archives differ +; only by this change. See docs/mbedtls-tls-footprint.md. +; +; Absolute path, hence local-only: not committable. + +[env:Heltec_v3_repeater_observer_mqtt] +platform_packages = + framework-arduinoespressif32 @ file:///Users/adam/framework-arduinoespressif32-tlsfix diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index 8486b68f..e1d5aa51 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -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; diff --git a/src/helpers/MQTTMessageBuilder.h b/src/helpers/MQTTMessageBuilder.h index 098b2d3b..496cebbc 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -5,6 +5,7 @@ #include "MeshCore.h" #include #include "MQTTPayloadBuilder.h" +#include "MQTTWireScratch.h" #include #include @@ -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, diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp index 06634625..19853af3 100644 --- a/src/helpers/MQTTPayloadBuilder.cpp +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -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(); root["origin"] = origin; diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h index 24e56124..7bca170d 100644 --- a/src/helpers/MQTTPayloadBuilder.h +++ b/src/helpers/MQTTPayloadBuilder.h @@ -61,6 +61,7 @@ public: ); static int buildRawMessage( + JsonDocument& doc, const char* origin, const char* origin_id, const char* timestamp, diff --git a/src/helpers/MQTTPresets.h b/src/helpers/MQTTPresets.h index f8362b57..6d196221 100644 --- a/src/helpers/MQTTPresets.h +++ b/src/helpers/MQTTPresets.h @@ -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 diff --git a/src/helpers/MQTTWireScratch.h b/src/helpers/MQTTWireScratch.h new file mode 100644 index 00000000..bc07182d --- /dev/null +++ b/src/helpers/MQTTWireScratch.h @@ -0,0 +1,47 @@ +#pragma once + +#include +#include + +// Sizing and validation for the scratch buffers that hold a serialized packet. +// Pure so the boundary conditions can be tested on the host: the firmware-side +// callers are MQTTMessageBuilder::packetToHex() and MQTTBridge::publishPacket(). +namespace MQTTWireScratch { + +// A serialized packet is header(1) + transport codes(0|4) + path_len(1) + +// path(<= MAX_PATH_SIZE) + payload(<= MAX_PACKET_PAYLOAD). Packet::writeTo() +// returns uint8_t, so MAX_TRANS_UNIT is the hard ceiling. +static const size_t kWireBytes = MAX_TRANS_UNIT; +// Two uppercase hex chars per byte, plus the NUL. +static const size_t kWireHexChars = 2 * MAX_TRANS_UNIT + 1; + +static_assert(1 + 4 + 1 + MAX_PATH_SIZE + MAX_PACKET_PAYLOAD <= MAX_TRANS_UNIT, + "serialized packet no longer fits MAX_TRANS_UNIT -- resize the wire scratch buffers"); + +// True when Packet::writeTo() can safely serialize `packet` into `dest_size` bytes. +// +// writeTo() trusts the packet's own length fields and cannot report an overrun (its +// return type is uint8_t), so the source fields must be checked as well as the +// destination: +// - payload_len drives an unchecked memcpy out of a MAX_PACKET_PAYLOAD array, and a +// corrupt value can still leave getRawLength() inside MAX_TRANS_UNIT. +// - path_len is written into a single wire byte, so anything above 255 is silently +// truncated and would disagree with getPathByteLen(). +// - the path encoding must be one writePath() will actually emit. It self-guards +// against overrunning the path array, but by writing nothing and returning 0, which +// is a correctness problem rather than a safety one: getRawLength() still counts the +// path, so an over-long or reserved encoding passes a destination-size check and +// then serializes to a truncated frame that gets published as the packet. The worst +// case is path_len 0xFF with no payload -- 254 counted bytes, 2 bytes emitted. +// isValidPathLen() rejects both the reserved 4-byte hash size and any +// count * size above MAX_PATH_SIZE, and is the same predicate Packet::readFrom() +// applies to every received packet, so no decodable packet is turned away. +inline bool canSerialize(const mesh::Packet& packet, size_t dest_size) { + if (packet.payload_len > MAX_PACKET_PAYLOAD) return false; + if (packet.path_len > 0xFF) return false; + if (!mesh::Packet::isValidPathLen((uint8_t)packet.path_len)) return false; + const int raw_len = packet.getRawLength(); + return raw_len > 0 && (size_t)raw_len <= dest_size; +} + +} // namespace MQTTWireScratch diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index ea757fe9..9914a14e 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -14,6 +14,7 @@ #include #include #include +#include #include #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(MQTTRuntimeBufferLifecycle::allocateIfMissing( _last_raw_data, LAST_RAW_DATA_SIZE, psram_malloc)); - _publish_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( - _publish_json_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc)); - _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( - _status_json_buffer, STATUS_JSON_BUFFER_SIZE, psram_malloc)); - MQTT_DEBUG_PRINTLN("Runtime buffers: raw=%s publish=%s status=%s", + _json_scratch_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( + _json_scratch_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc)); + MQTT_DEBUG_PRINTLN("Runtime buffers: raw=%s json=%s", _last_raw_data ? "PSRAM" : "unavailable", - _publish_json_buffer ? "PSRAM" : "stack fallback", - _status_json_buffer ? "PSRAM" : "stack fallback"); + _json_scratch_buffer ? "PSRAM" : "stack fallback"); #endif #if defined(WITH_MQTT_NEIGHBORS) @@ -779,12 +814,15 @@ void MQTTBridge::releaseRuntimeBuffers() { #if defined(BOARD_HAS_PSRAM) _last_raw_data = static_cast(MQTTRuntimeBufferLifecycle::release( _last_raw_data, psram_free)); - _publish_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( - _publish_json_buffer, psram_free)); - _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( - _status_json_buffer, psram_free)); + _json_scratch_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( + _json_scratch_buffer, psram_free)); #endif + // Drop the shared document's pools with the buffers. clear() destroys every pool + // and resets the list to its inline array; the next publish reallocates. Holding + // 4 KB of pool across a stopped bridge is pure overhead. + _json_scratch_doc.clear(); + #if defined(WITH_MQTT_NEIGHBORS) // Paired with the unconditional allocation in allocateRuntimeBuffers(). _neighbors_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( @@ -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(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(psram_malloc(PUBLISH_JSON_BUFFER_SIZE)); - } - if (_status_json_buffer == nullptr) { - _status_json_buffer = static_cast(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(MQTTRuntimeBufferLifecycle::allocateIfMissing( + slot.auth_token, AUTH_TOKEN_SIZE, psram_malloc)); + if (slot.auth_token == nullptr) { + MQTT_DEBUG_PRINTLN("MQTT%d: out of memory allocating auth token", index + 1); + return false; + } + // Initialise only a newly allocated buffer. Clearing on every call would discard a + // valid token at the start of each renewal, so a renewal that then failed inside + // JWTHelper would leave the slot with an empty password where it previously kept + // working credentials (JWTHelper writes the token only on success). + if (fresh) slot.auth_token[0] = '\0'; + return true; +} + +// Safe only once this slot's client is gone: setCredentials() gave the client this +// pointer, and esp-mqtt re-reads it from _mqtt_cfg on any later connect() that +// re-applies a dirtied config. See the MQTTSlot::auth_token comment. +void MQTTBridge::releaseSlotAuthToken(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; + MQTTSlot& slot = _slots[index]; + slot.auth_token = static_cast( + MQTTRuntimeBufferLifecycle::release(slot.auth_token, psram_free)); + slot.token_expires_at = 0; + slot.last_token_renewal = 0; } void MQTTBridge::destroySlotClients() { for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { MQTTSlot& slot = _slots[i]; - if (slot.client == nullptr) continue; - - if (slot.client->connected()) { - slot.client->disconnect(); + if (slot.client != nullptr) { + if (slot.client->connected()) { + slot.client->disconnect(); + } + #ifdef ESP_PLATFORM + vTaskDelay(pdMS_TO_TICKS(50)); + #else + delay(50); + #endif + delete slot.client; + slot.client = nullptr; } - #ifdef ESP_PLATFORM - vTaskDelay(pdMS_TO_TICKS(50)); - #else - delay(50); - #endif - delete slot.client; - slot.client = nullptr; + // Unconditional: only now is the token unreachable from the client's stored + // config, and a token without a client would otherwise leak. + releaseSlotAuthToken(i); } } -void MQTTBridge::setupSlot(int index) { - if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; +int MQTTBridge::activatedSlotCount() const { + int n = 0; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].enabled && _slots[i].initial_connect_done) n++; + } + return n; +} + +bool MQTTBridge::canActivateSlot(int index) const { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; + // Already holding a position (a reconfigure of a live slot) -- no new position needed. + if (_slots[index].enabled && _slots[index].initial_connect_done) return true; + return activatedSlotCount() < _max_active_slots; +} + +// Returns true only when the slot reached connect(). A false result leaves the slot +// enabled but not activated, so it holds no active-slot position and +// maintainSlotConnections() will retry it -- the allocation failures below are transient +// memory conditions, not permanent misconfiguration. +bool MQTTBridge::setupSlot(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; if (!slot.enabled) { teardownSlot(index); - return; + return false; } - // Persistent client is expected to have been allocated by initSlotClients(). - // If it hasn't, we can't proceed - bail loudly rather than silently leaking. - if (slot.client == nullptr) { - MQTT_DEBUG_PRINTLN("MQTT%d: setupSlot before initSlotClients() - skipping", index + 1); - return; + // Every failure below is a real attempt, so stamp it: the retry interval in + // maintainSlotConnections() measures from last_reconnect_attempt, which starts at 0 + // and is re-zeroed by teardownSlot(). Left unstamped, the gate degenerates to + // "uptime >= SLOT_SETUP_RETRY_INTERVAL" and a failure past that point is retried on + // the very next maintenance pass -- the same task iteration, for a live reconfigure. + // The reconnect ladder never reads this field for an unactivated slot (it is gated + // on initial_connect_done), so stamping here cannot perturb reconnect timing. + + // First setup for this slot allocates its persistent client; later ones reuse it. + if (!ensureSlotClient(index)) { + MQTT_DEBUG_PRINTLN("MQTT%d: client allocation failed - will retry", index + 1); + slot.last_reconnect_attempt = millis(); + return false; } // Reconfigure path: if we're re-applying (e.g. after a preset change), stop @@ -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(now_millis), + static_cast(_slots[i].last_reconnect_attempt)) + >= SLOT_SETUP_RETRY_INTERVAL) { + _slots[i].last_reconnect_attempt = now_millis; + setup_retry_this_cycle = true; + MQTT_DEBUG_PRINTLN("MQTT%d retrying deferred setup (int_heap=%d)", i + 1, + (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL)); + if (setupSlot(i)) { + // A successful setup ends in connect(), so it spends this cycle's single + // handshake allowance as well as arming the 15 s cross-slot guard. Without + // the local flag, a disconnected slot later in this same pass would start a + // second concurrent TLS handshake -- the contention the guard exists to + // prevent, and most damaging here because a failed allocation is why we are + // retrying at all. A failed setup launches nothing and so spends only + // setup_retry_this_cycle. + _last_slot_reconnect_ms = now_millis; + reconnect_attempted_this_cycle = true; + } + } + continue; + } + if (!_slots[i].client) continue; + maintainSlotConnection(i, now_millis, current_time, time_synced, reconnect_attempted_this_cycle, teardown_attempted_this_cycle); } } @@ -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(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(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(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) { diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 1810519e..4d59414b 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -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 _neighbors_secs_until_next; #endif - // JSON document scratch space - inline StaticJsonDocument keeps the pool off the MQTT - // task stack and eliminates two separate heap allocations (fragmentation reduction). - StaticJsonDocument _packet_json_doc; - StaticJsonDocument _status_json_doc; + // Routes the shared document's pools to PSRAM where the board has it, matching the + // neighbors document's allocator in MyMesh.cpp. ArduinoJson's default allocator is + // plain malloc(), which puts every per-publish pool block in internal DRAM next to + // the mbedTLS working set. A block is ARDUINOJSON_POOL_CAPACITY slots: these targets + // are 32-bit, so ARDUINOJSON_SLOT_ID_SIZE is 2 and that resolves to 128 slots = + // 1024 bytes per block, not the 4096 quoted near NEIGHBORS_DOC_POOL_BUDGET below + // (which describes a 64-bit configuration; its own byte measurements still stand). + struct JsonScratchAllocator : ArduinoJson::Allocator { + void* allocate(size_t size) override; + void deallocate(void* ptr) override; + void* reallocate(void* ptr, size_t new_size) override; + }; + JsonScratchAllocator _json_allocator; + + // Shared by the packet/raw/status builders, like _json_scratch_buffer above. + // Declared after _json_allocator so the allocator is constructed first. + // This was a StaticJsonDocument described as an inline pool; under ArduinoJson 7 + // that is a deprecated empty subclass of JsonDocument whose template argument only + // feeds capacity(), so the object is 64 bytes and every pool comes from the allocator. + JsonDocument _json_scratch_doc{&_json_allocator}; // Memory pressure monitoring (per-publish skip; see publishPacket()). // The broader fragmentation-recovery machinery was removed in Phase 4 of @@ -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(); diff --git a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp index 23b83c2e..733dbf07 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -185,8 +185,9 @@ TEST(MQTTPayloadBuilder, RxPacketOmitsUnknownNanScore) { TEST(MQTTPayloadBuilder, RawMessageHasExactContractAndEscapesData) { char buffer[512]; + JsonDocument doc; int len = MQTTPayloadBuilder::buildRawMessage( - "node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer)); + doc, "node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer)); ASSERT_GT(len, 0); EXPECT_EQ(static_cast(len), strlen(buffer)); @@ -238,8 +239,9 @@ TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) { EXPECT_STREQ("3c3d3e3f", parsed_path[15].as()); char raw_buffer[1024]; + JsonDocument raw_doc; int raw_len = MQTTPayloadBuilder::buildRawMessage( - "node", "0123456789ABCDEF", kTimestamp, raw.c_str(), + raw_doc, "node", "0123456789ABCDEF", kTimestamp, raw.c_str(), raw_buffer, sizeof(raw_buffer)); ASSERT_GT(raw_len, 0); JsonDocument parsed_raw; diff --git a/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp b/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp new file mode 100644 index 00000000..555778c0 --- /dev/null +++ b/test/test_mqtt_wire_scratch/test_mqtt_wire_scratch.cpp @@ -0,0 +1,133 @@ +// Boundary tests for the wire-format scratch sizing used by the MQTT raw/packet +// publish paths. Packet::writeTo() cannot report an overrun (uint8_t return) and +// trusts the packet's own length fields, so canSerialize() is what keeps it in +// bounds -- these cases pin the exact accept/reject edges. +#include + +#include "helpers/MQTTWireScratch.h" + +namespace { + +// A packet that serializes to the largest legal wire form: transport codes present, +// a full path, and a full payload. +mesh::Packet maxPacket() { + mesh::Packet p; + p.header = ROUTE_TYPE_TRANSPORT_DIRECT | (PAYLOAD_TYPE_TXT_MSG << PH_TYPE_SHIFT); + p.transport_codes[0] = 0x1234; + p.transport_codes[1] = 0x5678; + // The hop count field is 6 bits, so MAX_PATH_SIZE one-byte hops is NOT encodable + // (64 & 63 == 0). 32 hops of 2 bytes is the widest path that reaches MAX_PATH_SIZE. + p.setPathHashSizeAndCount(2, 32); + EXPECT_EQ(MAX_PATH_SIZE, p.getPathByteLen()); + p.payload_len = MAX_PACKET_PAYLOAD; + memset(p.payload, 0xAB, sizeof(p.payload)); + return p; +} + +} // namespace + +TEST(MQTTWireScratch, MaxLegalPacketFitsTheScratchBuffer) { + mesh::Packet p = maxPacket(); + // 1 header + 4 transport + 1 path_len + 64 path + 184 payload = 254. + EXPECT_EQ(254, p.getRawLength()); + EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); + + uint8_t buf[MQTTWireScratch::kWireBytes]; + const uint8_t written = p.writeTo(buf); + EXPECT_EQ(254, (int)written); + EXPECT_LE((size_t)written, sizeof(buf)); + // The hex buffer must hold two chars per byte plus the NUL. + EXPECT_GE(MQTTWireScratch::kWireHexChars, (size_t)written * 2 + 1); +} + +TEST(MQTTWireScratch, RejectsPayloadLenPastTheArrayEvenWhenEncodedLengthFits) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.setPathHashSizeAndCount(1, 0); + // getRawLength() == 2 + 0 + 185 == 187, comfortably inside MAX_TRANS_UNIT, but + // writeTo() would memcpy 185 bytes out of a 184-byte array. + p.payload_len = MAX_PACKET_PAYLOAD + 1; + EXPECT_LE(p.getRawLength(), (int)MQTTWireScratch::kWireBytes); + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); + + p.payload_len = MAX_PACKET_PAYLOAD; + EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} + +TEST(MQTTWireScratch, RejectsPathLenThatWouldTruncateIntoOneWireByte) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.payload_len = 4; + p.path_len = 0x100; // writeTo() stores this in a single byte + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} + +// The case a destination-size check cannot catch, and which an earlier version of +// these tests masked by using a payload big enough to push getRawLength() over the +// limit: 0xFF encodes 63 hops of 4 bytes, so with no payload the counted length is +// 254 -- inside the buffer -- while writePath() refuses the 252-byte path and writeTo() +// emits only the 2-byte header. Publishing that would put 4 hex chars in the `raw` +// field and call them the packet. +TEST(MQTTWireScratch, RejectsOverlongPathEvenWhenTheCountedLengthFits) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.path_len = 0xFF; + p.payload_len = 0; + + ASSERT_EQ(254, p.getRawLength()); + ASSERT_LE((size_t)p.getRawLength(), MQTTWireScratch::kWireBytes); + uint8_t buf[MQTTWireScratch::kWireBytes]; + ASSERT_EQ(2, (int)p.writeTo(buf)); + + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} + +// hash_size 4 is reserved: isValidPathLen() and therefore Packet::readFrom() reject +// it, so serializing one produces a frame no receiver can parse back -- even though the +// hop bytes fit and writePath() copies them happily. +TEST(MQTTWireScratch, RejectsReservedFourByteHashEncoding) { + mesh::Packet p; + p.header = ROUTE_TYPE_FLOOD; + p.setPathHashSizeAndCount(4, 2); + p.payload_len = 4; + + ASSERT_EQ(4, p.getPathHashSize()); + ASSERT_EQ(8, p.getPathByteLen()); // fits the path array + ASSERT_LE(p.getRawLength(), (int)MQTTWireScratch::kWireBytes); + ASSERT_FALSE(mesh::Packet::isValidPathLen((uint8_t)p.path_len)); + + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes)); +} + +TEST(MQTTWireScratch, DestinationEdgeIsInclusive) { + mesh::Packet p = maxPacket(); + const size_t exact = (size_t)p.getRawLength(); + EXPECT_TRUE(MQTTWireScratch::canSerialize(p, exact)); + EXPECT_FALSE(MQTTWireScratch::canSerialize(p, exact - 1)); +} + +// A zero-payload packet is a valid two-byte wire frame: header plus path length. +// Keep the serializer and parser contract aligned at that lower boundary. +TEST(MQTTWireScratch, ZeroPayloadPacketSerializesAndRoundTrips) { + 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; + ASSERT_TRUE(restored.readFrom(buf, written)); + EXPECT_EQ(p.header, restored.header); + EXPECT_EQ(0, restored.getPathByteLen()); + EXPECT_EQ(0, restored.payload_len); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}