fix(mqtt): write every owned field on every apply; recreate only where needed (F02)
The stale-configuration fix, and the reason the four preceding commits went first. `setupSlot()` applied configuration field by field, interleaved with the decisions that produced it, and cleared stale fields only when `slot.initial_connect_done` was set — which `teardownSlot()` clears at the top of every reconfigure. So live reconfiguration always skipped the cleanup. On a 5-slot board this was reproducible in one command: moving a slot from a JWT preset to an anonymous custom endpoint sent the previous configuration's `v1_<pubkey>` username in the CONNECT to the new broker (95-byte CONNECT captured at a throwaway broker on the LAN). The same config after a reboot sent 26 bytes with no username, which is what pinned it to the reconfigure path. That cleanup could not have worked anyway: it nulled the wrapper's pointers, and IDF's `esp_mqtt_set_if_config()` treats NULL as "leave unchanged", so an SDK-held credential survives it. An empty string does overwrite it, and leaves the CONNECT's username flag clear. So configuration is now decided in full first, then applied in one place: - every owned field is written on every apply — server, trust material, username, password — with absent credentials written as "" rather than omitted; - `mqttConfigRecreateDecision()` (previous commit) picks reuse or recreation. Credentials, auth-mode changes and endpoint moves within one scheme reuse the client, because a create/destroy cycle on the reconnect and renewal paths is the fork's documented internal-heap fragmentation driver. A transport (scheme) change, a trust-policy or CA change, and buffer-capacity growth recreate it, because those are the fields a write cannot safely replace; - `recreateSlotClient()` will not destroy a client whose stop was not proven — it quarantines it and fails the setup instead; - the applied configuration is recorded only after the client actually starts. Verified on hardware (Heltec V4, 5 live slots, both paths): - reuse path — a custom JWT slot (639-byte CONNECT carrying `v1_<pubkey>` plus the token) with its audience cleared now sends 26 bytes with no credentials, with no client recreation; - recreate path — `mqtt://` to `wss://` logged `recreating client (transport-change)` and reconnected. Also fixes a cap regression the hardware run caught: `canActivateSlot()` now counts the positions held by the *other* slots. A slot being reconfigured keeps its esp-mqtt task alive, so once the count included live clients it counted itself out of its own position and a full board refused to reconfigure a slot — disabling it instead. Asking "is there room for this slot" rather than "is there room for one more" is also what makes a resource-based count safe.
This commit is contained in:
1 parent
11f8cf1c58
commit
3afd7ebedb
2 files changed
+165
-35
No files matched your search
@@ -783,6 +783,10 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg
|
||||
_slots[i].last_reconnect_attempt = 0;
|
||||
_slots[i].last_log_time = 0;
|
||||
_slots[i].port = 1883;
|
||||
_slots[i].client_state = ClientState::Absent;
|
||||
_slots[i].generation = 0;
|
||||
_slots[i].applied_config = MqttEffectiveConfig();
|
||||
_slots[i].allocated_buffer_size = 0;
|
||||
_slot_reconfigure_pending[i] = false;
|
||||
_slot_force_jwt_mint[i] = false;
|
||||
_status_publish_pending[i] = false;
|
||||
@@ -1742,6 +1746,11 @@ bool MQTTBridge::ensureSlotClient(int index) {
|
||||
return false;
|
||||
}
|
||||
slot.client_state = ClientState::Configured;
|
||||
// The SDK allocates its buffers in esp_mqtt_client_init() (inside the first
|
||||
// connect()) from the size set before it, and never resizes them. Record what
|
||||
// this client will therefore own, so a later config that needs more capacity
|
||||
// is recognised as needing a new client rather than silently truncating.
|
||||
slot.allocated_buffer_size = kMqttClientBufferSize;
|
||||
slot.client->setAutoReconnect(false); // we handle reconnect with our own backoff
|
||||
|
||||
slot.client->onConnect([this, index](bool sessionPresent) {
|
||||
@@ -1918,9 +1927,12 @@ void MQTTBridge::destroySlotClients() {
|
||||
}
|
||||
}
|
||||
|
||||
int MQTTBridge::activatedSlotCount() const {
|
||||
int MQTTBridge::activatedSlotCount() const { return activatedSlotCountExcluding(-1); }
|
||||
|
||||
int MQTTBridge::activatedSlotCountExcluding(int skip) const {
|
||||
int n = 0;
|
||||
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
|
||||
if (i == skip) continue;
|
||||
const MQTTSlot& s = _slots[i];
|
||||
// Two ways to hold a position, and the cap must respect both. The original
|
||||
// one is the configuration view: enabled and through a successful setup.
|
||||
@@ -1939,9 +1951,14 @@ int MQTTBridge::activatedSlotCount() const {
|
||||
|
||||
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;
|
||||
// Count what the OTHER slots hold. A slot being set up or reconfigured must
|
||||
// not count itself out of its own position — and it can now hold resources
|
||||
// while being reconfigured, because a reconfigure keeps the esp-mqtt task
|
||||
// alive (TeardownReason::Reconfigure). Asking "is there room for this slot"
|
||||
// rather than "is there room for one more" is also what makes the count safe
|
||||
// to base on live resources instead of on initial_connect_done, which
|
||||
// teardown clears.
|
||||
return activatedSlotCountExcluding(index) < _max_active_slots;
|
||||
}
|
||||
|
||||
// Returns true only when the slot reached connect(). A false result leaves the slot
|
||||
@@ -1991,22 +2008,13 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
slot.client_state = ClientState::Disconnected;
|
||||
slot.generation++;
|
||||
}
|
||||
// Clear TLS verification fields so a stale CA-bundle attach or cert
|
||||
// pointer from a prior preset doesn't override the new one.
|
||||
esp_mqtt_client_config_t* cfg = slot.client->getMqttConfig();
|
||||
#if ESP_IDF_VERSION_MAJOR == 5
|
||||
cfg->broker.verification.certificate = nullptr;
|
||||
cfg->broker.verification.certificate_len = 0;
|
||||
cfg->broker.verification.crt_bundle_attach = nullptr;
|
||||
cfg->credentials.username = nullptr;
|
||||
cfg->credentials.authentication.password = nullptr;
|
||||
#else
|
||||
cfg->cert_pem = nullptr;
|
||||
cfg->cert_len = 0;
|
||||
cfg->crt_bundle_attach = nullptr;
|
||||
cfg->username = nullptr;
|
||||
cfg->password = nullptr;
|
||||
#endif
|
||||
// The config fields are NOT cleared here any more. Two reasons: this block
|
||||
// only ran when initial_connect_done was set, which teardownSlot() clears
|
||||
// first, so live reconfiguration always skipped it (F02); and nulling a
|
||||
// wrapper pointer cannot clear an SDK-held string anyway, because IDF's
|
||||
// esp_mqtt_set_if_config() treats NULL as "leave unchanged". Every owned
|
||||
// field is written unconditionally below instead, with absent credentials
|
||||
// written as "" — which does clear them.
|
||||
if (slot.auth_token) slot.auth_token[0] = '\0';
|
||||
slot.connected = false;
|
||||
slot.token_expires_at = 0;
|
||||
@@ -2032,11 +2040,25 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
#endif
|
||||
#endif
|
||||
|
||||
// What this slot should be configured with. Decided first, in full, and
|
||||
// applied in one place below: field-by-field application interleaved with
|
||||
// decisions is what let a stale credential survive a reconfigure (F02).
|
||||
// Every pointer here must outlive the client — esp-mqtt stores the pointer
|
||||
// and re-reads it whenever a later connect() re-applies the config — so these
|
||||
// are preset literals in flash, or slot/bridge members, never locals.
|
||||
const char* cfg_uri = nullptr;
|
||||
MqttAuth cfg_auth = MqttAuth::None;
|
||||
const char* cfg_user = nullptr;
|
||||
const char* cfg_pass = nullptr;
|
||||
MqttTrust cfg_trust = MqttTrust::Plaintext;
|
||||
const char* cfg_pem = nullptr;
|
||||
|
||||
if (slot.preset) {
|
||||
// Preset-based slot
|
||||
slot.client->setServer(slot.preset->server_url);
|
||||
cfg_uri = slot.preset->server_url;
|
||||
if (slot.preset->ca_cert) {
|
||||
slot.client->setCACert(slot.preset->ca_cert);
|
||||
cfg_trust = MqttTrust::PemCert;
|
||||
cfg_pem = slot.preset->ca_cert;
|
||||
}
|
||||
|
||||
// A JWT slot with no usable token would connect unauthenticated and be rejected.
|
||||
@@ -2048,7 +2070,9 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
slot.last_reconnect_attempt = millis();
|
||||
return false;
|
||||
}
|
||||
slot.client->setCredentials(_jwt_username, slot.auth_token);
|
||||
cfg_auth = MqttAuth::Jwt;
|
||||
cfg_user = _jwt_username;
|
||||
cfg_pass = slot.auth_token;
|
||||
} else if (slot.preset->auth_type == MQTT_AUTH_USERPASS) {
|
||||
const char* user = nullptr;
|
||||
const char* pass = slot.preset->userpass_password
|
||||
@@ -2062,7 +2086,9 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
user = slot.username;
|
||||
}
|
||||
if (user && user[0] != '\0' && pass && pass[0] != '\0') {
|
||||
slot.client->setCredentials(user, pass);
|
||||
cfg_auth = MqttAuth::UserPass;
|
||||
cfg_user = user;
|
||||
cfg_pass = pass;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
@@ -2115,7 +2141,7 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
}
|
||||
snprintf(slot.broker_uri, sizeof(slot.broker_uri), "%s://%s:%d", proto, slot.host, slot.port);
|
||||
}
|
||||
slot.client->setServer(slot.broker_uri);
|
||||
cfg_uri = slot.broker_uri;
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d custom broker URI: %s (host='%s', port=%u)",
|
||||
index + 1, slot.broker_uri, slot.host, (unsigned)slot.port);
|
||||
|
||||
@@ -2124,8 +2150,7 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
// a use-after-free race: connect() launches an async FreeRTOS task, and
|
||||
// calling setCACertBundle() again from a later slot would free the global
|
||||
// crts array while a prior slot's TLS handshake may still be reading it.
|
||||
bool needs_tls = (strncmp(slot.broker_uri, "mqtts://", 8) == 0 ||
|
||||
strncmp(slot.broker_uri, "wss://", 6) == 0);
|
||||
const bool needs_tls = mqttTransportIsEncrypted(mqttTransportFromUri(slot.broker_uri));
|
||||
if (needs_tls) {
|
||||
if (!s_ca_bundle_loaded) {
|
||||
size_t bundle_len = 0;
|
||||
@@ -2138,8 +2163,10 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
if (bundle_len > 0) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT global CA bundle init: embedded bundle (%u bytes)",
|
||||
(unsigned)bundle_len);
|
||||
// Load the bundle into the global s_crt_bundle via the first client.
|
||||
// This is a one-time operation; subsequent clients reuse via attachArduinoCACertBundle.
|
||||
// Load the bundle into the global s_crt_bundle via this client. The
|
||||
// load is global and one-shot (calling it again would free the crts
|
||||
// array while another slot's handshake may still be reading it); the
|
||||
// per-client attach pointer is set uniformly in the apply step below.
|
||||
slot.client->setCACertBundle(rootca_crt_bundle_start, bundle_len);
|
||||
s_ca_bundle_loaded = true;
|
||||
} else {
|
||||
@@ -2151,6 +2178,7 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
}
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d TLS verify: CA bundle %s", index + 1,
|
||||
s_ca_bundle_loaded ? "active" : "unavailable");
|
||||
if (s_ca_bundle_loaded) cfg_trust = MqttTrust::Bundle;
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d custom broker uses non-TLS transport", index + 1);
|
||||
}
|
||||
@@ -2163,11 +2191,19 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
slot.last_reconnect_attempt = millis();
|
||||
return false;
|
||||
}
|
||||
slot.client->setCredentials(_jwt_username, slot.auth_token);
|
||||
cfg_auth = MqttAuth::Jwt;
|
||||
cfg_user = _jwt_username;
|
||||
cfg_pass = 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);
|
||||
} else if (slot.username[0] != '\0') {
|
||||
cfg_auth = MqttAuth::UserPass;
|
||||
cfg_user = slot.username;
|
||||
cfg_pass = slot.password;
|
||||
}
|
||||
// No else: an anonymous endpoint gets empty credentials written to it, so
|
||||
// whatever the previous configuration left in the SDK is overwritten. This
|
||||
// exact transition (JWT preset -> anonymous custom) was observed sending
|
||||
// the old v1_<pubkey> username to the new broker.
|
||||
}
|
||||
|
||||
// Activation is now conditional on the client actually starting. A failed
|
||||
@@ -2176,6 +2212,49 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
// positions, the reconnect ladder (which is gated on activation) governed it,
|
||||
// and nothing retried the setup. Leaving it unactivated hands it to the
|
||||
// existing deferred-setup retry in maintainSlotConnections() instead.
|
||||
// --- one apply, every field ------------------------------------------------
|
||||
MqttEffectiveConfig desired;
|
||||
// Keepalive is recorded as 0: optimizeMqttClientConfig() owns it, it is
|
||||
// rewritten on every apply, and it plays no part in the recreate decision.
|
||||
if (!mqttBuildEffectiveConfig(cfg_uri, cfg_auth, cfg_user, cfg_pass, cfg_trust, cfg_pem,
|
||||
kMqttClientBufferSize, 0, &desired)) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d: unusable broker URI '%s' - not connecting", index + 1,
|
||||
cfg_uri ? cfg_uri : "(none)");
|
||||
slot.last_reconnect_attempt = millis();
|
||||
return false;
|
||||
}
|
||||
|
||||
// Recreate only where a field cannot be overwritten: a transport (scheme)
|
||||
// change, a trust-policy or CA change, or growth past the allocated buffer
|
||||
// capacity. Everything else — credentials, auth mode, an endpoint move within
|
||||
// one scheme — is reconfigured in place, because a client create/destroy cycle
|
||||
// on the reconnect/renewal path is the documented fragmentation driver.
|
||||
const MqttRecreateDecision decision =
|
||||
mqttConfigRecreateDecision(slot.applied_config, desired, slot.allocated_buffer_size);
|
||||
if (decision.recreate) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d recreating client (%s)", index + 1, decision.reason);
|
||||
if (!recreateSlotClient(index)) {
|
||||
slot.last_reconnect_attempt = millis();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// Write every owned field, in one place, whether or not it changed. Absent
|
||||
// credentials are written as "" rather than left alone: NULL means "leave
|
||||
// unchanged" to esp_mqtt_set_config(), while an empty string clears the
|
||||
// stored value and leaves the CONNECT's username flag clear.
|
||||
slot.client->setServer(cfg_uri);
|
||||
if (desired.trust == MqttTrust::PemCert) {
|
||||
slot.client->setCACert(desired.pem);
|
||||
slot.client->attachArduinoCACertBundle(false);
|
||||
} else if (desired.trust == MqttTrust::Bundle) {
|
||||
slot.client->attachArduinoCACertBundle(true);
|
||||
} else {
|
||||
slot.client->attachArduinoCACertBundle(false);
|
||||
}
|
||||
slot.client->setCredentials(mqttFieldOrEmpty(desired.username),
|
||||
mqttFieldOrEmpty(desired.password));
|
||||
|
||||
// Start or reconnect according to what the SDK client actually is, not what
|
||||
// the network is doing. esp_mqtt_client_start() fails on an already-started
|
||||
// client, so a reconfigure that kept the task (TeardownReason::Reconfigure)
|
||||
@@ -2188,12 +2267,43 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
slot.last_reconnect_attempt = millis();
|
||||
return false;
|
||||
}
|
||||
slot.client_state = ClientState::Starting;
|
||||
slot.generation++;
|
||||
// reconnectSlotClient() moved the state to Starting and bumped the generation.
|
||||
// Record what the client is now configured with, so the next apply can tell
|
||||
// an in-place change from a structural one.
|
||||
slot.applied_config = desired;
|
||||
slot.initial_connect_done = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
// Destroy and re-create this slot's client, for the configuration changes that
|
||||
// cannot be applied to a live SDK client (see mqttConfigRecreateDecision).
|
||||
//
|
||||
// The stop must be PROVEN before the object is freed: a client whose
|
||||
// esp_mqtt_client_stop() did not return OK still has a task, and deleting it
|
||||
// then is the use-after-free F01 is about. Such a client is quarantined and this
|
||||
// returns false — the slot stays unactivated and its resources are retained
|
||||
// until the node reboots, rather than being freed under a live task.
|
||||
bool MQTTBridge::recreateSlotClient(int index) {
|
||||
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false;
|
||||
MQTTSlot& slot = _slots[index];
|
||||
if (slot.client == nullptr) return ensureSlotClient(index);
|
||||
|
||||
teardownSlot(index, TeardownReason::Disable); // stop, do not just close the transport
|
||||
if (slot.client_state == ClientState::Quarantined) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT%d cannot recreate: previous stop unproven", index + 1);
|
||||
return false;
|
||||
}
|
||||
|
||||
delete slot.client;
|
||||
slot.client = nullptr;
|
||||
slot.client_state = ClientState::Absent;
|
||||
slot.applied_config = MqttEffectiveConfig(); // nothing is applied to a client that does not exist
|
||||
slot.allocated_buffer_size = 0;
|
||||
// The token buffer survives: the new client is configured from it below, and
|
||||
// its lifetime is the slot's, not the client's (see MQTTSlot::auth_token).
|
||||
return ensureSlotClient(index);
|
||||
}
|
||||
|
||||
// Disconnect the slot's MQTT client and clear per-connection state, but leave
|
||||
// the client object alive so a subsequent setupSlot() can reuse its mbedTLS
|
||||
// context. This is called both on reconfigure (preset change) and at shutdown;
|
||||
@@ -4877,7 +4987,7 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_
|
||||
// The cost is 384 bytes per client on a non-PSRAM board (at most 2 active
|
||||
// slots there), against a handshake that needs 16 KB of *contiguous* internal
|
||||
// DRAM — noise, and it buys the removal of a recreate case.
|
||||
static const int MQTT_CLIENT_BUFFER_SIZE = 896;
|
||||
static const int MQTT_CLIENT_BUFFER_SIZE = kMqttClientBufferSize;
|
||||
(void)needs_large_buffer; // kept: callers still express the intent
|
||||
|
||||
client->setBufferSize(MQTT_CLIENT_BUFFER_SIZE);
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
#include "helpers/MQTTPresets.h"
|
||||
#include "helpers/MQTTLifecycle.h"
|
||||
#include "helpers/AlertFaultPolicy.h"
|
||||
#include "helpers/MQTTEffectiveConfig.h"
|
||||
#include <atomic>
|
||||
|
||||
#ifdef WITH_SNMP
|
||||
@@ -91,6 +92,12 @@ public:
|
||||
private:
|
||||
static const size_t AUTH_TOKEN_SIZE = 768;
|
||||
|
||||
// Every client gets buffers sized for a JWT CONNECT (frame + 768-byte token),
|
||||
// on every board: the SDK fixes the allocation at client init and does not
|
||||
// resize it, so a per-slot size made the desired and allocated capacities
|
||||
// disagree (F03).
|
||||
static const uint16_t kMqttClientBufferSize = 896;
|
||||
|
||||
// NTP acceptance bounds. A reply outside them is rejected outright rather
|
||||
// than allowed to set the clock, the RTC and every JWT minted afterwards.
|
||||
static const uint16_t kNtpPort = 123;
|
||||
@@ -126,6 +133,12 @@ private:
|
||||
struct MQTTSlot {
|
||||
PsychicMqttClient* client;
|
||||
ClientState client_state;
|
||||
// What this client was last successfully configured with, and the buffer
|
||||
// capacity it actually owns (the SDK fixes that at client init). Together
|
||||
// they answer "can the next configuration be applied in place?" — the
|
||||
// question `initial_connect_done` was standing in for, wrongly.
|
||||
MqttEffectiveConfig applied_config;
|
||||
uint16_t allocated_buffer_size;
|
||||
// Bumped on every start/stop. Only used for diagnostics and log lines: the
|
||||
// accept/reject decision for a late callback is made on client_state, which
|
||||
// the bridge task owns.
|
||||
@@ -501,6 +514,10 @@ private:
|
||||
// This avoids delete/new cycles that shed ~40 KB of mbedTLS buffers per
|
||||
// reconfigure and fragment the internal heap on non-PSRAM boards.
|
||||
bool ensureSlotClient(int index); // Allocate this slot's persistent client + callbacks on first use
|
||||
// Stop, destroy and re-allocate this slot's client. Only for configuration
|
||||
// changes that cannot be applied to a live client, and only after the stop is
|
||||
// proven — false means the client is quarantined and must not be reused.
|
||||
bool recreateSlotClient(int index);
|
||||
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)
|
||||
// No force variant: the only caller that ever passed one was the dirty-stop
|
||||
@@ -514,6 +531,9 @@ private:
|
||||
// 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;
|
||||
// Positions held by every slot except `skip` (-1 for none). canActivateSlot()
|
||||
// excludes the candidate so a live reconfigure cannot fail its own cap check.
|
||||
int activatedSlotCountExcluding(int skip) const;
|
||||
bool canActivateSlot(int index) const;
|
||||
// force as in destroySlotClients(): skip the unbounded wait, dirty-stop path only.
|
||||
// Why a slot is being torn down. The distinction is not cosmetic:
|
||||
|
||||
Reference in new issue
Block a user