fix(mqtt): address review of the demand-driven slot work
[P1] A failed setup no longer strands the slot. setupSlot() returns bool and the startup loops count only successful activations, so a slot that fails on a client allocation neither consumes an active-slot position (starving a later healthy broker on capped hardware) nor sits dead forever: maintainSlotConnections() previously skipped clientless slots and the reconnect ladder is gated on initial_connect_done, so nothing retried it. It now retries an enabled but unactivated slot on a 60 s timer, one per cycle, gated on the same _slots_setup_done ordering so the NTP-deferred setup sequence is preserved. [P2] JWT setup no longer proceeds without a usable token. Both the preset and custom-audience paths returned after ignoring createSlotAuthToken()'s result, then called connect() and latched initial_connect_done -- so the token-allocation failure introduced by the previous commit produced an unauthenticated attempt exactly when memory was exhausted. They now return false and let the retry path handle it. [P2] ensureSlotAuthToken() no longer clears an existing token. It cleared unconditionally, so every renewal wiped the current token before JWTHelper ran; a renewal that then failed left an empty password where the inline buffer used to preserve working credentials (JWTHelper writes only on success). Only freshly allocated buffers are initialised now. [P2] Raw publications reuse the shared document. buildRawJSON() reached MQTTPayloadBuilder::buildRawMessage(), which constructed its own default JsonDocument and therefore malloc'd and freed an internal-heap variant pool per message -- on the highest-rate topic. The document is threaded through both builders and the bridge passes _json_scratch_doc. [P3] The writeTo() guard validates the source fields, not just the destination. A corrupt payload_len of MAX_PACKET_PAYLOAD + 1 still leaves getRawLength() inside MAX_TRANS_UNIT, so writeTo() read past packet->payload. Sizing and validation moved to a pure MQTTWireScratch header with host tests covering the accept/reject edges, matching the MQTTPacketFilter/MQTTConnectionPolicy pattern. Two findings fell out: MAX_PATH_SIZE one-byte hops is not encodable (the hop count is 6 bits, so 64 & 63 == 0; 32 two-byte hops is the widest real path), and a zero-payload packet serializes but does not survive readFrom() -- pinned as a test because it constrains any future wire-only queue. [P3] Corrected the pool-size comment: these targets are 32-bit, so ARDUINOJSON_SLOT_ID_SIZE is 2 and a pool block is 128 slots / 1024 bytes, not 4096. The 4096 figure came from a pre-existing comment near NEIGHBORS_DOC_POOL_BUDGET, which is left alone -- its byte measurements are empirical and still stand, only the block-size attribution is wrong. Activation is now centralized in activatedSlotCount()/canActivateSlot(), used by both startup loops, the retry path, and applySlotPreset(). That closes the pre-existing divergence where a live preset change called setupSlot() without consulting _max_active_slots, letting a non-PSRAM board reach three concurrent TLS sessions against a cap of two. BEHAVIOUR CHANGE: a reconfigure that would exceed the cap now logs and leaves the slot inactive instead of connecting. Reconfiguring an already-active slot still works, because teardownSlot() releases its position first. 272/272 native tests pass (5 new); both observer envs and an nRF52 repeater build clean. Flash 1593249 B non-PSRAM, 1555625 B PSRAM.
This commit is contained in:
1 parent
b93b6b1560
commit
e242cc6fd3
9 files changed
+272
-42
No files matched your search
@@ -96,6 +96,7 @@ int MQTTMessageBuilder::buildPacketMessage(
|
||||
}
|
||||
|
||||
int MQTTMessageBuilder::buildRawMessage(
|
||||
JsonDocument& doc,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
const char* timestamp,
|
||||
@@ -104,7 +105,7 @@ int MQTTMessageBuilder::buildRawMessage(
|
||||
size_t buffer_size
|
||||
) {
|
||||
return MQTTPayloadBuilder::buildRawMessage(
|
||||
origin, origin_id, timestamp, raw, buffer, buffer_size);
|
||||
doc, origin, origin_id, timestamp, raw, buffer, buffer_size);
|
||||
}
|
||||
|
||||
int MQTTMessageBuilder::buildNeighborsMessage(
|
||||
@@ -297,6 +298,7 @@ int MQTTMessageBuilder::buildPacketJSONFromRaw(
|
||||
}
|
||||
|
||||
int MQTTMessageBuilder::buildRawJSON(
|
||||
JsonDocument& doc,
|
||||
mesh::Packet* packet,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
@@ -316,7 +318,7 @@ int MQTTMessageBuilder::buildRawJSON(
|
||||
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) {
|
||||
@@ -355,9 +357,7 @@ void MQTTMessageBuilder::packetToHex(mesh::Packet* packet, char* hex, size_t hex
|
||||
// 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[WIRE_SCRATCH_SIZE];
|
||||
// writeTo() neither bounds-checks nor can report an overrun (it returns uint8_t),
|
||||
// so reject an over-long packet on getRawLength() before writing anything.
|
||||
if (packet->getRawLength() > (int)sizeof(raw_buf)) return;
|
||||
if (!canSerializePacket(packet, sizeof(raw_buf))) return;
|
||||
uint8_t raw_len = packet->writeTo(raw_buf);
|
||||
if (raw_len == 0) return;
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
#include "MeshCore.h"
|
||||
#include <ArduinoJson.h>
|
||||
#include "MQTTPayloadBuilder.h"
|
||||
#include "MQTTWireScratch.h"
|
||||
#include <Mesh.h>
|
||||
#include <Timezone.h>
|
||||
|
||||
@@ -21,13 +22,14 @@
|
||||
*/
|
||||
class MQTTMessageBuilder {
|
||||
public:
|
||||
// Wire-format scratch sizing, from the protocol maximum: Packet::writeTo() returns
|
||||
// uint8_t, so MAX_TRANS_UNIT is the hard ceiling, and hex is 2 chars/byte + NUL.
|
||||
static const size_t WIRE_SCRATCH_SIZE = MAX_TRANS_UNIT;
|
||||
static const size_t WIRE_HEX_SCRATCH_SIZE = 2 * MAX_TRANS_UNIT + 1;
|
||||
// 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_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");
|
||||
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).
|
||||
@@ -154,6 +156,7 @@ public:
|
||||
* @return Length of JSON string, or 0 on error
|
||||
*/
|
||||
static int buildRawMessage(
|
||||
JsonDocument& doc,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
const char* timestamp,
|
||||
@@ -240,6 +243,7 @@ public:
|
||||
* @return Length of JSON string, or 0 on error
|
||||
*/
|
||||
static int buildRawJSON(
|
||||
JsonDocument& doc,
|
||||
mesh::Packet* packet,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
|
||||
@@ -166,6 +166,7 @@ int MQTTPayloadBuilder::buildPacketMessage(
|
||||
}
|
||||
|
||||
int MQTTPayloadBuilder::buildRawMessage(
|
||||
JsonDocument& doc,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
const char* timestamp,
|
||||
@@ -173,7 +174,7 @@ int MQTTPayloadBuilder::buildRawMessage(
|
||||
char* buffer,
|
||||
size_t buffer_size
|
||||
) {
|
||||
JsonDocument doc;
|
||||
doc.clear();
|
||||
JsonObject root = doc.to<JsonObject>();
|
||||
|
||||
root["origin"] = origin;
|
||||
|
||||
@@ -61,6 +61,7 @@ public:
|
||||
);
|
||||
|
||||
static int buildRawMessage(
|
||||
JsonDocument& doc,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
const char* timestamp,
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
#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 bytes themselves need no check here: writePath() already refuses a
|
||||
// getPathByteLen() above MAX_PATH_SIZE.
|
||||
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;
|
||||
const int raw_len = packet.getRawLength();
|
||||
return raw_len > 0 && (size_t)raw_len <= dest_size;
|
||||
}
|
||||
|
||||
} // namespace MQTTWireScratch
|
||||
@@ -1399,11 +1399,10 @@ void MQTTBridge::mqttTaskLoop() {
|
||||
#endif
|
||||
|
||||
MQTT_DEBUG_PRINTLN("NTP synced, setting up MQTT slots (max %d active)...", _max_active_slots);
|
||||
int active_count = 0;
|
||||
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
|
||||
if (_slots[i].enabled) {
|
||||
if (active_count >= _max_active_slots) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached (no PSRAM)", i + 1, _max_active_slots);
|
||||
if (!canActivateSlot(i)) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", i + 1, _max_active_slots);
|
||||
_slots[i].enabled = false; // Disable so other loops skip it
|
||||
continue;
|
||||
}
|
||||
@@ -1412,8 +1411,10 @@ void MQTTBridge::mqttTaskLoop() {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d not ready - run '%s' to connect", i + 1, reason);
|
||||
continue;
|
||||
}
|
||||
setupSlot(i);
|
||||
active_count++;
|
||||
// A slot that fails to activate consumes no position and stays enabled, so
|
||||
// maintainSlotConnections() retries it and a later healthy broker is not
|
||||
// starved by it on a capped board.
|
||||
if (!setupSlot(i)) continue;
|
||||
// Stagger connections: 5s between slots to avoid simultaneous TLS handshakes
|
||||
// which compete for ~40KB internal heap each
|
||||
if (i < RUNTIME_MQTT_SLOTS - 1) {
|
||||
@@ -1682,13 +1683,18 @@ bool MQTTBridge::ensureSlotClient(int index) {
|
||||
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;
|
||||
}
|
||||
slot.auth_token[0] = '\0';
|
||||
// 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;
|
||||
}
|
||||
|
||||
@@ -1725,19 +1731,38 @@ void MQTTBridge::destroySlotClients() {
|
||||
}
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
// First setup for this slot allocates its persistent client; later ones reuse it.
|
||||
if (!ensureSlotClient(index)) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d: client allocation failed - skipping", index + 1);
|
||||
return;
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d: client allocation failed - will retry", index + 1);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Reconfigure path: if we're re-applying (e.g. after a preset change), stop
|
||||
@@ -1794,12 +1819,15 @@ 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 && 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);
|
||||
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
|
||||
@@ -1908,11 +1936,12 @@ 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 (allocates the buffer)
|
||||
createSlotAuthToken(index);
|
||||
if (slot.auth_token && 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);
|
||||
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);
|
||||
@@ -1921,6 +1950,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
|
||||
@@ -1988,8 +2018,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) ||
|
||||
@@ -1998,6 +2032,27 @@ 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)) _last_slot_reconnect_ms = now_millis;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (!_slots[i].client) continue;
|
||||
|
||||
maintainSlotConnection(i, now_millis, current_time, time_synced, reconnect_attempted_this_cycle, teardown_attempted_this_cycle);
|
||||
}
|
||||
}
|
||||
@@ -2515,6 +2570,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;
|
||||
@@ -2536,6 +2598,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);
|
||||
}
|
||||
}
|
||||
@@ -2722,10 +2791,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;
|
||||
}
|
||||
@@ -2733,7 +2801,6 @@ void MQTTBridge::loop() {
|
||||
continue;
|
||||
}
|
||||
setupSlot(i);
|
||||
active_count++;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3423,10 +3490,11 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx,
|
||||
} else {
|
||||
// Reconstruct wire-format bytes from packet (same as MQTTMessageBuilder::packetToHex).
|
||||
// Reached when the queued item carried no captured raw frame, so the "raw" hex field
|
||||
// is re-serialized rather than dropped. Length-guarded for the same reason.
|
||||
// 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 (packet->getRawLength() <= (int)sizeof(reconstructed)) {
|
||||
if (MQTTMessageBuilder::canSerializePacket(packet, sizeof(reconstructed))) {
|
||||
rlen = packet->writeTo(reconstructed);
|
||||
}
|
||||
if (rlen > 0) {
|
||||
@@ -3503,6 +3571,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
|
||||
);
|
||||
|
||||
|
||||
@@ -329,8 +329,11 @@ private:
|
||||
|
||||
// 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 (4096 bytes here) in
|
||||
// internal DRAM next to the mbedTLS working set.
|
||||
// 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;
|
||||
@@ -378,6 +381,10 @@ private:
|
||||
unsigned long _last_no_broker_log;
|
||||
static const unsigned long NO_BROKER_LOG_INTERVAL = 30000; // Log every 30 seconds max
|
||||
static const unsigned long SLOT_LOG_INTERVAL = 30000; // Log every 30 seconds max
|
||||
// Retry cadence for a slot whose setup failed on an allocation. Deliberately slower
|
||||
// than the first backoff rung: the failure means internal heap is exhausted, and a
|
||||
// retry that succeeds immediately launches a TLS handshake.
|
||||
static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000;
|
||||
unsigned long _last_config_warning; // Throttle configuration mismatch warnings
|
||||
static const unsigned long CONFIG_WARNING_INTERVAL = 300000; // Log every 5 minutes max
|
||||
|
||||
@@ -419,7 +426,13 @@ private:
|
||||
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);
|
||||
|
||||
@@ -171,8 +171,9 @@ TEST(MQTTPayloadBuilder, RxPacketOmitsUnknownNanScore) {
|
||||
|
||||
TEST(MQTTPayloadBuilder, RawMessageHasExactContractAndEscapesData) {
|
||||
char buffer[512];
|
||||
JsonDocument doc;
|
||||
int len = MQTTPayloadBuilder::buildRawMessage(
|
||||
"node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer));
|
||||
doc, "node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer));
|
||||
|
||||
ASSERT_GT(len, 0);
|
||||
EXPECT_EQ(static_cast<size_t>(len), strlen(buffer));
|
||||
@@ -224,8 +225,9 @@ TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) {
|
||||
EXPECT_STREQ("3c3d3e3f", parsed_path[15].as<const char*>());
|
||||
|
||||
char raw_buffer[1024];
|
||||
JsonDocument raw_doc;
|
||||
int raw_len = MQTTPayloadBuilder::buildRawMessage(
|
||||
"node", "0123456789ABCDEF", kTimestamp, raw.c_str(),
|
||||
raw_doc, "node", "0123456789ABCDEF", kTimestamp, raw.c_str(),
|
||||
raw_buffer, sizeof(raw_buffer));
|
||||
ASSERT_GT(raw_len, 0);
|
||||
JsonDocument parsed_raw;
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
// 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));
|
||||
|
||||
p.path_len = 0xFF;
|
||||
// Still rejected, but now on the destination check rather than truncation:
|
||||
// 0xFF encodes 63 hops of 4 bytes.
|
||||
EXPECT_GT(p.getRawLength(), (int)MQTTWireScratch::kWireBytes);
|
||||
EXPECT_FALSE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
}
|
||||
|
||||
TEST(MQTTWireScratch, DestinationEdgeIsInclusive) {
|
||||
mesh::Packet p = maxPacket();
|
||||
const size_t exact = (size_t)p.getRawLength();
|
||||
EXPECT_TRUE(MQTTWireScratch::canSerialize(p, exact));
|
||||
EXPECT_FALSE(MQTTWireScratch::canSerialize(p, exact - 1));
|
||||
}
|
||||
|
||||
// A zero-payload packet serializes fine but does not survive readFrom(), which
|
||||
// requires at least one payload byte. canSerialize() deliberately does not reject it:
|
||||
// the raw/packet publish paths only ever write the bytes out. Any future change that
|
||||
// reconstructs a Packet from queued wire bytes has to handle this asymmetry.
|
||||
TEST(MQTTWireScratch, ZeroPayloadPacketSerializesButDoesNotRoundTrip) {
|
||||
mesh::Packet p;
|
||||
p.header = ROUTE_TYPE_FLOOD;
|
||||
p.setPathHashSizeAndCount(1, 0);
|
||||
p.payload_len = 0;
|
||||
EXPECT_EQ(2, p.getRawLength());
|
||||
EXPECT_TRUE(MQTTWireScratch::canSerialize(p, MQTTWireScratch::kWireBytes));
|
||||
|
||||
uint8_t buf[MQTTWireScratch::kWireBytes];
|
||||
const uint8_t written = p.writeTo(buf);
|
||||
EXPECT_EQ(2, (int)written);
|
||||
|
||||
mesh::Packet restored;
|
||||
EXPECT_FALSE(restored.readFrom(buf, written));
|
||||
}
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
::testing::InitGoogleTest(&argc, argv);
|
||||
return RUN_ALL_TESTS();
|
||||
}
|
||||
Reference in new issue
Block a user