diff --git a/MQTT_OWNERSHIP.md b/MQTT_OWNERSHIP.md index f6a06ec4..57cd7561 100644 --- a/MQTT_OWNERSHIP.md +++ b/MQTT_OWNERSHIP.md @@ -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. diff --git a/docs/research/firmware_memory_budget.md b/docs/research/firmware_memory_budget.md index 595e934b..e430de07 100644 --- a/docs/research/firmware_memory_budget.md +++ b/docs/research/firmware_memory_budget.md @@ -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. diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.h b/lib/PsychicMqttClient/src/PsychicMqttClient.h index d1110323..f5b29f31 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.h +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.h @@ -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. diff --git a/scripts/check_firmware_ram.py b/scripts/check_firmware_ram.py index 8777d9b3..b2a5cead 100644 --- a/scripts/check_firmware_ram.py +++ b/scripts/check_firmware_ram.py @@ -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, diff --git a/src/helpers/MQTTConnectionAdmission.h b/src/helpers/MQTTConnectionAdmission.h index 7e282af0..e69ed8d3 100644 --- a/src/helpers/MQTTConnectionAdmission.h +++ b/src/helpers/MQTTConnectionAdmission.h @@ -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; } diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index cd3c82f1..564d670e 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -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 } diff --git a/test/test_firmware_ram.py b/test/test_firmware_ram.py index 16a97da2..57fb4a4e 100644 --- a/test/test_firmware_ram.py +++ b/test/test_firmware_ram.py @@ -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 = { diff --git a/test/test_mqtt_connection_admission.py b/test/test_mqtt_connection_admission.py index 177f0363..16db5ebc 100644 --- a/test/test_mqtt_connection_admission.py +++ b/test/test_mqtt_connection_admission.py @@ -16,6 +16,7 @@ POLICY_PREAMBLE = r''' #include #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(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)); diff --git a/test/test_mqtt_handshake_scheduler.py b/test/test_mqtt_handshake_scheduler.py index 0f7736de..f097624a 100644 --- a/test/test_mqtt_handshake_scheduler.py +++ b/test/test_mqtt_handshake_scheduler.py @@ -20,24 +20,43 @@ PREAMBLE = r''' #include #include #include +#include #include +#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 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 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() diff --git a/test/test_mqtt_network_timeout.py b/test/test_mqtt_network_timeout.py index dc26d962..0b24217c 100644 --- a/test/test_mqtt_network_timeout.py +++ b/test/test_mqtt_network_timeout.py @@ -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 _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{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( diff --git a/test/test_mqtt_publish_ack.py b/test/test_mqtt_publish_ack.py index 0ba2626f..ee59e4d7 100644 --- a/test/test_mqtt_publish_ack.py +++ b/test/test_mqtt_publish_ack.py @@ -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 #include #include +#include // 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 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(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(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()