mirror of
https://github.com/mikecarper/MeshCore.git
synced 2026-09-01 22:48:17 +00:00
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:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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) | -- | -- |
|
||||
|
||||
|
||||
@@ -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/<env>/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.
|
||||
@@ -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;
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
<b>MQTT Observer v1.16.0 (experimental)</b><ul><li>Up to 6 MQTT broker slots with built-in presets</li><li>Presets: Analyzer, MeshMapper, MeshRank, Waev, Meshomatic, CascadiaMesh, TennMesh, NashMesh, and more</li><li>JWT (Ed25519) and username/password authentication</li><li>Automatic reconnection with exponential backoff</li><li>After flashing, configure via serial console (115200 baud)</li></ul><p>See <a target='_blank' href='https://github.com/agessaman/MeshCore/blob/observer-firmware/MQTT_IMPLEMENTATION.md'>setup guide</a> for configuration instructions.</p>
|
||||
<b>MQTT Observer v1.17.1</b><ul><li>Based on MeshCore 1.17.1</li><li>Up to 6 MQTT broker slots with built-in community presets</li><li>Presets include Analyzer, MeshMapper, MeshRank, Waev, Meshomatic, CascadiaMesh, TennMesh, NashMesh, IdahoMesh, and more</li><li>JWT (Ed25519) and username/password authentication</li><li>Web config portal on the device, plus serial console (115200 baud)</li><li>Over-the-air updates within the channel you flashed</li><li>Neighbor discovery uplink on supported boards</li></ul><p>See <a target='_blank' href='https://github.com/agessaman/MeshCore/blob/observer-firmware/MQTT_IMPLEMENTATION.md'>setup guide</a> for configuration instructions.</p>
|
||||
|
||||
@@ -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
|
||||
@@ -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;
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -61,6 +61,7 @@ public:
|
||||
);
|
||||
|
||||
static int buildRawMessage(
|
||||
JsonDocument& doc,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
const char* timestamp,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
@@ -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) {
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
@@ -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<size_t>(len), strlen(buffer));
|
||||
@@ -238,8 +239,9 @@ TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) {
|
||||
EXPECT_STREQ("3c3d3e3f", parsed_path[15].as<const char*>());
|
||||
|
||||
char raw_buffer[1024];
|
||||
JsonDocument raw_doc;
|
||||
int raw_len = MQTTPayloadBuilder::buildRawMessage(
|
||||
"node", "0123456789ABCDEF", kTimestamp, raw.c_str(),
|
||||
raw_doc, "node", "0123456789ABCDEF", kTimestamp, raw.c_str(),
|
||||
raw_buffer, sizeof(raw_buffer));
|
||||
ASSERT_GT(raw_len, 0);
|
||||
JsonDocument parsed_raw;
|
||||
|
||||
@@ -0,0 +1,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 <gtest/gtest.h>
|
||||
|
||||
#include "helpers/MQTTWireScratch.h"
|
||||
|
||||
namespace {
|
||||
|
||||
// A packet that serializes to the largest legal wire form: transport codes present,
|
||||
// a full path, and a full payload.
|
||||
mesh::Packet maxPacket() {
|
||||
mesh::Packet p;
|
||||
p.header = ROUTE_TYPE_TRANSPORT_DIRECT | (PAYLOAD_TYPE_TXT_MSG << PH_TYPE_SHIFT);
|
||||
p.transport_codes[0] = 0x1234;
|
||||
p.transport_codes[1] = 0x5678;
|
||||
// The hop count field is 6 bits, so MAX_PATH_SIZE one-byte hops is NOT encodable
|
||||
// (64 & 63 == 0). 32 hops of 2 bytes is the widest path that reaches MAX_PATH_SIZE.
|
||||
p.setPathHashSizeAndCount(2, 32);
|
||||
EXPECT_EQ(MAX_PATH_SIZE, p.getPathByteLen());
|
||||
p.payload_len = MAX_PACKET_PAYLOAD;
|
||||
memset(p.payload, 0xAB, sizeof(p.payload));
|
||||
return p;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(MQTTWireScratch, MaxLegalPacketFitsTheScratchBuffer) {
|
||||
mesh::Packet p = maxPacket();
|
||||
// 1 header + 4 transport + 1 path_len + 64 path + 184 payload = 254.
|
||||
EXPECT_EQ(254, p.getRawLength());
|
||||
EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
|
||||
uint8_t buf[MQTTWireScratch::kWireBytes];
|
||||
const uint8_t written = p.writeTo(buf);
|
||||
EXPECT_EQ(254, (int)written);
|
||||
EXPECT_LE((size_t)written, sizeof(buf));
|
||||
// The hex buffer must hold two chars per byte plus the NUL.
|
||||
EXPECT_GE(MQTTWireScratch::kWireHexChars, (size_t)written * 2 + 1);
|
||||
}
|
||||
|
||||
TEST(MQTTWireScratch, RejectsPayloadLenPastTheArrayEvenWhenEncodedLengthFits) {
|
||||
mesh::Packet p;
|
||||
p.header = ROUTE_TYPE_FLOOD;
|
||||
p.setPathHashSizeAndCount(1, 0);
|
||||
// getRawLength() == 2 + 0 + 185 == 187, comfortably inside MAX_TRANS_UNIT, but
|
||||
// writeTo() would memcpy 185 bytes out of a 184-byte array.
|
||||
p.payload_len = MAX_PACKET_PAYLOAD + 1;
|
||||
EXPECT_LE(p.getRawLength(), (int)MQTTWireScratch::kWireBytes);
|
||||
EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
|
||||
p.payload_len = MAX_PACKET_PAYLOAD;
|
||||
EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
}
|
||||
|
||||
TEST(MQTTWireScratch, RejectsPathLenThatWouldTruncateIntoOneWireByte) {
|
||||
mesh::Packet p;
|
||||
p.header = ROUTE_TYPE_FLOOD;
|
||||
p.payload_len = 4;
|
||||
p.path_len = 0x100; // writeTo() stores this in a single byte
|
||||
EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
}
|
||||
|
||||
// The case a destination-size check cannot catch, and which an earlier version of
|
||||
// these tests masked by using a payload big enough to push getRawLength() over the
|
||||
// limit: 0xFF encodes 63 hops of 4 bytes, so with no payload the counted length is
|
||||
// 254 -- inside the buffer -- while writePath() refuses the 252-byte path and writeTo()
|
||||
// emits only the 2-byte header. Publishing that would put 4 hex chars in the `raw`
|
||||
// field and call them the packet.
|
||||
TEST(MQTTWireScratch, RejectsOverlongPathEvenWhenTheCountedLengthFits) {
|
||||
mesh::Packet p;
|
||||
p.header = ROUTE_TYPE_FLOOD;
|
||||
p.path_len = 0xFF;
|
||||
p.payload_len = 0;
|
||||
|
||||
ASSERT_EQ(254, p.getRawLength());
|
||||
ASSERT_LE((size_t)p.getRawLength(), MQTTWireScratch::kWireBytes);
|
||||
uint8_t buf[MQTTWireScratch::kWireBytes];
|
||||
ASSERT_EQ(2, (int)p.writeTo(buf));
|
||||
|
||||
EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
}
|
||||
|
||||
// hash_size 4 is reserved: isValidPathLen() and therefore Packet::readFrom() reject
|
||||
// it, so serializing one produces a frame no receiver can parse back -- even though the
|
||||
// hop bytes fit and writePath() copies them happily.
|
||||
TEST(MQTTWireScratch, RejectsReservedFourByteHashEncoding) {
|
||||
mesh::Packet p;
|
||||
p.header = ROUTE_TYPE_FLOOD;
|
||||
p.setPathHashSizeAndCount(4, 2);
|
||||
p.payload_len = 4;
|
||||
|
||||
ASSERT_EQ(4, p.getPathHashSize());
|
||||
ASSERT_EQ(8, p.getPathByteLen()); // fits the path array
|
||||
ASSERT_LE(p.getRawLength(), (int)MQTTWireScratch::kWireBytes);
|
||||
ASSERT_FALSE(mesh::Packet::isValidPathLen((uint8_t)p.path_len));
|
||||
|
||||
EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
}
|
||||
|
||||
TEST(MQTTWireScratch, DestinationEdgeIsInclusive) {
|
||||
mesh::Packet p = maxPacket();
|
||||
const size_t exact = (size_t)p.getRawLength();
|
||||
EXPECT_TRUE(MQTTWireScratch::canSerialize(p, exact));
|
||||
EXPECT_FALSE(MQTTWireScratch::canSerialize(p, exact - 1));
|
||||
}
|
||||
|
||||
// A zero-payload packet 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();
|
||||
}
|
||||
Reference in New Issue
Block a user