From 70052c5e5b722b1661ab53a769ee2df2bfbbd9b2 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Sat, 10 Oct 2026 14:43:57 -0700 Subject: [PATCH] Preserve MQTT TLS connection time and internal DMA headroom --- .github/workflows/run-unit-tests.yml | 4 + MQTT_OWNERSHIP.md | 7 + docs/research/firmware_memory_budget.md | 53 +++- docs/test-results/mqtt_tls_v4_regression.md | 82 +++++ lib/PsychicMqttClient/src/PsychicMqttClient.h | 1 + scripts/check_firmware_ram.py | 67 +++- src/helpers/MQTTConnectionAdmission.h | 57 ++++ src/helpers/MQTTConnectionPolicy.h | 4 + src/helpers/bridges/MQTTBridge.cpp | 231 +++++++++----- src/helpers/bridges/MQTTBridge.h | 14 +- test/test_firmware_ram.py | 84 ++++- test/test_mqtt_connection_admission.py | 296 ++++++++++++++++++ test/test_mqtt_diagnostics.py | 165 ++++++++++ .../test_mqtt_error_labels.cpp | 6 + test/test_mqtt_handshake_scheduler.py | 286 +++++++++++++++++ test/test_mqtt_network_timeout.py | 165 ++++++++++ test/test_mqtt_transport_results.py | 58 ++-- 17 files changed, 1461 insertions(+), 119 deletions(-) create mode 100644 docs/test-results/mqtt_tls_v4_regression.md create mode 100644 src/helpers/MQTTConnectionAdmission.h create mode 100644 test/test_mqtt_connection_admission.py create mode 100644 test/test_mqtt_diagnostics.py create mode 100644 test/test_mqtt_handshake_scheduler.py create mode 100644 test/test_mqtt_network_timeout.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 34fda74b4..6218469b2 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -192,6 +192,10 @@ jobs: run: | python3 -B test/test_mqtt_transport_results.py -v python3 -B test/test_mqtt_tls_security.py -v + python3 -B test/test_mqtt_network_timeout.py -v + python3 -B test/test_mqtt_connection_admission.py -v + python3 -B test/test_mqtt_handshake_scheduler.py -v + python3 -B test/test_mqtt_diagnostics.py -v - name: Verify DNS lease restoration and bounded diagnostics run: python3 -B test/test_network_dns.py -v diff --git a/MQTT_OWNERSHIP.md b/MQTT_OWNERSHIP.md index 1cbc8fb36..f6a06ec41 100644 --- a/MQTT_OWNERSHIP.md +++ b/MQTT_OWNERSHIP.md @@ -137,6 +137,13 @@ acknowledged cleanly in about 11.7 seconds within its 21-second budget. The residual Section 1/Section 2 instance-pointer reads remain deferred, so those consumers can still touch live or torn-down bridge state as tracked above. +The 2026-10-10 connection-timeout fix restores the SDK network budget to +10 seconds. The stop deadline now also allows the owner to finish one blocked +publish and each client to finish a network operation: `5 s + 10 s + +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. + ## Remaining target primitives - **Task notifications or a command queue** for one-way lifecycle / reconfigure diff --git a/docs/research/firmware_memory_budget.md b/docs/research/firmware_memory_budget.md index cc6a00cc6..9af18d9c2 100644 --- a/docs/research/firmware_memory_budget.md +++ b/docs/research/firmware_memory_budget.md @@ -13,6 +13,8 @@ The check reserves room for enabled runtime allocations as well as static data. A firmware image fitting its board's reported RAM total is insufficient: the display, packet pool, USB, Bluetooth workers and WiFi can allocate after startup. The T096 Full 1.17.1.5 report exposed this distinction. +MQTT uses an explicit startup minimum plus runtime connection admission. A +passing build report does not qualify its maximum TLS/WSS/outbox workload. ## What is counted @@ -38,7 +40,9 @@ keeps runtime UART enablement in the budget beside MQTT and ESP-NOW. T-LoRa V2.1-1.6 cannot fit all three transports: the qualified linked triple image had 190,488 available internal bytes against 196,984 required bytes under the earlier allocation estimates. The corrected equivalent requirement -is 193,400 bytes, still 2,912 bytes above capacity. +was 193,400 bytes with the earlier flat MQTT allowance, still 2,912 bytes above +capacity. Its configured 16 KiB MQTT worker stack adds another 8 KiB under the +current startup policy; these figures describe historical linked checks. Its ordinary release retains separate UART/ESP-NOW and MQTT/ESP-NOW Full images; routing capacities remain unchanged. A 160x80 ST7735 @@ -50,6 +54,53 @@ Headless and OLED devices use their own smaller totals. The JSON lists each component and checks the largest available region against the largest planned single allocation. +### MQTT startup minimum and runtime qualification + +The `mqtt_startup_allowance` component keeps a 24,576-byte internal startup +minimum for the default 8,192-byte MQTTBridge worker stack and two active +clients. Each SDK client task defaults to 6,144 bytes and its maximum configured +receive/transmit buffers total 1,792 bytes. These stacks and buffers total +24,064 bytes for two clients, leaving 512 bytes within the existing allowance. +This is a minimum allowance, not a complete allocation bound for the clients, +transport objects, TLS records, certificate processing, WebSockets or outboxes. +The independent 16 KiB transient margin remains in the policy. + +`MQTT_TASK_STACK_SIZE` above 8,192 bytes raises the startup requirement by its +actual increase and raises the required contiguous allocation when needed. +Smaller stacks and fewer runtime slots do not lower the existing allowance. +The default runtime slot array has six entries on PSRAM builds and three on +internal-only builds. The bridge can expose five active connections when PSRAM +is present and two otherwise; `MQTT_RUNTIME_SLOT_COUNT` can restrict that +capacity. A PSRAM board with unavailable PSRAM uses the two-connection cap. + +The report's separate `mqtt.maximum_slot_stack_buffer_allowance_bytes` records +48,384 bytes for five exposed slots or 24,576 bytes for two, with larger worker +stacks added to either. The five-slot figure extends the minimum by 7,936 bytes +for each additional client task and buffer pair. It is not added to the startup +gate: extra clients allocate lazily, and every connection attempt must pass +runtime admission. It still excludes the maximum TLS/WSS/outbox workload. +`maximum_runtime_load_qualified` and `physical_validation_performed` are false +in these build reports. A `passed` startup check must not be presented as +qualification of five simultaneous PSRAM connections or two internal TLS +connections. + +The `mqtt` report also names the production runtime admission thresholds. +Each attempt requires at least 16 KiB of DMA-capable internal free heap and a +4 KiB contiguous DMA block. A TLS attempt without PSRAM requires at least +60 KiB free and a 17,408-byte contiguous block. PSRAM allocation fallback must +leave the 16 KiB DMA reserve available. These thresholds describe free heap at +the time of admission; they are not additional linked capacity and the TLS +contiguous threshold is not a startup allocation requirement. Available linked +internal heap cannot establish how much DMA-capable heap remains or how it is +fragmented after boot. Only runtime observations can establish that. + +Maximum-load qualification still needs physical boot and sustained reconnect, +publish and configuration-change tests with the selected number of brokers, +transport schemes, certificates and enabled UART/browser/OTA features. Record +DMA free heap, largest DMA block and allocation failures throughout that +workload. Host admission tests and passing ELF checks provide engineering +evidence without claiming that hardware validation has been performed. + The W12 Full MQTT profile remains unqualified: the current linked image has 182,768 internal bytes available against 198,432 required, including the unchanged 16 KiB transient margin. Its standard repeater compiles and passes diff --git a/docs/test-results/mqtt_tls_v4_regression.md b/docs/test-results/mqtt_tls_v4_regression.md new file mode 100644 index 000000000..675479eaf --- /dev/null +++ b/docs/test-results/mqtt_tls_v4_regression.md @@ -0,0 +1,82 @@ + +

View this page on MeshCore Docs.

+ + +# ESP32 MQTT TLS connection regression + +## Report and diagnosis + +The Heltec V4 report contains two separate failures: + +- `0x8006`: ESP-TLS connection timeout opening the WSS brokers + `mqtt.waev.app:443` and `mqtt-v1.cascadiamesh.org:443`. +- `esp-aes: Failed to allocate memory`, followed by TLS write failure + `0x8018` and resend errors. + +The firmware configured a 2,500 ms network timeout to bound synchronous +publishes. ESP-MQTT also uses that setting for connection establishment, +including TLS and the WebSocket upgrade. It was not a write-only setting. + +The shipped ESP32-S3 AES driver allocates DMA-capable internal bounce buffers +for unaligned external-memory records, up to 1,600 bytes per buffer. Ample +PSRAM or an aggregate free-heap reading therefore does not establish that an +AES operation can succeed. The supplied log identifies an allocation failure +in this driver; the exact V4 heap history has not been captured in this run. + +The shipped Arduino 2.0.17/IDF 4.4.7 TLS allocator was checked in its archives: +TLS allocations use the configurable `mbedtls_calloc` function, so the +existing PSRAM redirect takes effect. Retaining a PsychicMqttClient does not +retain its TLS records: ESP-IDF frees/recreates the TLS session on reconnect. + +References: [ESP-IDF 4.4.7 AES implementation](https://github.com/espressif/esp-idf/blob/v4.4.7/components/mbedtls/port/aes/dma/esp_aes.c) +and [ESP-TLS errors](https://github.com/espressif/esp-idf/blob/v4.4.7/components/esp-tls/esp_tls_errors.h). + +## Firmware changes + +These changes apply to the shared ESP32 MQTT bridge, including repeater, +room-server and companion images that enable it. + +- Restore the SDK's 10,000 ms network timeout. A stalled synchronous publish + can now wait up to that same timeout; the radio loop runs separately. + Cooperative shutdown also includes that deadline in its stop budget. +- Schedule initial connections through the normal maintenance loop. Check + the actual pending attempt across SDK tasks, rather than assuming a + five-second startup delay or 15-second reconnect gap finished a handshake. +- Defer a new connection if DMA-capable internal memory has less than + 16 KiB free or a largest block below 4 KiB. TLS without working PSRAM + additionally needs 60 KiB free and a 17 KiB largest block. These are + conservative admission bounds, not a promise that future allocations succeed. +- Prefer PSRAM for TLS allocations and reject an internal fallback that + would consume the 16 KiB DMA reserve. Check allocation-size overflow. +- Keep renewal able to close an old TLS connection before checking memory + for its replacement. Stop an in-flight SDK client on reconfiguration. +- Require byte-addressable internal memory for string/JSON/queue buffer + fallbacks. The internal-memory capability alone also admits instruction RAM + that only supports aligned 32-bit access. +- Add DMA free/largest readings to `get mqtt.stats` and diagnostic logs. + Correct the operator-facing TLS error labels against SDK constants. + +Existing slot limits, certificate verification, authentication and backoff +remain in effect. Memory pressure defers additional slots instead of consuming +memory needed by existing connections and the radio. + +## Validation and limits + +The regression tests execute the production optimizer, adapter setters, +connection admission, TLS allocator fallback, scheduler and diagnostics with +controlled SDK calls and heap readings. They cover IDF 4/5 config layouts, +PSRAM and internal-only modes, fragmented heap, exhausted PSRAM, delayed +connection callbacks, retry timing and bounded replies. They do not execute +an ESP32 AES peripheral or qualify physical WiFi behavior. + +All 1,794 native tests and the 31 memory-budget tests passed. The focused +MQTT tests are also connected to the GitHub unit-test workflow. + +The VM reached both reported broker TLS endpoints with certificate +verification enabled. That is an endpoint check, not an ESP32 MQTT test. + +MercerWoodMesh reported offline and its SSH connection timed out during this +run. No live V4 flash, MQTT publication or soak result is claimed. The next +hardware check must exercise the reported WSS presets together, verify rising +publish counters and DMA headroom, then include a broker/WiFi reconnection +and power-saving idle run with no reset or AES allocation error. diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.h b/lib/PsychicMqttClient/src/PsychicMqttClient.h index c00573da0..f5b174b82 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.h +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.h @@ -128,6 +128,7 @@ public: /** * @brief Sets the network operation timeout in milliseconds. esp-mqtt aborts a + * connection attempt (including TLS/WebSocket setup) or a * network read/write (including a synchronous publish's socket write) if it does * not complete within this window. A lower value bounds how long a synchronous * QoS0 publish can block on a stalled/half-open socket before failing and letting diff --git a/scripts/check_firmware_ram.py b/scripts/check_firmware_ram.py index bc4586a05..41cf6e41b 100644 --- a/scripts/check_firmware_ram.py +++ b/scripts/check_firmware_ram.py @@ -1,10 +1,12 @@ #!/usr/bin/env python3 -"""Gate firmware builds on internal runtime RAM, including enabled startup features. +"""Gate firmware builds on internal RAM capacity and startup allowances. Uses real linker heap bounds on ARM and the linked ESP-IDF allocator tables on ESP32. These are capacity checks before dynamic allocation, not hardware soak -results. PSRAM, instruction-only RAM and reserved bootloader arenas never count -as internal heap. See docs/research/firmware_memory_budget.md for the budget policy. +results. MQTT uses a startup minimum with separate runtime admission; a passing +build does not qualify its maximum TLS/WSS/outbox load. PSRAM, instruction-only +RAM and reserved bootloader arenas never count as internal heap. See +docs/research/firmware_memory_budget.md for the budget policy. """ from __future__ import annotations @@ -32,6 +34,14 @@ DISPLAY_HEAP = { "GxEPDDisplay": 0, "E213Display": 8192, "E290Display": 8192, } +# Match MQTTBridge's default worker and the pinned ESP-MQTT client task/buffers. +# The legacy 24 KiB allowance covers two clients with 512 bytes beyond the known +# stacks/buffers. It is a startup minimum, not a bound on TLS/WSS/outbox memory. +MQTT_STARTUP_ALLOWANCE = 24576 +MQTT_WORKER_STACK_BYTES = 8192 +MQTT_CLIENT_STACK_BYTES = 6144 +MQTT_CLIENT_BUFFER_BYTES = 2 * 896 + def integer(defines, name, default): value = str(defines.get(name, default)).strip().strip("()") @@ -45,6 +55,41 @@ def integer(defines, name, default): return result +def mqtt_requirements(defines): + psram = "BOARD_HAS_PSRAM" in defines + slots = integer(defines, "MQTT_RUNTIME_SLOT_COUNT", 6 if psram else 3) + if not 1 <= slots <= 6: + raise ValueError("MQTT_RUNTIME_SLOT_COUNT must be between 1 and 6") + worker_stack = integer(defines, "MQTT_TASK_STACK_SIZE", MQTT_WORKER_STACK_BYTES) + if worker_stack == 0: + raise ValueError("MQTT_TASK_STACK_SIZE must be positive") + # Reducing slots or stack size cannot lower the existing allowance. Extra + # slots are lazy and each attempt must pass the production admission guard. + startup = MQTT_STARTUP_ALLOWANCE + max(0, worker_stack - MQTT_WORKER_STACK_BYTES) + active = min(slots, 5 if psram else 2) + potential = startup + max(0, active - 2) * (MQTT_CLIENT_STACK_BYTES + MQTT_CLIENT_BUFFER_BYTES) + return { + "budget_scope": "startup minimum with runtime connection admission", + "startup_allowance_bytes": startup, + "startup_active_slot_allowance": min(slots, 2), + "runtime_slot_count": slots, + "maximum_active_slots": active, + "maximum_active_slots_without_psram": min(slots, 2), + "worker_stack_bytes": worker_stack, + "client_task_stack_bytes": MQTT_CLIENT_STACK_BYTES, + "client_rx_tx_buffer_bytes": MQTT_CLIENT_BUFFER_BYTES, + "maximum_slot_stack_buffer_allowance_bytes": potential, + "dma_admission_free_bytes": 16384, + "dma_admission_contiguous_bytes": 4096, + "tls_internal_admission_free_bytes": 61440, + "tls_internal_admission_contiguous_bytes": 17408, + "maximum_runtime_load_qualified": False, + "physical_validation_performed": False, + "limitations": "Maximum TLS/WSS/outbox workload is not covered by the startup allowance; " + "runtime free DMA heap and fragmentation must satisfy connection admission.", + } + + def requirements(platform, defines, target): if platform not in SUPPORTED: raise ValueError(f"no memory policy for {platform}") @@ -64,6 +109,7 @@ def requirements(platform, defines, target): else: raise ValueError(f"add a runtime allocation budget for display {display!r}") parts = {} + mqtt = None combined_rak_ethernet = "RAK4631_COMBINED_ETHERNET" in defines if combined_rak_ethernet and (platform != "NRF52_PLATFORM" or not {"RAK_4631", "ETHERNET_ENABLED", "OTA_RAK_AUTO_STORE"}.issubset(defines)): @@ -93,7 +139,8 @@ def requirements(platform, defines, target): ble = 32768 if "BLE_PIN_CODE" in defines else 0 parts["wireless_stacks"] = max(wifi, ble) if "COMPANION_EXCLUSIVE_WIFI_BLE" in defines else wifi + ble if "WITH_MQTT_BRIDGE" in defines: - parts["mqtt_connections_buffers"] = 24576 + mqtt = mqtt_requirements(defines) + parts["mqtt_startup_allowance"] = mqtt["startup_allowance_bytes"] if "WITH_RS232_BRIDGE" in defines: # The UART bridge is allocated on enable, alongside WiFi/MQTT. # RS232Bridge.cpp bounds its object to 2304 bytes. Reserve 2560 @@ -240,8 +287,13 @@ def requirements(platform, defines, target): largest = max(largest, parts.get(name, 0)) if "room_browser_mailbox" in parts: largest = max(largest, 8192, parts["room_browser_mailbox"]) - return {"required_heap_bytes": required, "required_contiguous_bytes": largest, - "components": parts, "display": display, "full_companion": full} + if mqtt: + largest = max(largest, mqtt["worker_stack_bytes"], mqtt["client_task_stack_bytes"]) + result = {"required_heap_bytes": required, "required_contiguous_bytes": largest, + "components": parts, "display": display, "full_companion": full} + if mqtt: + result["mqtt"] = mqtt + return result def subtract_regions(regions, reserved): @@ -353,6 +405,9 @@ def check_firmware(elf_path, platform, mcu, defines, target, output=None): Path(output).write_text(json.dumps(report, indent=2) + "\n") print(f"Runtime RAM: {available:,} internal bytes available; {policy['required_heap_bytes']:,} required; " f"largest region {largest:,}; {'PASS' if passed else 'FAIL'}") + if "mqtt" in policy: + print("MQTT RAM scope: startup minimum with runtime admission; " + "maximum TLS/WSS/outbox load is not qualified by this build check.") if not passed: print("Runtime RAM check failed: insufficient heap for enabled features. " "Share cold buffers or reduce allocations before publishing this build.", file=sys.stderr) diff --git a/src/helpers/MQTTConnectionAdmission.h b/src/helpers/MQTTConnectionAdmission.h new file mode 100644 index 000000000..0625d2296 --- /dev/null +++ b/src/helpers/MQTTConnectionAdmission.h @@ -0,0 +1,57 @@ +#pragma once + +#include +#include +#include + +// Connection admission uses DMA-capable internal heap rather than total heap: +// PSRAM-backed TLS still needs internal AES bounce buffers and WiFi allocations. +// These are headroom bounds, not a guarantee that a later allocation succeeds. +namespace MQTTConnectionAdmission { + +static const size_t kDmaReserveBytes = 16384; +static const size_t kDmaLargestBytes = 4096; +static const size_t kTlsInternalFreeBytes = 61440; +static const size_t kTlsRecordAllocationBytes = 17408; + +static inline bool canStart(bool psram, bool tls, size_t dma_free, + size_t dma_largest) { + if (dma_free < kDmaReserveBytes || dma_largest < kDmaLargestBytes) { + return false; + } + return !tls || psram || + (dma_free >= kTlsInternalFreeBytes && + dma_largest >= kTlsRecordAllocationBytes); +} + +// Charge a PSRAM allocation's internal fallback against the DMA reserve even +// when the internal allocator might find a block with fewer capabilities. +static inline bool canFallback(size_t requested, size_t dma_free, + size_t dma_largest) { + return requested > 0 && requested <= dma_largest && + dma_free >= kDmaReserveBytes && + requested <= dma_free - kDmaReserveBytes; +} + +static inline bool checkedCallocSize(size_t count, size_t size, size_t& total) { + total = 0; + if (count == 0 || size == 0 || count > SIZE_MAX / size) return false; + total = count * size; + return true; +} + +// An explicit scheme controls the transport even on an unusual port. Bare +// custom hosts match setupSlot's port-based URI construction. +static inline bool requiresTls(const char* uri, uint16_t port) { + if (uri != nullptr) { + if (strncmp(uri, "mqtts://", 8) == 0 || strncmp(uri, "wss://", 6) == 0) { + return true; + } + if (strncmp(uri, "mqtt://", 7) == 0 || strncmp(uri, "ws://", 5) == 0) { + return false; + } + } + return port == 443 || port == 8883; +} + +} // namespace MQTTConnectionAdmission diff --git a/src/helpers/MQTTConnectionPolicy.h b/src/helpers/MQTTConnectionPolicy.h index b77c47b8b..49f128e33 100644 --- a/src/helpers/MQTTConnectionPolicy.h +++ b/src/helpers/MQTTConnectionPolicy.h @@ -9,6 +9,10 @@ // deterministic clock. namespace MQTTConnectionPolicy { +// esp-mqtt uses one timeout for TLS/WebSocket connection setup, CONNACK, and +// network reads/writes. Keep the SDK's ten-second connection budget; shortening +// it to bound synchronous publishes also cuts off slow TLS handshakes. +static const uint32_t kNetworkTimeoutMs = 10000UL; static const uint32_t kReconnectGuardMs = 15000UL; static const uint32_t kStableResetMs = 120000UL; static const uint32_t kCircuitBreakerProbeMs = 1800000UL; diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index c86e93878..6db2a79ea 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -1,9 +1,11 @@ #define MQTT_PRESETS_IMPLEMENTATION #include "MQTTBridge.h" +#include "MQTTErrorLabels.h" #ifdef ESP_PLATFORM #include #endif #include "../MQTTConnectionPolicy.h" +#include "../MQTTConnectionAdmission.h" #include "../MQTTMessageBuilder.h" #include "../MQTTPacketQueuePolicy.h" #include "../MQTTReplyFormat.h" @@ -30,6 +32,7 @@ #include #include #include +#include #include #include #include @@ -37,6 +40,22 @@ #include #include #include + +// Keep the host-tested labels tied to the SDK used by the firmware. In +// particular, 0x8008 is a clean TCP close and 0x8018 is an SSL write failure. +static_assert(MQTTErrorLabels::kTlsCannotResolveHostname == ESP_ERR_ESP_TLS_CANNOT_RESOLVE_HOSTNAME, "TLS DNS code changed"); +static_assert(MQTTErrorLabels::kTlsCannotCreateSocket == ESP_ERR_ESP_TLS_CANNOT_CREATE_SOCKET, "TLS socket code changed"); +static_assert(MQTTErrorLabels::kTlsUnsupportedProtoFamily == ESP_ERR_ESP_TLS_UNSUPPORTED_PROTOCOL_FAMILY, "TLS protocol code changed"); +static_assert(MQTTErrorLabels::kTlsFailedConnectToHost == ESP_ERR_ESP_TLS_FAILED_CONNECT_TO_HOST, "TLS connect code changed"); +static_assert(MQTTErrorLabels::kTlsSocketSetoptFailed == ESP_ERR_ESP_TLS_SOCKET_SETOPT_FAILED, "TLS socket option code changed"); +static_assert(MQTTErrorLabels::kTlsConnectionTimeout == ESP_ERR_ESP_TLS_CONNECTION_TIMEOUT, "TLS timeout code changed"); +static_assert(MQTTErrorLabels::kTlsTcpClosedFin == ESP_ERR_ESP_TLS_TCP_CLOSED_FIN, "TLS close code changed"); +static_assert(MQTTErrorLabels::kTlsMbedtlsCertPartlyOk == ESP_ERR_MBEDTLS_CERT_PARTLY_OK, "TLS partial certificate code changed"); +static_assert(MQTTErrorLabels::kTlsMbedtlsSetHostname == ESP_ERR_MBEDTLS_SSL_SET_HOSTNAME_FAILED, "TLS hostname code changed"); +static_assert(MQTTErrorLabels::kTlsMbedtlsX509ParseFailed == ESP_ERR_MBEDTLS_X509_CRT_PARSE_FAILED, "TLS certificate parse code changed"); +static_assert(MQTTErrorLabels::kTlsMbedtlsSslSetupFailed == ESP_ERR_MBEDTLS_SSL_SETUP_FAILED, "TLS setup code changed"); +static_assert(MQTTErrorLabels::kTlsMbedtlsSslWriteFailed == ESP_ERR_MBEDTLS_SSL_WRITE_FAILED, "TLS write code changed"); +static_assert(MQTTErrorLabels::kTlsMbedtlsHandshakeFailed == ESP_ERR_MBEDTLS_SSL_HANDSHAKE_FAILED, "TLS handshake code changed"); #endif // Effective MQTT origin: empty mqtt_origin follows node_name; otherwise mqtt_origin override (quotes stripped). @@ -248,7 +267,7 @@ static void* psram_malloc(size_t size) { #if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) void* p = heap_caps_malloc(size, MALLOC_CAP_SPIRAM); if (p != nullptr) return p; - p = heap_caps_malloc(size, MALLOC_CAP_INTERNAL); + p = heap_caps_malloc(size, MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT); return p; #else return malloc(size); @@ -256,11 +275,20 @@ static void* psram_malloc(size_t size) { } static void* psram_calloc(size_t n, size_t size) { - if (n == 0 || size == 0) return nullptr; + size_t bytes = 0; + if (!MQTTConnectionAdmission::checkedCallocSize(n, size, bytes)) return nullptr; #if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) void* p = heap_caps_calloc(n, size, MALLOC_CAP_SPIRAM); if (p != nullptr) return p; - return heap_caps_calloc(n, size, MALLOC_CAP_INTERNAL); + // TLS can recover after an allocation failure; starving WiFi or the AES DMA + // buffers can instead break every connected broker. Preserve that reserve + // when PSRAM is unavailable or full, rather than silently draining DRAM. + const uint32_t caps = MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA; + if (!MQTTConnectionAdmission::canFallback(bytes, + heap_caps_get_free_size(caps), heap_caps_get_largest_free_block(caps))) { + return nullptr; + } + return heap_caps_calloc(n, size, caps); #else return calloc(n, size); #endif @@ -285,7 +313,7 @@ static void* psram_realloc(void* ptr, size_t new_size) { if (p != nullptr) return p; // A block that fell back to internal DRAM on allocation (PSRAM exhausted) cannot // be grown in PSRAM; retry there rather than reporting failure. - return heap_caps_realloc(ptr, new_size, MALLOC_CAP_INTERNAL); + return heap_caps_realloc(ptr, new_size, MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT); #else return realloc(ptr, new_size); #endif @@ -460,8 +488,16 @@ void MQTTBridge::formatMqttStatsReply(char* buf, size_t bufsize) { // drops=/: outbox-cap drops vs. memory-pressure skips. int pos = 0; - replyAppendf(buf, bufsize, &pos, "> Free=%d Max=%d q:%d/%d Outbox=%u drops=%lu/%d", - (int)ESP.getFreeHeap(), (int)ESP.getMaxAllocHeap(), + replyAppendf(buf, bufsize, &pos, "> Free=%d Max=%d", + (int)ESP.getFreeHeap(), (int)ESP.getMaxAllocHeap()); +#ifdef ESP_PLATFORM + // Internal heap can include regions that hardware AES cannot use. Its + // unaligned PSRAM records need a DMA bounce buffer even with ample PSRAM. + replyAppendf(buf, bufsize, &pos, " DMA=%u/%u", + (unsigned)heap_caps_get_free_size(MALLOC_CAP_DMA), + (unsigned)heap_caps_get_largest_free_block(MALLOC_CAP_DMA)); +#endif + replyAppendf(buf, bufsize, &pos, " q:%d/%d Outbox=%u drops=%lu/%d", q, MAX_QUEUE_SIZE, (unsigned)outbox_total, outbox_drops, b->_skipped_publishes); // filt=: packets the per-slot type filters rejected before the queue. @@ -580,17 +616,7 @@ const char* MQTTBridge::wifiReasonStr(uint8_t reason) { } const char* MQTTBridge::tlsErrorStr(int32_t err) { - switch (err) { - case 0x8001: return "DNS failed"; - case 0x8002: return "socket error"; - case 0x8004: return "connect refused"; - case 0x8006: return "TLS timeout"; - case 0x8008: return "connection timeout"; - case 0x800B: return "cert verify failed"; - case 0x8010: return "mbedTLS error"; - case 0x801A: return "TLS handshake failed"; - default: return nullptr; - } + return MQTTErrorLabels::tlsError(err); } void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) { @@ -739,7 +765,14 @@ static const uint32_t MQTT_STOP_TIMEOUT_PER_SLOT_MS = 8000; // ~5-6 s measured // still budgets for the base teardown. static inline uint32_t mqttStopTimeoutForSlots(int slots) { if (slots < 1) slots = 1; - return MQTT_STOP_TIMEOUT_BASE_MS + MQTT_STOP_TIMEOUT_PER_SLOT_MS * (uint32_t)slots; + // The owner may first finish one blocked publish, and stopping an SDK task + // may wait for its current network operation. Include the connection budget + // as well as the measured teardown allowance so restoring slow-link support + // cannot force-delete a client that is still within its valid timeout. + const uint32_t per_slot = MQTT_STOP_TIMEOUT_PER_SLOT_MS > MQTTConnectionPolicy::kNetworkTimeoutMs + ? MQTT_STOP_TIMEOUT_PER_SLOT_MS : MQTTConnectionPolicy::kNetworkTimeoutMs; + return MQTT_STOP_TIMEOUT_BASE_MS + MQTTConnectionPolicy::kNetworkTimeoutMs + + per_slot * static_cast(slots); } // --------------------------------------------------------------------------- @@ -1213,8 +1246,8 @@ void MQTTBridge::begin() { // MQTT client objects are NOT allocated here. setupSlot() creates one on a slot's // first setup, so unconfigured and capped-off slots never cost their ~1.3 KB of // internal DRAM. Once created a client lives for the bridge's lifetime, so the - // reconfigure/reconnect paths still reuse the same mbedTLS context instead of - // churning ~40 KB of internal heap per cycle. + // reconfigure/reconnect paths reuse its MQTT buffers, transport handles and + // callbacks. ESP-IDF still destroys and recreates the TLS session on a close. // Sync the lifecycle Coordinator to Running now that all resources exist and // the task is created. Driven only on the success path: the failure rollbacks @@ -1662,34 +1695,13 @@ void MQTTBridge::mqttTaskLoop() { MQTT_DEBUG_PRINTLN("mbedTLS allocator redirected to PSRAM"); #endif - MQTT_DEBUG_PRINTLN("NTP synced, setting up MQTT slots (max %d active)...", _max_active_slots); - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (_stop_requested) break; - if (_slots[i].enabled) { - if (!canActivateSlot(i)) { - MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", i + 1, _max_active_slots); - _slots[i].enabled = false; // Disable so other loops skip it - continue; - } - char reason[80]; - if (!isSlotReady(i, reason, sizeof(reason))) { - MQTT_DEBUG_PRINTLN("MQTT%d not ready - run '%s' to connect", i + 1, reason); - continue; - } - // A slot that fails to activate consumes no position and stays enabled, so - // maintainSlotConnections() retries it and a later healthy broker is not - // starved by it on a capped board. - if (!setupSlot(i)) continue; - // Stagger connections: 5s between slots to avoid simultaneous TLS handshakes - // which compete for ~40KB internal heap each - if (i < RUNTIME_MQTT_SLOTS - 1) { - if (!waitUnlessStopping(5000)) break; - } - } - } + // Connection maintenance starts each ready slot without blocking packet + // processing. A fixed five-second sleep did not serialize the async SDK + // handshakes: slow TLS/WSS sessions could still overlap the next slot. + MQTT_DEBUG_PRINTLN("NTP synced, scheduling MQTT slots (max %d active)...", _max_active_slots); } - // A stop during a stagger must not fall through into reconfiguration, + // A stop during deferred setup must not fall through into reconfiguration, // publication, or connection maintenance before the next loop iteration. if (_stop_requested) continue; @@ -1865,10 +1877,10 @@ void MQTTBridge::mqttTaskLoop() { // initial_connect_done), so slots that are unconfigured or capped off never get one. // // Once created the object lives until destroySlotClients(): reconfiguring a slot -// (preset change, JWT renewal, reconnect) reuses it, so the mbedTLS context and its -// ~40 KB of internal-heap buffers are allocated once instead of every reconfigure. -// That context is created by connect(), not by this constructor, so deferring the -// allocation to first use costs nothing beyond the object itself. +// (preset change, JWT renewal, reconnect) reuses its MQTT buffers, transport handles +// and callbacks. ESP-IDF's transport still frees the TLS session on disconnect and +// creates a new one for reconnect. Deferring client creation avoids reserving +// wrapper and MQTT storage for slots that cannot run. bool MQTTBridge::ensureSlotClient(int index) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; @@ -1954,6 +1966,14 @@ bool MQTTBridge::ensureSlotClient(int index) { MQTT_DEBUG_PRINTLN("MQTT%d error: tls=%d, tls_stack=%d, sock=%d, type=%d", index + 1, error.esp_tls_last_esp_err, error.esp_tls_stack_err, error.esp_transport_sock_errno, error.error_type); +#ifdef ESP_PLATFORM + MQTT_DEBUG_PRINTLN("MQTT%d error heap: internal=%u/%u DMA=%u/%u", + index + 1, + (unsigned)heap_caps_get_free_size(MALLOC_CAP_INTERNAL), + (unsigned)heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL), + (unsigned)heap_caps_get_free_size(MALLOC_CAP_DMA), + (unsigned)heap_caps_get_largest_free_block(MALLOC_CAP_DMA)); +#endif } else { MQTT_DEBUG_PRINTLN("MQTT%d error: type=%d", index + 1, error.error_type); } @@ -2039,12 +2059,40 @@ bool MQTTBridge::canActivateSlot(int index) const { return activatedSlotCount() < _max_active_slots; } +bool MQTTBridge::hasPendingSlotConnection() const { + // Callbacks run on esp-mqtt tasks. Time spacing alone cannot establish that + // an earlier handshake has finished, particularly on a slow WiFi link. + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + if (_slot_attempt_pending[i].load(std::memory_order_acquire)) return true; + } + return false; +} + +bool MQTTBridge::canStartSlotConnection(int index) const { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS || hasPendingSlotConnection()) return false; +#ifdef ESP_PLATFORM + const MQTTSlot& slot = _slots[index]; + 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); + const uint32_t caps = MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA; + return MQTTConnectionAdmission::canStart( + ESP.getPsramSize() > 0, tls, heap_caps_get_free_size(caps), + heap_caps_get_largest_free_block(caps)); +#else + return true; +#endif +} + // Returns true only when the slot reached connect(). A false result leaves the slot // enabled but not activated, so it holds no active-slot position and // maintainSlotConnections() will retry it -- the allocation failures below are transient // memory conditions, not permanent misconfiguration. bool MQTTBridge::setupSlot(int index) { if (_stop_requested.load(std::memory_order_acquire)) return false; + // A CLI reconfigure can arrive before the deferred startup pass. Wait for + // that pass to install the TLS allocator and establish usable clock state. + if (!_slots_setup_done) return false; if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; @@ -2052,6 +2100,7 @@ bool MQTTBridge::setupSlot(int index) { teardownSlot(index); return false; } + if (!canStartSlotConnection(index)) return false; // Every failure below is a real attempt, so stamp it: the retry interval in // maintainSlotConnections() measures from last_reconnect_attempt, which starts at 0 @@ -2069,8 +2118,8 @@ bool MQTTBridge::setupSlot(int index) { } // Reconfigure path: if we're re-applying (e.g. after a preset change), stop - // the existing connection cleanly first. The client object (and its mbedTLS - // context) is reused; setCredentials / setServer below overwrite the config + // the existing connection cleanly first. The client object and its + // transport handles are reused; setCredentials / setServer overwrite the config // fields in place before connect() restarts the ESP-IDF client. if (slot.initial_connect_done) { if (slot.client->connected()) { @@ -2285,15 +2334,16 @@ bool MQTTBridge::setupSlot(int index) { } // Disconnect the slot's MQTT client and clear per-connection state, but leave -// the client object alive so a subsequent setupSlot() can reuse its mbedTLS -// context. This is called both on reconfigure (preset change) and at shutdown; +// the client object alive so a subsequent setupSlot() can reuse its MQTT storage +// and transport handles. TLS session storage is freed by the transport close. +// This is called both on reconfigure (preset change) and at shutdown; // destruction of the underlying client happens once in destroySlotClients(). void MQTTBridge::teardownSlot(int index, bool force) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[index]; _slot_attempt_pending[index] = false; - if (slot.client && (force || slot.client->connected())) { + if (slot.client && (force || slot.client->isStarted())) { if (force) { slot.client->forceStop(); } else { @@ -2328,6 +2378,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { if (_stop_requested.load(std::memory_order_acquire)) return ESP_ERR_INVALID_STATE; MQTTSlot& slot = _slots[index]; if (slot.client == nullptr) return ESP_ERR_INVALID_STATE; + if (!canStartSlotConnection(index)) return ESP_ERR_NO_MEM; const bool was_pending = _slot_attempt_pending[index]; const bool starting = !slot.client->isStarted(); @@ -2375,11 +2426,12 @@ void MQTTBridge::maintainSlotConnections() { // Only allow one reconnect attempt per maintenance cycle to avoid // multiple simultaneous TLS handshakes blocking the network stack. - // Time-based guard: block reconnects if any slot reconnected within the last 15 s, - // ensuring the previous TLS handshake (and its Core-0-expensive completion events) - // finish before the next slot begins its own handshake. - bool reconnect_attempted_this_cycle = MQTTConnectionPolicy::reconnectGuardActive( - static_cast(now_millis), static_cast(_last_slot_reconnect_ms)); + // Keep a 15-second gap, and also wait for the pending attempt's callback. + // The gap alone cannot ensure a slow TLS/WSS handshake has completed. + bool reconnect_attempted_this_cycle = _last_slot_reconnect_ms != 0 && + MQTTConnectionPolicy::reconnectGuardActive( + static_cast(now_millis), static_cast(_last_slot_reconnect_ms)); + reconnect_attempted_this_cycle |= hasPendingSlotConnection(); // Only allow one full teardown+setup per cycle to limit heap fragmentation // when multiple slots fail simultaneously bool teardown_attempted_this_cycle = false; @@ -2405,11 +2457,18 @@ void MQTTBridge::maintainSlotConnections() { // without this the slot stayed dead until a reconfigure or reboot. Only retried // after the initial pass has run, so the NTP-deferred setup order is preserved. if (!_slots[i].initial_connect_done) { + if (_slots_setup_done && !canActivateSlot(i)) { + // Preserve the inactive classification for the spare slot after all + // supported active positions have been assigned. + _slots[i].enabled = false; + continue; + } if (_slots_setup_done && !setup_retry_this_cycle && !reconnect_attempted_this_cycle && - isSlotReady(i) && canActivateSlot(i) && - MQTTConnectionPolicy::elapsedMs(static_cast(now_millis), + isSlotReady(i) && canActivateSlot(i) && canStartSlotConnection(i) && + (_slots[i].last_reconnect_attempt == 0 || + MQTTConnectionPolicy::elapsedMs(static_cast(now_millis), static_cast(_slots[i].last_reconnect_attempt)) - >= SLOT_SETUP_RETRY_INTERVAL) { + >= SLOT_SETUP_RETRY_INTERVAL)) { _slots[i].last_reconnect_attempt = now_millis; setup_retry_this_cycle = true; MQTT_DEBUG_PRINTLN("MQTT%d retrying deferred setup (int_heap=%d)", i + 1, @@ -2473,7 +2532,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns bool can_attempt_renewal = MQTTConnectionPolicy::renewalAttemptAllowed( static_cast(now_millis), static_cast(slot.last_token_renewal)); - if (token_needs_renewal && can_attempt_renewal) { + if (token_needs_renewal && can_attempt_renewal && !reconnect_attempted && + !hasPendingSlotConnection()) { slot.last_token_renewal = now_millis; unsigned long old_token_expires_at = slot.token_expires_at; @@ -2529,10 +2589,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns } } - // Phase 4 (MQTT memory-defrag): the MIN_TLS_HEAP preflight was a workaround - // for the fragmentation caused by per-reconnect mbedTLS allocations. With - // persistent clients (Phase 1), the mbedTLS context is allocated once at - // startup and the preflight is no longer necessary. + // Admission now checks DMA-capable memory, rather than the former aggregate + // MIN_TLS_HEAP check. ESP-IDF reallocates TLS sessions on each reconnect. const auto prepareJwtReconnect = [&](bool force_mint, int backoff_level) { const bool has_token = slot.auth_token && slot.auth_token[0] != '\0'; @@ -2594,7 +2652,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // Periodic probe for circuit-breaker-tripped slots (recovery from transient outages) // Attempts a single reconnect every 30 minutes to see if the server has come back - if (slot.circuit_breaker_tripped && !reconnect_attempted) { + if (slot.circuit_breaker_tripped && !reconnect_attempted && + canStartSlotConnection(index)) { unsigned long probe_elapsed = MQTTConnectionPolicy::elapsedMs( static_cast(now_millis), static_cast(slot.last_reconnect_attempt)); if (MQTTConnectionPolicy::circuitBreakerProbeDue( @@ -2618,7 +2677,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // Reconnect with exponential backoff (for disconnected slots that already have valid config) // Only one reconnect per maintenance cycle to prevent TLS handshakes from blocking other slots - if (!slot.connected && slot.initial_connect_done && !slot.circuit_breaker_tripped && !reconnect_attempted) { + if (!slot.connected && slot.initial_connect_done && !slot.circuit_breaker_tripped && + !reconnect_attempted && canStartSlotConnection(index)) { if (MQTTConnectionPolicy::reconnectDue( static_cast(now_millis), static_cast(slot.last_reconnect_attempt), slot.reconnect_backoff, static_cast(index))) { @@ -3943,11 +4003,10 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, // Memory pressure check: Skip publishes when there's not enough contiguous // heap for the publish itself (JSON buffer + esp-mqtt outbox frame + WiFi TX - // path). Headroom only - NOT an mbedTLS preflight: persistent clients keep - // their TLS contexts allocated for the bridge lifetime, so the old ~52 KB - // "reserve space for reconnect" guard is obsolete post Phase 1. Publish - // payload is capped at PUBLISH_JSON_BUFFER_SIZE (2 KB); 8 KB is a safe - // ceiling including esp-mqtt frame overhead and transient TCP buffers. + // path). This guard covers publish headroom, not the separate TLS session + // allocations on reconnect or AES DMA allocations. Publish payload is capped + // at PUBLISH_JSON_BUFFER_SIZE (2 KB); the threshold includes frame overhead + // and transient TCP buffers. #ifdef ESP32 #if defined(BOARD_HAS_PSRAM) static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 16000; @@ -4638,6 +4697,9 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) { (!_slots[i].preset && _slots[i].audience[0] != '\0'); if (_slots[i].enabled && slot_jwt && _slots[i].client) { if (_slots[i].token_expires_at > 0 && current_time > _slots[i].token_expires_at) { + // Leave the expiry stale while another connection owns admission; + // maintenance will renew/bounce it after that attempt completes. + if (hasPendingSlotConnection()) continue; MQTT_DEBUG_PRINTLN("MQTT%d token stale after time correction, re-creating", i + 1); const MQTTConnectionPolicy::StaleTokenAction action = MQTTConnectionPolicy::classifyStaleToken( @@ -5034,12 +5096,11 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_ client->setBufferSize(MQTT_CLIENT_BUFFER_SIZE); - // Bound how long a synchronous QoS0 publish (see publishToSlot) can block the MQTT - // task on a stalled/half-open socket before esp-mqtt aborts the write. Default is 10s; - // 2.5s lets a first stall resolve fast (write fails -> slot flips to disconnected -> - // subsequent packets skip it) without holding up publishing to the other slots. Mesh - // RX (Core 1) and the WiFi/TCP stack are unaffected by this block regardless. - client->setNetworkTimeout(2500); + // esp-mqtt shares this timeout across TLS/WebSocket setup, CONNACK, and writes. + // Retain the SDK's ten-second budget so a slow WSS handshake can complete when + // other slots are active. A stalled synchronous QoS0 publish may also wait up to + // this budget; mesh RX runs independently on Core 1. + client->setNetworkTimeout(MQTTConnectionPolicy::kNetworkTimeoutMs); // Dormant safety net: cap the esp-mqtt outbox for any residual async QoS0 path. QoS0 // packets now publish synchronously (store=false, no outbox), so this normally never @@ -5090,9 +5151,17 @@ void MQTTBridge::logMemoryStatus() { } } } +#ifdef ESP_PLATFORM + MQTT_DEBUG_PRINTLN("Memory: Free=%d, Max=%d, DMA=%u/%u, Queue=%d/%d, Outbox=%u | pub(ok/err) %s", + ESP.getFreeHeap(), ESP.getMaxAllocHeap(), + (unsigned)heap_caps_get_free_size(MALLOC_CAP_DMA), + (unsigned)heap_caps_get_largest_free_block(MALLOC_CAP_DMA), + _queue_count, MAX_QUEUE_SIZE, (unsigned)outbox_total, pub_detail); +#else MQTT_DEBUG_PRINTLN("Memory: Free=%d, Max=%d, Queue=%d/%d, Outbox=%u | pub(ok/err) %s", ESP.getFreeHeap(), ESP.getMaxAllocHeap(), _queue_count, MAX_QUEUE_SIZE, (unsigned)outbox_total, pub_detail); +#endif } // --------------------------------------------------------------------------- diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 115988eaa..783d69cab 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -258,7 +258,7 @@ private: // Set before starting/reconnecting, cleared by connect/failed-attempt/stop. // Must be set before the SDK call because its event task can callback at once. - volatile bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS]; // Pending on-connect status publish: set from the onConnect callback (which // runs on the esp-mqtt event task, NOT this bridge task), consumed by the MQTT @@ -401,8 +401,8 @@ private: // Memory pressure monitoring (per-publish skip; see publishPacket()). // The broader fragmentation-recovery machinery was removed in Phase 4 of - // the MQTT memory-defrag work - persistent MQTT clients no longer churn - // the heap, so gray-zone / critical-restart trackers are unnecessary. + // the MQTT memory-defrag work. Persistent clients reduce MQTT storage churn; + // ESP-IDF still reallocates TLS sessions when a connection closes. unsigned long _last_memory_check; bool _memory_pressure = false; // Cached max-alloc verdict; re-sampled at most once per interval in publishPacket() so the heap walk isn't paid per-packet under pressure int _skipped_publishes; // Exposed via SNMP; count of publishes skipped when max_alloc is too low @@ -479,9 +479,9 @@ private: // - setupSlot() ensures the client exists, then configures it (server, // credentials, CA) and calls connect(). Safe to call again to reconfigure. // - teardownSlot() only disconnects - it never deletes the client. Leaves - // the mbedTLS/transport state ready for a subsequent setupSlot(). - // This avoids delete/new cycles that shed ~40 KB of mbedTLS buffers per - // reconfigure and fragment the internal heap on non-PSRAM boards. + // MQTT storage and transport handles ready for a subsequent setupSlot(). + // This avoids wrapper/configuration allocation cycles. TLS sessions and their + // record buffers are still freed and recreated by ESP-IDF on reconnect. bool ensureSlotClient(int index); // Allocate this slot's persistent client + callbacks on first use bool ensureSlotAuthToken(int index); // Allocate this slot's JWT token buffer on first token creation void releaseSlotAuthToken(int index);// Free the token buffer (only with the client -- see MQTTSlot) @@ -493,6 +493,8 @@ private: // exceeded by one route while another enforces it. int activatedSlotCount() const; bool canActivateSlot(int index) const; + bool hasPendingSlotConnection() const; + bool canStartSlotConnection(int index) const; void teardownSlot(int index, bool force = false); esp_err_t reconnectSlotClient(int index); void maintainSlotConnections(); // Maintain all slot connections (token renewal, reconnect) diff --git a/test/test_firmware_ram.py b/test/test_firmware_ram.py index aa0792d51..21c03d737 100644 --- a/test/test_firmware_ram.py +++ b/test/test_firmware_ram.py @@ -42,14 +42,15 @@ def elf32(path, symbols, data=b"", address=0x3F400000): return FirmwareElf(path) -def esp_fixture(path, modern=False, fragmented=False): +def esp_fixture(path, modern=False, fragmented=False, internal_length=0x10000): address = 0x3F400000 caps = [0x804, 0x804, 0x404, 0x803, 0x8804] data = bytearray(struct.pack(" +#include +#include +#include "helpers/MQTTConnectionAdmission.h" +namespace Admission = MQTTConnectionAdmission; +''' + +HEAP_PREAMBLE = POLICY_PREAMBLE + r''' +#define ESP_PLATFORM 1 +#define BOARD_HAS_PSRAM 1 +constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2, MALLOC_CAP_SPIRAM = 4; +static size_t dma_free = 100000, dma_largest = 20000; +static size_t heap_queries = 0, psram_allocations = 0, internal_allocations = 0; +static bool psram_fails = false, internal_fails = false; +static unsigned char psram_storage[4096], internal_storage[4096]; +size_t heap_caps_get_free_size(uint32_t caps) { + assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); + ++heap_queries; + return dma_free; +} +size_t heap_caps_get_largest_free_block(uint32_t caps) { + assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); + ++heap_queries; + return dma_largest; +} +void* heap_caps_calloc(size_t count, size_t size, uint32_t caps) { + assert(count && size && count <= SIZE_MAX / size); + if (caps == MALLOC_CAP_SPIRAM) { + ++psram_allocations; + return psram_fails ? nullptr : psram_storage; + } + assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); + ++internal_allocations; + return internal_fails ? nullptr : internal_storage; +} +''' + +CONNECTION_PREAMBLE = HEAP_PREAMBLE + r''' +#include +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 MQTTBridge { + struct MQTTSlot { + const Preset* preset = nullptr; + char broker_uri[128] = {}, host[64] = "broker"; + uint16_t port = 8883; + } _slots[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + bool hasPendingSlotConnection() const; + bool canStartSlotConnection(int index) const; +}; +''' + + +class MqttConnectionAdmissionTest(unittest.TestCase): + def compile_run(self, source): + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler, "a host C++ compiler is required") + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-admission-") as temporary: + work = Path(temporary) + path, binary = work / "admission.cpp", work / "admission" + path.write_text(source, encoding="utf-8") + result = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + "-I", str(ROOT / "src"), str(path), "-o", str(binary)], + capture_output=True, text=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(binary)], cwd=work, capture_output=True, + text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_connection_admission_checks_both_total_and_contiguous_dma_heap(self): + self.compile_run(POLICY_PREAMBLE + r''' +int main() { + // A large aggregate heap cannot compensate for a fragmented DMA region. + assert(!Admission::canStart(true, true, 100000, 4095)); + assert(!Admission::canStart(true, true, 16383, 20000)); + assert(Admission::canStart(true, true, 16384, 4096)); + assert(Admission::canStart(false, false, 16384, 4096)); + assert(!Admission::canStart(false, false, 16383, 4096)); + assert(!Admission::canStart(false, false, 16384, 4095)); + + // Without PSRAM, TLS also needs a full record allocation and session budget. + assert(!Admission::canStart(false, true, 61439, 20000)); + assert(!Admission::canStart(false, true, 100000, 17407)); + assert(Admission::canStart(false, true, 61440, 17408)); + assert(Admission::canStart(true, false, 16384, 4096)); +} +''') + + def test_internal_fallback_preserves_dma_reserve_and_rejects_fragmentation(self): + self.compile_run(POLICY_PREAMBLE + r''' +int main() { + assert(Admission::canFallback(1600, 16384 + 1600, 1600)); + assert(!Admission::canFallback(1600, 16384 + 1599, 1600)); + assert(!Admission::canFallback(1600, 100000, 1599)); + assert(!Admission::canFallback(1600, 16383, 100000)); + assert(!Admission::canFallback(1, 0, 100000)); + assert(!Admission::canFallback(0, 100000, 100000)); + assert(!Admission::canFallback(SIZE_MAX, SIZE_MAX, SIZE_MAX)); + assert(Admission::canFallback(SIZE_MAX - 16384, SIZE_MAX, SIZE_MAX)); +} +''') + + def test_calloc_size_rejects_zero_and_overflow_without_allocator_calls(self): + self.compile_run(POLICY_PREAMBLE + r''' +int main() { + size_t total = 123; + assert(!Admission::checkedCallocSize(0, 16, total) && total == 0); + total = 123; + assert(!Admission::checkedCallocSize(16, 0, total) && total == 0); + total = 123; + assert(!Admission::checkedCallocSize(SIZE_MAX, 2, total) && total == 0); + assert(Admission::checkedCallocSize(SIZE_MAX, 1, total) && total == SIZE_MAX); + assert(Admission::checkedCallocSize(1, SIZE_MAX, total) && total == SIZE_MAX); + assert(Admission::checkedCallocSize(2, SIZE_MAX / 2, total)); + assert(total == SIZE_MAX - 1); + assert(Admission::checkedCallocSize(4, 896, total) && total == 3584); +} +''') + + def test_explicit_uri_scheme_takes_precedence_over_port(self): + self.compile_run(POLICY_PREAMBLE + r''' +int main() { + assert(Admission::requiresTls("wss://broker:1883/mqtt", 1883)); + assert(Admission::requiresTls("mqtts://broker:1883", 1883)); + assert(!Admission::requiresTls("ws://broker:443/mqtt", 443)); + assert(!Admission::requiresTls("mqtt://broker:8883", 8883)); + assert(Admission::requiresTls("broker", 443)); + assert(Admission::requiresTls("broker", 8883)); + assert(!Admission::requiresTls("broker", 1883)); + assert(!Admission::requiresTls("broker", 9443)); + assert(Admission::requiresTls(nullptr, 443)); + assert(!Admission::requiresTls(nullptr, 0)); + assert(Admission::requiresTls("", 8883)); +} +''') + + def test_actual_calloc_fallback_preserves_dma_headroom_and_handles_failure(self): + source = HEAP_PREAMBLE + method("src/helpers/bridges/MQTTBridge.cpp", + "static void* psram_calloc(size_t n, size_t size)") + self.compile_run(source + r''' +int main() { + assert(psram_calloc(0, 16) == nullptr); + assert(psram_calloc(16, 0) == nullptr); + assert(psram_calloc(SIZE_MAX, 2) == nullptr); + assert(psram_allocations == 0 && internal_allocations == 0 && heap_queries == 0); + + // A successful external allocation needs no internal fallback or reserve. + dma_free = dma_largest = 0; + assert(psram_calloc(4, 400) == psram_storage); + assert(psram_allocations == 1 && internal_allocations == 0 && heap_queries == 0); + psram_fails = true; + dma_free = 16384 + 1599; dma_largest = 20000; + assert(psram_calloc(4, 400) == nullptr && internal_allocations == 0); + dma_free = 100000; dma_largest = 1599; + assert(psram_calloc(4, 400) == nullptr && internal_allocations == 0); + dma_free = 16383; dma_largest = 20000; + assert(psram_calloc(1, 1) == nullptr && internal_allocations == 0); + + dma_free = 16384 + 1600; dma_largest = 1600; + assert(psram_calloc(4, 400) == internal_storage && internal_allocations == 1); + internal_fails = true; + assert(psram_calloc(4, 400) == nullptr && internal_allocations == 2); +} +''') + + def test_actual_byte_allocators_use_byte_capable_fallback_and_preserve_failed_resize(self): + source = POLICY_PREAMBLE + r''' +#include +#define ESP_PLATFORM 1 +#define BOARD_HAS_PSRAM 1 +constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_SPIRAM = 4, MALLOC_CAP_8BIT = 8; +static bool psram_fails = false, internal_fails = false; +static size_t malloc_calls = 0, realloc_calls = 0, free_calls = 0; +static unsigned char original[32], psram_storage[64], internal_storage[64]; +static void* freed = nullptr; +void* allocation(uint32_t caps) { + if (caps == MALLOC_CAP_SPIRAM) return psram_fails ? nullptr : psram_storage; + // INTERNAL alone can select 32-bit-only IRAM on ESP32-S3. + assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT)); + return internal_fails ? nullptr : internal_storage; +} +void* heap_caps_malloc(size_t size, uint32_t caps) { + assert(size == 32); + ++malloc_calls; + return allocation(caps); +} +void* heap_caps_realloc(void* ptr, size_t size, uint32_t caps) { + assert(ptr == original && size == 48); + ++realloc_calls; + return allocation(caps); +} +void heap_caps_free(void* ptr) { ++free_calls; freed = ptr; } +''' + for signature in ("static void* psram_malloc(size_t size)", + "static void psram_free(void* ptr)", + "static void* psram_realloc(void* ptr, size_t new_size)"): + source += method("src/helpers/bridges/MQTTBridge.cpp", signature) + self.compile_run(source + r''' +int main() { + assert(psram_malloc(0) == nullptr && malloc_calls == 0); + assert(psram_malloc(32) == psram_storage && malloc_calls == 1); + psram_fails = true; + assert(psram_malloc(32) == internal_storage && malloc_calls == 3); + internal_fails = true; + assert(psram_malloc(32) == nullptr && malloc_calls == 5 && free_calls == 0); + + memset(original, 0xa5, sizeof(original)); + psram_fails = internal_fails = false; + assert(psram_realloc(original, 48) == psram_storage && realloc_calls == 1); + psram_fails = true; + assert(psram_realloc(original, 48) == internal_storage && realloc_calls == 3); + internal_fails = true; + assert(psram_realloc(original, 48) == nullptr && realloc_calls == 5); + assert(free_calls == 0); + for (unsigned char byte : original) assert(byte == 0xa5); + + assert(psram_realloc(original, 0) == nullptr); + assert(free_calls == 1 && freed == original && realloc_calls == 5); + assert(psram_realloc(nullptr, 0) == nullptr && free_calls == 1); +} +''') + + def test_actual_preflight_uses_detected_psram_transport_and_dma_caps(self): + source = CONNECTION_PREAMBLE + method("src/helpers/bridges/MQTTBridge.cpp", + "bool MQTTBridge::hasPendingSlotConnection() const") + source += method("src/helpers/bridges/MQTTBridge.cpp", + "bool MQTTBridge::canStartSlotConnection(int index) const") + self.compile_run(source + r''' +int main() { + MQTTBridge bridge; + assert(!bridge.canStartSlotConnection(-1)); + assert(!bridge.canStartSlotConnection(RUNTIME_MQTT_SLOTS)); + assert(heap_queries == 0); + + // BOARD_HAS_PSRAM does not establish that physical PSRAM initialized. + ESP.size = 0; dma_free = 61440; dma_largest = 17408; + assert(bridge.canStartSlotConnection(0)); + dma_free = 61439; + assert(!bridge.canStartSlotConnection(0)); + dma_free = 100000; dma_largest = 17407; + assert(!bridge.canStartSlotConnection(0)); + ESP.size = 2097152; dma_free = 16384; dma_largest = 4096; + assert(bridge.canStartSlotConnection(0)); + dma_largest = 4095; + assert(!bridge.canStartSlotConnection(0)); + + // An explicit plaintext URI overrides a TLS port. The effective URI wins + // over the raw host; a preset's URI wins over both custom fields. + ESP.size = 0; dma_free = 16384; dma_largest = 4096; + strcpy(bridge._slots[0].host, "mqtt://broker:8883"); + assert(bridge.canStartSlotConnection(0)); + strcpy(bridge._slots[0].broker_uri, "wss://broker/mqtt"); + assert(!bridge.canStartSlotConnection(0)); + const Preset plaintext{"ws://broker:443/mqtt"}; + bridge._slots[0].preset = &plaintext; + assert(bridge.canStartSlotConnection(0)); + const Preset tls{"mqtts://broker:1883"}; + bridge._slots[0].preset = &tls; + assert(!bridge.canStartSlotConnection(0)); + ESP.size = 2097152; + assert(bridge.canStartSlotConnection(0)); + + // Every pending TLS attempt, including the target's own attempt, blocks all + // other starts until an event callback clears it. No heap walk is needed. + for (int pending = 0; pending < RUNTIME_MQTT_SLOTS; ++pending) { + bridge._slot_attempt_pending[pending].store(true, std::memory_order_release); + const size_t before = heap_queries; + for (int index = 0; index < RUNTIME_MQTT_SLOTS; ++index) { + assert(!bridge.canStartSlotConnection(index)); + } + assert(heap_queries == before); + bridge._slot_attempt_pending[pending].store(false, std::memory_order_release); + assert(bridge.canStartSlotConnection(0)); + } +} +''') + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mqtt_diagnostics.py b/test/test_mqtt_diagnostics.py new file mode 100644 index 000000000..bfcaf823a --- /dev/null +++ b/test/test_mqtt_diagnostics.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""Exercise production MQTT diagnostics with distinct internal and DMA heaps.""" +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] + + +def method(signature): + source = (ROOT / "src/helpers/bridges/MQTTBridge.cpp").read_text(encoding="utf-8") + start = source.index(signature) + opening = source.index("{", start) + depth = 1 + pos = opening + 1 + while depth: + depth += (source[pos] == "{") - (source[pos] == "}") + pos += 1 + return source[start:pos] + "\n" + + +PREAMBLE = r''' +#include +#include +#include +#include +#include +#include "helpers/MQTTReplyFormat.h" +#include "helpers/MQTTConnectionHealth.h" +#include "helpers/bridges/MQTTErrorLabels.h" +constexpr int RUNTIME_MQTT_SLOTS = 5; +constexpr unsigned MALLOC_CAP_DMA = 1U << 3; +constexpr unsigned MALLOC_CAP_INTERNAL = 1U << 11; +static unsigned dma_queries = 0; +static size_t dma_free = 8128, dma_largest = 1536; +size_t heap_caps_get_free_size(unsigned caps) { + assert(caps == MALLOC_CAP_DMA); + ++dma_queries; + return dma_free; +} +size_t heap_caps_get_largest_free_block(unsigned caps) { + assert(caps == MALLOC_CAP_DMA); + ++dma_queries; + return dma_largest; +} +struct { + int getFreeHeap() const { return 64000; } + int getMaxAllocHeap() const { return 30000; } +} ESP; +unsigned uxQueueMessagesWaiting(void*) { return 3; } +static char debug_line[512]; +void capture_debug(const char* format, ...) { + va_list args; + va_start(args, format); + vsnprintf(debug_line, sizeof(debug_line), format, args); + va_end(args); +} +#define MQTT_DEBUG_PRINTLN(...) capture_debug(__VA_ARGS__) +struct MQTTPrefs {}; +struct Client { + unsigned long getOutboxSize() const { return 0; } + unsigned long getOutboxDrops() const { return 0; } + unsigned long getPublishOk() const { return 123; } + unsigned long getPublishErr() const { return 4; } +}; +struct MQTTBridge { + struct Slot { + Client* client = nullptr; + bool enabled = true, connected = true; + }; + Slot _slots[RUNTIME_MQTT_SLOTS]; + bool _initialized = true; + void* _packet_queue_handle = reinterpret_cast(1); + int _queue_count = 3, _skipped_publishes = 0; + unsigned long _filtered_packets = 0; + static constexpr int MAX_QUEUE_SIZE = 50; + bool isSlotReady(int) const { return true; } + static void formatMqttStatsReply(char*, size_t); + static const char* tlsErrorStr(int32_t); + void logMemoryStatus(); +}; +static MQTTBridge* s_mqtt_bridge_instance = nullptr; +''' + +MAIN = r''' +int main() { + // Use the public production method, not the pure table alone. Unknown SDK + // codes retain raw-number formatting, and write failures stay distinct. + assert(strcmp(MQTTBridge::tlsErrorStr(0x8006), "connection timeout") == 0); + assert(strcmp(MQTTBridge::tlsErrorStr(0x8008), "server closed connection") == 0); + assert(strcmp(MQTTBridge::tlsErrorStr(0x8018), "TLS write failed") == 0); + assert(MQTTBridge::tlsErrorStr(0x800B) == nullptr); + + char reply[160]; + MQTTBridge::formatMqttStatsReply(reply, sizeof(reply)); + assert(strcmp(reply, "> (bridge not running)") == 0); + assert(dma_queries == 0); + + MQTTBridge bridge; + Client clients[RUNTIME_MQTT_SLOTS]; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) bridge._slots[i].client = &clients[i]; + s_mqtt_bridge_instance = &bridge; + MQTTBridge::formatMqttStatsReply(reply, sizeof(reply)); + assert(strstr(reply, "Free=64000 Max=30000") != nullptr); +#ifdef ESP_PLATFORM + // A large general internal block must not mask the smaller AES DMA heap. + assert(strstr(reply, "DMA=8128/1536") != nullptr); + assert(dma_queries == 2); +#else + assert(strstr(reply, "DMA=") == nullptr); + assert(dma_queries == 0); +#endif + // Normal five-slot stats still fit the fixed LoRa reply budget. + assert(strstr(reply, "s5=123/4") != nullptr); + + bridge.logMemoryStatus(); +#ifdef ESP_PLATFORM + assert(strstr(debug_line, "DMA=8128/1536") != nullptr); +#else + assert(strstr(debug_line, "DMA=") == nullptr); +#endif + + // Large counters and new heap fields never write beyond a short reply. + struct { char reply[32]; unsigned char guard[8]; } bounded; + memset(&bounded, 0xA5, sizeof(bounded)); + bridge._filtered_packets = 4294967295UL; + dma_free = 4294967295UL; + dma_largest = 4294967295UL; + MQTTBridge::formatMqttStatsReply(bounded.reply, sizeof(bounded.reply)); + assert(bounded.reply[sizeof(bounded.reply) - 1] == '\0'); + for (unsigned char byte : bounded.guard) assert(byte == 0xA5); + MQTTBridge::formatMqttStatsReply(nullptr, 32); + MQTTBridge::formatMqttStatsReply(bounded.reply, 0); +} +''' + + +class MqttDiagnosticsTests(unittest.TestCase): + def test_production_stats_and_labels(self): + compiler = shutil.which("g++") + if not compiler: + self.skipTest("g++ is unavailable") + source = PREAMBLE + "\n".join(method(signature) for signature in ( + "void MQTTBridge::formatMqttStatsReply(", + "const char* MQTTBridge::tlsErrorStr(", + "void MQTTBridge::logMemoryStatus(", + )) + MAIN + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-diagnostics-") as directory: + test_source = Path(directory) / "diagnostics.cpp" + test_source.write_text(source, encoding="ascii") + for esp in (False, True): + with self.subTest(esp=esp): + executable = Path(directory) / ("esp" if esp else "other") + command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + "-I", str(ROOT / "src"), str(test_source), "-o", str(executable)] + if esp: + command.insert(1, "-DESP_PLATFORM=1") + subprocess.run(command, check=True, capture_output=True, text=True) + subprocess.run([str(executable)], check=True, capture_output=True, text=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mqtt_error_labels/test_mqtt_error_labels.cpp b/test/test_mqtt_error_labels/test_mqtt_error_labels.cpp index 604789bcd..7b87236eb 100644 --- a/test/test_mqtt_error_labels/test_mqtt_error_labels.cpp +++ b/test/test_mqtt_error_labels/test_mqtt_error_labels.cpp @@ -29,6 +29,12 @@ TEST(MQTTErrorLabels, TlsErrorsMatchTheSdkMeaning) { EXPECT_TRUE(labelIs(tlsError(0x8010), "cert chain partly parsed")); // was "mbedTLS error" EXPECT_TRUE(labelIs(tlsError(0x8006), "connection timeout")); EXPECT_TRUE(labelIs(tlsError(0x8001), "DNS failed")); + EXPECT_TRUE(labelIs(tlsError(0x8012), "SNI hostname rejected")); + EXPECT_TRUE(labelIs(tlsError(0x8015), "cert parse failed")); + EXPECT_TRUE(labelIs(tlsError(0x8017), "TLS setup failed")); + // A hardware AES allocation failure returns -1 through the TLS write path, + // which the SDK records as 0x8018 rather than a handshake or auth failure. + EXPECT_TRUE(labelIs(tlsError(0x8018), "TLS write failed")); EXPECT_TRUE(labelIs(tlsError(0x801A), "TLS handshake failed")); } diff --git a/test/test_mqtt_handshake_scheduler.py b/test/test_mqtt_handshake_scheduler.py new file mode 100644 index 000000000..5bca0ab50 --- /dev/null +++ b/test/test_mqtt_handshake_scheduler.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +"""Exercise production MQTT scheduling with delayed callbacks and bounded heaps. + +The actual maintenance/admission/reconnect methods run with a deterministic +clock. Hardware, SDK task starts, and token generation are controllable seams. +""" +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +from test_mqtt_transport_results import method + +ROOT = Path(__file__).resolve().parents[1] + +PREAMBLE = r''' +#include +#include +#include +#include +#include +#include +#include "helpers/MQTTConnectionAdmission.h" +#include "helpers/MQTTConnectionHealth.h" +#include "helpers/MQTTConnectionPolicy.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 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; +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 { 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; +} +size_t heap_caps_get_largest_free_block(uint32_t caps) { + assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_largest; +} +const char* esp_err_to_name(int) { return "mock"; } +struct Preset { + const char* server_url = "wss://broker/mqtt"; + int auth_type = 0; + unsigned long token_lifetime = 3300; + bool enforce_exp = false; +}; +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 isStarted() const { return started; } + bool connected() const { return live; } + int connect() { ++starts; started = true; return ESP_OK; } + int reconnect() { ++reconnects; return ESP_OK; } + void softDisconnect() { + ++bounces; live = false; + // Releasing the old internal TLS session creates room for its replacement. + dma_free += 44000; dma_largest = 20000; + } + void setCredentials(const char*, const char*) { ++credentials; } +}; +struct MQTTBridge { + struct MQTTSlot { + bool enabled = true, connected = false, initial_connect_done = false; + const Preset* preset = nullptr; + PsychicMqttClient* client = nullptr; + char host[64] = "broker", broker_uri[128] = {}, audience[64] = {}; + uint16_t port = 8883; + char* auth_token = nullptr; + unsigned long token_expires_at = 0, last_token_renewal = 0; + unsigned long last_reconnect_attempt = 0, connected_at_ms = 0; + uint8_t reconnect_backoff = 0, max_backoff_failures = 0; + bool circuit_breaker_tripped = false; + uint32_t start_failures = 0; + } _slots[RUNTIME_MQTT_SLOTS]; + PsychicMqttClient clients[RUNTIME_MQTT_SLOTS]; + char tokens[RUNTIME_MQTT_SLOTS][16] = {}; + bool setup_succeeds[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true}; + bool ready[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true}; + std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _stop_requested{false}, _ntp_synced{true}; + bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {}; + void* _identity = this; + bool _slots_setup_done = true; + unsigned long _last_slot_reconnect_ms = 0; + int _max_active_slots = 5, mints = 0; + const char* _jwt_username = "jwt"; + static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000; + std::vector setups; + bool isSlotReady(int index) { return ready[index]; } + bool setupSlot(int index) { + setups.push_back(index); + _slots[index].client = &clients[index]; + if (!setup_succeeds[index]) return false; + if (reconnectSlotClient(index) != ESP_OK) return false; + _slots[index].initial_connect_done = true; + return true; + } + bool createSlotAuthToken(int index) { + ++mints; + strcpy(tokens[index], "fresh"); + _slots[index].auth_token = tokens[index]; + _slots[index].token_expires_at = wall_seconds + 3300; + return true; + } + void complete(int index) { + _slot_attempt_pending[index].store(false, std::memory_order_release); + _slots[index].connected = clients[index].live = true; + } + void only(int index) { + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) _slots[i].enabled = i == index; + } + int activatedSlotCount() const; + bool canActivateSlot(int index) const; + bool hasPendingSlotConnection() const; + bool canStartSlotConnection(int index) const; + esp_err_t reconnectSlotClient(int index); + unsigned long slotTokenLifetime(int index) const; + void maintainSlotConnections(); + void maintainSlotConnection(int, unsigned long, unsigned long, bool, bool&, bool&); +}; +''' + + +class MqttHandshakeSchedulerTest(unittest.TestCase): + def fixture(self): + bridge = "src/helpers/bridges/MQTTBridge.cpp" + return PREAMBLE + "\n".join(method(bridge, signature) for signature in ( + "int MQTTBridge::activatedSlotCount() const", + "bool MQTTBridge::canActivateSlot(int index) const", + "bool MQTTBridge::hasPendingSlotConnection() const", + "bool MQTTBridge::canStartSlotConnection(int index) const", + "esp_err_t MQTTBridge::reconnectSlotClient(int index)", + "unsigned long MQTTBridge::slotTokenLifetime(int index) const", + "void MQTTBridge::maintainSlotConnections()", + "void MQTTBridge::maintainSlotConnection(")) + + def compile_run(self, main): + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler, "a host C++ compiler is required") + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-scheduler-") as temporary: + work = Path(temporary) + source, binary = work / "scheduler.cpp", work / "scheduler" + source.write_text(self.fixture() + main, encoding="utf-8") + result = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-I", str(ROOT / "src"), + str(source), "-o", str(binary)], capture_output=True, text=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(binary)], cwd=work, capture_output=True, + text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_first_start_is_immediate_and_delayed_callback_serializes_following_slots(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + clock_ms = 1000; + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector{0}); + assert(bridge._slot_attempt_pending[0]); + // The time guard has expired, but the earlier asynchronous handshake is + // still pending. Ordinary maintenance must not launch another SDK request. + clock_ms = 16000; + bridge.maintainSlotConnections(); + clock_ms = 3600000; + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector{0}); + assert(bridge.clients[0].starts == 1 && bridge.clients[0].reconnects == 0); + bridge.complete(0); + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector({0, 1})); + assert(bridge._slot_attempt_pending[1]); + bridge.complete(1); + clock_ms += 14999; + bridge.maintainSlotConnections(); + assert(bridge.setups.size() == 2); + ++clock_ms; + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector({0, 1, 2})); + bridge._stop_requested = true; + bridge.complete(2); + clock_ms += 60000; + bridge.maintainSlotConnections(); + assert(bridge.setups.size() == 3); +} +''') + + def test_setup_failure_waits_full_minute_even_when_client_was_allocated(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); bridge.setup_succeeds[0] = false; + clock_ms = 1000; + bridge.maintainSlotConnections(); + assert(bridge.setups.size() == 1 && bridge._slots[0].client); + assert(!bridge._slots[0].initial_connect_done && bridge._slots[0].enabled); + assert(!bridge.hasPendingSlotConnection()); + clock_ms = 1050; + bridge.maintainSlotConnections(); + clock_ms = 60999; + bridge.maintainSlotConnections(); + assert(bridge.setups.size() == 1); + clock_ms = 61000; bridge.setup_succeeds[0] = true; + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector({0, 0})); + assert(bridge._slots[0].initial_connect_done && bridge.hasPendingSlotConnection()); +} +''') + + def test_failed_slot_does_not_starve_healthy_slot_or_consume_active_cap(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge._max_active_slots = 1; bridge.setup_succeeds[0] = false; + clock_ms = 1000; + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector{0}); + assert(bridge.activatedSlotCount() == 0); + clock_ms = 1050; + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector({0, 1})); + assert(bridge.activatedSlotCount() == 1); + for (int i = 2; i < RUNTIME_MQTT_SLOTS; ++i) assert(!bridge._slots[i].enabled); + bridge.complete(1); clock_ms += 60000; + bridge.maintainSlotConnections(); + assert(!bridge._slots[0].enabled && bridge.setups.size() == 2); +} +''') + + def test_low_dma_memory_defers_setup_without_consuming_retry_or_capacity(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); clock_ms = 1000; dma_free = 16383; + bridge.maintainSlotConnections(); + assert(bridge.setups.empty() && bridge._slots[0].last_reconnect_attempt == 0); + assert(bridge._slots[0].enabled && bridge.activatedSlotCount() == 0); + dma_free = 100000; dma_largest = 4095; + bridge.maintainSlotConnections(); + assert(bridge.setups.empty()); + dma_largest = 4096; + bridge.maintainSlotConnections(); + assert(bridge.setups == std::vector{0} && bridge.activatedSlotCount() == 1); +} +''') + + def test_internal_tls_renewal_releases_old_session_before_new_memory_admission(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); clock_ms = 70000; ESP.size = 0; + const Preset jwt{"wss://broker/mqtt", MQTT_AUTH_JWT, 3300, true}; + 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.token_expires_at = wall_seconds + 120; + // Before releasing the current TLS session, the full admission check fails. + dma_free = 20000; dma_largest = 10000; + assert(!bridge.canStartSlotConnection(0)); + bridge.maintainSlotConnections(); + assert(bridge.mints == 1 && slot.client->bounces == 1); + assert(slot.client->reconnects == 1 && bridge._slot_attempt_pending[0]); + assert(slot.client->credentials == 1 && dma_free == 64000); + + // Another pending slot still prevents renewal from tearing down a live link. + bridge.complete(0); bridge._slot_attempt_pending[1] = true; + slot.token_expires_at = wall_seconds + 120; clock_ms += 60000; + bridge.maintainSlotConnections(); + assert(bridge.mints == 1 && slot.client->bounces == 1); +} +''') + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mqtt_network_timeout.py b/test/test_mqtt_network_timeout.py new file mode 100644 index 000000000..174bd99f8 --- /dev/null +++ b/test/test_mqtt_network_timeout.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""Exercise production MQTT configuration for both SDK layouts and RAM profiles. + +The real optimizer and adapter setters run on the host. SDK field layouts are +stand-ins; this checks the timeout written to SDK configuration, not a handshake. +""" +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +from test_mqtt_transport_results import method + +ROOT = Path(__file__).resolve().parents[1] + +PREAMBLE = r''' +#include +#include +#include "helpers/MQTTConnectionPolicy.h" +struct esp_mqtt_client_config_t { +#if ESP_IDF_VERSION_MAJOR == 5 + struct { int keepalive = 0, message_retransmit_timeout = 0; } session; + struct { int timeout_ms = 0; } network; + struct { int size = 0, out_size = 0; } buffer; +#else + int keepalive = 0, message_retransmit_timeout = 0; + int network_timeout_ms = 0, buffer_size = 0; +#endif +}; +struct PsychicMqttClient { + esp_mqtt_client_config_t _mqtt_cfg; + bool _config_dirty = false; + size_t _outbox_limit = 0; + PsychicMqttClient& setKeepAlive(int); + PsychicMqttClient& setMessageRetransmitTimeout(int); + PsychicMqttClient& setBufferSize(int); + PsychicMqttClient& setNetworkTimeout(int); + PsychicMqttClient& setOutboxLimit(size_t); + esp_mqtt_client_config_t* getMqttConfig(); +}; +struct MQTTBridge { + void optimizeMqttClientConfig(PsychicMqttClient*, bool); +}; +''' + +MAIN = r''' +int main() { + MQTTBridge bridge; + bridge.optimizeMqttClientConfig(nullptr, true); + PsychicMqttClient client; + // Reconfiguration must restore the connection budget after a shorter value, + // for both small user/password and large JWT CONNECT buffers. + for (bool needs_large : {false, true, false}) { + client.setNetworkTimeout(2500); + client._config_dirty = false; + bridge.optimizeMqttClientConfig(&client, needs_large); + assert(client._config_dirty); +#if ESP_IDF_VERSION_MAJOR == 5 + const int timeout = client._mqtt_cfg.network.timeout_ms; + const int buffer_size = client._mqtt_cfg.buffer.size; + const int keepalive = client._mqtt_cfg.session.keepalive; + assert(client._mqtt_cfg.buffer.out_size == buffer_size); + assert(client._mqtt_cfg.session.message_retransmit_timeout == 15000); +#else + const int timeout = client._mqtt_cfg.network_timeout_ms; + const int buffer_size = client._mqtt_cfg.buffer_size; + const int keepalive = client._mqtt_cfg.keepalive; + assert(client._mqtt_cfg.message_retransmit_timeout == 15000); +#endif + assert(timeout == 10000); + assert(timeout == MQTTConnectionPolicy::kNetworkTimeoutMs); +#ifdef BOARD_HAS_PSRAM + assert(buffer_size == 896 && keepalive == 45); + assert(client._outbox_limit == 16384); +#else + assert(buffer_size == (needs_large ? 896 : 512) && keepalive == 75); + assert(client._outbox_limit == 8192); +#endif + } +} +''' + + +class MqttNetworkTimeoutTest(unittest.TestCase): + def fixture(self): + adapter = "lib/PsychicMqttClient/src/PsychicMqttClient.cpp" + source = PREAMBLE + "#include \n" + source += "\n".join(method(adapter, signature) for signature in ( + "PsychicMqttClient &PsychicMqttClient::setKeepAlive(", + "PsychicMqttClient &PsychicMqttClient::setMessageRetransmitTimeout(", + "PsychicMqttClient &PsychicMqttClient::setBufferSize(", + "PsychicMqttClient &PsychicMqttClient::setNetworkTimeout(", + "PsychicMqttClient &PsychicMqttClient::setOutboxLimit(", + "esp_mqtt_client_config_t *PsychicMqttClient::getMqttConfig(")) + source += method("src/helpers/bridges/MQTTBridge.cpp", + "void MQTTBridge::optimizeMqttClientConfig(") + return source + MAIN + + def compile_run(self, source, expect_rejection=False): + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler, "a host C++ compiler is required") + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-timeout-") as temporary: + work = Path(temporary) + path = work / "config.cpp" + path.write_text(source, encoding="utf-8") + for major in (4, 5): + for psram in (False, True): + with self.subTest(idf_major=major, psram=psram): + binary = work / f"config-{major}-{int(psram)}" + command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + "-Wno-unused-parameter", + f"-DESP_IDF_VERSION_MAJOR={major}", + "-I", str(ROOT / "src"), str(path), "-o", str(binary)] + if psram: + command.insert(1, "-DBOARD_HAS_PSRAM=1") + compiled = subprocess.run(command, capture_output=True, + text=True, timeout=60) + self.assertEqual(compiled.returncode, 0, compiled.stderr) + result = subprocess.run([str(binary)], cwd=work, capture_output=True, + text=True, timeout=10) + if expect_rejection: + self.assertNotEqual(result.returncode, 0, + result.stdout + result.stderr) + self.assertIn("Assertion", result.stderr) + else: + self.assertEqual(result.returncode, 0, + result.stdout + result.stderr) + + def test_production_optimizer_applies_connection_budget_to_every_config(self): + self.compile_run(self.fixture()) + + def test_previous_publish_timeout_fails_configuration_regression(self): + source = self.fixture() + old = source.replace( + "client->setNetworkTimeout(MQTTConnectionPolicy::kNetworkTimeoutMs);", + "client->setNetworkTimeout(2500);", 1) + self.assertNotEqual(old, source) + self.compile_run(old, expect_rejection=True) + + def test_production_stop_budget_covers_shared_network_timeout_for_every_slot(self): + bridge = ROOT / "src/helpers/bridges/MQTTBridge.cpp" + source = bridge.read_text(encoding="utf-8") + start = source.index("static const uint32_t MQTT_STOP_TIMEOUT_BASE_MS") + end = source.index("static inline uint32_t mqttStopTimeoutForSlots(int slots)") + fixture = '#include \n#include "helpers/MQTTConnectionPolicy.h"\n' + fixture += source[start:end] + fixture += method("src/helpers/bridges/MQTTBridge.cpp", + "static inline uint32_t mqttStopTimeoutForSlots(int slots)") + fixture += r''' +int main() { + assert(mqttStopTimeoutForSlots(0) == 25000); + assert(mqttStopTimeoutForSlots(-1) == 25000); + assert(mqttStopTimeoutForSlots(2) == 35000); + assert(mqttStopTimeoutForSlots(5) == 65000); + assert(mqttStopTimeoutForSlots(6) == 75000); + assert(mqttStopTimeoutForSlots(6) > UINT16_MAX); + assert(mqttStopTimeoutForSlots(16) == 175000); +} +''' + self.compile_run(fixture) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mqtt_transport_results.py b/test/test_mqtt_transport_results.py index 77e8048f1..100993887 100644 --- a/test/test_mqtt_transport_results.py +++ b/test/test_mqtt_transport_results.py @@ -90,8 +90,10 @@ struct MQTTBridge { uint32_t start_failures=0; }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; - volatile bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; std::atomic _stop_requested{false}; + bool hasPendingSlotConnection() const; + bool canStartSlotConnection(int index) const; esp_err_t reconnectSlotClient(int index); }; ''' @@ -135,10 +137,19 @@ int main() { reconnect_result=ESP_FAIL; assert(b.reconnectSlotClient(0)==ESP_FAIL); assert(!b._slot_attempt_pending[0] && b._slots[0].start_failures==0); + int attempts = reconnect_calls; b._slot_attempt_pending[0]=true; - assert(b.reconnectSlotClient(0)==ESP_FAIL); + assert(b.reconnectSlotClient(0)==ESP_ERR_NO_MEM && reconnect_calls==attempts); assert(b._slot_attempt_pending[0] && b._slots[0].start_failures==0); - sdk_callback=[&]() { b._slot_attempt_pending[0]=false; }; + b._slot_attempt_pending[0]=false; + b._slot_attempt_pending[1]=true; + assert(b.reconnectSlotClient(0)==ESP_ERR_NO_MEM && reconnect_calls==attempts); + assert(!b._slot_attempt_pending[0] && b._slot_attempt_pending[1]); + b._slot_attempt_pending[1]=false; + sdk_callback=[&]() { + assert(b._slot_attempt_pending[0]); + b._slot_attempt_pending[0]=false; + }; assert(b.reconnectSlotClient(0)==ESP_FAIL && !b._slot_attempt_pending[0]); sdk_callback=nullptr; c._started=false; @@ -259,7 +270,7 @@ struct MQTTBridge { } void maintainSlotConnections() { work(); - if (scenario == 10) _stop_requested = true; + if (scenario == 5 || scenario == 10) _stop_requested = true; } void processPacketQueue() { work(); ++queue_calls; @@ -292,9 +303,9 @@ void vTaskDelay(TickType_t ticks) { task_sleeps.push_back(duration); elapsed_ms += duration; if (elapsed_ms >= stop_at_ms) active_bridge->_stop_requested = true; - // The no-stop scenario exits after one complete worker iteration. Stop is - // raised during its ordinary idle delay, after all setup and publications. - if (scenario == 3 && active_bridge->setups.size() == RUNTIME_MQTT_SLOTS) + // The no-stop scenario exits after one complete worker iteration. Startup + // only schedules slots; connection maintenance now owns their async starts. + if (scenario == 3 && active_bridge->queue_calls == 1) active_bridge->_stop_requested = true; assert(elapsed_ms < 30000); // bound a missing cooperative exit deterministically } @@ -313,11 +324,11 @@ int main(int argc, char** argv) { active_bridge = &bridge; switch (scenario) { case 0: stop_at_ms = 187; break; // stop while WiFi settles - case 1: stop_at_ms = 1173; break; // stop during first TLS stagger - case 2: stop_at_ms = 6191; break; // stop during second TLS stagger + case 1: stop_at_ms = 1173; break; // stop during ordinary service + case 2: stop_at_ms = 6191; break; // stop during later ordinary service case 3: break; // preserve every normal delay case 4: bridge._stop_requested = true; break; - case 5: break; // setup callback requests stop + case 5: break; // maintenance callback requests stop case 6: bridge._ntp_force_requested = true; break; case 7: bridge._ntp_sync_pending = true; bridge._ntp_synced = false; break; case 8: case 9: case 10: case 11: break; // stop raised during worker callbacks @@ -327,28 +338,23 @@ int main(int argc, char** argv) { assert(bridge._stop_acked && bridge.late_work == 0); assert(bridge.teardown == std::vector({ "ntp", "slot0", "slot1", "slot2", "destroy"})); + assert(bridge.setups.empty()); // no blocking startup slot loop remains if (scenario == 3) { - assert(bridge.setups == std::vector({0, 1, 2})); - assert(bridge.setup_times == std::vector({1000, 6000, 11000})); + assert(bridge._slots_setup_done); assert(bridge.reconfigures == RUNTIME_MQTT_SLOTS); assert(bridge.publishes >= RUNTIME_MQTT_SLOTS && bridge.queue_calls == 1); } else { - const size_t wanted = scenario >= 8 ? RUNTIME_MQTT_SLOTS - : scenario == 0 || scenario == 4 || scenario >= 6 ? 0 - : scenario == 2 ? 2 : 1; - assert(bridge.setups.size() == wanted); - if (scenario >= 8) { + if (scenario >= 8 || scenario == 5) { assert(bridge.reconfigures == (scenario == 8 ? 1 : RUNTIME_MQTT_SLOTS)); assert(bridge.publishes == (scenario == 8 ? 0 : scenario == 9 ? 1 : RUNTIME_MQTT_SLOTS)); assert(bridge.queue_calls == (scenario == 11 ? 1 : 0)); - } else { + } else if (scenario != 1 && scenario != 2) { assert(bridge.reconfigures == 0 && bridge.publishes == 0 && bridge.queue_calls == 0); } - const uint32_t poll_bound = std::max(uint32_t{20}, tick_ms); + const uint32_t poll_bound = std::max(uint32_t{50}, tick_ms); assert(std::all_of(task_sleeps.begin(), task_sleeps.end(), [poll_bound](uint32_t duration) { return duration <= poll_bound; })); if (scenario == 4) assert(elapsed_ms == 0 && task_sleeps.empty()); - else if (scenario >= 8) assert(elapsed_ms == 11000); else if (scenario >= 5) assert(elapsed_ms == 1000); else assert(elapsed_ms >= stop_at_ms && elapsed_ms - stop_at_ms < poll_bound); } @@ -368,6 +374,7 @@ ASYNC_STOP_PREAMBLE = r''' #include #include #include "helpers/MQTTLifecycle.h" +#include "helpers/MQTTConnectionPolicy.h" #define ESP_PLATFORM 1 #define MQTT_DEBUG_PRINTLN(...) ((void)0) #define pdMS_TO_TICKS(ms) (ms) @@ -483,7 +490,10 @@ int main() { clock_ms = UINT32_MAX - 500; elapsed_ms = 0; sleeps = 0; bridge_waiting = &bridge; bridge.requestStop(); - ack_after_ms = at_deadline ? bridge._lifecycle.stopTimeoutMs() : 8000; + assert(bridge._lifecycle.stopTimeoutMs() == 45000); + // A ten-second blocked publish plus healthy client teardown can exceed + // the former 29-second budget for three slots without requiring a kill. + ack_after_ms = at_deadline ? bridge._lifecycle.stopTimeoutMs() : 31000; bridge.end(); // synchronous OTA barrier joins the existing request assert(elapsed_ms == ack_after_ms && sleeps > 0 && bridge.ops.signals == 1); assert(!bridge._lifecycle.stopTimedOut()); @@ -525,7 +535,7 @@ class MqttTransportResultsTests(unittest.TestCase): result = subprocess.run([str(exe)], text=True, capture_output=True, timeout=5) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) - def test_actual_worker_interrupts_startup_and_stagger_without_late_work(self): + def test_actual_worker_interrupts_startup_and_service_without_late_work(self): compiler = shutil.which("g++") or shutil.which("c++") self.assertIsNotNone(compiler, "a host C++ compiler is required") bridge = "src/helpers/bridges/MQTTBridge.cpp" @@ -555,6 +565,10 @@ class MqttTransportResultsTests(unittest.TestCase): "esp_err_t PsychicMqttClient::applyConfig()", "esp_err_t PsychicMqttClient::connect()", "esp_err_t PsychicMqttClient::reconnect()")) + source += method("src/helpers/bridges/MQTTBridge.cpp", + "bool MQTTBridge::hasPendingSlotConnection() const") + source += method("src/helpers/bridges/MQTTBridge.cpp", + "bool MQTTBridge::canStartSlotConnection(int index) const") source += method("src/helpers/bridges/MQTTBridge.cpp", "esp_err_t MQTTBridge::reconnectSlotClient(int index)") + MAIN with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-results-") as temp: