Fix MQTT recovery admission and retained client buffer sizing

This commit is contained in:
mikecarper committed 2026-10-10 15:29:50 -07:00
1 parent beb5b69a80
commit 110b26c737
11 files changed
+556 -63

No files matched your search

+7
View File
@@ -225,6 +225,9 @@ class FirmwareRamTest(unittest.TestCase):
self.assertEqual(profile["cold_start_allocation_allowance_bytes"], 16384)
self.assertEqual(profile["cold_start_dma_admission_free_bytes"], 32768)
self.assertEqual(profile["cold_start_dma_admission_contiguous_bytes"], 8192)
self.assertEqual(profile["restart_allocation_allowance_bytes"], 8192)
self.assertEqual(profile["restart_dma_admission_free_bytes"], 24576)
self.assertEqual(profile["restart_dma_admission_contiguous_bytes"], 8192)
self.assertFalse(profile["maximum_runtime_load_qualified"])
self.assertFalse(profile["physical_validation_performed"])
ordinary = ram.requirements("ESP32_PLATFORM", {"BOARD_HAS_PSRAM": 1}, "v4_companion")
@@ -280,6 +283,8 @@ class FirmwareRamTest(unittest.TestCase):
("client_task_priority", "kClientTaskPriority"),
("cold_start_allocation_allowance_bytes", "kColdStartAllowanceBytes"),
("cold_start_dma_admission_contiguous_bytes", "kColdStartLargestBytes"),
("restart_allocation_allowance_bytes", "kRestartAllowanceBytes"),
("restart_dma_admission_contiguous_bytes", "kRestartLargestBytes"),
("dma_admission_free_bytes", "kDmaReserveBytes"),
("dma_admission_contiguous_bytes", "kDmaLargestBytes"),
("tls_internal_admission_free_bytes", "kTlsInternalFreeBytes"),
@@ -289,6 +294,8 @@ class FirmwareRamTest(unittest.TestCase):
self.assertIn(f"{name} = {profile[key]};", source)
self.assertEqual(profile["cold_start_dma_admission_free_bytes"],
profile["dma_admission_free_bytes"] + profile["cold_start_allocation_allowance_bytes"])
self.assertEqual(profile["restart_dma_admission_free_bytes"],
profile["dma_admission_free_bytes"] + profile["restart_allocation_allowance_bytes"])
def test_combined_rak_budgets_uart_beside_exclusive_ethernet_and_ota(self):
defines = {
+36 -16
View File
@@ -16,6 +16,7 @@ POLICY_PREAMBLE = r'''
#include <cstdint>
#include "helpers/MQTTConnectionAdmission.h"
namespace Admission = MQTTConnectionAdmission;
using MemoryState = Admission::ClientMemoryState;
'''
HEAP_PREAMBLE = POLICY_PREAMBLE + r'''
@@ -54,7 +55,8 @@ constexpr int RUNTIME_MQTT_SLOTS = 6;
struct { size_t size = 0; size_t getPsramSize() const { return size; } } ESP;
struct Preset { const char* server_url; };
struct Client {
bool started = false;
bool initialized = false, started = false;
bool isInitialized() const { return initialized; }
bool isStarted() const { return started; }
};
struct MQTTBridge {
@@ -92,20 +94,23 @@ class MqttConnectionAdmissionTest(unittest.TestCase):
self.compile_run(POLICY_PREAMBLE + r'''
int main() {
// A large aggregate heap cannot compensate for a fragmented DMA region.
assert(!Admission::canStart(true, true, false, 100000, 4095));
assert(!Admission::canStart(true, true, false, 16383, 20000));
assert(Admission::canStart(true, true, false, 16384, 4096));
assert(Admission::canStart(false, false, false, 16384, 4096));
assert(!Admission::canStart(false, false, false, 16383, 4096));
assert(!Admission::canStart(false, false, false, 16384, 4095));
assert(!Admission::canStart(true, true, MemoryState::Started, 100000, 4095));
assert(!Admission::canStart(true, true, MemoryState::Started, 16383, 20000));
assert(Admission::canStart(true, true, MemoryState::Started, 16384, 4096));
assert(Admission::canStart(false, false, MemoryState::Started, 16384, 4096));
assert(!Admission::canStart(false, false, MemoryState::Started, 16383, 4096));
assert(!Admission::canStart(false, false, MemoryState::Started, 16384, 4095));
// Without PSRAM, TLS also needs a full record allocation and session budget.
for (int cold = 0; cold <= 1; ++cold) {
assert(!Admission::canStart(false, true, cold, 61439, 20000));
assert(!Admission::canStart(false, true, cold, 100000, 17407));
assert(Admission::canStart(false, true, cold, 61440, 17408));
const MemoryState states[] = {MemoryState::Uninitialized, MemoryState::Stopped,
MemoryState::Started};
for (MemoryState state : states) {
assert(!Admission::canStart(false, true, state, 61439, 20000));
assert(!Admission::canStart(false, true, state, 100000, 17407));
assert(Admission::canStart(false, true, state, 61440, 17408));
}
assert(Admission::canStart(true, false, false, 16384, 4096));
assert(Admission::canStart(true, false, MemoryState::Started, 16384, 4096));
assert(!Admission::canStart(true, false, static_cast<MemoryState>(255), SIZE_MAX, SIZE_MAX));
// A new SDK task needs its stack and ordinary startup allocations in
// addition to the DMA reserve, including plaintext and PSRAM-backed TLS.
@@ -113,12 +118,19 @@ int main() {
assert(Admission::kClientTaskPriority == 5);
assert(Admission::kColdStartAllowanceBytes == 16384);
assert(Admission::kColdStartLargestBytes == 8192);
assert(Admission::kRestartAllowanceBytes == 8192);
assert(Admission::kRestartLargestBytes == 8192);
for (int psram = 0; psram <= 1; ++psram) {
for (int tls = 0; tls <= 1; ++tls) {
if (!psram && tls) continue; // Larger TLS thresholds checked above.
assert(!Admission::canStart(psram, tls, true, 32767, 20000));
assert(!Admission::canStart(psram, tls, true, 100000, 8191));
assert(Admission::canStart(psram, tls, true, 32768, 8192));
assert(!Admission::canStart(psram, tls, MemoryState::Uninitialized, 32767, 20000));
assert(!Admission::canStart(psram, tls, MemoryState::Uninitialized, 100000, 8191));
assert(Admission::canStart(psram, tls, MemoryState::Uninitialized, 32768, 8192));
// An initialized handle retains its ordinary SDK buffers across a stop.
// Only the replacement task/TCB is charged beside the DMA reserve.
assert(!Admission::canStart(psram, tls, MemoryState::Stopped, 24575, 20000));
assert(!Admission::canStart(psram, tls, MemoryState::Stopped, 100000, 8191));
assert(Admission::canStart(psram, tls, MemoryState::Stopped, 24576, 8192));
}
}
}
@@ -281,7 +293,7 @@ int main() {
assert(!bridge.canStartSlotConnection(0));
Client client;
bridge._slots[0].client = &client;
assert(!bridge.canStartSlotConnection(0)); // Allocated, but task not started.
assert(!bridge.canStartSlotConnection(0)); // Wrapper allocated; SDK not initialized.
dma_free = 32767; dma_largest = 20000;
assert(!bridge.canStartSlotConnection(0));
dma_free = 32768; dma_largest = 8191;
@@ -289,6 +301,14 @@ int main() {
dma_largest = 8192;
assert(bridge.canStartSlotConnection(0));
// An initialized but stopped handle needs its task again, not new SDK buffers.
client.initialized = true; dma_free = 24575; dma_largest = 20000;
assert(!bridge.canStartSlotConnection(0));
dma_free = 24576; dma_largest = 8191;
assert(!bridge.canStartSlotConnection(0));
dma_largest = 8192;
assert(bridge.canStartSlotConnection(0));
// A warm reconnect retains the SDK task/buffers and uses the smaller floor.
client.started = true; dma_free = 16384; dma_largest = 4096;
assert(bridge.canStartSlotConnection(0));
+137 -8
View File
@@ -20,24 +20,43 @@ PREAMBLE = r'''
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <functional>
#include <vector>
#include "helpers/AlertFaultPolicy.h"
#include "helpers/MQTTConnectionAdmission.h"
#include "helpers/MQTTConnectionHealth.h"
#include "helpers/MQTTConnectionPolicy.h"
#include "helpers/WiFiPowerSave.h"
#define ESP_PLATFORM 1
#define MQTT_DEBUG_PRINTLN(...) ((void)0)
using esp_err_t = int;
constexpr int ESP_OK = 0, ESP_ERR_INVALID_ARG = 1, ESP_ERR_INVALID_STATE = 2;
constexpr int ESP_ERR_NO_MEM = 3, RUNTIME_MQTT_SLOTS = 6, WL_CONNECTED = 1;
constexpr int WL_DISCONNECTED = 0, WIFI_POWER_11dBm = 11;
using wl_status_t = int;
using wifi_ps_type_t = int;
constexpr int WIFI_PS_NONE = 0, WIFI_PS_MAX_MODEM = 1, WIFI_PS_MIN_MODEM = 2;
namespace mesh { namespace wifi {
constexpr bool kPrimaryEspNowRadio = false;
bool enforceStationChannel() { return true; }
} }
bool mqttStationWiFiMutationAllowed() { return true; }
void esp_wifi_set_ps(wifi_ps_type_t) {}
constexpr int MQTT_AUTH_JWT = 1;
constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2;
static uint32_t clock_ms = 1000;
static unsigned long wall_seconds = 1800000000;
static size_t dma_free = 100000, dma_largest = 20000;
static unsigned long s_wifi_connected_at = 0;
uint32_t millis() { return clock_ms; }
unsigned long fake_time(void*) { return wall_seconds; }
#define time fake_time
struct { int status() const { return WL_CONNECTED; } } WiFi;
struct {
int state = WL_CONNECTED;
int status() const { return state; }
void setTxPower(int) {}
void disconnect() { state = WL_DISCONNECTED; }
} WiFi;
struct { size_t size = 2097152; size_t getPsramSize() const { return size; } } ESP;
size_t heap_caps_get_free_size(uint32_t caps) {
assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_free;
@@ -56,12 +75,28 @@ bool mqttPresetEnforcesTokenExp(const Preset* preset) {
return preset && preset->enforce_exp;
}
struct PsychicMqttClient {
bool started = false, live = false;
int starts = 0, reconnects = 0, bounces = 0, credentials = 0;
bool initialized = false, started = false, live = false, model_task_heap = false;
int starts = 0, reconnects = 0, bounces = 0, credentials = 0, stops = 0;
std::function<void()> disconnected;
bool isInitialized() const { return initialized; }
bool isStarted() const { return started; }
bool connected() const { return live; }
int connect() { ++starts; started = true; return ESP_OK; }
int connect() {
++starts;
if (model_task_heap) {
assert(initialized && !started && dma_free >= 6536);
dma_free -= 6536;
}
initialized = started = true;
return ESP_OK;
}
int reconnect() { ++reconnects; return ESP_OK; }
void disconnect() {
++stops;
if (model_task_heap && started) dma_free += 6536;
started = live = false;
if (disconnected) disconnected();
}
void softDisconnect() {
++bounces; live = false;
// Releasing the old internal TLS session creates room for its replacement.
@@ -92,8 +127,19 @@ struct MQTTBridge {
bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {};
void* _identity = this;
bool _slots_setup_done = true;
bool _wifi_status_initialized = true, _manage_wifi = false;
wl_status_t _last_wifi_status = WL_CONNECTED;
unsigned long _last_wifi_check = 0, _last_wifi_reconnect_attempt = 0;
uint8_t _wifi_reconnect_backoff_attempt = 0, _wifi_power_save = 1;
struct { bool canonical_wifi = false; } _node_info;
struct Observer { uint8_t wifi_power_save = 1; } observer;
Observer* _obs = &observer;
AlertFaultPolicy::OutageSnapshot _wifi_outage{};
unsigned long _last_slot_reconnect_ms = 0;
int _max_active_slots = 5, mints = 0;
AlertFaultPolicy::OutageSnapshot wifiOutage() const { return _wifi_outage; }
void setWifiOutage(AlertFaultPolicy::OutageSnapshot snapshot) { _wifi_outage = snapshot; }
void beginWiFiStation() {}
const char* _jwt_username = "jwt";
static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000;
std::vector<int> setups;
@@ -127,6 +173,7 @@ struct MQTTBridge {
esp_err_t reconnectSlotClient(int index);
unsigned long slotTokenLifetime(int index) const;
void maintainSlotConnections();
bool handleWiFiConnection(unsigned long);
void maintainSlotConnection(int, unsigned long, unsigned long, bool, bool&, bool&);
};
'''
@@ -142,6 +189,7 @@ class MqttHandshakeSchedulerTest(unittest.TestCase):
"bool MQTTBridge::canStartSlotConnection(int index) const",
"esp_err_t MQTTBridge::reconnectSlotClient(int index)",
"unsigned long MQTTBridge::slotTokenLifetime(int index) const",
"bool MQTTBridge::handleWiFiConnection(unsigned long now)",
"void MQTTBridge::maintainSlotConnections()",
"void MQTTBridge::maintainSlotConnection("))
@@ -268,9 +316,10 @@ int main() {
stopped.client = &bridge.clients[0];
retained.client = &bridge.clients[1];
stopped.initial_connect_done = retained.initial_connect_done = true;
stopped.client->initialized = retained.client->initialized = true;
retained.client->started = true;
// An existing handle with a stopped task needs the cold-start allowance.
// An existing handle with a stopped task needs the task-restart allowance.
// Its deferral must leave this cycle available for the retained SDK task.
assert(!bridge.canStartSlotConnection(0));
assert(bridge.canStartSlotConnection(1));
@@ -281,11 +330,11 @@ int main() {
assert(retained.client->starts == 0 && retained.client->reconnects == 1);
assert(bridge._slot_attempt_pending[1]);
// Completion and newly available cold-start headroom allow a real SDK start,
// Completion and newly available task-restart headroom allow an SDK start,
// with no extra delay introduced by the earlier memory-only deferral.
bridge.complete(1);
clock_ms += 15000;
dma_free = 32768; dma_largest = 8192;
dma_free = 24576; dma_largest = 8192;
bridge.maintainSlotConnections();
assert(stopped.client->starts == 1 && stopped.client->reconnects == 0);
assert(stopped.client->isStarted() && bridge._slot_attempt_pending[0]);
@@ -303,7 +352,7 @@ int main() {
auto& slot = bridge._slots[0];
slot.preset = &jwt; slot.client = &bridge.clients[0];
slot.initial_connect_done = slot.connected = true;
slot.client->started = slot.client->live = true;
slot.client->initialized = slot.client->started = slot.client->live = true;
slot.token_expires_at = wall_seconds + 120;
// Before releasing the current TLS session, the full admission check fails.
dma_free = 20000; dma_largest = 10000;
@@ -321,6 +370,86 @@ int main() {
}
''')
def test_wifi_loss_restarts_three_retained_clients_without_cold_buffer_double_charge(self):
self.compile_run(r'''
int main() {
MQTTBridge bridge;
clock_ms = 70000;
// The live V4's three-broker measurement left about 22.7 KiB DMA heap.
// The SDK seam uses the pinned 6144-byte stack and 344-byte TCB, plus a
// simulated 48-byte allocator allowance. Transport/config buffers remain
// owned by initialized handles after stop; SDK heap calls are not executed.
dma_free = 22768; dma_largest = 22516;
for (int i = 0; i < 3; ++i) {
auto& slot = bridge._slots[i];
slot.client = &bridge.clients[i];
slot.initial_connect_done = slot.connected = true;
slot.client->initialized = slot.client->started = slot.client->live = true;
slot.client->model_task_heap = true;
slot.client->disconnected = [&bridge, i] { bridge._slots[i].connected = false; };
}
bridge._slots[5].enabled = false;
WiFi.state = WL_DISCONNECTED;
assert(!bridge.handleWiFiConnection(clock_ms));
assert(bridge.wifiOutage().down && dma_free == 42376);
for (int i = 0; i < 3; ++i) {
assert(bridge.clients[i].isInitialized() && !bridge.clients[i].isStarted());
assert(bridge.clients[i].stops == 1 && !bridge._slots[i].connected);
assert(!bridge._slot_attempt_pending[i]);
}
bridge.maintainSlotConnections();
for (int i = 0; i < 3; ++i) assert(bridge.clients[i].starts == 0);
clock_ms += 11000;
WiFi.state = WL_CONNECTED;
assert(bridge.handleWiFiConnection(clock_ms));
assert(!bridge.wifiOutage().down);
// Restart admission still fails safely if either aggregate or contiguous
// headroom is inadequate. Neither failure consumes retry/backoff state.
dma_free = 24575;
bridge.maintainSlotConnections();
dma_free = 24576; dma_largest = 8191;
bridge.maintainSlotConnections();
for (int i = 0; i < 3; ++i) {
assert(bridge.clients[i].starts == 0 && bridge._slots[i].last_reconnect_attempt == 0);
assert(bridge._slots[i].reconnect_backoff == 0 && bridge._slots[i].start_failures == 0);
}
dma_free = 42376; dma_largest = 22516;
bridge.maintainSlotConnections();
assert(bridge.clients[0].starts == 1 && dma_free == 35840);
assert(bridge.hasPendingSlotConnection());
clock_ms += 3600000;
bridge.maintainSlotConnections();
assert(bridge.clients[1].starts == 0 && bridge.clients[2].starts == 0);
bridge.complete(0);
bridge.maintainSlotConnections();
assert(bridge.clients[1].starts == 1 && dma_free == 29304);
clock_ms += 15000;
bridge.maintainSlotConnections();
assert(bridge.clients[2].starts == 0); // Callback, not time, ends a pending attempt.
bridge.complete(1);
assert(!MQTTConnectionAdmission::canStart(
true, true, MQTTConnectionAdmission::ClientMemoryState::Uninitialized,
dma_free, dma_largest));
assert(bridge.canStartSlotConnection(2));
bridge.maintainSlotConnections();
assert(bridge.clients[2].starts == 1 && dma_free == 22768);
bridge.complete(2);
clock_ms += 15000;
bridge.maintainSlotConnections();
for (int i = 0; i < 3; ++i) {
assert(bridge._slots[i].connected && bridge.clients[i].isStarted());
assert(bridge.clients[i].starts == 1 && bridge.clients[i].reconnects == 0);
assert(bridge._slots[i].start_failures == 0);
}
// Additional clients still need the full first-start allowance and wait.
assert(bridge.setups.empty());
assert(!bridge.canStartSlotConnection(3) && !bridge.canStartSlotConnection(4));
assert(!bridge.hasPendingSlotConnection());
}
''')
if __name__ == "__main__":
unittest.main()
+202 -1
View File
@@ -11,6 +11,8 @@ import tempfile
import unittest
from test_mqtt_transport_results import method
from test_mqtt_publish_ack import (ARDUINO_HEADER, MQTT_HEADER,
PREAMBLE as ADAPTER_PREAMBLE)
ROOT = Path(__file__).resolve().parents[1]
@@ -86,13 +88,145 @@ int main() {
assert(buffer_size == 896 && keepalive == 45);
assert(client._outbox_limit == 16384);
#else
assert(buffer_size == (needs_large ? 896 : 512) && keepalive == 75);
assert(buffer_size == 896 && keepalive == 75);
assert(client._outbox_limit == 8192);
#endif
}
}
'''
RECONFIGURE_PREAMBLE = r'''
#include "helpers/MQTTConnectionAdmission.h"
#include "helpers/MQTTConnectionPolicy.h"
#define MQTT_DEBUG_PRINTLN(...) ((void)0)
#define pdMS_TO_TICKS(value) (value)
constexpr int RUNTIME_MQTT_SLOTS = 2, MQTT_AUTH_JWT = 1;
constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2;
static size_t dma_free = 100000, dma_largest = 20000;
size_t heap_caps_get_free_size(uint32_t caps) {
assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_free;
}
size_t heap_caps_get_largest_free_block(uint32_t caps) {
assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_largest;
}
struct { size_t getPsramSize() const {
#ifdef BOARD_HAS_PSRAM
return 2097152;
#else
return 0;
#endif
} } ESP;
constexpr const char* MQTT_PRESET_NONE = "none";
constexpr const char* MQTT_PRESET_CUSTOM = "custom";
struct MQTTPresetDef { const char* server_url; int auth_type; };
static const MQTTPresetDef plain_preset{"mqtt://broker", 0};
static const MQTTPresetDef jwt_preset{"mqtt://broker", MQTT_AUTH_JWT};
const MQTTPresetDef* findMQTTPreset(const char* name) {
if (strcmp(name, "plain") == 0) return &plain_preset;
if (strcmp(name, "jwt") == 0) return &jwt_preset;
return nullptr;
}
bool mqttPresetNeedsSlotCredentials(const MQTTPresetDef*) { return false; }
bool customEndpointComplete(const char* host, uint16_t port) { return host[0] && port; }
struct MQTTPrefs {
char mqtt_slot_host[RUNTIME_MQTT_SLOTS][64] = {"custom-broker", ""};
uint16_t mqtt_slot_port[RUNTIME_MQTT_SLOTS] = {1883, 0};
char mqtt_slot_username[RUNTIME_MQTT_SLOTS][32] = {"custom-user", ""};
char mqtt_slot_password[RUNTIME_MQTT_SLOTS][32] = {"custom-password", ""};
char mqtt_slot_audience[RUNTIME_MQTT_SLOTS][32] = {"custom-audience", ""};
};
struct MQTTBridge {
struct MQTTSlot {
PsychicMqttClient* client = nullptr;
bool enabled = false, connected = false, initial_connect_done = false;
const MQTTPresetDef* preset = nullptr;
char host[64] = {}, broker_uri[128] = {}, username[32] = {}, password[32] = {};
char audience[32] = {}, *auth_token = nullptr;
uint16_t port = 1883;
unsigned long token_expires_at = 0, last_token_renewal = 0;
unsigned long last_reconnect_attempt = 0, last_log_time = 0, last_deferred_log_ms = 0;
uint8_t reconnect_backoff = 0, max_backoff_failures = 0;
bool circuit_breaker_tripped = false;
} _slots[RUNTIME_MQTT_SLOTS];
std::atomic<bool> _stop_requested{false}, _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {};
bool _initialized = true;
int _max_active_slots = 2;
MQTTPrefs prefs;
MQTTPrefs* _obs = &prefs;
bool canActivateSlot(int) const { return true; }
bool isSlotReady(int, char*, size_t) const { return true; }
bool hasPendingSlotConnection() const;
bool canStartSlotConnection(int) const;
void optimizeMqttClientConfig(PsychicMqttClient*, bool);
void teardownSlot(int, bool force = false);
void applySlotPreset(int, const char*);
bool setupSlot(int index) {
auto& slot = _slots[index];
// The setup seam executes the actual optimizer, adapter init/start, and
// production admission. Token creation and network callbacks are seams.
if (!canStartSlotConnection(index)) return false;
if (!slot.client) slot.client = new PsychicMqttClient;
optimizeMqttClientConfig(slot.client,
(slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) || slot.audience[0]);
slot.client->_mqtt_cfg.uri = "mqtt://broker";
slot.client->_mqtt_cfg.broker.address.uri = "mqtt://broker";
if (slot.client->connect() != ESP_OK) return false;
esp_mqtt_event_t event{MQTT_EVENT_CONNECTED, 0, false};
sdk_event_handler(sdk_handler_args, "MQTT", event.event_id, &event);
slot.initial_connect_done = slot.connected = true;
return true;
}
};
'''
RECONFIGURE_MAIN = r'''
int main() {
MQTTBridge bridge;
auto& slot = bridge._slots[0];
bridge.applySlotPreset(0, "plain");
assert(slot.client && slot.initial_connect_done && slot.client->isStarted());
assert(slot.client->_buffer_capacity == 896);
char token[8] = "old-jwt";
slot.auth_token = token;
bridge.applySlotPreset(0, "jwt");
assert(token[0] == '\0' && slot.preset == &jwt_preset);
assert(slot.client->_buffer_capacity == 896);
bridge.applySlotPreset(0, "plain");
assert(initialized_buffer_sizes == std::vector<int>{896});
assert(init_calls == 1 && destroy_calls == 0 && slot.client->_buffer_capacity == 896);
// A custom broker's audience requests JWT-size buffers through the same
// explicit-reconfiguration path and synchronizes its saved fields.
bridge.applySlotPreset(0, "custom");
assert(slot.preset == nullptr && slot.client->_buffer_capacity == 896);
assert(strcmp(slot.host, "custom-broker") == 0 && slot.port == 1883);
assert(strcmp(slot.audience, "custom-audience") == 0);
assert(strcmp(slot.username, "custom-user") == 0);
assert(strcmp(slot.password, "custom-password") == 0);
bridge.applySlotPreset(0, "none");
assert(!slot.enabled && !slot.initial_connect_done);
assert(slot.client && slot.client->isInitialized() && !slot.client->isStarted());
assert(init_calls == 1 && destroy_calls == 0);
// Reconfiguration retains the initialized client on both board profiles.
// It needs the stopped-task floor and never silently shrinks its buffers.
dma_free = 24575; dma_largest = 20000;
bridge.applySlotPreset(0, "plain");
assert(slot.client && !slot.client->isStarted() && !slot.initial_connect_done);
assert(init_calls == 1 && destroy_calls == 0);
dma_free = 24576;
assert(bridge.setupSlot(0));
assert(slot.client->_buffer_capacity == 896 && init_calls == 1);
const int destructions_before_guard = destroy_calls;
bridge.applySlotPreset(-1, "jwt");
bridge.applySlotPreset(RUNTIME_MQTT_SLOTS, "jwt");
bridge._stop_requested = true;
bridge.applySlotPreset(0, "jwt");
assert(destroy_calls == destructions_before_guard && slot.client->isStarted());
delete slot.client;
slot.client = nullptr;
}
'''
class MqttNetworkTimeoutTest(unittest.TestCase):
def fixture(self):
@@ -143,6 +277,73 @@ class MqttNetworkTimeoutTest(unittest.TestCase):
def test_production_optimizer_applies_connection_budget_to_every_config(self):
self.compile_run(self.fixture())
def test_production_preset_reconfiguration_retains_uniform_sdk_and_reassembly_buffers(self):
compiler = shutil.which("g++") or shutil.which("clang++")
self.assertIsNotNone(compiler, "a host C++ compiler is required")
adapter = "lib/PsychicMqttClient/src/PsychicMqttClient.cpp"
# Reuse the actual-header SDK seam; this bridge fixture needs the full
# preset reconfiguration path rather than the publish diagnostic shell.
preamble = ADAPTER_PREAMBLE.split("constexpr int RUNTIME_MQTT_SLOTS = 2;", 1)[0]
preamble = preamble.replace(
'void vTaskDelay(unsigned) { assert(false && "stub disconnect must deliver its event"); }',
'void vTaskDelay(unsigned ticks) { assert(ticks == 50); }')
source = "#define ESP_PLATFORM 1\n" + preamble + RECONFIGURE_PREAMBLE
source += "\n".join(method(adapter, signature) for signature in (
"PsychicMqttClient::PsychicMqttClient()",
"PsychicMqttClient::~PsychicMqttClient()",
"PsychicMqttClient &PsychicMqttClient::setKeepAlive(",
"PsychicMqttClient &PsychicMqttClient::setMessageRetransmitTimeout(",
"PsychicMqttClient &PsychicMqttClient::setBufferSize(",
"PsychicMqttClient &PsychicMqttClient::setNetworkTimeout(",
"PsychicMqttClient &PsychicMqttClient::setOutboxLimit(",
"PsychicMqttClient &PsychicMqttClient::setTaskStackAndPriority(",
"esp_mqtt_client_config_t *PsychicMqttClient::getMqttConfig(",
"bool PsychicMqttClient::connected()",
"esp_err_t PsychicMqttClient::applyConfig()",
"esp_err_t PsychicMqttClient::connect()",
"void PsychicMqttClient::disconnect()",
"void PsychicMqttClient::forceStop()",
"int PsychicMqttClient::subscribe(",
"void PsychicMqttClient::_onMqttEventStatic(",
"void PsychicMqttClient::_onMqttEvent(",
"void PsychicMqttClient::_onConnect(",
"void PsychicMqttClient::_onDisconnect(",
"void PsychicMqttClient::_onSubscribe(",
"void PsychicMqttClient::_onUnsubscribe(",
"void PsychicMqttClient::_onPublish(",
))
source += "\n".join(method("src/helpers/bridges/MQTTBridge.cpp", signature)
for signature in (
"bool MQTTBridge::hasPendingSlotConnection() const",
"bool MQTTBridge::canStartSlotConnection(int index) const",
"void MQTTBridge::optimizeMqttClientConfig(",
"void MQTTBridge::teardownSlot(",
"void MQTTBridge::applySlotPreset(",
)) + RECONFIGURE_MAIN
with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-reconfigure-") as temporary:
work = Path(temporary)
(work / "Arduino.h").write_text(ARDUINO_HEADER, encoding="ascii")
(work / "mqtt_client.h").write_text(MQTT_HEADER, encoding="ascii")
(work / "esp_crt_bundle.h").write_text("#pragma once\n", encoding="ascii")
path = work / "reconfigure.cpp"
path.write_text(source, encoding="ascii")
for major in (4, 5):
for psram in (False, True):
with self.subTest(idf_major=major, psram=psram):
binary = work / f"reconfigure-{major}-{int(psram)}"
command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
"-Wno-unused-parameter", "-DARDUINO_ARCH_ESP32=1",
f"-DESP_IDF_VERSION_MAJOR={major}",
"-I", str(work),
"-I", str(ROOT / "lib/PsychicMqttClient/src"),
"-I", str(ROOT / "src"), str(path), "-o", str(binary)]
if psram:
command.insert(1, "-DBOARD_HAS_PSRAM=1")
result = subprocess.run(command, capture_output=True, text=True, timeout=60)
self.assertEqual(result.returncode, 0, result.stderr)
result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
def test_previous_publish_timeout_fails_configuration_regression(self):
source = self.fixture()
old = source.replace(
+74 -10
View File
@@ -50,9 +50,13 @@ enum {
};
struct esp_mqtt_client_config_t {
const char* uri;
int buffer_size;
int buffer_size, keepalive, message_retransmit_timeout, network_timeout_ms;
int task_stack, task_prio;
struct { struct { const char* uri; } address; } broker;
struct { int size; } buffer;
struct { int size, out_size; } buffer;
struct { int keepalive, message_retransmit_timeout; } session;
struct { int timeout_ms; } network;
struct { int stack_size, priority; } task;
};
struct esp_mqtt_error_codes_t { int unused; };
struct esp_mqtt_event_t { int event_id, msg_id; bool session_present; };
@@ -68,6 +72,7 @@ PREAMBLE = r'''
#include <cstring>
#include <functional>
#include <utility>
#include <vector>
// Preinclude the standard headers so this seam only opens adapter visibility.
#define private public
#include "PsychicMqttClient.h"
@@ -90,25 +95,35 @@ const char* esp_err_to_name(int) { return "mock"; }
static int publish_result = 0, enqueue_result = 41, outbox_size = 0;
static int publish_calls = 0, enqueue_calls = 0, config_calls = 0;
static int start_calls = 0, reconnect_calls = 0, receive_events = 0, error_events = 0;
static int init_calls = 0, destroy_calls = 0, stop_calls = 0;
static int start_result = ESP_OK, register_result = ESP_OK;
static bool fail_init = false;
static std::vector<int> initialized_buffer_sizes;
using EventHandler = void(*)(void*, esp_event_base_t, int32_t, void*);
static EventHandler sdk_event_handler = nullptr;
static void* sdk_handler_args = nullptr;
void* esp_mqtt_client_init(esp_mqtt_client_config_t*) {
return reinterpret_cast<void*>(1);
void* esp_mqtt_client_init(esp_mqtt_client_config_t* config) {
++init_calls;
#if ESP_IDF_VERSION_MAJOR == 5
initialized_buffer_sizes.push_back(config->buffer.size);
#else
initialized_buffer_sizes.push_back(config->buffer_size);
#endif
return fail_init ? nullptr : reinterpret_cast<void*>(1);
}
int esp_mqtt_client_register_event(void*, int, EventHandler handler, void* args) {
sdk_event_handler = handler;
sdk_handler_args = args;
return ESP_OK;
return register_result;
}
int esp_mqtt_client_destroy(void*) { return ESP_OK; }
int esp_mqtt_client_destroy(void*) { ++destroy_calls; return ESP_OK; }
int esp_mqtt_set_config(void*, esp_mqtt_client_config_t*) {
++config_calls;
return ESP_OK;
}
int esp_mqtt_client_start(void*) { ++start_calls; return ESP_OK; }
int esp_mqtt_client_start(void*) { ++start_calls; return start_result; }
int esp_mqtt_client_reconnect(void*) { ++reconnect_calls; return ESP_OK; }
int esp_mqtt_client_stop(void*) { return ESP_OK; }
int esp_mqtt_client_stop(void*) { ++stop_calls; return ESP_OK; }
int esp_mqtt_client_disconnect(void*) {
assert(sdk_event_handler != nullptr);
esp_mqtt_event_t event{MQTT_EVENT_DISCONNECTED, 0, false};
@@ -318,7 +333,7 @@ int main() {
class MqttPublishAckTests(unittest.TestCase):
def test_production_publish_events_and_slot_diagnostics(self):
def compile_run(self, main):
compiler = shutil.which("g++")
if not compiler:
self.skipTest("g++ is unavailable")
@@ -331,6 +346,7 @@ class MqttPublishAckTests(unittest.TestCase):
"esp_err_t PsychicMqttClient::connect()",
"esp_err_t PsychicMqttClient::reconnect()",
"void PsychicMqttClient::disconnect()",
"void PsychicMqttClient::softDisconnect(",
"void PsychicMqttClient::forceStop()",
"int PsychicMqttClient::subscribe(",
"int PsychicMqttClient::publish(",
@@ -349,7 +365,7 @@ class MqttPublishAckTests(unittest.TestCase):
)) + "\n".join(method(BRIDGE, signature) for signature in (
"const char* MQTTBridge::tlsErrorStr(",
"void MQTTBridge::formatSlotDiagReply(",
)) + MAIN
)) + main
with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-publish-ack-") as directory:
fixture = Path(directory)
(fixture / "Arduino.h").write_text(ARDUINO_HEADER, encoding="ascii")
@@ -373,6 +389,54 @@ class MqttPublishAckTests(unittest.TestCase):
result = subprocess.run([str(executable)], capture_output=True, text=True)
self.assertEqual(result.returncode, 0, result.stderr)
def test_production_publish_events_and_slot_diagnostics(self):
self.compile_run(MAIN)
def test_production_initialized_getter_tracks_sdk_creation_and_retained_stops(self):
self.compile_run(r'''
int main() {
{
PsychicMqttClient client;
assert(!client.isInitialized() && !client.isStarted());
assert(client.connect() == ESP_ERR_INVALID_STATE);
assert(!client.isInitialized() && !client.isStarted());
client._mqtt_cfg.uri = "mqtt://broker";
client._mqtt_cfg.broker.address.uri = "mqtt://broker";
fail_init = true;
assert(client.connect() == ESP_ERR_NO_MEM);
assert(!client.isInitialized() && !client.isStarted());
fail_init = false;
register_result = ESP_ERR_INVALID_STATE;
assert(client.connect() == ESP_ERR_INVALID_STATE);
assert(!client.isInitialized() && !client.isStarted() && destroy_calls == 1);
register_result = ESP_OK;
start_result = ESP_ERR_NO_MEM;
assert(client.connect() == ESP_ERR_NO_MEM);
assert(client.isInitialized() && !client.isStarted());
const int retained_init_calls = init_calls;
start_result = ESP_OK;
assert(client.connect() == ESP_OK);
assert(client.isInitialized() && client.isStarted());
esp_mqtt_event_t connected{MQTT_EVENT_CONNECTED, 0, false};
sdk_event_handler(sdk_handler_args, "MQTT", connected.event_id, &connected);
client.disconnect();
assert(client.isInitialized() && !client.isStarted() && !client.connected());
assert(stop_calls == 1 && init_calls == retained_init_calls);
assert(client.connect() == ESP_OK);
assert(client.isInitialized() && client.isStarted() && init_calls == retained_init_calls);
sdk_event_handler(sdk_handler_args, "MQTT", connected.event_id, &connected);
client.softDisconnect();
assert(client.isInitialized() && client.isStarted() && !client.connected());
client.forceStop();
assert(client.isInitialized() && !client.isStarted());
assert(init_calls == retained_init_calls && destroy_calls == 1);
}
assert(destroy_calls == 2);
PsychicMqttClient fresh;
assert(!fresh.isInitialized() && !fresh.isStarted());
}
''')
if __name__ == "__main__":
unittest.main()