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 5a031f7674
commit 937f8610c7
11 files changed
+556 -63

No files matched your search

+27 -1
View File
@@ -144,6 +144,30 @@ max(8 s, 10 s) * enabled_slots` (35 seconds for two slots). A clean
acknowledgment still returns immediately. This deadline change is covered by
synthetic tests; no new physical teardown measurement is claimed.
Connection admission reads SDK initialization and task-start state on the
bridge owner task. `PsychicMqttClient::isInitialized()` reports whether its SDK
handle still owns configuration, transport objects and MQTT buffers;
`isStarted()` separately reports whether its SDK task is running. Callbacks
change connection state but do not assign or clear the SDK handle.
`MQTTConnectionAdmission::ClientMemoryState` uses those distinct lifetimes:
uninitialized clients require 32 KiB free DMA heap and an 8 KiB block,
initialized stopped clients require 24 KiB and an 8 KiB block, and started
clients require 16 KiB and a 4 KiB block. The stopped-client allowance reserves
8 KiB for recreating the explicit 6,144-byte SDK task stack plus TCB and
allocator overhead while retaining the 16 KiB AES reserve. WiFi-loss shutdown
still stops the SDK task and retains its handle; this admission distinction
permits its restart without charging initialization buffers twice. A failed
initialization that discards the SDK handle continues to use the full cold
allowance. Internal-only TLS keeps its additional 60 KiB/17,408-byte thresholds.
All ESP32 clients use 896-byte MQTT buffers from their first initialization,
including non-JWT slots on internal-only boards. The SDK does not resize those
buffers when applying configuration, and the wrapper reassembly buffer also
keeps its initial capacity. Uniform sizing therefore permits later JWT
reconfiguration while retaining the client, its callbacks and the slot's token
storage. Explicit reconfiguration and WiFi recovery keep the existing client
lifetime; neither introduces deletion while Core 1 diagnostics may read it.
## Remaining target primitives
- **Task notifications or a command queue** for one-way lifecycle / reconfigure
@@ -181,7 +205,9 @@ Phase 5 as a pure state machine plus a narrow injected `Ops` seam
### Hardware characterization result
WSS teardown was measured as roughly 5-6 seconds per slot and is sequential.
Production now sets the coordinator timeout to `5 s + 8 s * enabled_slots`.
That characterization used the coordinator timeout `5 s + 8 s * enabled_slots`.
The current deadline also includes the SDK network and blocked-publish budgets
described above.
Representative non-PSRAM and PSRAM start/stop matrices found no downward heap
or largest-block trend; multi-day soak, task-stack high-water, and exhaustive
callback ordering remain outside that run.
+18 -1
View File
@@ -91,7 +91,24 @@ and an 8 KiB contiguous DMA block. This includes a separate 16 KiB allowance
for its SDK task stack/TCB, receive/transmit/reassembly buffers and transport
or configuration objects, alongside the 16 KiB AES reserve. Those ordinary
SDK allocations occur after admission and do not use the TLS allocation guard.
A retained SDK handle with a stopped task receives the same cold allowance.
A retained initialized SDK handle with a stopped task requires 24 KiB free and
an 8 KiB contiguous DMA block. Its separate 8 KiB restart allowance covers the
pinned 6 KiB task stack plus TCB and allocator overhead. SDK stop retains its
receive/transmit buffers, configuration, transport objects and wrapper
reassembly buffer, so restarting that handle does not need to reserve their
first-initialization allocation again. A wrapper whose SDK initialization has
not completed still receives the full cold allowance. This distinction also
applies after WiFi loss stops the SDK tasks but keeps their handles.
All ESP32 clients use 896-byte receive, transmit and wrapper reassembly buffers
from first initialization. Applying configuration cannot grow the retained SDK
buffers or wrapper reassembly capacity, so uniform sizing permits later JWT
reconfiguration without replacing the client. This adds 1,152 bytes to a
non-JWT client's three buffers on an internal-only board compared with 512-byte
buffers. The startup allowance already uses 896-byte receive/transmit buffers,
and cold admission covers the ordinary allocations. WiFi recovery and explicit
reconfiguration continue to retain clients.
A warm reconnect keeps its running task and buffers; it requires at least
16 KiB free and a 4 KiB contiguous DMA block. The per-client SDK stack is
independent of `MQTT_TASK_STACK_SIZE`, which sizes the existing bridge worker.
@@ -376,6 +376,10 @@ public:
/** True after a successful start and before a stop. */
bool isStarted() const { return _started; }
/** True while SDK buffers/configuration remain allocated, including after
* stop. Read on the same owner task as connect/reconnect/disconnect. */
bool isInitialized() const { return _client != nullptr; }
/**
* @brief Forcefully stops the MQTT client and disconnects from the server.
* This does not trigger the onDisconnect callbacks.
+3
View File
@@ -83,6 +83,9 @@ def mqtt_requirements(defines):
"cold_start_allocation_allowance_bytes": 16384,
"cold_start_dma_admission_free_bytes": 32768,
"cold_start_dma_admission_contiguous_bytes": 8192,
"restart_allocation_allowance_bytes": 8192,
"restart_dma_admission_free_bytes": 24576,
"restart_dma_admission_contiguous_bytes": 8192,
"dma_admission_free_bytes": 16384,
"dma_admission_contiguous_bytes": 4096,
"tls_internal_admission_free_bytes": 61440,
+30 -4
View File
@@ -20,14 +20,40 @@ static const size_t kClientTaskStackBytes = 6144;
static const int kClientTaskPriority = 5;
static const size_t kColdStartAllowanceBytes = 16384;
static const size_t kColdStartLargestBytes = 8192;
// SDK stop retains buffers/configuration but releases its task. Restart only
// needs the pinned 6 KiB stack plus TCB/allocator headroom, not another SDK init.
static const size_t kRestartAllowanceBytes = 8192;
static const size_t kRestartLargestBytes = 8192;
static const size_t kTlsInternalFreeBytes = 61440;
static const size_t kTlsRecordAllocationBytes = 17408;
static inline bool canStart(bool psram, bool tls, bool cold_start,
enum class ClientMemoryState : uint8_t {
Uninitialized,
Stopped,
Started,
};
static inline bool canStart(bool psram, bool tls, ClientMemoryState state,
size_t dma_free, size_t dma_largest) {
const size_t free_required = kDmaReserveBytes +
(cold_start ? kColdStartAllowanceBytes : 0);
const size_t largest_required = cold_start ? kColdStartLargestBytes : kDmaLargestBytes;
size_t allocation_allowance;
size_t largest_required;
switch (state) {
case ClientMemoryState::Uninitialized:
allocation_allowance = kColdStartAllowanceBytes;
largest_required = kColdStartLargestBytes;
break;
case ClientMemoryState::Stopped:
allocation_allowance = kRestartAllowanceBytes;
largest_required = kRestartLargestBytes;
break;
case ClientMemoryState::Started:
allocation_allowance = 0;
largest_required = kDmaLargestBytes;
break;
default:
return false;
}
const size_t free_required = kDmaReserveBytes + allocation_allowance;
if (dma_free < free_required || dma_largest < largest_required) {
return false;
}
+18 -22
View File
@@ -2088,14 +2088,16 @@ bool MQTTBridge::canStartSlotConnection(int index) const {
const char* uri = slot.preset ? slot.preset->server_url
: (slot.broker_uri[0] ? slot.broker_uri : slot.host);
const bool tls = MQTTConnectionAdmission::requiresTls(uri, slot.port);
// A never-created or stopped SDK client still allocates a task and ordinary
// SDK buffers after this check. Retained handles with a stopped task are
// conservatively charged the same allowance as first starts. A started
// reconnect keeps that task/buffer memory and needs only handshake headroom.
const bool cold_start = slot.client == nullptr || !slot.client->isStarted();
// SDK stop retains its buffers/configuration, so a retained stopped client
// only needs a new task. First initialization also allocates SDK objects and
// buffers; a started reconnect keeps both. These getters run on their owner.
using MQTTConnectionAdmission::ClientMemoryState;
const ClientMemoryState memory_state = slot.client == nullptr || !slot.client->isInitialized()
? ClientMemoryState::Uninitialized
: (slot.client->isStarted() ? ClientMemoryState::Started : ClientMemoryState::Stopped);
const uint32_t caps = MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA;
return MQTTConnectionAdmission::canStart(
ESP.getPsramSize() > 0, tls, cold_start, heap_caps_get_free_size(caps),
ESP.getPsramSize() > 0, tls, memory_state, heap_caps_get_free_size(caps),
heap_caps_get_largest_free_block(caps));
#else
return true;
@@ -2833,7 +2835,7 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload
// even light packet load the outbox pins at its cap and drops ~20-30%. A synchronous
// write bypasses that drain ceiling entirely and does not store in the outbox. It can
// block the (Core-0, prio-1) MQTT task on a stalled socket, but only up to
// network_timeout_ms (lowered in optimizeMqttClientConfig); mesh RX (Core 1) and the
// network_timeout_ms (configured in optimizeMqttClientConfig); mesh RX (Core 1) and the
// WiFi/TCP stack (higher-prio system tasks) are unaffected, and a failed write flips
// the slot to disconnected so subsequent packets skip it.
// - QoS 1 (low-rate retained status): async, so it keeps the durable outbox + retransmit.
@@ -5108,15 +5110,12 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_
// preserving at-least-once delivery while capping duplicates at one.
client->setMessageRetransmitTimeout(15000);
// Buffer sizing: 896 is the minimum safe size for JWT clients (CONNECT + 768-byte JWT).
// On PSRAM boards, use a uniform size to reduce fragmentation from mixed allocations.
// On non-PSRAM boards, use smaller buffers for non-JWT slots to reduce heap usage and
// leave smaller holes during teardown/recreate cycles.
#if defined(BOARD_HAS_PSRAM)
// SDK buffers and wrapper reassembly are sized at first initialization; changing
// configuration cannot grow them on a retained client. Use the JWT-safe size
// from the first start so explicit non-JWT -> JWT reconfiguration stays valid
// without destroying a client that another core may be reading for diagnostics.
(void)needs_large_buffer;
static const int MQTT_CLIENT_BUFFER_SIZE = 896;
#else
const int MQTT_CLIENT_BUFFER_SIZE = needs_large_buffer ? 896 : 512;
#endif
client->setBufferSize(MQTT_CLIENT_BUFFER_SIZE);
@@ -5140,13 +5139,10 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_
esp_mqtt_client_config_t* config = client->getMqttConfig();
if (config) {
#if defined(ESP_IDF_VERSION_MAJOR) && ESP_IDF_VERSION_MAJOR >= 5
// Keep the output buffer (used to build the CONNECT/PUBLISH frames) in lockstep
// with the input buffer. setBufferSize() above only sets buffer.size, so out_size
// must be set here. The previous conditional only ever shrank out_size or set it
// from 0 - when a slot was reconfigured from a small non-JWT buffer (512) up to
// the JWT buffer (896), out_size stayed at 512 and the JWT CONNECT frame (username
// + ~537-768B token) overflowed it, producing esp-mqtt "Connect message cannot be
// created". Always matching MQTT_CLIENT_BUFFER_SIZE fixes the grow case.
// Match CONNECT/PUBLISH output capacity to input capacity at initialization.
// setBufferSize() only sets buffer.size on IDF5; set out_size explicitly.
// Updating config alone cannot resize an initialized SDK output buffer,
// so the uniform JWT-safe size above must be used from the first start.
config->buffer.out_size = MQTT_CLIENT_BUFFER_SIZE;
#endif
}
+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()