rich/MeshCore
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4 from agessaman/fix/malloc-remove

Browse Source 
Refactor PsychicMqttClient to improve memory management and callback …
This commit is contained in:
Adam Gessaman authored and GitHub committed 2026-04-21 21:24:54 -07:00
2 parents 1ff9437f73 e9ff1ae055
commit 0ad7c33549
5 files changed
+571 -715
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

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