dandri/HaloKeymind
1
0
Fork 0
mirror of https://github.com/mikecarper/MeshCore.git synced 2026-09-17 00:34:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor PsychicMqttClient to improve memory management and callback handling. Replace dynamic memory allocations with fixed-size arrays for callbacks, enhancing performance and reducing fragmentation. Introduce inline storage for topics and optimize buffer allocation during connection setup. Update version to 0.2.2 to reflect changes.

Browse Source
This commit is contained in:
agessaman
2026-04-21 20:43:09 -07:00
parent 1ff9437f73
commit e9ff1ae055
5 changed files with 571 additions and 715 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/PsychicMqttClient/src/PsychicMqttClient.cpp
+213 -150
View File
@@ -1,5 +1,7 @@
#include "PsychicMqttClient.h"
#include <string.h>
static const char *TAG = "🐙";
static void log_error_if_nonzero(const char *message, int error_code)
@@ -13,36 +15,26 @@ static void log_error_if_nonzero(const char *message, int error_code)
PsychicMqttClient::PsychicMqttClient() : _mqtt_cfg()
{
memset(&_mqtt_cfg, 0, sizeof(_mqtt_cfg));
_topic[0] = '\0';
}
PsychicMqttClient::~PsychicMqttClient()
{
disconnect();
esp_mqtt_client_destroy(_client);
if (_client != nullptr)
{
esp_mqtt_client_destroy(_client);
_client = nullptr;
}
// Free memory in _buffer and _topic
if (_buffer != nullptr)
{
free(_buffer);
_buffer = nullptr; // Set to nullptr to avoid dangling pointers
_buffer = nullptr;
_buffer_capacity = 0;
}
if (_topic != nullptr)
{
free(_topic);
_topic = nullptr; // Set to nullptr to avoid dangling pointers
}
// Free memory in _onMessageUserCallbacks
for (auto &callback : _onMessageUserCallbacks)
{
if (callback.topic != nullptr)
{
free(callback.topic); // Free the dynamically allocated topic
callback.topic = nullptr;
}
}
_onMessageUserCallbacks.clear(); // Clear the vector
// Topic storage is inline; nothing to free.
// Subscription entries have inline topic storage; nothing to free.
}
PsychicMqttClient &PsychicMqttClient::setKeepAlive(int keepAlive)
@@ -224,53 +216,118 @@ PsychicMqttClient &PsychicMqttClient::setServer(const char *uri)
PsychicMqttClient &PsychicMqttClient::onConnect(OnConnectUserCallback callback)
{
_onConnectUserCallbacks.push_back(callback);
if (_onConnectUserCallbackCount < PSYCHIC_MAX_CONNECT_CB)
{
_onConnectUserCallbacks[_onConnectUserCallbackCount++] = std::move(callback);
}
else
{
ESP_LOGE(TAG, "onConnect callback list full (max=%d)", PSYCHIC_MAX_CONNECT_CB);
}
return *this;
}
PsychicMqttClient &PsychicMqttClient::onDisconnect(OnDisconnectUserCallback callback)
{
_onDisconnectUserCallbacks.push_back(callback);
if (_onDisconnectUserCallbackCount < PSYCHIC_MAX_DISCONNECT_CB)
{
_onDisconnectUserCallbacks[_onDisconnectUserCallbackCount++] = std::move(callback);
}
else
{
ESP_LOGE(TAG, "onDisconnect callback list full (max=%d)", PSYCHIC_MAX_DISCONNECT_CB);
}
return *this;
}
PsychicMqttClient &PsychicMqttClient::onSubscribe(OnSubscribeUserCallback callback)
{
_onSubscribeUserCallbacks.push_back(callback);
if (_onSubscribeUserCallbackCount < PSYCHIC_MAX_SUBSCRIBE_CB)
{
_onSubscribeUserCallbacks[_onSubscribeUserCallbackCount++] = std::move(callback);
}
else
{
ESP_LOGE(TAG, "onSubscribe callback list full (max=%d)", PSYCHIC_MAX_SUBSCRIBE_CB);
}
return *this;
}
PsychicMqttClient &PsychicMqttClient::onUnsubscribe(OnUnsubscribeUserCallback callback)
{
_onUnsubscribeUserCallbacks.push_back(callback);
if (_onUnsubscribeUserCallbackCount < PSYCHIC_MAX_UNSUBSCRIBE_CB)
{
_onUnsubscribeUserCallbacks[_onUnsubscribeUserCallbackCount++] = std::move(callback);
}
else
{
ESP_LOGE(TAG, "onUnsubscribe callback list full (max=%d)", PSYCHIC_MAX_UNSUBSCRIBE_CB);
}
return *this;
}
PsychicMqttClient &PsychicMqttClient::onMessage(OnMessageUserCallback callback)
{
OnMessageUserCallback_t subscription = {nullptr, 0, callback};
_onMessageUserCallbacks.push_back(subscription);
if (_onMessageUserCallbackCount < PSYCHIC_MAX_MESSAGE_CB)
{
OnMessageUserCallback_t &slot = _onMessageUserCallbacks[_onMessageUserCallbackCount++];
slot.topic[0] = '\0';
slot.qos = 0;
slot.callback = std::move(callback);
slot.has_topic = false;
slot.used = true;
}
else
{
ESP_LOGE(TAG, "onMessage callback list full (max=%d)", PSYCHIC_MAX_MESSAGE_CB);
}
return *this;
}
PsychicMqttClient &PsychicMqttClient::onTopic(const char *topic, int qos, OnMessageUserCallback callback)
{
OnMessageUserCallback_t subscription = {strcpy((char *)malloc(strlen(topic) + 1), topic), qos, callback};
_onMessageUserCallbacks.push_back(subscription);
if (_onMessageUserCallbackCount >= PSYCHIC_MAX_MESSAGE_CB)
{
ESP_LOGE(TAG, "onTopic subscription list full (max=%d)", PSYCHIC_MAX_MESSAGE_CB);
return *this;
}
OnMessageUserCallback_t &slot = _onMessageUserCallbacks[_onMessageUserCallbackCount++];
size_t tlen = strnlen(topic, PSYCHIC_MAX_TOPIC_LEN - 1);
memcpy(slot.topic, topic, tlen);
slot.topic[tlen] = '\0';
slot.qos = qos;
slot.callback = std::move(callback);
slot.has_topic = true;
slot.used = true;
if (_connected)
subscribe(topic, qos);
subscribe(slot.topic, qos);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onPublish(OnPublishUserCallback callback)
{
_onPublishUserCallbacks.push_back(callback);
if (_onPublishUserCallbackCount < PSYCHIC_MAX_PUBLISH_CB)
{
_onPublishUserCallbacks[_onPublishUserCallbackCount++] = std::move(callback);
}
else
{
ESP_LOGE(TAG, "onPublish callback list full (max=%d)", PSYCHIC_MAX_PUBLISH_CB);
}
return *this;
}
PsychicMqttClient &PsychicMqttClient::onError(OnErrorUserCallback callback)
{
_onErrorUserCallbacks.push_back(callback);
if (_onErrorUserCallbackCount < PSYCHIC_MAX_ERROR_CB)
{
_onErrorUserCallbacks[_onErrorUserCallbackCount++] = std::move(callback);
}
else
{
ESP_LOGE(TAG, "onError callback list full (max=%d)", PSYCHIC_MAX_ERROR_CB);
}
return *this;
}
@@ -287,14 +344,30 @@ void PsychicMqttClient::connect()
ESP_LOGE(TAG, "MQTT URI not set.");
return;
}
int desired_buffer = _mqtt_cfg.buffer.size > 0 ? _mqtt_cfg.buffer.size : 1024;
#else
if (_mqtt_cfg.uri == nullptr)
{
ESP_LOGE(TAG, "MQTT URI not set.");
return;
}
int desired_buffer = _mqtt_cfg.buffer_size > 0 ? _mqtt_cfg.buffer_size : 1024;
#endif
// Lazily size the reassembly buffer once for the client's lifetime.
// Messages larger than this will be rejected rather than triggering
// per-message heap allocation.
if (_buffer == nullptr)
{
_buffer_capacity = (size_t)desired_buffer;
_buffer = (char *)malloc(_buffer_capacity + 1);
if (_buffer == nullptr)
{
ESP_LOGE(TAG, "Failed to allocate reassembly buffer (%u bytes)", (unsigned)_buffer_capacity);
_buffer_capacity = 0;
}
}
if (_client == nullptr)
{
_client = esp_mqtt_client_init(&_mqtt_cfg);
@@ -468,25 +541,26 @@ void PsychicMqttClient::_onConnect(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
// Resubscribe to all topics
for (auto topic : _onMessageUserCallbacks)
// Resubscribe to all registered topics.
for (uint8_t i = 0; i < _onMessageUserCallbackCount; ++i)
{
if (topic.topic != nullptr)
subscribe(topic.topic, topic.qos);
OnMessageUserCallback_t &sub = _onMessageUserCallbacks[i];
if (sub.used && sub.has_topic)
subscribe(sub.topic, sub.qos);
}
for (auto callback : _onConnectUserCallbacks)
for (uint8_t i = 0; i < _onConnectUserCallbackCount; ++i)
{
callback(event->session_present);
_onConnectUserCallbacks[i](event->session_present);
}
}
void PsychicMqttClient::_onDisconnect(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
for (auto callback : _onDisconnectUserCallbacks)
for (uint8_t i = 0; i < _onDisconnectUserCallbackCount; ++i)
{
callback(event->session_present);
_onDisconnectUserCallbacks[i](event->session_present);
}
_stopMqttClient = true;
}
@@ -494,110 +568,106 @@ void PsychicMqttClient::_onDisconnect(esp_mqtt_event_handle_t &event)
void PsychicMqttClient::_onSubscribe(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
for (auto callback : _onSubscribeUserCallbacks)
for (uint8_t i = 0; i < _onSubscribeUserCallbackCount; ++i)
{
callback(event->msg_id);
_onSubscribeUserCallbacks[i](event->msg_id);
}
}
void PsychicMqttClient::_onUnsubscribe(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
for (auto callback : _onUnsubscribeUserCallbacks)
for (uint8_t i = 0; i < _onUnsubscribeUserCallbackCount; ++i)
{
callback(event->msg_id);
_onUnsubscribeUserCallbacks[i](event->msg_id);
}
}
void PsychicMqttClient::_onMessage(esp_mqtt_event_handle_t &event)
{
// ESP_LOGI(TAG, "MQTT_EVENT_DATA");
// printf("MSG_ID=%d\r\n", event->msg_id);
// printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
// printf("DATA=%.*s\r\n", event->data_len, event->data);
// printf("DATA_LEN=%d\r\n", event->data_len);
// printf("TOTAL_DATA_LEN=%d\r\n", event->total_data_len);
// printf("CURRENT_DATA_OFFSET=%d\r\n", event->current_data_offset);
// Check if we are dealing with a simple message
// Single-message path: payload fits in the event. No heap use.
if (event->total_data_len == event->data_len)
{
ESP_LOGV(TAG, "MQTT_EVENT_DATA_SINGLE");
// Copy the characters from data->data_ptr to c-string
char payload[event->data_len + 1];
memcpy(payload, (char *)event->data, event->data_len);
memcpy(payload, event->data, event->data_len);
payload[event->data_len] = '\0';
ESP_LOGV(TAG, "Payload=%s", payload);
char topic[event->topic_len + 1];
memcpy(topic, (char *)event->topic, event->topic_len);
topic[event->topic_len] = '\0';
ESP_LOGV(TAG, "Topic=%s", topic);
size_t tlen = (size_t)event->topic_len;
if (tlen >= sizeof(_topic)) tlen = sizeof(_topic) - 1;
char topic[sizeof(_topic)];
memcpy(topic, event->topic, tlen);
topic[tlen] = '\0';
for (auto callback : _onMessageUserCallbacks)
for (uint8_t i = 0; i < _onMessageUserCallbackCount; ++i)
{
if (callback.topic == nullptr || _isTopicMatch(topic, callback.topic))
OnMessageUserCallback_t &cb = _onMessageUserCallbacks[i];
if (!cb.used) continue;
if (!cb.has_topic || _isTopicMatch(topic, cb.topic))
{
callback.callback(topic, payload, event->retain, event->qos, event->dup);
cb.callback(topic, payload, event->retain, event->qos, event->dup);
}
}
return;
}
// Check if we are dealing with a first multipart message
else if (event->current_data_offset == 0)
// Multipart first chunk: remember the topic, begin reassembly in _buffer.
if (event->current_data_offset == 0)
{
ESP_LOGV(TAG, "MQTT_EVENT_DATA_MULTIPART_FIRST");
// Allocate memory for the buffer
_buffer = (char *)malloc(event->total_data_len + 1);
// Copy the characters from even->data to _buffer
memcpy(_buffer, (char *)event->data, event->data_len);
if (_buffer == nullptr)
{
ESP_LOGE(TAG, "multipart message but no reassembly buffer allocated");
return;
}
if ((size_t)event->total_data_len > _buffer_capacity)
{
ESP_LOGE(TAG, "multipart message size %d exceeds reassembly buffer %u", event->total_data_len, (unsigned)_buffer_capacity);
return;
}
memcpy(_buffer, event->data, event->data_len);
// Store the topic for later use, as it is only sent with the first message
_topic = (char *)malloc(event->topic_len + 1);
memcpy(_topic, (char *)event->topic, event->topic_len);
_topic[event->topic_len] = '\0';
size_t tlen = (size_t)event->topic_len;
if (tlen >= sizeof(_topic)) tlen = sizeof(_topic) - 1;
memcpy(_topic, event->topic, tlen);
_topic[tlen] = '\0';
return;
}
// Check if we are on the last message
else if (event->current_data_offset + event->data_len == event->total_data_len)
// Multipart final chunk: finalize and dispatch.
if (event->current_data_offset + event->data_len == event->total_data_len)
{
ESP_LOGV(TAG, "MQTT_EVENT_DATA_MULTIPART_LAST");
// Copy the characters from even->data to _buffer
memcpy(_buffer + event->current_data_offset, (char *)event->data, event->data_len);
if (_buffer == nullptr) return;
if ((size_t)(event->current_data_offset + event->data_len) > _buffer_capacity) return;
memcpy(_buffer + event->current_data_offset, event->data, event->data_len);
_buffer[event->total_data_len] = '\0';
ESP_LOGV(TAG, "Topic=%s", _topic);
ESP_LOGV(TAG, "Payload=%s", _buffer);
for (auto callback : _onMessageUserCallbacks)
for (uint8_t i = 0; i < _onMessageUserCallbackCount; ++i)
{
if (callback.topic == nullptr || _isTopicMatch(_topic, callback.topic))
OnMessageUserCallback_t &cb = _onMessageUserCallbacks[i];
if (!cb.used) continue;
if (!cb.has_topic || _isTopicMatch(_topic, cb.topic))
{
callback.callback(_topic, _buffer, event->retain, event->qos, event->dup);
cb.callback(_topic, _buffer, event->retain, event->qos, event->dup);
}
}
// Free the memory
free(_buffer);
_buffer = nullptr;
free(_topic);
_topic = nullptr;
return;
}
// Otherwise, we are in the middle of the message
else
{
// copy the characters from even->data to _buffer
memcpy(_buffer + event->current_data_offset, (char *)event->data, event->data_len);
ESP_LOGV(TAG, "MQTT_EVENT_DATA_MULTIPART");
}
// Multipart middle chunk: copy into _buffer at the correct offset.
if (_buffer == nullptr) return;
if ((size_t)(event->current_data_offset + event->data_len) > _buffer_capacity) return;
memcpy(_buffer + event->current_data_offset, event->data, event->data_len);
ESP_LOGV(TAG, "MQTT_EVENT_DATA_MULTIPART");
}
void PsychicMqttClient::_onPublish(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
for (auto callback : _onPublishUserCallbacks)
for (uint8_t i = 0; i < _onPublishUserCallbackCount; ++i)
{
callback(event->msg_id);
_onPublishUserCallbacks[i](event->msg_id);
}
}
@@ -611,77 +681,70 @@ void PsychicMqttClient::_onError(esp_mqtt_event_handle_t &event)
log_error_if_nonzero("captured as transport's socket errno", event->error_handle->esp_transport_sock_errno);
ESP_LOGI(TAG, "Last errno string (%s)", strerror(event->error_handle->esp_transport_sock_errno));
for (auto callback : _onErrorUserCallbacks)
for (uint8_t i = 0; i < _onErrorUserCallbackCount; ++i)
{
callback(*event->error_handle);
_onErrorUserCallbacks[i](*event->error_handle);
}
}
}
// Zero-allocation MQTT topic/subscription match. Supports '+' single-level
// wildcard and trailing '#' multi-level wildcard per MQTT 3.1.1 spec.
bool PsychicMqttClient::_isTopicMatch(const char *topic, const char *subscription)
{
ESP_LOGD(TAG, "Match topic: %s with subscription: %s", topic, subscription);
if (topic == nullptr || subscription == nullptr) return false;
String topicStr(topic);
String subscriptionStr(subscription);
const char *t = topic;
const char *s = subscription;
// Check if the subscription is a pure wildcard
if (subscriptionStr == "#" || subscriptionStr == "+")
for (;;)
{
ESP_LOGV(TAG, "Subscription is a pure wildcard --> MATCH");
return true;
}
// Check if the topic is a simple match
if (topicStr == subscriptionStr)
{
ESP_LOGV(TAG, "Topic is a direct match --> MATCH");
return true;
}
// Split the topic and subscription into tokens
int topicIndex = 0;
int subscriptionIndex = 0;
int lastTopicIndex = topicStr.lastIndexOf('/');
int lastSubscriptionIndex = subscriptionStr.lastIndexOf('/');
String topicToken = topicStr.substring(topicIndex, topicStr.indexOf('/', topicIndex));
String subscriptionToken = subscriptionStr.substring(subscriptionIndex, subscriptionStr.indexOf('/', subscriptionIndex));
ESP_LOGV(TAG, "Initial topic token: %s, subscription token: %s", topicToken.c_str(), subscriptionToken.c_str());
ESP_LOGV(TAG, "Last topic index: %d, last subscription index: %d", lastTopicIndex, lastSubscriptionIndex);
while (topicToken.length() > 0 && subscriptionToken.length() > 0)
{
ESP_LOGV(TAG, "Comparing topic token: %s with subscription token: %s", topicToken.c_str(), subscriptionToken.c_str());
if (subscriptionToken == "#")
{
ESP_LOGV(TAG, "Subscription token is # --> MATCH");
// '#' at the start of a level matches everything remaining.
if (s[0] == '#' && s[1] == '\0')
return true;
}
if (subscriptionToken != "+" && topicToken != subscriptionToken)
// Find end of the current level in both strings.
const char *t_end = strchr(t, '/');
const char *s_end = strchr(s, '/');
size_t t_len = t_end ? (size_t)(t_end - t) : strlen(t);
size_t s_len = s_end ? (size_t)(s_end - s) : strlen(s);
bool level_matches;
if (s_len == 1 && s[0] == '+')
{
ESP_LOGV(TAG, "Tokens do not match and subscription token is not + --> NO MATCH");
// '+' matches exactly one level (any content, including empty).
level_matches = true;
}
else
{
level_matches = (t_len == s_len) && (memcmp(t, s, t_len) == 0);
}
if (!level_matches)
return false;
// Advance past this level.
bool t_done = (t_end == nullptr);
bool s_done = (s_end == nullptr);
if (t_done && s_done)
return true;
if (s_done)
{
// Subscription ended but topic has more levels → no match,
// unless the subscription ended with '+' and topic also has no
// more levels (already handled above).
return false;
}
ESP_LOGV(TAG, "Current token index: topic: %d, subscription: %d", topicIndex, subscriptionIndex);
if (topicIndex == lastTopicIndex + 1 || subscriptionIndex == lastSubscriptionIndex + 1)
if (t_done)
{
ESP_LOGV(TAG, "End of tokens. Topic: %s, Subscription: %s", topicStr.substring(topicIndex).c_str(), subscriptionStr.substring(subscriptionIndex).c_str());
bool match = ((subscriptionToken == "+" && topicIndex == lastTopicIndex + 1) || topicStr.substring(topicIndex) == subscriptionStr.substring(subscriptionIndex));
ESP_LOGV(TAG, "End of tokens. Match: %s", match ? "true" : "false");
return match;
// Topic ended but subscription has more levels. Only matches if
// the remainder is exactly '#'.
return (s_end[1] == '#' && s_end[2] == '\0');
}
topicIndex = topicStr.indexOf('/', topicIndex) + 1;
subscriptionIndex = subscriptionStr.indexOf('/', subscriptionIndex) + 1;
ESP_LOGV(TAG, "Next token index: topic: %d, subscription: %d", topicIndex, subscriptionIndex);
topicToken = topicStr.substring(topicIndex, topicStr.indexOf('/', topicIndex));
subscriptionToken = subscriptionStr.substring(subscriptionIndex, subscriptionStr.indexOf('/', subscriptionIndex));
ESP_LOGV(TAG, "Next topic token: %s, subscription token: %s", topicToken.c_str(), subscriptionToken.c_str());
t = t_end + 1;
s = s_end + 1;
}
return false;
}
lib/PsychicMqttClient/src/PsychicMqttClient.h
+59 -12
View File
@@ -32,16 +32,15 @@
*/
#include <functional>
#include <vector>
#include "Arduino.h"
#include "mqtt_client.h"
#include "esp_crt_bundle.h"
#define PSYCHIC_MQTT_CLIENT_VERSION_STR "0.2.1"
#define PSYCHIC_MQTT_CLIENT_VERSION_STR "0.2.2"
#define PSYCHIC_MQTT_CLIENT_VERSION_MAJOR 0
#define PSYCHIC_MQTT_CLIENT_VERSION_MINOR 2
#define PSYCHIC_MQTT_CLIENT_VERSION_PATCH 1
#define PSYCHIC_MQTT_CLIENT_VERSION_PATCH 2
#ifndef ARDUINO_ARCH_ESP32
#error "This library only supports boards with an ESP32 processor."
@@ -56,11 +55,40 @@ typedef std::function<void(char *topic, char *payload, int retain, int qos, bool
typedef std::function<void(int msgId)> OnPublishUserCallback;
typedef std::function<void(esp_mqtt_error_codes_t error)> OnErrorUserCallback;
// Fixed caps sized for MeshCore's usage patterns. All callback and subscription
// storage is inline in the client object — zero dynamic allocation on register.
#ifndef PSYCHIC_MAX_CONNECT_CB
#define PSYCHIC_MAX_CONNECT_CB 4
#endif
#ifndef PSYCHIC_MAX_DISCONNECT_CB
#define PSYCHIC_MAX_DISCONNECT_CB 4
#endif
#ifndef PSYCHIC_MAX_SUBSCRIBE_CB
#define PSYCHIC_MAX_SUBSCRIBE_CB 2
#endif
#ifndef PSYCHIC_MAX_UNSUBSCRIBE_CB
#define PSYCHIC_MAX_UNSUBSCRIBE_CB 2
#endif
#ifndef PSYCHIC_MAX_MESSAGE_CB
#define PSYCHIC_MAX_MESSAGE_CB 4
#endif
#ifndef PSYCHIC_MAX_PUBLISH_CB
#define PSYCHIC_MAX_PUBLISH_CB 2
#endif
#ifndef PSYCHIC_MAX_ERROR_CB
#define PSYCHIC_MAX_ERROR_CB 4
#endif
#ifndef PSYCHIC_MAX_TOPIC_LEN
#define PSYCHIC_MAX_TOPIC_LEN 128
#endif
typedef struct
{
char *topic;
char topic[PSYCHIC_MAX_TOPIC_LEN];
int qos;
OnMessageUserCallback callback;
bool has_topic; // false = match all topics (onMessage)
bool used;
} OnMessageUserCallback_t;
/**
@@ -378,20 +406,39 @@ private:
bool _connected = false;
bool _stopMqttClient = false;
// Multipart message reassembly. _buffer is lazily allocated at connect() time
// to match the configured buffer size, then reused for the client's lifetime.
// _topic is inline storage, never heap-allocated.
char *_buffer = nullptr;
char *_topic = nullptr;
size_t _buffer_capacity = 0;
char _topic[PSYCHIC_MAX_TOPIC_LEN];
static void _onMqttEventStatic(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data);
void _onMqttEvent(esp_event_base_t base, int32_t event_id, void *event_data);
bool _isTopicMatch(const char *topic, const char *subscription);
std::vector<OnConnectUserCallback> _onConnectUserCallbacks;
std::vector<OnDisconnectUserCallback> _onDisconnectUserCallbacks;
std::vector<OnSubscribeUserCallback> _onSubscribeUserCallbacks;
std::vector<OnUnsubscribeUserCallback> _onUnsubscribeUserCallbacks;
std::vector<OnMessageUserCallback_t> _onMessageUserCallbacks;
std::vector<OnPublishUserCallback> _onPublishUserCallbacks;
std::vector<OnErrorUserCallback> _onErrorUserCallbacks;
// Fixed-size callback storage. Each slot is "used" once registered; we never
// unregister so we only ever append. Iteration uses the count, not sizeof.
OnConnectUserCallback _onConnectUserCallbacks[PSYCHIC_MAX_CONNECT_CB];
uint8_t _onConnectUserCallbackCount = 0;
OnDisconnectUserCallback _onDisconnectUserCallbacks[PSYCHIC_MAX_DISCONNECT_CB];
uint8_t _onDisconnectUserCallbackCount = 0;
OnSubscribeUserCallback _onSubscribeUserCallbacks[PSYCHIC_MAX_SUBSCRIBE_CB];
uint8_t _onSubscribeUserCallbackCount = 0;
OnUnsubscribeUserCallback _onUnsubscribeUserCallbacks[PSYCHIC_MAX_UNSUBSCRIBE_CB];
uint8_t _onUnsubscribeUserCallbackCount = 0;
OnMessageUserCallback_t _onMessageUserCallbacks[PSYCHIC_MAX_MESSAGE_CB];
uint8_t _onMessageUserCallbackCount = 0;
OnPublishUserCallback _onPublishUserCallbacks[PSYCHIC_MAX_PUBLISH_CB];
uint8_t _onPublishUserCallbackCount = 0;
OnErrorUserCallback _onErrorUserCallbacks[PSYCHIC_MAX_ERROR_CB];
uint8_t _onErrorUserCallbackCount = 0;
void _onBeforeConnect(esp_mqtt_event_handle_t &event_data, esp_mqtt_client_handle_t &client);
void _onConnect(esp_mqtt_event_handle_t &event_data);
src/helpers/bridges/MQTTBridge.cpp
+269 -528
View File
@@ -292,13 +292,16 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCCloc
_queue_count(0),
_last_status_publish(0), _last_status_retry(0), _status_interval(300000),
_ntp_client(_ntp_udp, "pool.ntp.org", 0, 60000), _last_ntp_sync(0), _ntp_synced(false), _ntp_sync_pending(false), _slots_setup_done(false), _max_active_slots(RUNTIME_MQTT_SLOTS),
_timezone(nullptr), _last_raw_len(0), _last_snr(0), _last_rssi(0), _last_raw_timestamp(0),
// Default to UTC; setRules() will be called from syncTimeWithNTP when a
// non-UTC timezone string is configured. Timezone has no default ctor,
// so we must pass rules here.
_timezone_storage(TimeChangeRule{"UTC", Last, Sun, Mar, 0, 0}, TimeChangeRule{"UTC", Last, Sun, Mar, 0, 0}),
_timezone(&_timezone_storage),
_last_raw_len(0), _last_snr(0), _last_rssi(0), _last_raw_timestamp(0),
_identity(identity),
_cached_has_connected_slots(false),
_last_memory_check(0), _skipped_publishes(0), _last_fragmentation_recovery(0),
_fragmentation_pressure_since(0), _last_critical_check_run(0),
_last_no_broker_log(0), _queue_disconnected_since(0), _all_tripped_since(0), _critical_heap_since(0),
_stuck_below_tls_since(0), _post_recovery_escalation_deadline(0),
_last_memory_check(0), _skipped_publishes(0),
_last_no_broker_log(0), _queue_disconnected_since(0),
_last_config_warning(0),
_dispatcher(nullptr), _radio(nullptr), _board(nullptr), _ms(nullptr),
#ifdef WITH_SNMP
@@ -592,6 +595,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();
_initialized = true;
s_mqtt_bridge_instance = this;
MQTT_DEBUG_PRINTLN("MQTT Bridge initialized");
@@ -643,16 +651,17 @@ void MQTTBridge::end() {
memset(_packet_queue, 0, sizeof(_packet_queue));
#endif
// Teardown all slots
// Disconnect and delete persistent MQTT clients. teardownSlot() intentionally
// only disconnects; destruction happens here so the mbedTLS contexts survive
// the reconfigure/reconnect hot path.
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
teardownSlot(i);
}
destroySlotClients();
// Clean up timezone object to prevent memory leak
if (_timezone) {
delete _timezone;
_timezone = nullptr;
}
// Timezone is inline class storage (_timezone_storage) since Phase 3 of
// the MQTT memory-defrag work — nothing to delete. _timezone always
// points at &_timezone_storage and stays valid for the bridge lifetime.
// Free PSRAM-backed buffers (non-PSRAM builds use inline class arrays — no free needed)
#if defined(BOARD_HAS_PSRAM)
@@ -907,14 +916,6 @@ void MQTTBridge::mqttTaskLoop() {
_last_status_publish = now;
_last_status_retry = 0;
MQTT_DEBUG_PRINTLN("Status published successfully, next publish in %lu ms", _status_interval);
// If we're in the hole but just proved connectivity, recover sooner than the dedicated pressure timer
size_t max_alloc = ESP.getMaxAllocHeap();
if (max_alloc < 58000 && (now - _last_fragmentation_recovery) > 300000) {
_last_fragmentation_recovery = now;
_fragmentation_pressure_since = 0;
MQTT_DEBUG_PRINTLN("Fragmentation recovery after status (max_alloc=%d)", (int)max_alloc);
recreateMqttClientsForFragmentationRecovery();
}
} else {
MQTT_DEBUG_PRINTLN("Status publish failed, will retry in %lu ms", STATUS_RETRY_INTERVAL);
}
@@ -922,8 +923,6 @@ void MQTTBridge::mqttTaskLoop() {
}
}
runCriticalMemoryCheckAndRecovery();
// Update cached connection status periodically (every 5 seconds)
// This ensures cache stays accurate even if callbacks miss updates
static unsigned long last_slot_status_update = 0;
@@ -946,6 +945,77 @@ 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;
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;
_slots[index].reconnect_backoff = 0;
_slots[index].max_backoff_failures = 0;
_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;
updateCachedConnectionStatus();
publishStatusToSlot(index);
});
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();
}
_slots[index].connected = false;
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.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);
}
});
}
}
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();
}
#ifdef ESP_PLATFORM
vTaskDelay(pdMS_TO_TICKS(50));
#else
delay(50);
#endif
delete slot.client;
slot.client = nullptr;
}
}
void MQTTBridge::setupSlot(int index) {
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return;
MQTTSlot& slot = _slots[index];
@@ -955,11 +1025,47 @@ void MQTTBridge::setupSlot(int index) {
return;
}
// Don't recreate if already exists
if (slot.client) return;
// 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;
}
// Reconfigure path: if we're re-applying (e.g. after a preset change), stop
// the existing connection cleanly first. The client object (and its mbedTLS
// context) is reused; setCredentials / setServer below overwrite the config
// fields in place before connect() restarts the ESP-IDF client.
if (slot.initial_connect_done) {
if (slot.client->connected()) {
slot.client->disconnect();
}
// 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
slot.auth_token[0] = '\0';
slot.connected = false;
slot.token_expires_at = 0;
slot.last_token_renewal = 0;
slot.reconnect_backoff = 0;
slot.max_backoff_failures = 0;
slot.circuit_breaker_tripped = false;
slot.last_reconnect_attempt = 0;
}
slot.client = new PsychicMqttClient();
slot.client->setAutoReconnect(false); // We handle reconnect with our own backoff logic
bool uses_jwt = (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) || slot.audience[0] != '\0';
optimizeMqttClientConfig(slot.client, uses_jwt); // sets keepalive (45s PSRAM, 75s non-PSRAM)
#ifndef MQTT_FORCE_KEEPALIVE_45
@@ -973,43 +1079,6 @@ void MQTTBridge::setupSlot(int index) {
#endif
#endif
// Callbacks (capture index by value)
slot.client->onConnect([this, index](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1);
_slots[index].connected = true;
_slots[index].reconnect_backoff = 0;
_slots[index].max_backoff_failures = 0;
_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;
updateCachedConnectionStatus();
publishStatusToSlot(index);
});
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();
}
_slots[index].connected = false;
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.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);
}
});
if (slot.preset) {
// Preset-based slot
slot.client->setServer(slot.preset->server_url);
@@ -1134,26 +1203,24 @@ void MQTTBridge::setupSlot(int index) {
slot.initial_connect_done = true;
}
// 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;
// destruction of the underlying client happens once in destroySlotClients().
void MQTTBridge::teardownSlot(int index) {
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return;
MQTTSlot& slot = _slots[index];
if (slot.client) {
if (slot.client->connected()) {
slot.client->disconnect();
}
if (slot.client && slot.client->connected()) {
slot.client->disconnect();
#ifdef ESP_PLATFORM
vTaskDelay(pdMS_TO_TICKS(50));
#else
delay(50);
#endif
delete slot.client;
slot.client = nullptr;
}
// Invalidate auth token (inline buffer — just zero the first byte)
slot.auth_token[0] = '\0';
slot.connected = false;
slot.initial_connect_done = false;
slot.broker_uri[0] = '\0';
@@ -1283,24 +1350,10 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
}
}
// Pre-flight: don't attempt TLS if there isn't enough contiguous internal heap.
// Each failed attempt allocates ~42 KB, fails, and frees slightly fragmented — over many
// cycles this degrades max_alloc from 70 KB to unusable. Block early to stop accumulation.
#ifdef ESP_PLATFORM
{
static const size_t MIN_TLS_HEAP = 45000;
static const unsigned long DEFERRED_LOG_INTERVAL_MS = 30000UL;
size_t avail = heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL);
if (avail < MIN_TLS_HEAP) {
if (now_millis - slot.last_deferred_log_ms >= DEFERRED_LOG_INTERVAL_MS || slot.last_deferred_log_ms == 0) {
slot.last_deferred_log_ms = now_millis;
MQTT_DEBUG_PRINTLN("MQTT%d connect deferred: max_block=%d < %d (heap too fragmented)",
index + 1, (int)avail, (int)MIN_TLS_HEAP);
}
return;
}
}
#endif
// Phase 4 (MQTT memory-defrag): the MIN_TLS_HEAP preflight was a workaround
// for the fragmentation caused by per-reconnect mbedTLS allocations. With
// persistent clients (Phase 1), the mbedTLS context is allocated once at
// startup and the preflight is no longer necessary.
// Periodic probe for circuit-breaker-tripped slots (recovery from transient outages)
// Attempts a single reconnect every 30 minutes to see if the server has come back
@@ -1319,39 +1372,16 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
_radio ? _radio->getRadioState() : -1,
(_radio && _radio->getLastRecvMillis() > 0) ? (_ms->getMillis() - _radio->getLastRecvMillis()) : 0);
if (slot_uses_jwt) {
unsigned long current_time = time(nullptr);
bool token_still_valid = slot.token_expires_at > 0 &&
current_time < slot.token_expires_at &&
(slot.token_expires_at - current_time) > 120; // >2 min remaining
if (token_still_valid) {
// Lightweight reconnect — reuse existing client but refresh JWT for fresh iat
if (createSlotAuthToken(index)) {
slot.client->setCredentials(_jwt_username, slot.auth_token);
MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (fresh token)", index + 1);
}
slot.client->connect();
} else {
// Token expired — regenerate token but avoid teardown+setup
// which would allocate new TLS context on potentially fragmented heap
if (slot.client) {
if (createSlotAuthToken(index)) {
slot.client->setCredentials(_jwt_username, slot.auth_token);
MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (regenerated expired token)", index + 1);
}
slot.client->connect();
} else {
// Client was destroyed — must do full setup
bool saved_tripped = slot.circuit_breaker_tripped;
MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (full setup, no client)", index + 1);
teardownSlot(index);
setupSlot(index);
_slots[index].circuit_breaker_tripped = saved_tripped;
_slots[index].last_reconnect_attempt = now_millis;
}
// 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.
if (createSlotAuthToken(index)) {
slot.client->setCredentials(_jwt_username, slot.auth_token);
MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (fresh token)", index + 1);
}
slot.client->reconnect();
} else {
slot.client->connect();
slot.client->reconnect();
}
// If the connect callback fires and sets slot.connected = true,
// it will clear circuit_breaker_tripped via the onConnect handler
@@ -1388,43 +1418,19 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
reconnect_attempted = true;
_last_slot_reconnect_ms = now_millis;
if (slot_uses_jwt) {
unsigned long current_time = time(nullptr);
bool token_still_valid = slot.token_expires_at > 0 &&
current_time < slot.token_expires_at &&
(slot.token_expires_at - current_time) > 120; // >2 min remaining
if (token_still_valid) {
// Token valid — always lightweight reconnect regardless of backoff level.
// Avoids creating a new TLS session (teardown+setup) which can race with
// WiFi association and cause drops when multiple slots do it simultaneously.
if (createSlotAuthToken(index)) {
slot.client->setCredentials(_jwt_username, slot.auth_token);
MQTT_DEBUG_PRINTLN("MQTT%d reconnect (fresh token, backoff %d)", index + 1, slot.reconnect_backoff);
} else {
MQTT_DEBUG_PRINTLN("MQTT%d reconnect (token refresh failed, backoff %d)", index + 1, slot.reconnect_backoff);
}
slot.client->reconnect();
// Always lightweight reconnect on the persistent client. A stale/expired
// token is handled by regenerating it in place and updating credentials
// — no teardown is needed because the client and its mbedTLS context
// persist for the bridge lifetime.
if (createSlotAuthToken(index)) {
slot.client->setCredentials(_jwt_username, slot.auth_token);
MQTT_DEBUG_PRINTLN("MQTT%d reconnect (fresh token, backoff %d)", index + 1, slot.reconnect_backoff);
} else {
// Token expired — full teardown to get fresh TLS context + credentials
if (teardown_attempted) {
// Defer to next cycle to limit heap fragmentation from simultaneous teardowns
slot.last_reconnect_attempt = now_millis;
return;
}
teardown_attempted = true;
uint8_t saved_backoff = slot.reconnect_backoff;
uint8_t saved_failures = slot.max_backoff_failures;
MQTT_DEBUG_PRINTLN("MQTT%d full teardown+setup (token expired, backoff %d)", index + 1, saved_backoff);
teardownSlot(index);
setupSlot(index);
_slots[index].reconnect_backoff = saved_backoff;
_slots[index].max_backoff_failures = saved_failures;
_slots[index].last_reconnect_attempt = now_millis;
MQTT_DEBUG_PRINTLN("MQTT%d reconnect (token refresh failed, backoff %d)", index + 1, slot.reconnect_backoff);
}
slot.client->reconnect();
} else {
// Non-JWT slots — always lightweight reconnect on existing client.
// recreateMqttClientsForFragmentationRecovery() handles teardown when
// memory pressure warrants it, without the timing hazard here.
// Non-JWT slots — lightweight reconnect on existing client.
MQTT_DEBUG_PRINTLN("MQTT%d reconnect (non-JWT, backoff %d)", index + 1, slot.reconnect_backoff);
slot.client->reconnect();
}
@@ -2025,33 +2031,12 @@ void MQTTBridge::loop() {
}
}
// Check if status hasn't been published successfully for too long
if (_status_enabled && _last_status_publish != 0) {
unsigned long now = millis();
unsigned long time_since_last_success = (now >= _last_status_publish) ?
(now - _last_status_publish) :
(ULONG_MAX - _last_status_publish + now + 1);
const unsigned long MAX_FAILURE_TIME_MS = 600000; // 10 minutes
if (time_since_last_success > MAX_FAILURE_TIME_MS) {
static unsigned long last_reinit_log = 0;
if (now - last_reinit_log > 300000) {
MQTT_DEBUG_PRINTLN("CRITICAL: Status publish has been failing for %lu ms (>%lu ms), forcing MQTT session reinitialization",
time_since_last_success, MAX_FAILURE_TIME_MS);
last_reinit_log = now;
}
recreateMqttClientsForFragmentationRecovery();
_last_status_publish = 0;
_last_status_retry = 0;
MQTT_DEBUG_PRINTLN("MQTT session reinitialized (clients recreated) - reconnection on next loop");
}
}
// Phase 4 (MQTT memory-defrag): the "recreate on prolonged status failure"
// path and the periodic runCriticalMemoryCheckAndRecovery() call have been
// removed. They were both symptoms of the heap churn introduced by
// delete/new cycles of the MQTT client; with persistent clients the
// allocator stays healthy and these recovery hooks aren't required.
}
#ifdef ESP_PLATFORM
runCriticalMemoryCheckAndRecovery();
#endif
#endif
}
@@ -2332,12 +2317,18 @@ void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx,
float snr, float rssi) {
if (!packet) return;
// Memory pressure check: Skip publishes when heap is severely fragmented
// Memory pressure check: Skip publishes when there's not enough contiguous
// heap for the publish itself (JSON buffer + esp-mqtt outbox frame + WiFi TX
// path). Headroom only — NOT an mbedTLS preflight: persistent clients keep
// their TLS contexts allocated for the bridge lifetime, so the old ~52 KB
// "reserve space for reconnect" guard is obsolete post Phase 1. Publish
// payload is capped at PUBLISH_JSON_BUFFER_SIZE (2 KB); 8 KB is a safe
// ceiling including esp-mqtt frame overhead and transient TCP buffers.
#ifdef ESP32
#if defined(BOARD_HAS_PSRAM)
static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 60000;
static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 16000;
#else
static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 52000;
static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 8000;
#endif
unsigned long now = millis();
if (now - _last_memory_check > 5000) {
@@ -2574,272 +2565,6 @@ void MQTTBridge::storeRawRadioData(const uint8_t* raw_data, int len, float snr,
}
}
// ---------------------------------------------------------------------------
// Memory management
// ---------------------------------------------------------------------------
#ifdef ESP_PLATFORM
void MQTTBridge::runCriticalMemoryCheckAndRecovery() {
const unsigned long CRITICAL_CHECK_INTERVAL_MS = 60000;
#if defined(BOARD_HAS_PSRAM)
const unsigned long PRESSURE_WINDOW_CRITICAL_MS = 180000;
const unsigned long PRESSURE_WINDOW_MODERATE_MS = 300000;
const unsigned long RECOVERY_THROTTLE_MS = 300000;
const size_t PRESSURE_THRESHOLD_CRITICAL = 58000;
const size_t PRESSURE_THRESHOLD_MODERATE = 70000;
const size_t HARD_RECOVERY_THRESHOLD = 54000;
#else
// Non-PSRAM boards run closer to the edge; use lower thresholds and longer windows.
const unsigned long PRESSURE_WINDOW_CRITICAL_MS = 300000;
const unsigned long PRESSURE_WINDOW_MODERATE_MS = 900000;
const unsigned long RECOVERY_THROTTLE_MS = 600000;
const size_t PRESSURE_THRESHOLD_CRITICAL = 50000;
const size_t PRESSURE_THRESHOLD_MODERATE = 56000;
const size_t HARD_RECOVERY_THRESHOLD = 46000;
#endif
const unsigned long CRITICAL_LOG_INTERVAL_MS = 900000;
unsigned long now = millis();
if (now - _last_critical_check_run < CRITICAL_CHECK_INTERVAL_MS) {
return;
}
_last_critical_check_run = now;
size_t free_h = ESP.getFreeHeap();
size_t max_alloc = ESP.getMaxAllocHeap();
size_t internal_free = heap_caps_get_free_size(MALLOC_CAP_INTERNAL);
size_t internal_max_block = heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL);
#ifdef MQTT_MEMORY_DEBUG
unsigned long spiram_f = 0;
#ifdef BOARD_HAS_PSRAM
spiram_f = heap_caps_get_free_size(MALLOC_CAP_SPIRAM);
#endif
agentLogHeap("MQTTBridge.cpp:runCriticalMemoryCheckAndRecovery", "critical_memory_check", "H1_H4", free_h, max_alloc, internal_free, spiram_f);
#endif
// Pressure timer: track how long max_alloc has been below moderate threshold
if (max_alloc >= PRESSURE_THRESHOLD_MODERATE) {
_fragmentation_pressure_since = 0;
} else {
if (_fragmentation_pressure_since == 0) {
_fragmentation_pressure_since = now;
}
}
// Rate-limited diagnostic logging (every 15 min)
static unsigned long last_critical_log = 0;
if (now - last_critical_log >= CRITICAL_LOG_INTERVAL_MS) {
last_critical_log = now;
// Always log heap state including internal heap (critical for diagnosing
// repeater hangs caused by internal heap exhaustion masked by PSRAM)
MQTT_DEBUG_PRINTLN("Heap: free=%d max=%d int_free=%d int_max=%d",
(int)free_h, (int)max_alloc, (int)internal_free, (int)internal_max_block);
if (max_alloc < PRESSURE_THRESHOLD_CRITICAL) {
MQTT_DEBUG_PRINTLN("CRITICAL: Low memory! Free: %d, Max: %d", (int)free_h, (int)max_alloc);
} else if (max_alloc < PRESSURE_THRESHOLD_MODERATE) {
MQTT_DEBUG_PRINTLN("WARNING: Memory pressure. Free: %d, Max: %d", (int)free_h, (int)max_alloc);
}
// Internal heap pressure check (PSRAM boards only — total heap can look fine
// while internal heap is exhausted, starving WiFi driver and Core 1)
#if defined(BOARD_HAS_PSRAM)
const size_t INTERNAL_HEAP_CRITICAL = 40000;
const size_t INTERNAL_BLOCK_CRITICAL = 20000;
if (internal_free < INTERNAL_HEAP_CRITICAL || internal_max_block < INTERNAL_BLOCK_CRITICAL) {
MQTT_DEBUG_PRINTLN("CRITICAL: Internal heap low! int_free=%d int_max_block=%d",
(int)internal_free, (int)internal_max_block);
}
#endif
// Log slot client count
int n_active = 0;
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
if (_slots[i].client != nullptr) n_active++;
}
MQTT_DEBUG_PRINTLN("MQTT clients active: %d", n_active);
}
// Dedicated recovery
unsigned long required_window_ms = (max_alloc < PRESSURE_THRESHOLD_CRITICAL)
? PRESSURE_WINDOW_CRITICAL_MS
: PRESSURE_WINDOW_MODERATE_MS;
bool allow_recovery = !_cached_has_connected_slots || max_alloc < HARD_RECOVERY_THRESHOLD;
if (_fragmentation_pressure_since != 0 &&
allow_recovery &&
(now - _fragmentation_pressure_since) >= required_window_ms &&
(now - _last_fragmentation_recovery) >= RECOVERY_THROTTLE_MS) {
_last_fragmentation_recovery = now;
_fragmentation_pressure_since = 0;
MQTT_DEBUG_PRINTLN("Fragmentation recovery: recreating MQTT clients (max_alloc=%d, pressure %lu min)", (int)max_alloc, (unsigned long)(required_window_ms / 60000));
recreateMqttClientsForFragmentationRecovery();
// Arm post-recovery escalation: if max_alloc hasn't recovered above MIN_TLS_HEAP
// within 60 s, the two-phase recovery failed and a clean reboot is the only option.
_post_recovery_escalation_deadline = now + 60000UL;
}
// Gray-zone trigger: MIN_TLS_HEAP (45000) is the pre-flight floor below which new
// TLS handshakes are blocked. If we sit below that floor with zero connected slots,
// no reconnect can succeed and we'll spin indefinitely. After 180 s of that state,
// force the two-phase recovery; the old pressure-window path only triggers after
// much longer windows and doesn't always apply when the gray-zone trap is active
// above PRESSURE_THRESHOLD_MODERATE but below MIN_TLS_HEAP.
static const size_t MIN_TLS_HEAP = 45000;
static const unsigned long STUCK_TRIGGER_MS = 180000UL; // 3 min
static const unsigned long STUCK_COOLDOWN_MS = 600000UL; // 10 min between attempts
bool in_gray_zone = (!_cached_has_connected_slots && max_alloc < MIN_TLS_HEAP);
if (in_gray_zone) {
if (_stuck_below_tls_since == 0) {
_stuck_below_tls_since = now;
} else if ((now - _stuck_below_tls_since) >= STUCK_TRIGGER_MS &&
(now - _last_fragmentation_recovery) >= STUCK_COOLDOWN_MS) {
MQTT_DEBUG_PRINTLN("Gray-zone recovery: no slots connected, max_alloc=%d < %d for %lu s",
(int)max_alloc, (int)MIN_TLS_HEAP, (now - _stuck_below_tls_since) / 1000);
_last_fragmentation_recovery = now;
_stuck_below_tls_since = 0;
_fragmentation_pressure_since = 0;
recreateMqttClientsForFragmentationRecovery();
_post_recovery_escalation_deadline = now + 60000UL;
}
} else {
_stuck_below_tls_since = 0;
}
// Post-recovery escalation: if the two-phase recovery just ran and 60 s later
// we're still stuck below the TLS floor with no slots connected, the recovery
// failed (either the allocator couldn't coalesce or something else is pinning
// the heap). Fall through to a clean reboot rather than spin forever.
if (_post_recovery_escalation_deadline != 0 && now >= _post_recovery_escalation_deadline) {
bool still_stuck = !_cached_has_connected_slots && max_alloc < MIN_TLS_HEAP;
if (still_stuck) {
MQTT_DEBUG_PRINTLN("CRITICAL: two-phase recovery failed to restore TLS viability "
"(max_alloc=%d < %d, no slots connected) — restarting.",
(int)max_alloc, (int)MIN_TLS_HEAP);
delay(100);
ESP.restart();
}
_post_recovery_escalation_deadline = 0;
}
// Last resort: if ALL enabled slots have circuit breakers tripped for >1 hour,
// heap is likely too fragmented for TLS to ever succeed — reboot.
bool all_tripped = true;
int enabled_count = 0;
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
if (_slots[i].enabled) {
enabled_count++;
if (!_slots[i].circuit_breaker_tripped) {
all_tripped = false;
break;
}
}
}
if (enabled_count > 0 && all_tripped) {
if (_all_tripped_since == 0) {
_all_tripped_since = now;
} else if ((now - _all_tripped_since) >= 3600000UL) {
MQTT_DEBUG_PRINTLN("All MQTT slots circuit-breaker tripped for >1 hour. Restarting ESP.");
delay(100);
ESP.restart();
}
} else {
_all_tripped_since = 0;
}
// Hard restart when max_alloc has been critically low for an extended period.
// The all-tripped check above misses "one slot connected, one slot failing" scenarios
// where _cached_has_connected_slots=true keeps circuit_breaker_tripped from firing.
// Below the TLS viability floor, further reconnect attempts are futile and only
// accumulate fragmentation. A clean reboot is the reliable recovery path.
#if defined(BOARD_HAS_PSRAM)
const size_t CRITICAL_RESTART_THRESHOLD = 45000;
#else
// Non-PSRAM: raised from 35000 to 40000 so we actually restart when stuck in
// the gray zone (observed stuck value 35828 > 35000 previously evaded restart
// while still below MIN_TLS_HEAP=45000, trapping the device in a no-recovery loop).
const size_t CRITICAL_RESTART_THRESHOLD = 40000;
#endif
const unsigned long CRITICAL_RESTART_WINDOW_MS = 300000; // 5 minutes
if (max_alloc < CRITICAL_RESTART_THRESHOLD) {
if (_critical_heap_since == 0) {
_critical_heap_since = now;
} else if ((now - _critical_heap_since) >= CRITICAL_RESTART_WINDOW_MS) {
MQTT_DEBUG_PRINTLN("CRITICAL: max_alloc=%d below %d for >5 min — restarting to recover heap.",
(int)max_alloc, (int)CRITICAL_RESTART_THRESHOLD);
delay(100);
ESP.restart();
}
} else {
_critical_heap_since = 0;
}
}
#endif
void MQTTBridge::recreateMqttClientsForFragmentationRecovery() {
// Two-phase recovery that reproduces what a WiFi drop achieves accidentally:
// ALL mbedTLS contexts are destroyed before any new one is allocated, giving the
// allocator a clean "no TLS context resident" moment so free blocks can coalesce.
//
// The previous per-slot teardown+setup loop held the earlier slot's fresh TLS
// context resident while later slots were still being torn down, defeating
// coalescing. It also skipped connected slots, so a "one up, one stuck" scenario
// was effectively a no-op — the connected slot's TLS allocation kept pinning
// the heap fragmented.
uint8_t saved_backoff[RUNTIME_MQTT_SLOTS] = {0};
uint8_t saved_failures[RUNTIME_MQTT_SLOTS] = {0};
#ifdef ESP_PLATFORM
size_t before_alloc = heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL);
MQTT_DEBUG_PRINTLN("Fragmentation recovery: tearing down all slots (max_alloc=%d)", (int)before_alloc);
#endif
// Phase 1: tear down EVERY enabled slot, regardless of connected state.
// Saving backoff state prevents teardownSlot() from clobbering it back to zero,
// which would otherwise let circuit-breaker logic never trip after recovery.
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
if (_slots[i].enabled) {
saved_backoff[i] = _slots[i].reconnect_backoff;
saved_failures[i] = _slots[i].max_backoff_failures;
teardownSlot(i);
}
}
// Settle: let the ESP-IDF MQTT task finish releasing transport and mbedTLS
// buffers before any new TLS context is allocated. Without this, the first
// setupSlot can race teardown cleanup and re-fragment at the same offset.
#ifdef ESP_PLATFORM
vTaskDelay(pdMS_TO_TICKS(500));
size_t after_teardown_alloc = heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL);
MQTT_DEBUG_PRINTLN("Fragmentation recovery: all slots torn down, settled (max_alloc=%d)",
(int)after_teardown_alloc);
#else
delay(500);
#endif
// Phase 2: staggered setup so two TLS handshakes don't fire back-to-back and
// immediately re-fragment. 5 s between slots matches the boot-time stagger.
int active_count = 0;
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
if (_slots[i].enabled) {
if (active_count >= _max_active_slots) {
break;
}
setupSlot(i);
_slots[i].reconnect_backoff = saved_backoff[i];
_slots[i].max_backoff_failures = saved_failures[i];
active_count++;
#ifdef ESP_PLATFORM
if (active_count < _max_active_slots && (i + 1) < RUNTIME_MQTT_SLOTS) {
vTaskDelay(pdMS_TO_TICKS(5000));
}
#endif
}
}
updateCachedConnectionStatus();
}
// ---------------------------------------------------------------------------
// NTP time sync
// ---------------------------------------------------------------------------
@@ -2938,32 +2663,32 @@ void MQTTBridge::syncTimeWithNTP() {
for (int i = 0; i < _max_active_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) {
// Check if the slot's token was created with a stale time
// (token_expires_at would be far in the past relative to current time)
if (_slots[i].enabled && slot_jwt && _slots[i].client) {
// Token created before NTP corrected the clock — refresh credentials
// in place and reconnect the persistent client. No teardown needed.
if (_slots[i].token_expires_at > 0 && current_time > _slots[i].token_expires_at) {
MQTT_DEBUG_PRINTLN("MQTT%d token stale after time correction, re-creating", i + 1);
teardownSlot(i);
setupSlot(i);
if (createSlotAuthToken(i)) {
_slots[i].client->setCredentials(_jwt_username, _slots[i].auth_token);
}
_slots[i].client->reconnect();
}
}
}
}
// Set timezone from string (with DST support) - only if changed
// Set timezone from string (with DST support) — only if changed.
// Reuses the inline _timezone_storage via setRules() instead of
// deleting/newing a Timezone, which was a per-change heap alloc pair.
static char last_timezone[64] = "";
if (strcmp(_prefs->timezone_string, last_timezone) != 0) {
if (_timezone) {
delete _timezone;
_timezone = nullptr;
}
Timezone* tz = createTimezoneFromString(_prefs->timezone_string);
if (tz) {
_timezone = tz;
} else {
TimeChangeRule dst_rule, std_rule;
if (!timezoneRulesFromString(_prefs->timezone_string, dst_rule, std_rule)) {
TimeChangeRule utc = {"UTC", Last, Sun, Mar, 0, 0};
_timezone = new Timezone(utc, utc);
dst_rule = utc;
std_rule = utc;
}
_timezone_storage.setRules(dst_rule, std_rule);
strncpy(last_timezone, _prefs->timezone_string, sizeof(last_timezone) - 1);
last_timezone[sizeof(last_timezone) - 1] = '\0';
}
@@ -2980,115 +2705,131 @@ void MQTTBridge::syncTimeWithNTP() {
// Timezone helper
// ---------------------------------------------------------------------------
Timezone* MQTTBridge::createTimezoneFromString(const char* tz_string) {
// Create Timezone objects for common IANA timezone strings
// Populates dst_out and std_out with the DST/standard TimeChangeRules for the
// given timezone string. Returns true on match, false on unknown strings. Zero
// heap allocation — the caller then passes these into Timezone::setRules() on
// an existing Timezone object.
bool MQTTBridge::timezoneRulesFromString(const char* tz_string, TimeChangeRule& dst_out, TimeChangeRule& std_out) {
// GCC refuses to implicitly build a TimeChangeRule temporary from a bare
// braced-init-list on the right-hand side of operator= (the aggregate has a
// char[6] member). Name the type explicitly so a proper temporary is formed.
// North America
if (strcmp(tz_string, "America/Los_Angeles") == 0 || strcmp(tz_string, "America/Vancouver") == 0) {
TimeChangeRule pst = {"PST", First, Sun, Nov, 2, -480}; // UTC-8
TimeChangeRule pdt = {"PDT", Second, Sun, Mar, 2, -420}; // UTC-7
return new Timezone(pdt, pst);
std_out = TimeChangeRule{"PST", First, Sun, Nov, 2, -480};
dst_out = TimeChangeRule{"PDT", Second, Sun, Mar, 2, -420};
return true;
} else if (strcmp(tz_string, "America/Denver") == 0) {
TimeChangeRule mst = {"MST", First, Sun, Nov, 2, -420}; // UTC-7
TimeChangeRule mdt = {"MDT", Second, Sun, Mar, 2, -360}; // UTC-6
return new Timezone(mdt, mst);
std_out = TimeChangeRule{"MST", First, Sun, Nov, 2, -420};
dst_out = TimeChangeRule{"MDT", Second, Sun, Mar, 2, -360};
return true;
} else if (strcmp(tz_string, "America/Chicago") == 0) {
TimeChangeRule cst = {"CST", First, Sun, Nov, 2, -360}; // UTC-6
TimeChangeRule cdt = {"CDT", Second, Sun, Mar, 2, -300}; // UTC-5
return new Timezone(cdt, cst);
std_out = TimeChangeRule{"CST", First, Sun, Nov, 2, -360};
dst_out = TimeChangeRule{"CDT", Second, Sun, Mar, 2, -300};
return true;
} else if (strcmp(tz_string, "America/New_York") == 0 || strcmp(tz_string, "America/Toronto") == 0) {
TimeChangeRule est = {"EST", First, Sun, Nov, 2, -300}; // UTC-5
TimeChangeRule edt = {"EDT", Second, Sun, Mar, 2, -240}; // UTC-4
return new Timezone(edt, est);
std_out = TimeChangeRule{"EST", First, Sun, Nov, 2, -300};
dst_out = TimeChangeRule{"EDT", Second, Sun, Mar, 2, -240};
return true;
} else if (strcmp(tz_string, "America/Anchorage") == 0) {
TimeChangeRule akst = {"AKST", First, Sun, Nov, 2, -540}; // UTC-9
TimeChangeRule akdt = {"AKDT", Second, Sun, Mar, 2, -480}; // UTC-8
return new Timezone(akdt, akst);
std_out = TimeChangeRule{"AKST", First, Sun, Nov, 2, -540};
dst_out = TimeChangeRule{"AKDT", Second, Sun, Mar, 2, -480};
return true;
} else if (strcmp(tz_string, "Pacific/Honolulu") == 0) {
TimeChangeRule hst = {"HST", Last, Sun, Oct, 2, -600}; // UTC-10 (no DST)
return new Timezone(hst, hst);
TimeChangeRule hst = {"HST", Last, Sun, Oct, 2, -600};
dst_out = hst; std_out = hst;
return true;
// Europe
} else if (strcmp(tz_string, "Europe/London") == 0) {
TimeChangeRule gmt = {"GMT", Last, Sun, Oct, 2, 0}; // UTC+0
TimeChangeRule bst = {"BST", Last, Sun, Mar, 1, 60}; // UTC+1
return new Timezone(bst, gmt);
std_out = TimeChangeRule{"GMT", Last, Sun, Oct, 2, 0};
dst_out = TimeChangeRule{"BST", Last, Sun, Mar, 1, 60};
return true;
} else if (strcmp(tz_string, "Europe/Paris") == 0 || strcmp(tz_string, "Europe/Berlin") == 0) {
TimeChangeRule cet = {"CET", Last, Sun, Oct, 3, 60}; // UTC+1
TimeChangeRule cest = {"CEST", Last, Sun, Mar, 2, 120}; // UTC+2
return new Timezone(cest, cet);
std_out = TimeChangeRule{"CET", Last, Sun, Oct, 3, 60};
dst_out = TimeChangeRule{"CEST", Last, Sun, Mar, 2, 120};
return true;
} else if (strcmp(tz_string, "Europe/Moscow") == 0) {
TimeChangeRule msk = {"MSK", Last, Sun, Oct, 3, 180}; // UTC+3 (no DST since 2014)
return new Timezone(msk, msk);
TimeChangeRule msk = {"MSK", Last, Sun, Oct, 3, 180};
dst_out = msk; std_out = msk;
return true;
// Asia
} else if (strcmp(tz_string, "Asia/Tokyo") == 0) {
TimeChangeRule jst = {"JST", Last, Sun, Oct, 2, 540}; // UTC+9 (no DST)
return new Timezone(jst, jst);
TimeChangeRule jst = {"JST", Last, Sun, Oct, 2, 540};
dst_out = jst; std_out = jst;
return true;
} else if (strcmp(tz_string, "Asia/Shanghai") == 0 || strcmp(tz_string, "Asia/Hong_Kong") == 0) {
TimeChangeRule cst = {"CST", Last, Sun, Oct, 2, 480}; // UTC+8 (no DST)
return new Timezone(cst, cst);
TimeChangeRule cst = {"CST", Last, Sun, Oct, 2, 480};
dst_out = cst; std_out = cst;
return true;
} else if (strcmp(tz_string, "Asia/Kolkata") == 0) {
TimeChangeRule ist = {"IST", Last, Sun, Oct, 2, 330}; // UTC+5:30 (no DST)
return new Timezone(ist, ist);
TimeChangeRule ist = {"IST", Last, Sun, Oct, 2, 330};
dst_out = ist; std_out = ist;
return true;
} else if (strcmp(tz_string, "Asia/Dubai") == 0) {
TimeChangeRule gst = {"GST", Last, Sun, Oct, 2, 240}; // UTC+4 (no DST)
return new Timezone(gst, gst);
TimeChangeRule gst = {"GST", Last, Sun, Oct, 2, 240};
dst_out = gst; std_out = gst;
return true;
// Australia
} else if (strcmp(tz_string, "Australia/Sydney") == 0 || strcmp(tz_string, "Australia/Melbourne") == 0) {
TimeChangeRule aest = {"AEST", First, Sun, Apr, 3, 600}; // UTC+10
TimeChangeRule aedt = {"AEDT", First, Sun, Oct, 2, 660}; // UTC+11
return new Timezone(aedt, aest);
std_out = TimeChangeRule{"AEST", First, Sun, Apr, 3, 600};
dst_out = TimeChangeRule{"AEDT", First, Sun, Oct, 2, 660};
return true;
} else if (strcmp(tz_string, "Australia/Perth") == 0) {
TimeChangeRule awst = {"AWST", Last, Sun, Oct, 2, 480}; // UTC+8 (no DST)
return new Timezone(awst, awst);
TimeChangeRule awst = {"AWST", Last, Sun, Oct, 2, 480};
dst_out = awst; std_out = awst;
return true;
// Timezone abbreviations (with DST handling)
} else if (strcmp(tz_string, "PDT") == 0 || strcmp(tz_string, "PST") == 0) {
TimeChangeRule pst = {"PST", First, Sun, Nov, 2, -480};
TimeChangeRule pdt = {"PDT", Second, Sun, Mar, 2, -420};
return new Timezone(pdt, pst);
std_out = TimeChangeRule{"PST", First, Sun, Nov, 2, -480};
dst_out = TimeChangeRule{"PDT", Second, Sun, Mar, 2, -420};
return true;
} else if (strcmp(tz_string, "MDT") == 0 || strcmp(tz_string, "MST") == 0) {
TimeChangeRule mst = {"MST", First, Sun, Nov, 2, -420};
TimeChangeRule mdt = {"MDT", Second, Sun, Mar, 2, -360};
return new Timezone(mdt, mst);
std_out = TimeChangeRule{"MST", First, Sun, Nov, 2, -420};
dst_out = TimeChangeRule{"MDT", Second, Sun, Mar, 2, -360};
return true;
} else if (strcmp(tz_string, "CDT") == 0 || strcmp(tz_string, "CST") == 0) {
TimeChangeRule cst = {"CST", First, Sun, Nov, 2, -360};
TimeChangeRule cdt = {"CDT", Second, Sun, Mar, 2, -300};
return new Timezone(cdt, cst);
std_out = TimeChangeRule{"CST", First, Sun, Nov, 2, -360};
dst_out = TimeChangeRule{"CDT", Second, Sun, Mar, 2, -300};
return true;
} else if (strcmp(tz_string, "EDT") == 0 || strcmp(tz_string, "EST") == 0) {
TimeChangeRule est = {"EST", First, Sun, Nov, 2, -300};
TimeChangeRule edt = {"EDT", Second, Sun, Mar, 2, -240};
return new Timezone(edt, est);
std_out = TimeChangeRule{"EST", First, Sun, Nov, 2, -300};
dst_out = TimeChangeRule{"EDT", Second, Sun, Mar, 2, -240};
return true;
} else if (strcmp(tz_string, "BST") == 0 || strcmp(tz_string, "GMT") == 0) {
TimeChangeRule gmt = {"GMT", Last, Sun, Oct, 2, 0};
TimeChangeRule bst = {"BST", Last, Sun, Mar, 1, 60};
return new Timezone(bst, gmt);
std_out = TimeChangeRule{"GMT", Last, Sun, Oct, 2, 0};
dst_out = TimeChangeRule{"BST", Last, Sun, Mar, 1, 60};
return true;
} else if (strcmp(tz_string, "CEST") == 0 || strcmp(tz_string, "CET") == 0) {
TimeChangeRule cet = {"CET", Last, Sun, Oct, 3, 60};
TimeChangeRule cest = {"CEST", Last, Sun, Mar, 2, 120};
return new Timezone(cest, cet);
std_out = TimeChangeRule{"CET", Last, Sun, Oct, 3, 60};
dst_out = TimeChangeRule{"CEST", Last, Sun, Mar, 2, 120};
return true;
// UTC and simple offsets
} else if (strcmp(tz_string, "UTC") == 0) {
TimeChangeRule utc = {"UTC", Last, Sun, Mar, 0, 0};
return new Timezone(utc, utc);
dst_out = utc; std_out = utc;
return true;
} else if (strncmp(tz_string, "UTC", 3) == 0) {
int offset = atoi(tz_string + 3);
TimeChangeRule utc_offset = {"UTC", Last, Sun, Mar, 0, offset * 60};
return new Timezone(utc_offset, utc_offset);
dst_out = utc_offset; std_out = utc_offset;
return true;
} else if (strncmp(tz_string, "GMT", 3) == 0) {
int offset = atoi(tz_string + 3);
TimeChangeRule gmt_offset = {"GMT", Last, Sun, Mar, 0, offset * 60};
return new Timezone(gmt_offset, gmt_offset);
} else if (strncmp(tz_string, "+", 1) == 0 || strncmp(tz_string, "-", 1) == 0) {
dst_out = gmt_offset; std_out = gmt_offset;
return true;
} else if (tz_string[0] == '+' || tz_string[0] == '-') {
int offset = atoi(tz_string);
TimeChangeRule offset_tz = {"TZ", Last, Sun, Mar, 0, offset * 60};
return new Timezone(offset_tz, offset_tz);
dst_out = offset_tz; std_out = offset_tz;
return true;
} else {
MQTT_DEBUG_PRINTLN("Unknown timezone: %s", tz_string);
return nullptr;
return false;
}
}
src/helpers/bridges/MQTTBridge.h
+29 -24
View File
@@ -177,7 +177,12 @@ private:
// Pending slot reconfigure: set from CLI (Core 1), processed by MQTT task (Core 0)
volatile bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS];
// Timezone handling
// Timezone handling.
// _timezone_storage is inline class storage (zero heap) that is reconfigured
// via setRules() whenever the preferred timezone string changes. _timezone
// is a stable alias pointer to &_timezone_storage so existing call sites
// that accept a Timezone* keep working without modification.
Timezone _timezone_storage;
Timezone* _timezone;
// Core 1-only staging: written by storeRawRadioData(), consumed by queuePacket().
@@ -221,12 +226,12 @@ private:
StaticJsonDocument<PUBLISH_JSON_BUFFER_SIZE> _packet_json_doc;
StaticJsonDocument<STATUS_JSON_BUFFER_SIZE> _status_json_doc;
// Memory pressure monitoring
// Memory pressure monitoring (per-publish skip; see publishPacket()).
// The broader fragmentation-recovery machinery was removed in Phase 4 of
// the MQTT memory-defrag work — persistent MQTT clients no longer churn
// the heap, so gray-zone / critical-restart trackers are unnecessary.
unsigned long _last_memory_check;
int _skipped_publishes; // Count of skipped publishes due to memory pressure
unsigned long _last_fragmentation_recovery; // Throttle: 5 min between recovery runs
unsigned long _fragmentation_pressure_since; // 0 = not under pressure
unsigned long _last_critical_check_run; // Throttle: run unified check at most every 60s
int _skipped_publishes; // Exposed via SNMP; count of publishes skipped when max_alloc is too low
// Status publish retry tracking
unsigned long _last_status_retry; // Track last retry attempt (separate from successful publish)
@@ -242,17 +247,6 @@ private:
unsigned long _queue_disconnected_since; // 0 = has connected slots
static const unsigned long QUEUE_STALE_MS = 300000UL; // Flush queue after 5 min disconnected
// Recovery: restart ESP after prolonged total failure
unsigned long _all_tripped_since; // 0 = not all tripped
unsigned long _critical_heap_since; // 0 = max_alloc is above critical threshold
// Gray-zone detector: max_alloc below the TLS viability floor (MIN_TLS_HEAP)
// with no connected slots — used to trigger the two-phase recovery before the
// hard restart check, and to arm a 60 s escalation deadline if that recovery
// fails to restore viability.
unsigned long _stuck_below_tls_since; // 0 = max_alloc above MIN_TLS_HEAP or slots connected
unsigned long _post_recovery_escalation_deadline; // 0 = not armed; else abs ms deadline for restart if still stuck
#ifdef WITH_SNMP
MeshSNMPAgent* _snmp_agent;
#endif
@@ -285,8 +279,20 @@ private:
bool substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size);
// Internal methods - slot management
void setupSlot(int index); // Create/destroy client for a slot based on its preset
void teardownSlot(int index); // Disconnect and free slot resources
// 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().
// - 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.
// - 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)
void destroySlotClients(); // Delete all persistent clients (shutdown only)
void setupSlot(int index); // Configure and connect the slot's existing client
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
@@ -295,10 +301,6 @@ private:
void publishStatusToSlot(int index);
void updateCachedConnectionStatus();
#ifdef ESP_PLATFORM
void runCriticalMemoryCheckAndRecovery();
#endif
void recreateMqttClientsForFragmentationRecovery();
void processPacketQueue();
bool publishStatus(); // Returns true if status was successfully published
bool handleWiFiConnection(unsigned long now);
@@ -318,7 +320,10 @@ private:
bool isAnySlotConnected();
void syncTimeWithNTP();
void refreshNTP(); // Lightweight periodic NTP refresh (non-blocking)
Timezone* createTimezoneFromString(const char* tz_string);
// Populates dst_out/std_out with TimeChangeRules for the given IANA or
// abbreviation string. Returns false if the string is not recognized
// (callers should fall back to UTC). Zero-allocation.
static bool timezoneRulesFromString(const char* tz_string, TimeChangeRule& dst_out, TimeChangeRule& std_out);
void checkConfigurationMismatch();
bool isIATAValid() const;
bool isSlotReady(int index, char* reason_buf = nullptr, size_t reason_size = 0) const;
variants/heltec_v3/platformio.ini
+1 -1
View File
@@ -126,7 +126,7 @@ build_flags =
-D MAX_MQTT_BROKERS=3
-D MQTT_MAX_PACKET_SIZE=1024
-D MQTT_DEBUG=1
; -D MQTT_MEMORY_DEBUG=1
-D MQTT_MEMORY_DEBUG=1
; Keep default observer profile less verbose to reduce runtime contention.
; -D MESH_PACKET_LOGGING=1
; -D MESH_DEBUG=1