From 2fc2f409d4e78ae6805d5edddd92a6f899a731e6 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 27 Oct 2026 11:28:47 -0700 Subject: [PATCH] Keep MQTT and WebConfig stable under internal memory pressure --- .github/workflows/run-unit-tests.yml | 4 + .../src/PsychicMqttClient.cpp | 74 +++- lib/PsychicMqttClient/src/PsychicMqttClient.h | 26 +- src/helpers/MQTTConnectionAdmission.h | 1 + src/helpers/MQTTPublishAdmission.h | 55 +++ src/helpers/MQTTStatusPublishPolicy.h | 32 ++ src/helpers/bridges/MQTTBridge.cpp | 402 +++++------------- src/helpers/bridges/MQTTBridge.h | 25 +- src/helpers/esp32/WebConfigJsonAllocator.h | 85 ++++ src/helpers/esp32/WebConfigServer.cpp | 192 +++++++-- src/helpers/esp32/WebConfigServer.h | 4 + test/fixtures/webconfig_ap_start.cpp | 1 + test/test_mqtt_connection_admission.py | 79 ++-- test/test_mqtt_handshake_scheduler.py | 20 +- test/test_mqtt_network_timeout.py | 11 + test/test_mqtt_publish_ack.py | 78 +++- test/test_mqtt_publish_admission.py | 137 ++++++ test/test_mqtt_reassembly_buffer.py | 214 ++++++++++ test/test_mqtt_slot_status.py | 228 ++++++++++ test/test_mqtt_transport_results.py | 18 +- test/test_mqtt_wifi_network_power.py | 1 + test/test_webconfig_json_allocator.py | 213 ++++++++++ 22 files changed, 1494 insertions(+), 406 deletions(-) create mode 100644 src/helpers/MQTTPublishAdmission.h create mode 100644 src/helpers/MQTTStatusPublishPolicy.h create mode 100644 src/helpers/esp32/WebConfigJsonAllocator.h create mode 100644 test/test_mqtt_publish_admission.py create mode 100644 test/test_mqtt_reassembly_buffer.py create mode 100644 test/test_mqtt_slot_status.py create mode 100644 test/test_webconfig_json_allocator.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index ca217bc92..8789644f7 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -528,7 +528,11 @@ jobs: python3 -B test/test_mqtt_network_timeout.py -v python3 -B test/test_mqtt_disconnect.py -v python3 -B test/test_mqtt_connection_admission.py -v + python3 -B test/test_mqtt_publish_admission.py -v + python3 -B test/test_mqtt_reassembly_buffer.py -v + python3 -B test/test_webconfig_json_allocator.py -v python3 -B test/test_mqtt_handshake_scheduler.py -v + python3 -B test/test_mqtt_slot_status.py -v python3 -B test/test_mqtt_transport_health.py -v python3 -B test/test_mqtt_diagnostics.py -v python3 -B test/test_mqtt_advert_verify.py -v diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp index 54dc75b49..f5ea57b9b 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.cpp +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.cpp @@ -3,6 +3,10 @@ #include "PsychicMqttClient.h" #include +#if defined(BOARD_HAS_PSRAM) +#include +#include +#endif static const char *TAG = "[OCTOPUS]"; @@ -27,11 +31,16 @@ PsychicMqttClient::~PsychicMqttClient() { esp_mqtt_client_destroy(_client); _client = nullptr; + _outbox_size.store(0, std::memory_order_relaxed); } if (_buffer != nullptr) { +#if defined(BOARD_HAS_PSRAM) + heap_caps_free(_buffer); +#else free(_buffer); +#endif _buffer = nullptr; _buffer_capacity = 0; } @@ -428,7 +437,23 @@ esp_err_t PsychicMqttClient::connect() if (_buffer == nullptr) { _buffer_capacity = (size_t)desired_buffer; +#if defined(BOARD_HAS_PSRAM) + // This character buffer is CPU-only; no atomics, DMA or ISR access. + // Retain it through reconnects just like the ordinary heap allocation. + const size_t bytes = _buffer_capacity + 1; + _buffer = static_cast(heap_caps_malloc(bytes, MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT)); + if (_buffer == nullptr) + { + const uint32_t dma_caps = MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA; + if (MQTTConnectionAdmission::canFallback(bytes, + heap_caps_get_free_size(dma_caps), heap_caps_get_largest_free_block(dma_caps))) + { + _buffer = static_cast(heap_caps_malloc(bytes, dma_caps | MALLOC_CAP_8BIT)); + } + } +#else _buffer = (char *)malloc(_buffer_capacity + 1); +#endif if (_buffer == nullptr) { ESP_LOGE(TAG, "Failed to allocate reassembly buffer (%u bytes)", (unsigned)_buffer_capacity); @@ -517,6 +542,7 @@ void PsychicMqttClient::disconnect() } _connected = false; _started = false; + _outbox_size.store(0, std::memory_order_relaxed); ESP_LOGI(TAG, "MQTT client stopped."); } @@ -561,9 +587,16 @@ void PsychicMqttClient::forceStop() { ESP_LOGI(TAG, "Forced stop MQTT client."); } - ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_stop(_client)); + const esp_err_t result = esp_mqtt_client_stop(_client); + ESP_ERROR_CHECK_WITHOUT_ABORT(result); + if (result != ESP_OK) + { + ESP_LOGW(TAG, "Forced MQTT stop failed: %s", esp_err_to_name(result)); + return; + } _connected = false; _started = false; + _outbox_size.store(0, std::memory_order_relaxed); ESP_LOGI(TAG, "MQTT client forcefully stopped."); } @@ -604,18 +637,18 @@ int PsychicMqttClient::publish(const char *topic, int qos, bool retain, const ch // own (esp-mqtt frees entries only on send-ack or time-based expiry), so on a // stalled uplink QoS0 frames pile up on internal heap without limit. Cap it // here: if the outbox is already at/over _outbox_limit, drop this QoS0 message - // and report -2 ("outbox full") so the caller can retry/drop. QoS1/2 (durable, - // low-rate) are never gated. - if (qos == 0 && _outbox_limit > 0 && _client != nullptr && - esp_mqtt_client_get_outbox_size(_client) >= _outbox_limit) + // and report -2 ("outbox full") so the caller can retry/drop. Apply the + // same bound to durable status messages on a stalled connection. + const size_t outbox_size = refreshOutboxSize(); + if (_outbox_limit > 0 && _client != nullptr && outbox_size >= _outbox_limit) { _outbox_drops++; static unsigned long last_full_log = 0; unsigned long now = millis(); if (now - last_full_log > 5000) { - ESP_LOGW(TAG, "Outbox at cap (%u bytes); dropping QoS0 message to topic %s", - (unsigned)_outbox_limit, topic); + ESP_LOGW(TAG, "Outbox at cap (%u bytes); dropping QoS%d message to topic %s", + (unsigned)_outbox_limit, qos, topic); last_full_log = now; } return -2; @@ -625,6 +658,7 @@ int PsychicMqttClient::publish(const char *topic, int qos, bool retain, const ch bool store_in_outbox = true; int result = esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, store_in_outbox); if (result < 0) _publish_err++; else _publish_ok++; + refreshOutboxSize(); return result; } else @@ -635,6 +669,7 @@ int PsychicMqttClient::publish(const char *topic, int qos, bool retain, const ch // callers can surface delivery health without per-message logging. int result = esp_mqtt_client_publish(_client, topic, payload, length, qos, retain); if (result < 0) _publish_err++; else _publish_ok++; + if (qos > 0) refreshOutboxSize(); else serviceOutboxDiagnostics(); return result; } } @@ -661,10 +696,24 @@ PsychicMqttClient &PsychicMqttClient::setOutboxLimit(size_t bytes) size_t PsychicMqttClient::getOutboxSize() { - if (_client == nullptr) - return 0; - int size = esp_mqtt_client_get_outbox_size(_client); - return size > 0 ? (size_t)size : 0; + return _outbox_size.load(std::memory_order_relaxed); +} + +size_t PsychicMqttClient::refreshOutboxSize() +{ + const int size = _client ? esp_mqtt_client_get_outbox_size(_client) : 0; + const size_t bytes = size > 0 ? static_cast(size) : 0; + _outbox_size.store(bytes, std::memory_order_relaxed); + _outbox_sampled_at.store(static_cast(millis()), std::memory_order_relaxed); + return bytes; +} + +void PsychicMqttClient::serviceOutboxDiagnostics() +{ + if (!_started.load(std::memory_order_relaxed) || !getOutboxSize()) return; + const uint32_t now = static_cast(millis()); + if (now - _outbox_sampled_at.load(std::memory_order_relaxed) >= 1000U) + refreshOutboxSize(); } size_t PsychicMqttClient::getOutboxLimit() @@ -862,6 +911,9 @@ void PsychicMqttClient::_onMessage(esp_mqtt_event_handle_t &event) void PsychicMqttClient::_onPublish(esp_mqtt_event_handle_t &event) { + // The SDK removes the acknowledged item before dispatch and its API mutex + // is recursive, so this callback can sample without waiting on another task. + refreshOutboxSize(); _publish_ack.fetch_add(1, std::memory_order_relaxed); ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id); for (uint8_t i = 0; i < _onPublishUserCallbackCount; ++i) diff --git a/lib/PsychicMqttClient/src/PsychicMqttClient.h b/lib/PsychicMqttClient/src/PsychicMqttClient.h index e1bced0fa..717789473 100644 --- a/lib/PsychicMqttClient/src/PsychicMqttClient.h +++ b/lib/PsychicMqttClient/src/PsychicMqttClient.h @@ -436,15 +436,15 @@ public: esp_mqtt_client_config_t *getMqttConfig(); /** - * @brief Caps the size of the esp-mqtt outbox for async QoS 0 publishes. + * @brief Caps the size of the esp-mqtt outbox for asynchronous publishes. * * QoS 0 async publishes are forced into the esp-mqtt outbox (store=true) so * packet topics keep flowing, but the outbox has no size bound of its own -- * it only frees entries on send-ack or time-based expiry. On a stalled uplink * the entries accumulate on internal heap without limit. When the current - * outbox size is at/over this cap, publish() drops the new QoS 0 message + * outbox size is at/over this cap, publish() drops the new message * (returns -2) instead of enqueuing it, applying backpressure. 0 disables the - * cap. QoS 1/2 publishes are never gated by this. + * cap. The publish owner samples the SDK immediately before admission. * * @param bytes Maximum outbox size in bytes, or 0 to disable. * @return A reference to the PsychicMqttClient instance. @@ -452,13 +452,18 @@ public: PsychicMqttClient &setOutboxLimit(size_t bytes); /** - * @brief Returns the current esp-mqtt outbox size in bytes (0 if the client - * is not initialized). Useful for diagnostics/backpressure monitoring. + * @brief Returns the last owner-sampled outbox size without taking the SDK + * mutex. Publish/ACK events update it; nonempty idle outboxes refresh within + * one second when the owner calls serviceOutboxDiagnostics(). * * @return Current outbox size in bytes. */ size_t getOutboxSize(); + // Call only from the client's network owner, never a radio/CLI reader. + // Expiry can remove queued items without a broker acknowledgment event. + void serviceOutboxDiagnostics(); + /** * @brief Returns the configured outbox cap in bytes (0 = disabled). Lets * callers confirm the cap is actually applied to this client. @@ -466,7 +471,7 @@ public: size_t getOutboxLimit(); /** - * @brief Returns the cumulative count of QoS0 messages dropped because the + * @brief Returns the cumulative count of async messages dropped because the * outbox was at/over the cap. Monotonic; useful for backpressure diagnostics. */ unsigned long getOutboxDrops(); @@ -500,10 +505,12 @@ private: bool _config_dirty = true; std::atomic _started{false}; - // Runtime cap on the esp-mqtt outbox for QoS 0 async publishes (bytes). + // Runtime cap on the esp-mqtt outbox for async publishes (bytes). // 0 = disabled. Enforced in publish(); not an esp-mqtt config field. size_t _outbox_limit = 0; - // Cumulative count of QoS0 publishes dropped because the outbox hit the cap. + std::atomic _outbox_size{0}; + std::atomic _outbox_sampled_at{0}; + // Cumulative count of async publishes dropped because the outbox hit the cap. unsigned long _outbox_drops = 0; // Cumulative publish accept/fail counts (any QoS, sync or async path). unsigned long _publish_ok = 0; @@ -513,6 +520,8 @@ private: // Multipart message reassembly. _buffer is lazily allocated at connect() time // to match the configured buffer size, then reused for the client's lifetime. + // PSRAM builds prefer external byte storage; internal fallback preserves + // the network/AES DMA reserve. The client object and task stacks stay internal. // _topic is inline storage, never heap-allocated. char *_buffer = nullptr; size_t _buffer_capacity = 0; @@ -523,6 +532,7 @@ private: // Configuration is transactional: never start/reconnect on stale credentials // after a failed esp_mqtt_set_config(). esp_err_t applyConfig(); + size_t refreshOutboxSize(); bool _isTopicMatch(const char *topic, const char *subscription); // Fixed-size callback storage. Each slot is "used" once registered; we never diff --git a/src/helpers/MQTTConnectionAdmission.h b/src/helpers/MQTTConnectionAdmission.h index e69ed8d3f..1338e7157 100644 --- a/src/helpers/MQTTConnectionAdmission.h +++ b/src/helpers/MQTTConnectionAdmission.h @@ -18,6 +18,7 @@ static const size_t kDmaLargestBytes = 4096; // SDK default or confusing this per-client task with the MQTTBridge worker. static const size_t kClientTaskStackBytes = 6144; static const int kClientTaskPriority = 5; +static const int kClientBufferBytes = 896; static const size_t kColdStartAllowanceBytes = 16384; static const size_t kColdStartLargestBytes = 8192; // SDK stop retains buffers/configuration but releases its task. Restart only diff --git a/src/helpers/MQTTPublishAdmission.h b/src/helpers/MQTTPublishAdmission.h new file mode 100644 index 000000000..0594ac401 --- /dev/null +++ b/src/helpers/MQTTPublishAdmission.h @@ -0,0 +1,55 @@ +#pragma once + +#include "MQTTConnectionAdmission.h" +#include +#include + +namespace MQTTPublishAdmission { + +// QoS0 writes use the SDK's fixed buffer; QoS1 can resend an entire outbox +// record. Keep the existing DMA reserve and account for larger queued records. +static const size_t kLegacyLargestBytes = 8000; +static const size_t kFramingAllowanceBytes = 128; +static const uint32_t kSampleIntervalMs = 5000; + +struct Heap { + bool psram = false; + size_t internal_largest = 0; + size_t dma_free = 0; + size_t dma_largest = 0; +}; + +inline bool canPublish(const Heap& heap, size_t payload_size = 0, + size_t topic_size = 0, uint8_t qos = 0) { + if (payload_size > SIZE_MAX - kFramingAllowanceBytes || + topic_size > SIZE_MAX - kFramingAllowanceBytes - payload_size) return false; + size_t record = qos ? payload_size + topic_size + kFramingAllowanceBytes : 0; + if (record > MQTTConnectionAdmission::kTlsRecordAllocationBytes) + record = MQTTConnectionAdmission::kTlsRecordAllocationBytes; + if (!heap.psram) { + const size_t required = record > kLegacyLargestBytes ? record : kLegacyLargestBytes; + return heap.internal_largest >= required; + } + const size_t largest = record > MQTTConnectionAdmission::kDmaLargestBytes + ? record : MQTTConnectionAdmission::kDmaLargestBytes; + const size_t free = MQTTConnectionAdmission::kDmaReserveBytes + + largest - MQTTConnectionAdmission::kDmaLargestBytes; + return heap.dma_free >= free && heap.dma_largest >= largest; +} + +struct State { + Heap heap; + uint32_t sampled_at = 0; + bool sampled = false; + bool needsSample(uint32_t now) const { + return !sampled || now - sampled_at >= kSampleIntervalMs; + } + void update(uint32_t now, const Heap& snapshot) { + heap = snapshot; + sampled_at = now; + sampled = true; + } + void invalidate() { sampled = false; } +}; + +} // namespace MQTTPublishAdmission diff --git a/src/helpers/MQTTStatusPublishPolicy.h b/src/helpers/MQTTStatusPublishPolicy.h new file mode 100644 index 000000000..a6415953a --- /dev/null +++ b/src/helpers/MQTTStatusPublishPolicy.h @@ -0,0 +1,32 @@ +#pragma once + +#include + +namespace MQTTStatusPublishPolicy { + +// Bridge-task owned. A failed destination keeps its own retry deadline; +// another broker accepting the same snapshot cannot postpone that retry. +struct State { + uint32_t attempted_at = 0; + uint32_t accepted_at = 0; + bool has_attempt = false; + bool has_accepted = false; + bool pending = false; + + void request() { pending = true; has_attempt = false; } + + bool due(uint32_t now, uint32_t interval, uint32_t retry_interval, + bool periodic_enabled) const { + if (pending) return !has_attempt || uint32_t(now - attempted_at) >= retry_interval; + return periodic_enabled && (!has_accepted || uint32_t(now - accepted_at) >= interval); + } + + void attempted(uint32_t now, bool accepted) { + attempted_at = now; + has_attempt = true; + pending = !accepted; + if (accepted) { accepted_at = now; has_accepted = true; } + } +}; + +} // namespace MQTTStatusPublishPolicy diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 7c8429c2f..6e4966a3f 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -287,6 +287,11 @@ 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; + const uint32_t caps = MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA; + if (!MQTTConnectionAdmission::canFallback(size, + heap_caps_get_free_size(caps), heap_caps_get_largest_free_block(caps))) { + return nullptr; + } p = heap_caps_malloc(size, MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT); return p; #else @@ -329,11 +334,16 @@ static void* psram_realloc(void* ptr, size_t new_size) { return nullptr; } #if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) - void* p = heap_caps_realloc(ptr, new_size, MALLOC_CAP_SPIRAM); - 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 | MALLOC_CAP_8BIT); + if (!ptr) return psram_malloc(new_size); + const size_t previous = heap_caps_get_allocated_size(ptr); + // ArduinoJson requires shrinking to succeed. Keep the old block rather + // than moving a shrink into scarce internal RAM when PSRAM is exhausted. + if (new_size <= previous) return ptr; + void* replacement = psram_malloc(new_size); + if (!replacement) return nullptr; + memcpy(replacement, ptr, previous); + psram_free(ptr); + return replacement; #else return realloc(ptr, new_size); #endif @@ -859,7 +869,7 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs, _node_info(node_info), _manage_wifi(manage_wifi), _queue_count(0), - _last_status_publish(0), _last_status_retry(0), _status_interval(300000), + _status_interval(300000), _ntp_client(_ntp_udp, effectiveNtpPrimary(obs), 0, 60000), _last_ntp_sync(0), _ntp_refresh_retry_at(0), _ntp_refresh_started_at(0), _ntp_refresh_previous_epoch(0), @@ -889,7 +899,7 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs, #endif _identity(identity), _cached_has_connected_slots(false), - _last_memory_check(0), _skipped_publishes(0), + _skipped_publishes(0), _last_no_broker_log(0), _queue_disconnected_since(0), _last_config_warning(0), _dispatcher(nullptr), _radio(nullptr), _board(nullptr), _ms(nullptr), @@ -2015,19 +2025,9 @@ void MQTTBridge::mqttTaskLoop() { } if (_stop_requested) continue; - // Publish on-connect status for slots whose onConnect callback fired since - // the last loop. Raised on the esp-mqtt event task, consumed here on the - // bridge task so the shared status doc/buffer/origin are only ever touched - // from Core 0 (see the onConnect handler / A2). Clear before publishing so a - // reconnect during the publish re-arms for the next loop rather than being - // lost; publishStatusToSlot() re-checks slot.connected and no-ops if dropped. - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (_stop_requested) break; - if (_status_publish_pending[i]) { - _status_publish_pending[i] = false; - publishStatusToSlot(i); - } - } + // Status deadlines belong to each slot, including reconnect requests. + // All JSON construction stays on this bridge task. + maintainStatus(static_cast(now)); if (_stop_requested) continue; // Maintain slot connections (token renewal, reconnect with backoff) @@ -2082,63 +2082,6 @@ void MQTTBridge::mqttTaskLoop() { // Automatic boot/reconnect/daily NTP is serviced above as one bounded // asynchronous operation. Explicit administrator diagnostics remain separate. - // Publish status updates (handle millis() overflow correctly). - // Read the toggle live from prefs (like mqtt.packets/rx/tx below) so a - // CLI/web `set mqtt.status` change applies without a bridge restart. - if (_obs->mqtt_status_enabled) { - bool has_destinations = _cached_has_connected_slots; - - // Early exit if no destinations - skip all the expensive logic below - if (!has_destinations) { - if (_last_status_retry != 0) { - _last_status_retry = 0; - } - } else { - bool should_publish = false; - - // First, check if we need to respect retry interval (prevents spam when publish keeps failing) - if (_last_status_retry != 0) { - unsigned long retry_elapsed = (now >= _last_status_retry) ? - (now - _last_status_retry) : - (ULONG_MAX - _last_status_retry + now + 1); - if (retry_elapsed < STATUS_RETRY_INTERVAL) { - should_publish = false; - } else { - should_publish = true; - } - } else { - if (_last_status_publish == 0) { - should_publish = true; - } else { - unsigned long elapsed = (now >= _last_status_publish) ? - (now - _last_status_publish) : - (ULONG_MAX - _last_status_publish + now + 1); - should_publish = (elapsed >= _status_interval); - } - } - - if (should_publish) { - if (_last_status_publish != 0) { - unsigned long elapsed = (now >= _last_status_publish) ? - (now - _last_status_publish) : - (ULONG_MAX - _last_status_publish + now + 1); - MQTT_DEBUG_PRINTLN("Status publish timer expired (elapsed: %lu ms, interval: %lu ms)", elapsed, _status_interval); - } else { - MQTT_DEBUG_PRINTLN("Status publish attempt (first publish or retry)"); - } - - _last_status_retry = now; - if (publishStatus()) { - _last_status_publish = now; - _last_status_retry = 0; - MQTT_DEBUG_PRINTLN("Status published successfully, next publish in %lu ms", _status_interval); - } else { - MQTT_DEBUG_PRINTLN("Status publish failed, will retry in %lu ms", STATUS_RETRY_INTERVAL); - } - } - } - } - // Refresh the aggregate cache periodically; owner maintenance above // reconciles adapter state and bounds missed terminal callbacks. static unsigned long last_slot_status_update = 0; @@ -2393,6 +2336,17 @@ bool MQTTBridge::canStartSlotConnection(int index) const { // 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. +void MQTTBridge::setSlotClientId(int index) { + MQTTSlot& slot = _slots[index]; + // esp-mqtt's default derives only from the MAC, so two slots on one broker + // otherwise evict each other's sessions. The adapter retains this pointer. + const uint64_t chip_id = ESP.getEfuseMac(); + snprintf(slot.client_id, sizeof(slot.client_id), "meshcore%012llXs%u", + static_cast(chip_id & 0xFFFFFFFFFFFFULL), + static_cast(index + 1)); + slot.client->setClientId(slot.client_id); +} + 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 @@ -2465,6 +2419,7 @@ bool MQTTBridge::setupSlot(int index) { _slot_force_jwt_mint[index] = false; } + setSlotClientId(index); bool uses_jwt = (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) || slot.audience[0] != '\0'; optimizeMqttClientConfig(slot.client, uses_jwt); // sets keepalive (45s PSRAM, 75s non-PSRAM) #ifndef MQTT_FORCE_KEEPALIVE_45 @@ -2798,6 +2753,7 @@ void MQTTBridge::serviceSlotTransportHealth(uint32_t now) { const bool pending = _slot_attempt_pending[i].load(std::memory_order_acquire); if (!pending) health.completeAttempt(); if (slot.client) { + slot.client->serviceOutboxDiagnostics(); health.observeAck(static_cast(slot.client->getPublishAck())); if (_slot_transport_fault_pending[i].load(std::memory_order_acquire)) { health.requestRecovery(MQTTTransportHealth::RecoveryReason::TransportError); @@ -3226,6 +3182,30 @@ bool MQTTBridge::createSlotAuthToken(int index) { return false; } +bool MQTTBridge::hasPublishHeadroom(size_t payload_len, size_t topic_len, uint8_t qos) { +#ifdef ESP_PLATFORM + const uint32_t now = static_cast(millis()); + if (_publish_memory.needsSample(now)) { + MQTTPublishAdmission::Heap heap; +#if defined(BOARD_HAS_PSRAM) + heap.psram = ESP.getPsramSize() > 0; +#endif + if (heap.psram) { + heap.dma_free = heap_caps_get_free_size(MALLOC_CAP_DMA); + heap.dma_largest = heap_caps_get_largest_free_block(MALLOC_CAP_DMA); + } else { + heap.internal_largest = ESP.getMaxAllocHeap(); + } + _publish_memory.update(now, heap); + } + if (!MQTTPublishAdmission::canPublish(_publish_memory.heap, payload_len, topic_len, qos)) { + ++_skipped_publishes; + return false; + } +#endif + return true; +} + bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload, size_t payload_len, bool retained, uint8_t qos) { if (_stop_requested.load(std::memory_order_acquire)) return false; if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; @@ -3247,6 +3227,8 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload return false; } + if (!hasPublishHeadroom(payload_len, topic ? strlen(topic) : 0, qos)) return false; + // Publish path by QoS: // - QoS 0 (high-rate packets/raw): SYNCHRONOUS (async=false -> esp_mqtt_client_publish), // which writes straight to the socket. The async/outbox path drains only one queued @@ -3268,6 +3250,7 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload const uint32_t ack_before = static_cast(slot.client->getPublishAck()); int result = slot.client->publish(topic, qos, retained, payload, (int)payload_len, async); if (result < 0) { + _publish_memory.invalidate(); if (!async) { slot.transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::PublishFailure); updateSlotTransportDiag(index, static_cast(millis())); @@ -3345,102 +3328,26 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_ topic_buf, buf_size); } -void MQTTBridge::publishStatusToSlot(int index) { - if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return; - MQTTSlot& slot = _slots[index]; - if (!slot.client || !slot.connected) return; +bool MQTTBridge::publishStatusToSlot(int index) { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; + return publishStatus(static_cast(1u << index)); +} - refreshOriginFromPrefs(); - - // Build per-slot topic (handles IATA check for meshcore, token check for meshrank) - char status_topic[128]; - if (!buildTopicForSlot(index, MSG_STATUS, status_topic, sizeof(status_topic))) { - return; // Slot is missing required topic configuration - } - - // Reuse pre-allocated buffer to avoid heap alloc/free churn under memory pressure. - // _json_scratch_doc/_json_scratch_buffer/_origin are shared with publishStatus() and - // with the packet/raw paths; every one of them runs only on the bridge task (this - // function is reached solely via the _status_publish_pending consumer in - // mqttTaskLoop, never from the onConnect callback thread - see A2), so the accesses - // are serialized and need no mutex. - #if defined(BOARD_HAS_PSRAM) - char fallback_status_buffer[STATUS_JSON_BUFFER_SIZE]; - char* json_buffer = (_json_scratch_buffer != nullptr) ? _json_scratch_buffer : fallback_status_buffer; - #else - char* json_buffer = _json_scratch_buffer; - #endif - - char origin_id[65]; - char timestamp[40]; - char radio_info[64]; - - // Status timestamp: UTC with explicit +00:00 offset, same as packet/raw JSON - // `timestamp` (system clock is UTC - SNTP offset 0; prefs Timezone is separate). - struct timeval now_tv; - gettimeofday(&now_tv, nullptr); - MQTTMessageBuilder::formatIsoTimestampForMqtt(now_tv.tv_sec, now_tv.tv_usec, _timezone, timestamp, sizeof(timestamp)); - - formatRadioInfo(_node_info.freq ? *_node_info.freq : 0.0f, - _node_info.bw ? *_node_info.bw : 0.0f, - _node_info.sf ? *_node_info.sf : 0, - _node_info.cr ? *_node_info.cr : 0, - radio_info, sizeof(radio_info)); - - strncpy(origin_id, _device_id, sizeof(origin_id) - 1); - origin_id[sizeof(origin_id) - 1] = '\0'; - - char client_version[64]; - getClientVersion(client_version, sizeof(client_version)); - - // Collect stats on-demand if sources are available - int battery_mv = -1; - int uptime_secs = -1; - int errors = -1; - int noise_floor = -999; - int tx_air_secs = -1; - int rx_air_secs = -1; - int recv_errors = -1; - int packets_sent = -1; - int packets_received = -1; - - if (_board) battery_mv = _board->getBattMilliVolts(); - if (_ms) uptime_secs = _ms->getMillis() / 1000; - if (_dispatcher) { - errors = _dispatcher->getErrFlags(); - tx_air_secs = _dispatcher->getTotalAirTime() / 1000; - rx_air_secs = _dispatcher->getReceiveAirTime() / 1000; - packets_sent = (int)(_dispatcher->getNumSentFlood() + _dispatcher->getNumSentDirect()); - packets_received = (int)(_dispatcher->getNumRecvFlood() + _dispatcher->getNumRecvDirect()); - } - if (_radio) { - noise_floor = (int16_t)_radio->getNoiseFloor(); - recv_errors = (int)_radio->getPacketsRecvErrors(); - } - - // Internal heap free (for diagnosing repeater hangs from internal heap exhaustion) - int internal_heap_free = (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL); - - int len = MQTTMessageBuilder::buildStatusMessage( - _json_scratch_doc, - _origin, origin_id, _board_model, _firmware_version, radio_info, - client_version, "online", timestamp, json_buffer, STATUS_JSON_BUFFER_SIZE, - battery_mv, uptime_secs, errors, _queue_count, noise_floor, - tx_air_secs, rx_air_secs, recv_errors, internal_heap_free, - packets_sent, packets_received, - repeatStatus(), collectConnHealth() - ); - - if (len > 0) { - // Honor the preset's retain policy, matching publishStatus() -- brokers that - // set allow_retain=false (e.g. waev) reject retained publishes, so this - // on-connect status must not force retain=true. Custom slots default to - // non-retained here too, keeping both status paths consistent. - bool use_retain = slot.preset ? slot.preset->allow_retain : false; - if (!publishToSlot(index, status_topic, json_buffer, static_cast(len), use_retain, 1)) { - MQTT_DEBUG_PRINTLN("MQTT%d status publish failed", index + 1); - } +void MQTTBridge::maintainStatus(uint32_t now) { + uint8_t due_slots = 0; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + if (_stop_requested.load(std::memory_order_acquire)) return; + MQTTSlot& slot = _slots[i]; + if (_status_publish_pending[i].exchange(false, std::memory_order_acq_rel)) + slot.status_publish.request(); + if (!slot.enabled || !slot.client || !slot.connected) continue; + if (!slot.status_publish.due(now, static_cast(_status_interval), + static_cast(STATUS_RETRY_INTERVAL), _obs->mqtt_status_enabled)) continue; + char topic[128]; + if (buildTopicForSlot(i, MSG_STATUS, topic, sizeof(topic))) + due_slots |= static_cast(1u << i); } + if (due_slots) publishStatus(due_slots); } MQTTConnHealth MQTTBridge::collectConnHealth() const { @@ -3886,65 +3793,7 @@ void MQTTBridge::loop() { refreshNTP(); } - // Publish status updates (handle millis() overflow correctly). - // Read the toggle live from prefs so a CLI/web `set mqtt.status` change - // applies without a bridge restart. - if (_obs->mqtt_status_enabled) { - bool has_destinations = _cached_has_connected_slots; - - if (has_destinations) { - unsigned long now = millis(); - bool should_publish = false; - - if (_last_status_retry != 0) { - unsigned long retry_elapsed = (now >= _last_status_retry) ? - (now - _last_status_retry) : - (ULONG_MAX - _last_status_retry + now + 1); - if (retry_elapsed >= STATUS_RETRY_INTERVAL) { - should_publish = true; - } - } else { - if (_last_status_publish == 0) { - should_publish = true; - } else { - unsigned long elapsed = (now >= _last_status_publish) ? - (now - _last_status_publish) : - (ULONG_MAX - _last_status_publish + now + 1); - should_publish = (elapsed >= _status_interval); - } - } - - if (should_publish) { - if (_last_status_publish != 0) { - unsigned long elapsed = (now >= _last_status_publish) ? - (now - _last_status_publish) : - (ULONG_MAX - _last_status_publish + now + 1); - MQTT_DEBUG_PRINTLN("Status publish timer expired (elapsed: %lu ms, interval: %lu ms)", elapsed, _status_interval); - } else { - MQTT_DEBUG_PRINTLN("Status publish attempt (first publish or retry)"); - } - - _last_status_retry = now; - if (publishStatus()) { - _last_status_publish = now; - _last_status_retry = 0; - MQTT_DEBUG_PRINTLN("Status published successfully, next publish in %lu ms", _status_interval); - } else { - MQTT_DEBUG_PRINTLN("Status publish failed, will retry in %lu ms", STATUS_RETRY_INTERVAL); - } - } - } else { - if (_last_status_retry != 0) { - _last_status_retry = 0; - } - } - - // Phase 4 (MQTT memory-defrag): the "recreate on prolonged status failure" - // path and the periodic runCriticalMemoryCheckAndRecovery() call have been - // removed. They were both symptoms of the heap churn introduced by - // delete/new cycles of the MQTT client; with persistent clients the - // allocator stays healthy and these recovery hooks aren't required. - } + maintainStatus(static_cast(millis())); #endif } @@ -4379,11 +4228,18 @@ bool MQTTBridge::shouldQueuePacketType(uint8_t packet_type, bool& filtered) { return false; } -bool MQTTBridge::publishStatus() { - if (!_cached_has_connected_slots) { +bool MQTTBridge::publishStatus(uint8_t slot_mask) { + if (!slot_mask) { return false; } + const uint32_t attempt_time = static_cast(millis()); + // Include JSON/allocation failures in the same per-slot backoff. A failed + // build must not be repeated every worker tick under memory pressure. + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) + if (slot_mask & static_cast(1u << i)) + _slots[i].status_publish.attempted(attempt_time, false); + refreshOriginFromPrefs(); // Reuse pre-allocated buffer to avoid heap alloc/free churn under memory pressure. @@ -4454,30 +4310,24 @@ bool MQTTBridge::publishStatus() { repeatStatus(), collectConnHealth() ); - if (len > 0) { - bool published = false; - bool any_slot_wants_status = false; - char topic[128]; - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (_slots[i].enabled && _slots[i].client && _slots[i].connected) { - if (buildTopicForSlot(i, MSG_STATUS, topic, sizeof(topic))) { - any_slot_wants_status = true; - bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false; - if (publishToSlot(i, topic, json_buffer, (size_t)len, use_retain, 1)) { - published = true; - } - } - } - } - // If no connected slot accepts status topics, treat as success to avoid - // infinite retry loops - if (published || !any_slot_wants_status) { - if (published) MQTT_DEBUG_PRINTLN("Status published"); - return true; + if (len <= 0) return false; + bool all_accepted = true; + char topic[128]; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + if (_stop_requested.load(std::memory_order_acquire)) return false; + if (!(slot_mask & static_cast(1u << i))) continue; + MQTTSlot& slot = _slots[i]; + bool accepted = false; + if (slot.enabled && slot.client && slot.connected + && buildTopicForSlot(i, MSG_STATUS, topic, sizeof(topic))) { + const bool retained = slot.preset ? slot.preset->allow_retain : false; + accepted = publishToSlot(i, topic, json_buffer, static_cast(len), retained, 1); } + slot.status_publish.attempted(attempt_time, accepted); + all_accepted = all_accepted && accepted; + if (!accepted) MQTT_DEBUG_PRINTLN("MQTT%d status deferred; retry in %lu ms", i + 1, STATUS_RETRY_INTERVAL); } - - return false; + return all_accepted; } bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, @@ -4496,40 +4346,9 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, refreshOriginFromPrefs(); - // 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). 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; - #else - static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 8000; - #endif - unsigned long now = millis(); - // Re-sample max-alloc at most once per interval and cache the verdict. - // getMaxAllocHeap() walks the heap free-list, so it must not run per packet. - // The previous code only advanced _last_memory_check on the healthy path, so - // under sustained pressure the guard stayed open and it walked the heap on - // EVERY packet -- the opposite of throttling (A15). Caching the verdict keeps - // the "skip publishes while memory is low" protection but pays for the walk - // only once per interval; _last_memory_check is now advanced on both paths. - if (now - _last_memory_check > 5000) { - _last_memory_check = now; - size_t max_alloc = ESP.getMaxAllocHeap(); - _memory_pressure = (max_alloc < PUBLISH_SKIP_MAX_ALLOC_THRESHOLD); - if (_memory_pressure) { - MQTT_DEBUG_PRINTLN("MQTT: memory pressure, skipping publishes (Max alloc: %d, threshold: %d, skipped: %d)", - (int)max_alloc, (int)PUBLISH_SKIP_MAX_ALLOC_THRESHOLD, _skipped_publishes); - } - } - if (_memory_pressure) { - _skipped_publishes++; - return false; - } - #endif + // JSON pools live in PSRAM. The shared admission keeps the existing DMA + // reserve for writes without requiring a new 16 KiB internal allocation. + if (!hasPublishHeadroom()) return false; // Use pre-allocated buffer; stack fallback only when PSRAM heap alloc may be null. #if defined(BOARD_HAS_PSRAM) @@ -4634,6 +4453,8 @@ bool MQTTBridge::publishRaw(mesh::Packet* packet, bool& has_eligible_target) { // raw): intentionally complete, and no JSON is built. if (!has_eligible_target) return false; + if (!hasPublishHeadroom()) return false; + refreshOriginFromPrefs(); #if defined(BOARD_HAS_PSRAM) @@ -5605,7 +5426,10 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_ // from the first start so explicit non-JWT -> JWT reconfiguration stays valid // without destroying a client that another core may be reading for diagnostics. (void)needs_large_buffer; - static const int MQTT_CLIENT_BUFFER_SIZE = 896; + static const int MQTT_CLIENT_BUFFER_SIZE = MQTTConnectionAdmission::kClientBufferBytes; + static_assert(MQTT_CLIENT_BUFFER_SIZE + MQTTPublishAdmission::kFramingAllowanceBytes + <= MQTTConnectionAdmission::kDmaLargestBytes, + "Publish admission must cover the SDK's QoS0 wire buffer"); client->setBufferSize(MQTT_CLIENT_BUFFER_SIZE); diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 589e2135c..436cdd264 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -13,6 +13,8 @@ #include "helpers/MQTTConnectionPolicy.h" #include "helpers/MQTTConnectionHealth.h" #include "helpers/MQTTTransportHealth.h" +#include "helpers/MQTTPublishAdmission.h" +#include "helpers/MQTTStatusPublishPolicy.h" #include "helpers/MQTTPayloadBuilder.h" #include "helpers/MQTTPacketFilter.h" #include "helpers/MQTTPresets.h" @@ -108,6 +110,7 @@ private: // Connection slot - each slot holds one MQTT connection struct MQTTSlot { PsychicMqttClient* client; + char client_id[32]; // Persistent: the adapter retains this pointer const MQTTPresetDef* preset; // Points to MQTT_PRESETS[] entry, nullptr for custom/none bool enabled; // true when preset is not "none" bool connected; // Updated in callbacks @@ -163,6 +166,7 @@ private: // Only the bridge worker mutates this state; SDK callbacks raise atomic // latches instead. Retained across reconnects to expose recovery history. MQTTTransportHealth::State transport_health; + MQTTStatusPublishPolicy::State status_publish; }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; @@ -182,7 +186,6 @@ private: bool _raw_enabled; bool _rx_enabled; uint8_t _tx_mode; // 0=off, 1=all TX, 2=self-advert only - unsigned long _last_status_publish; unsigned long _status_interval; // Packet queue for offline scenarios @@ -278,8 +281,8 @@ private: // task (Core 0). publishStatusToSlot() touches the shared status doc/buffer/ // origin that publishStatus() also uses, so it must run only on the bridge // task -- the callback just raises this flag. Same idiom as - // _slot_reconfigure_pending; a single-byte volatile store/load is atomic. - volatile bool _status_publish_pending[RUNTIME_MQTT_SLOTS]; + // _slot_reconfigure_pending; exchange consumes each reconnect edge safely. + std::atomic _status_publish_pending[RUNTIME_MQTT_SLOTS]; // CLI-requested forced NTP sync, marshalled onto the MQTT task (Core 0). // All NTP I/O (_ntp_client, configTime) must run on Core 0; the CLI thread @@ -412,21 +415,18 @@ private: // feeds capacity(), so the object is 64 bytes and every pool comes from the allocator. JsonDocument _json_scratch_doc{&_json_allocator}; - // Memory pressure monitoring (per-publish skip; see publishPacket()). + // Owner-task admission shared by packet, raw and status publication. // The broader fragmentation-recovery machinery was removed in Phase 4 of // 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 + MQTTPublishAdmission::State _publish_memory; + int _skipped_publishes; // Exposed via SNMP; attempts suppressed by memory admission // Packets rejected by the per-slot filters before reaching the queue. Written // on Core 1 (radio callbacks), read on Core 0 for `mqtt.stats`; a torn read of // a diagnostic counter is harmless, so no atomic is warranted. unsigned long _filtered_packets = 0; unsigned long _bad_sig_adverts = 0; // RX adverts dropped by mqtt.advert.verify - // Status publish retry tracking - unsigned long _last_status_retry; // Track last retry attempt (separate from successful publish) static const unsigned long STATUS_RETRY_INTERVAL = 30000; // Retry every 30 seconds if failed // Device identity for JWT token creation @@ -512,6 +512,7 @@ private: void releaseSlotAuthToken(int index);// Free the token buffer (only with the client -- see MQTTSlot) void destroySlotClients(bool force = false); bool setupSlot(int index); // Configure and connect the slot; false = not activated + void setSlotClientId(int index); // Single definition of "this slot holds one of the _max_active_slots positions": // it is enabled and has been through a successful setupSlot(). Startup, the // setup-retry path, and live reconfigure all gate on these so the cap cannot be @@ -533,13 +534,15 @@ private: // knows it, and passing it avoids re-scanning up to 2 KB of JSON per destination // slot (and up to NEIGHBORS_JSON_BUFFER_SIZE per neighbor snapshot). bool publishToSlot(int index, const char* topic, const char* payload, size_t payload_len, bool retained = false, uint8_t qos = 0); + bool hasPublishHeadroom(size_t payload_len = 0, size_t topic_len = 0, uint8_t qos = 0); bool publishToAllSlots(const char* topic, const char* payload, size_t payload_len, bool retained = false, uint8_t qos = 0); - void publishStatusToSlot(int index); + bool publishStatusToSlot(int index); + void maintainStatus(uint32_t now); MQTTConnHealth collectConnHealth() const; void updateCachedConnectionStatus(); void processPacketQueue(); - bool publishStatus(); // Returns true if status was successfully published + bool publishStatus(uint8_t slot_mask); // True when every selected destination accepted it bool handleWiFiConnection(unsigned long now); // FreeRTOS task function (runs on Core 0) diff --git a/src/helpers/esp32/WebConfigJsonAllocator.h b/src/helpers/esp32/WebConfigJsonAllocator.h new file mode 100644 index 000000000..424de541b --- /dev/null +++ b/src/helpers/esp32/WebConfigJsonAllocator.h @@ -0,0 +1,85 @@ +#pragma once + +#include +#include +#include + +#if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) +#include +#include +#endif + +namespace mesh { namespace webconfig { + +// JSON pools and strings are CPU workspaces, never SDK/DMA buffers. Avoid +// ordinary malloc's small-allocation preference for scarce internal RAM. +class JsonAllocator : public ArduinoJson::Allocator { +public: + void* allocate(size_t size) override { + if (!size) return nullptr; +#if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) + void* memory = heap_caps_malloc(size, MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT); + if (memory) return memory; + const uint32_t caps = MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA; + if (!MQTTConnectionAdmission::canFallback(size, heap_caps_get_free_size(caps), + heap_caps_get_largest_free_block(caps))) return nullptr; + return heap_caps_malloc(size, MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT); +#else + return malloc(size); +#endif + } + + void deallocate(void* memory) override { +#if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) + heap_caps_free(memory); +#else + free(memory); +#endif + } + + void* reallocate(void* memory, size_t size) override { + if (!memory) return allocate(size); + if (!size) { deallocate(memory); return nullptr; } +#if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) + const size_t previous = heap_caps_get_allocated_size(memory); + // ArduinoJson shrinks pools and string buffers after parsing. Keeping the + // original allocation is valid and cannot turn a shrink into a failure or + // move PSRAM data into internal RAM. It is freed with the short-lived doc. + if (size <= previous) return memory; + void* replacement = allocate(size); + if (!replacement) return nullptr; // realloc failure leaves the old data valid + memcpy(replacement, memory, previous); + deallocate(memory); + return replacement; +#else + return realloc(memory, size); +#endif + } + + static JsonAllocator* instance() { + static JsonAllocator allocator; + return &allocator; + } +}; + +#if defined(ESP_PLATFORM) && defined(BOARD_HAS_PSRAM) +// Match ArduinoJson 7's compatibility wrappers, including their nominal +// capacity() values. Pool growth and every existing route limit stay intact. +class DynamicJsonDocument : public ArduinoJson::JsonDocument { + size_t capacity_; +public: + explicit DynamicJsonDocument(size_t capacity) + : ArduinoJson::JsonDocument(JsonAllocator::instance()), capacity_(capacity) {} + size_t capacity() const { return capacity_; } +}; +template class StaticJsonDocument : public ArduinoJson::JsonDocument { +public: + StaticJsonDocument() : ArduinoJson::JsonDocument(JsonAllocator::instance()) {} + size_t capacity() const { return Capacity; } +}; +#else +using DynamicJsonDocument = ArduinoJson::DynamicJsonDocument; +template using StaticJsonDocument = ArduinoJson::StaticJsonDocument; +#endif + +} } // namespace mesh::webconfig diff --git a/src/helpers/esp32/WebConfigServer.cpp b/src/helpers/esp32/WebConfigServer.cpp index 201e73162..8fd949ae8 100644 --- a/src/helpers/esp32/WebConfigServer.cpp +++ b/src/helpers/esp32/WebConfigServer.cpp @@ -2364,7 +2364,12 @@ void WebConfigServer::handleStatus(AsyncWebServerRequest* req) { // duplicated in the page, which cannot know how this build was sized. doc["max_cmds"] = MAX_BATCH; + if (doc.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } AsyncResponseStream* res = req->beginResponseStream("application/json"); + if (!res) { req->send(503); return; } serializeJson(doc, *res); req->send(res); } @@ -2407,7 +2412,13 @@ void WebConfigServer::handleLogin(AsyncWebServerRequest* req) { } const char* body = (const char*)req->_tempObject; DynamicJsonDocument doc(256); - if (!body || deserializeJson(doc, body) != DeserializationError::Ok) { + const auto json_error = body ? deserializeJson(doc, body) + : DeserializationError(DeserializationError::InvalidInput); + if (json_error == DeserializationError::NoMemory) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } + if (json_error != DeserializationError::Ok) { req->send(400, "application/json", "{\"error\":\"bad request\"}"); return; } @@ -2601,7 +2612,12 @@ void WebConfigServer::handleConfigGet(AsyncWebServerRequest* req) { #endif } + if (doc.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } AsyncResponseStream* res = req->beginResponseStream("application/json"); + if (!res) { req->send(503); return; } serializeJson(doc, *res); req->send(res); } @@ -2615,7 +2631,13 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { } const char* body = (const char*)req->_tempObject; DynamicJsonDocument doc(6144); - if (!body || deserializeJson(doc, body) != DeserializationError::Ok) { + const auto json_error = body ? deserializeJson(doc, body) + : DeserializationError(DeserializationError::InvalidInput); + if (json_error == DeserializationError::NoMemory) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } + if (json_error != DeserializationError::Ok) { req->send(400, "application/json", "{\"error\":\"bad json\"}"); return; } @@ -2650,6 +2672,10 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { ack["state"] = WebConfigBatch::replayStateName(bstate); ack["count"] = _batch_count; ack["reqid"] = (const char*)_batch_reqid; + if (ack.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(ack, out); req->send(202, "application/json", out); @@ -2659,6 +2685,10 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { StaticJsonDocument<96> busy; busy["error"] = "busy"; busy["reqid"] = (const char*)_batch_reqid; + if (busy.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(busy, out); req->send(409, "application/json", out); @@ -2699,6 +2729,10 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { StaticJsonDocument<128> err; err["error"] = "bad key"; err["key"] = safe_key; + if (err.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(err, out); req->send(400, "application/json", out); @@ -2710,6 +2744,10 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { StaticJsonDocument<128> err; err["error"] = "ESP-NOW channel must be 1-13"; err["key"] = key; + if (err.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(err, out); req->send(400, "application/json", out); @@ -2754,6 +2792,10 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { StaticJsonDocument<128> ed; ed["error"] = "value too long"; ed["key"] = safe_key; + if (ed.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(ed, out); req->send(400, "application/json", out); @@ -2766,6 +2808,17 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { req->send(400, "application/json", "{\"error\":\"no changes\"}"); return; } + StaticJsonDocument<96> ack; + ack["state"] = "pending"; + ack["count"] = count; + ack["reqid"] = reqid; + if (ack.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } + String out; + serializeJson(ack, out); + _batch_kind = BATCH_CONFIG; _batch_count = count; _batch_next = 0; @@ -2787,12 +2840,6 @@ void WebConfigServer::handleConfigPost(AsyncWebServerRequest* req) { "WC: config POST accepted, %d cmds, reboot=%d\n", count, (int)reboot_after); - StaticJsonDocument<96> ack; - ack["state"] = "pending"; - ack["count"] = count; - ack["reqid"] = (const char*)_batch_reqid; - String out; - serializeJson(ack, out); req->send(202, "application/json", out); } @@ -2836,6 +2883,10 @@ void WebConfigServer::handleConfigResult(AsyncWebServerRequest* req) { StaticJsonDocument<64> idle; idle["state"] = "idle"; idle["reqid"] = requested_reqid; + if (idle.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(idle, out); req->send(200, "application/json", out); @@ -2850,6 +2901,10 @@ void WebConfigServer::handleConfigResult(AsyncWebServerRequest* req) { pending["state"] = "pending"; pending["reqid"] = (const char*)_batch_reqid; pending["reauth_required"] = _batch_reauth_required && !authenticated; + if (pending.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(pending, out); req->send(200, "application/json", out); @@ -2865,11 +2920,6 @@ void WebConfigServer::handleConfigResult(AsyncWebServerRequest* req) { WebConfigBatch::doneReportsReboot(_batch_reboot, _batch_all_ok); doc["all_ok"] = _batch_all_ok; doc["reauth_required"] = _batch_reauth_required && !authenticated; - if (_batch_reauth_required && !_batch_reauth_delivered) { - _batch_reauth_delivered = true; - if (_ap_rekey_generation) - _ap_rekey_after = WebConfigBatch::scheduleAt(millis(), 3000); - } doc["reqid"] = (const char*)_batch_reqid; if (_setup_wifi_handoff_ip[0]) { doc["wifi_ip"] = (const char*)_setup_wifi_handoff_ip; @@ -2884,19 +2934,20 @@ void WebConfigServer::handleConfigResult(AsyncWebServerRequest* req) { } // State stays DONE (re-readable) until the next POST claims the slot. const uint32_t result_now = millis(); - if (WebConfigBatch::shouldArmConfirmReboot( + const bool arm_reboot = WebConfigBatch::shouldArmConfirmReboot( toSpecState(_batch_state), _batch_reboot, - _batch_all_ok, _batch_reboot_armed)) { + _batch_all_ok, _batch_reboot_armed); + uint32_t confirmed_reboot_at = 0; + if (arm_reboot) { // Confirmation delivered. The handoff page remains rendered in the // browser after the AP disappears, so it only needs a one-second response // flush before reboot. Armed once; re-reads cannot push the deadline out. - _batch_reboot_armed = true; const bool setup_handoff = _setup_wifi_handoff_ip[0] != 0; const uint32_t reboot_delay = setup_handoff ? WebConfigBatch::kSetupHandoffRebootConfirmMs : WebConfigBatch::kRebootConfirmMs; doc["reboot_in_ms"] = reboot_delay; - _reboot_at = setup_handoff + confirmed_reboot_at = setup_handoff ? WebConfigBatch::confirmSetupHandoffRebootAt(result_now) : WebConfigBatch::confirmRebootAt(result_now); } else if (WebConfigBatch::doneReportsReboot( @@ -2907,7 +2958,21 @@ void WebConfigServer::handleConfigResult(AsyncWebServerRequest* req) { result_now, _reboot_at) ? 1u : _reboot_at - result_now; } + if (doc.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } AsyncResponseStream* res = req->beginResponseStream("application/json"); + if (!res) { req->send(503); return; } + if (_batch_reauth_required && !_batch_reauth_delivered) { + _batch_reauth_delivered = true; + if (_ap_rekey_generation) + _ap_rekey_after = WebConfigBatch::scheduleAt(millis(), 3000); + } + if (arm_reboot) { + _batch_reboot_armed = true; + _reboot_at = confirmed_reboot_at; + } serializeJson(doc, *res); req->send(res); } @@ -3023,7 +3088,12 @@ void WebConfigServer::handleTerminalPost(AsyncWebServerRequest* req) { return; } DynamicJsonDocument doc(2048); - if (deserializeJson(doc, static_cast(req->_tempObject))) { + const auto json_error = deserializeJson(doc, static_cast(req->_tempObject)); + if (json_error == DeserializationError::NoMemory) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } + if (json_error) { req->send(400, "application/json", "{\"error\":\"bad json\"}"); return; } @@ -3117,7 +3187,12 @@ void WebConfigServer::handleTerminalOutput(AsyncWebServerRequest* req) { doc["seq"] = _terminal->queue.sequence; doc["done"] = !_terminal->queue.pending; doc["closed"] = _terminal->closed; + if (doc.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } AsyncResponseStream* response = req->beginResponseStream("application/json"); + if (!response) { req->send(503); return; } response->addHeader("Cache-Control", "no-store"); serializeJson(doc, *response); req->send(response); @@ -3146,7 +3221,13 @@ void WebConfigServer::handleCliPost(AsyncWebServerRequest* req) { } const char* body = (const char*)req->_tempObject; DynamicJsonDocument doc(6144); - if (!body || deserializeJson(doc, body) != DeserializationError::Ok) { + const auto json_error = body ? deserializeJson(doc, body) + : DeserializationError(DeserializationError::InvalidInput); + if (json_error == DeserializationError::NoMemory) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } + if (json_error != DeserializationError::Ok) { req->send(400, "application/json", "{\"error\":\"bad json\"}"); return; } @@ -3180,6 +3261,10 @@ void WebConfigServer::handleCliPost(AsyncWebServerRequest* req) { ack["state"] = (bstate == WebConfigBatch::State::Done) ? "done" : "running"; ack["total"] = _batch_count; ack["reqid"] = (const char*)_batch_reqid; + if (ack.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(ack, out); req->send(202, "application/json", out); @@ -3189,6 +3274,10 @@ void WebConfigServer::handleCliPost(AsyncWebServerRequest* req) { StaticJsonDocument<96> bd; bd["error"] = "busy"; bd["reqid"] = (const char*)_batch_reqid; + if (bd.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(bd, out); req->send(409, "application/json", out); @@ -3204,6 +3293,10 @@ void WebConfigServer::handleCliPost(AsyncWebServerRequest* req) { StaticJsonDocument<96> ed; ed["error"] = "too many commands"; ed["max"] = MAX_BATCH; + if (ed.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(ed, out); req->send(413, "application/json", out); @@ -3241,6 +3334,10 @@ void WebConfigServer::handleCliPost(AsyncWebServerRequest* req) { if (why) { StaticJsonDocument<256> ed; ed["error"] = why; + if (ed.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(ed, out); req->send(400, "application/json", out); @@ -3270,6 +3367,17 @@ void WebConfigServer::handleCliPost(AsyncWebServerRequest* req) { return; } + StaticJsonDocument<96> ack; + ack["state"] = "running"; + ack["total"] = count; + ack["reqid"] = reqid; + if (ack.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } + String out; + serializeJson(ack, out); + _batch_kind = BATCH_CLI; _batch_count = count; _batch_next = 0; @@ -3284,12 +3392,6 @@ void WebConfigServer::handleCliPost(AsyncWebServerRequest* req) { "WC: cli POST accepted, %d cmds, reboot=%d\n", count, (int)defer_reboot); - StaticJsonDocument<96> ack; - ack["state"] = "running"; - ack["total"] = count; - ack["reqid"] = (const char*)_batch_reqid; - String out; - serializeJson(ack, out); req->send(202, "application/json", out); } @@ -3343,6 +3445,10 @@ void WebConfigServer::handleCliResult(AsyncWebServerRequest* req) { StaticJsonDocument<64> idle; idle["state"] = "idle"; idle["reqid"] = requested_reqid; + if (idle.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } String out; serializeJson(idle, out); req->send(200, "application/json", out); @@ -3379,24 +3485,30 @@ void WebConfigServer::handleCliResult(AsyncWebServerRequest* req) { : WebConfigBatch::cliWriteSucceeded(_batch[i].reply) ? "OK" : "Sign in again to read this result"; } + const bool arm_reboot = final_read && WebConfigBatch::shouldArmConfirmReboot( + toSpecState(_batch_state), _batch_reboot, _batch_all_ok, _batch_reboot_armed); if (final_read) { - if (_batch_reauth_required && !_batch_reauth_delivered) { - _batch_reauth_delivered = true; - if (_ap_rekey_generation) - _ap_rekey_after = WebConfigBatch::scheduleAt(millis(), 3000); - } doc["all_ok"] = _batch_all_ok; const bool rebooting = WebConfigBatch::cliRebootAllowed(_batch_reboot, _batch_all_ok); doc["reboot"] = rebooting; // Tell the operator why a `reboot` they asked for is not happening. if (_batch_reboot && !_batch_all_ok) doc["reboot_withheld"] = true; - if (WebConfigBatch::shouldArmConfirmReboot(toSpecState(_batch_state), _batch_reboot, - _batch_all_ok, _batch_reboot_armed)) { - _batch_reboot_armed = true; - _reboot_at = WebConfigBatch::confirmRebootAt(millis()); - } + } + if (doc.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; } AsyncResponseStream* res = req->beginResponseStream("application/json"); + if (!res) { req->send(503); return; } + if (final_read && _batch_reauth_required && !_batch_reauth_delivered) { + _batch_reauth_delivered = true; + if (_ap_rekey_generation) + _ap_rekey_after = WebConfigBatch::scheduleAt(millis(), 3000); + } + if (arm_reboot) { + _batch_reboot_armed = true; + _reboot_at = WebConfigBatch::confirmRebootAt(millis()); + } serializeJson(doc, *res); req->send(res); } @@ -3452,7 +3564,12 @@ void WebConfigServer::handleScan(AsyncWebServerRequest* req) { net["enc"] = WiFi.encryptionType(i) != WIFI_AUTH_OPEN; net["channel"] = WiFi.channel(i); } + if (doc.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } AsyncResponseStream* res = req->beginResponseStream("application/json"); + if (!res) { req->send(503); return; } serializeJson(doc, *res); req->send(res); } @@ -3485,7 +3602,12 @@ void WebConfigServer::handlePresets(AsyncWebServerRequest* req) { o["needs"] = "none"; } } + if (doc.overflowed()) { + req->send(503, "application/json", "{\"error\":\"not enough memory\"}"); + return; + } AsyncResponseStream* res = req->beginResponseStream("application/json"); + if (!res) { req->send(503); return; } serializeJson(doc, *res); req->send(res); #else diff --git a/src/helpers/esp32/WebConfigServer.h b/src/helpers/esp32/WebConfigServer.h index eb7b31c0b..96bbad80a 100644 --- a/src/helpers/esp32/WebConfigServer.h +++ b/src/helpers/esp32/WebConfigServer.h @@ -42,6 +42,7 @@ #include #include #include +#include "WebConfigJsonAllocator.h" class AsyncWebServer; class AsyncWebServerRequest; @@ -75,6 +76,9 @@ class WebTerminalStream; #endif class WebConfigServer { + using DynamicJsonDocument = mesh::webconfig::DynamicJsonDocument; + template + using StaticJsonDocument = mesh::webconfig::StaticJsonDocument; public: enum Mode : uint8_t { MODE_OFF = 0, MODE_SETUP, MODE_CONNECTING, MODE_LAN }; diff --git a/test/fixtures/webconfig_ap_start.cpp b/test/fixtures/webconfig_ap_start.cpp index a943e73fb..7f9d56195 100644 --- a/test/fixtures/webconfig_ap_start.cpp +++ b/test/fixtures/webconfig_ap_start.cpp @@ -318,6 +318,7 @@ struct DynamicJsonDocument { std::string state; unsigned networks = 0; explicit DynamicJsonDocument(unsigned) {} + bool overflowed() const { return false; } JsonValue operator[](const char*) { return {&state}; } JsonArray createNestedArray(const char*) { return {&networks}; } }; diff --git a/test/test_mqtt_connection_admission.py b/test/test_mqtt_connection_admission.py index e47334027..4a668d05c 100644 --- a/test/test_mqtt_connection_admission.py +++ b/test/test_mqtt_connection_admission.py @@ -221,55 +221,52 @@ int main() { #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; +constexpr uint32_t MALLOC_CAP_INTERNAL=1, MALLOC_CAP_DMA=2, + MALLOC_CAP_SPIRAM=4, MALLOC_CAP_8BIT=8; +static bool psram_fails=false, internal_fails=false; +static size_t dma_free=100000,dma_largest=100000,internal_calls=0,free_calls=0,malloc_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; +static void* freed=nullptr; +size_t heap_caps_get_free_size(uint32_t caps){assert(caps==3);return dma_free;} +size_t heap_caps_get_largest_free_block(uint32_t caps){assert(caps==3);return dma_largest;} +size_t heap_caps_get_allocated_size(void*ptr){assert(ptr==original);return sizeof(original);} +void* heap_caps_malloc(size_t size,uint32_t caps){ + assert(size==32||size==48);++malloc_calls; + if(caps==MALLOC_CAP_SPIRAM)return psram_fails?nullptr:psram_storage; + assert(caps==(MALLOC_CAP_INTERNAL|MALLOC_CAP_8BIT)); + ++internal_calls;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; } +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); +int main(){ + assert(psram_malloc(0)==nullptr&&malloc_calls==0); + assert(psram_malloc(32)==psram_storage&&internal_calls==0); + psram_fails=true;dma_free=16384+31; + assert(psram_malloc(32)==nullptr&&internal_calls==0); + dma_free=16384+32;dma_largest=31; + assert(psram_malloc(32)==nullptr&&internal_calls==0); + dma_largest=32;assert(psram_malloc(32)==internal_storage&&internal_calls==1); + memset(original,0xa5,sizeof(original)); + const auto before=malloc_calls; + dma_free=0;assert(psram_realloc(original,16)==original&&malloc_calls==before); + assert(psram_realloc(original,48)==nullptr&&free_calls==0); + for(unsigned char byte:original)assert(byte==0xa5); + psram_fails=false; + assert(psram_realloc(original,48)==psram_storage&&freed==original&&free_calls==1); + assert(!memcmp(original,psram_storage,sizeof(original))); + psram_fails=true;dma_free=16384+48;dma_largest=48; + assert(psram_realloc(original,48)==internal_storage&&free_calls==2); + assert(!memcmp(original,internal_storage,sizeof(original))); + internal_fails=true; + assert(psram_realloc(original,48)==nullptr&&free_calls==2); + for(unsigned char byte:original)assert(byte==0xa5); + assert(psram_realloc(original,0)==nullptr&&free_calls==3); + assert(psram_realloc(nullptr,0)==nullptr&&free_calls==3); } ''') diff --git a/test/test_mqtt_handshake_scheduler.py b/test/test_mqtt_handshake_scheduler.py index c48ec1d55..43682ae46 100644 --- a/test/test_mqtt_handshake_scheduler.py +++ b/test/test_mqtt_handshake_scheduler.py @@ -26,9 +26,11 @@ PREAMBLE = r''' #include #include "helpers/AlertFaultPolicy.h" #include "helpers/MQTTConnectionAdmission.h" +#include "helpers/MQTTPublishAdmission.h" #include "helpers/MQTTConnectionHealth.h" #include "helpers/MQTTConnectionPolicy.h" #include "helpers/MQTTTransportHealth.h" +#include "helpers/MQTTStatusPublishPolicy.h" #include "helpers/WiFiPowerSave.h" #define ESP_PLATFORM 1 #define MQTT_DEBUG_PRINTLN(...) ((void)0) @@ -158,6 +160,7 @@ struct PsychicMqttClient { dma_free += 44000; dma_largest = 20000; } void setCredentials(const char*, const char*) { ++credentials; } + void serviceOutboxDiagnostics() {} }; struct MQTTBridge { struct MQTTSlot { @@ -179,8 +182,10 @@ struct MQTTBridge { int last_sock_errno = 0; unsigned long last_error_time = 0; MQTTTransportHealth::State transport_health; + MQTTStatusPublishPolicy::State status_publish; } _slots[RUNTIME_MQTT_SLOTS]; PsychicMqttClient clients[RUNTIME_MQTT_SLOTS]; + MQTTPublishAdmission::State _publish_memory; 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}; @@ -191,7 +196,7 @@ struct MQTTBridge { std::atomic _slot_health_reason[RUNTIME_MQTT_SLOTS] = {}; std::atomic _slot_health_recovering[RUNTIME_MQTT_SLOTS] = {}; std::atomic _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS] = {}; - bool _status_publish_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _status_publish_pending[RUNTIME_MQTT_SLOTS] = {}; std::atomic _stop_requested{false}, _ntp_synced{true}; bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {}; void* _identity = this; @@ -206,7 +211,7 @@ struct MQTTBridge { float *freq = nullptr, *bw = nullptr; int *sf = nullptr, *cr = nullptr; } _node_info; - struct Observer { uint8_t wifi_power_save = 1; } observer; + struct Observer { uint8_t wifi_power_save = 1; bool mqtt_status_enabled = true; } observer; Observer* _obs = &observer; AlertFaultPolicy::OutageSnapshot _wifi_outage{}; unsigned long _last_slot_reconnect_ms = 0; @@ -217,6 +222,8 @@ struct MQTTBridge { const char* _jwt_username = "jwt"; static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000; static const unsigned long SLOT_LOG_INTERVAL = 60000; + static const unsigned long STATUS_RETRY_INTERVAL = 30000; + unsigned long _status_interval = 300000; static constexpr int STATUS_JSON_BUFFER_SIZE = 1024, MSG_STATUS = 0; const char *_timezone = "UTC", *_origin = "node", *_device_id = "id"; const char *_board_model = "V4", *_firmware_version = "test"; @@ -276,7 +283,10 @@ struct MQTTBridge { void updateCachedConnectionStatus() {} bool ensureSlotClient(int index); bool publishToSlot(int, const char*, const char*, size_t, bool, uint8_t); - void publishStatusToSlot(int); + bool hasPublishHeadroom(size_t = 0, size_t = 0, uint8_t = 0) { return true; } + bool publishStatusToSlot(int); + bool publishStatus(uint8_t); + void maintainStatus(uint32_t); void only(int index) { for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) _slots[i].enabled = i == index; } @@ -310,7 +320,9 @@ class MqttHandshakeSchedulerTest(unittest.TestCase): "void MQTTBridge::updateSlotTransportDiag(", "bool MQTTBridge::ensureSlotClient(", "bool MQTTBridge::publishToSlot(", - "void MQTTBridge::publishStatusToSlot(", + "bool MQTTBridge::publishStatusToSlot(", + "bool MQTTBridge::publishStatus(", + "void MQTTBridge::maintainStatus(", "unsigned long MQTTBridge::slotTokenLifetime(int index) const", "bool MQTTBridge::handleWiFiConnection(unsigned long now)", "void MQTTBridge::maintainSlotConnections()", diff --git a/test/test_mqtt_network_timeout.py b/test/test_mqtt_network_timeout.py index 5d3fc1a20..bef58c9ea 100644 --- a/test/test_mqtt_network_timeout.py +++ b/test/test_mqtt_network_timeout.py @@ -20,6 +20,7 @@ PREAMBLE = r''' #include #include #include "helpers/MQTTConnectionAdmission.h" +#include "helpers/MQTTPublishAdmission.h" #include "helpers/MQTTConnectionPolicy.h" #include "helpers/MQTTTransportHealth.h" struct esp_mqtt_client_config_t { @@ -98,12 +99,19 @@ int main() { RECONFIGURE_PREAMBLE = r''' #include "helpers/MQTTConnectionAdmission.h" +#include "helpers/MQTTPublishAdmission.h" #include "helpers/MQTTConnectionPolicy.h" #include "helpers/MQTTTransportHealth.h" #define MQTT_DEBUG_PRINTLN(...) ((void)0) #define pdMS_TO_TICKS(value) (value) constexpr int RUNTIME_MQTT_SLOTS = 2, MQTT_AUTH_JWT = 1; constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2; +constexpr uint32_t MALLOC_CAP_SPIRAM = 4, MALLOC_CAP_8BIT = 8; +void* heap_caps_malloc(size_t bytes, uint32_t caps) { + assert(caps == (MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT)); + return std::malloc(bytes); +} +void heap_caps_free(void* memory) { std::free(memory); } static size_t dma_free = 100000, dma_largest = 20000; size_t heap_caps_get_free_size(uint32_t caps) { assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_free; @@ -329,6 +337,9 @@ class MqttNetworkTimeoutTest(unittest.TestCase): "void PsychicMqttClient::_onDisconnect(", "void PsychicMqttClient::_onSubscribe(", "void PsychicMqttClient::_onUnsubscribe(", + "size_t PsychicMqttClient::getOutboxSize()", + "size_t PsychicMqttClient::refreshOutboxSize()", + "void PsychicMqttClient::serviceOutboxDiagnostics()", "void PsychicMqttClient::_onPublish(", )) source += "\n".join(method("src/helpers/bridges/MQTTBridge.cpp", signature) diff --git a/test/test_mqtt_publish_ack.py b/test/test_mqtt_publish_ack.py index d37c0c682..eee866fc4 100644 --- a/test/test_mqtt_publish_ack.py +++ b/test/test_mqtt_publish_ack.py @@ -94,6 +94,8 @@ unsigned long millis() { return now_ms; } void vTaskDelay(unsigned) { assert(false && "stub disconnect must deliver its event"); } const char* esp_err_to_name(int) { return "mock"; } static int publish_result = 0, enqueue_result = 41, outbox_size = 0; +static int outbox_queries = 0, enqueue_added_size = 0, stop_result = ESP_OK; +static bool sdk_mutex_blocked = false, sdk_callback_owner = false; static int publish_calls = 0, enqueue_calls = 0, config_calls = 0; static int start_calls = 0, reconnect_calls = 0, receive_events = 0, error_events = 0; static int init_calls = 0, destroy_calls = 0, stop_calls = 0; @@ -124,7 +126,7 @@ int esp_mqtt_set_config(void*, esp_mqtt_client_config_t*) { } int esp_mqtt_client_start(void*) { ++start_calls; return start_result; } int esp_mqtt_client_reconnect(void*) { ++reconnect_calls; return ESP_OK; } -int esp_mqtt_client_stop(void*) { ++stop_calls; return ESP_OK; } +int esp_mqtt_client_stop(void*) { ++stop_calls; if(stop_result==ESP_OK)outbox_size=0;return stop_result; } int esp_mqtt_client_disconnect(void*) { assert(sdk_event_handler != nullptr); esp_mqtt_event_t event{MQTT_EVENT_DISCONNECTED, 0, false}; @@ -132,7 +134,10 @@ int esp_mqtt_client_disconnect(void*) { return ESP_OK; } int esp_mqtt_client_subscribe(void*, const char*, int) { return 7; } -int esp_mqtt_client_get_outbox_size(void*) { return outbox_size; } +int esp_mqtt_client_get_outbox_size(void*) { + assert(!sdk_mutex_blocked || sdk_callback_owner); + ++outbox_queries;return outbox_size; +} int esp_mqtt_client_publish(void*, const char*, const char*, int, int, bool) { ++publish_calls; return publish_result; @@ -140,6 +145,7 @@ int esp_mqtt_client_publish(void*, const char*, const char*, int, int, bool) { int esp_mqtt_client_enqueue(void*, const char*, const char*, int, int, bool, bool store) { assert(store); ++enqueue_calls; + if(enqueue_result>=0)outbox_size+=enqueue_added_size; return enqueue_result; } void PsychicMqttClient::_onMessage(esp_mqtt_event_handle_t&) { ++receive_events; } @@ -360,6 +366,9 @@ class MqttPublishAckTests(unittest.TestCase): "int PsychicMqttClient::subscribe(", "int PsychicMqttClient::publish(", "PsychicMqttClient &PsychicMqttClient::setOutboxLimit(", + "size_t PsychicMqttClient::getOutboxSize()", + "size_t PsychicMqttClient::refreshOutboxSize()", + "void PsychicMqttClient::serviceOutboxDiagnostics()", "unsigned long PsychicMqttClient::getOutboxDrops()", "unsigned long PsychicMqttClient::getPublishOk()", "unsigned long PsychicMqttClient::getPublishErr()", @@ -401,6 +410,71 @@ class MqttPublishAckTests(unittest.TestCase): def test_production_publish_events_and_slot_diagnostics(self): self.compile_run(MAIN) + def test_outbox_snapshot_never_takes_cli_lock_and_tracks_enqueue_ack_and_expiry(self): + self.compile_run(r''' +static void dispatch(PsychicMqttClient& client,int id){ + esp_mqtt_event_t event{id,41,false}; + PsychicMqttClient::_onMqttEventStatic(&client,"MQTT",id,&event); +} +int main(){ + PsychicMqttClient client; + assert(client.getOutboxSize()==0&&outbox_queries==0); + client._mqtt_cfg.uri="mqtt://broker";client._mqtt_cfg.broker.address.uri="mqtt://broker"; + assert(client.connect()==ESP_OK); + dispatch(client,MQTT_EVENT_CONNECTED); + now_ms=1000;enqueue_added_size=120; + assert(client.publish("status",1,false,"body",4,true)==41); + assert(client.getOutboxSize()==120); + int queries=outbox_queries; + sdk_mutex_blocked=true; + for(int i=0;i<1000;++i)assert(client.getOutboxSize()==120); + assert(outbox_queries==queries); // A radio/CLI reader never waits for the SDK. + sdk_callback_owner=true;outbox_size=0; + dispatch(client,MQTT_EVENT_PUBLISHED); // recursive SDK-owned callback after removal + assert(client.getPublishAck()==1&&client.getOutboxSize()==0&&outbox_queries==queries+1); + sdk_callback_owner=false;sdk_mutex_blocked=false; + queries=outbox_queries; + for(int i=0;i<20;++i){now_ms+=1000;client.serviceOutboxDiagnostics();} + assert(outbox_queries==queries); // Quiet/empty outboxes never enter the SDK. + + // Admission must use the SDK's fresh value, even when the CLI cache is old. + client.setOutboxLimit(200);outbox_size=200; + const int enqueued=enqueue_calls; + assert(client.getOutboxSize()==0); + assert(client.publish("status",1,false,"body",4,true)==-2&&enqueue_calls==enqueued); + assert(client.getOutboxSize()==200&&client.getOutboxDrops()==1); + outbox_size=0; + assert(client.publish("status",1,false,"body",4,true)==41); + assert(client.getOutboxSize()==120); + const unsigned long accepted=client.getPublishOk(),acks=client.getPublishAck(); + queries=outbox_queries; + outbox_size=0;now_ms+=999;client.serviceOutboxDiagnostics(); + assert(client.getOutboxSize()==120&&outbox_queries==queries); + ++now_ms;client.serviceOutboxDiagnostics(); + assert(client.getOutboxSize()==0&&outbox_queries==queries+1); + assert(client.getPublishOk()==accepted&&client.getPublishAck()==acks); // expiry is no ACK + + now_ms=UINT32_MAX-500; + assert(client.publish("status",1,false,"body",4,true)==41); + queries=outbox_queries;outbox_size=0; + now_ms=498;client.serviceOutboxDiagnostics();assert(outbox_queries==queries); + now_ms=499;client.serviceOutboxDiagnostics();assert(outbox_queries==queries+1); + assert(client.getOutboxSize()==0); + + assert(client.publish("status",1,false,"body",4,true)==41); + client.softDisconnect(); + assert(client.isStarted()&&!client.connected()&&client.getOutboxSize()==120); + stop_result=ESP_ERR_INVALID_STATE; + client.forceStop(); + assert(client.isStarted()&&client.getOutboxSize()==120); // failed stop retains ownership/cache + stop_result=ESP_OK;client.forceStop(); + assert(!client.isStarted()&&!client.connected()&&client.getOutboxSize()==0); + assert(client.connect()==ESP_OK);dispatch(client,MQTT_EVENT_CONNECTED); + assert(client.publish("status",1,false,"body",4,true)==41); + client.disconnect();assert(!client.isStarted()&&client.getOutboxSize()==0); +} +''') + def test_production_initialized_getter_tracks_sdk_creation_and_retained_stops(self): self.compile_run(r''' int main() { diff --git a/test/test_mqtt_publish_admission.py b/test/test_mqtt_publish_admission.py new file mode 100644 index 000000000..e4e2517ed --- /dev/null +++ b/test/test_mqtt_publish_admission.py @@ -0,0 +1,137 @@ +"""Run the production packet/raw/status admission paths with bounded heaps.""" +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +from test_mqtt_publish_ack import method + +ROOT = Path(__file__).resolve().parents[1] +BRIDGE = 'src/helpers/bridges/MQTTBridge.cpp' +HARNESS = r''' +#include +#include +#include +#include +#include +#include "helpers/MQTTPublishAdmission.h" +#include "helpers/MQTTTransportHealth.h" +#define ESP_PLATFORM 1 +#define BOARD_HAS_PSRAM 1 +#define MQTT_DEBUG_PRINTLN(...) ((void)0) +static uint32_t clock_ms=0; +unsigned long millis(){return clock_ms;} +constexpr int MALLOC_CAP_DMA=1, RUNTIME_MQTT_SLOTS=1; +static size_t dma_free=15348,dma_largest=9716,heap_queries=0; +size_t heap_caps_get_free_size(int){++heap_queries;return dma_free;} +size_t heap_caps_get_largest_free_block(int){++heap_queries;return dma_largest;} +struct Esp { size_t size=8000000; size_t getPsramSize(){return size;} + size_t getMaxAllocHeap(){++heap_queries;return dma_largest;} } ESP; +namespace mesh { struct Packet { uint8_t getPayloadType(){return 7;} + uint8_t writeTo(uint8_t*){return 2;} }; } +static int builds=0; +namespace MQTTMessageBuilder { +constexpr size_t WIRE_SCRATCH_SIZE=255; +bool canSerializePacket(mesh::Packet*,size_t){return true;} +template int buildPacketJSONFromRaw(Args...){++builds;return 1;} +template int buildPacketJSON(Args...){++builds;return 1;} +template int buildRawJSON(Args...){++builds;return 1;} +} +struct Radio {float packetScore(float,int){return 0;}}; +struct Client {int calls=0,result=0;uint32_t ack=0; + uint32_t getPublishAck(){return ack;} + int publish(const char*,uint8_t,bool,const char*,int,bool){++calls;return result;} +}; +struct MQTTBridge { + struct MQTTSlot {Client* client=nullptr;bool enabled=true,connected=true; + unsigned long last_log_time=0;MQTTTransportHealth::State transport_health;}; + MQTTSlot _slots[1]; + std::atomic _stop_requested{false}; + std::atomic _slot_health_reset_pending[1]{},_slot_transport_fault_pending[1]{}; + MQTTPublishAdmission::State _publish_memory; + int _skipped_publishes=0; + static constexpr unsigned long SLOT_LOG_INTERVAL=30000; + static constexpr size_t PUBLISH_JSON_BUFFER_SIZE=2048; + enum {MSG_PACKETS,MSG_RAW}; + char scratch[2048]{},*_json_scratch_buffer=scratch; + int _json_scratch_doc=0; + char _device_id[65]="ABC",_origin[32]="test";int _timezone=0; + uint8_t* _last_raw_data=nullptr;int _last_raw_len=0; + unsigned long _last_raw_timestamp=0;float _last_snr=0,_last_rssi=0; + Radio* _radio=nullptr;uint8_t eligible=1; + uint8_t eligiblePacketSlots(uint8_t,int){return eligible;} + void refreshOriginFromPrefs(){} + bool buildTopicForSlot(int,int,char*topic,size_t){strcpy(topic,"topic");return true;} + void updateSlotTransportDiag(int,uint32_t){} + bool hasPublishHeadroom(size_t=0,size_t=0,uint8_t=0); + bool publishToSlot(int,const char*,const char*,size_t,bool=false,uint8_t=0); + bool publishPacket(mesh::Packet*,bool,bool&,const uint8_t*=nullptr,int=0,float=0,float=0); + bool publishRaw(mesh::Packet*,bool&); +}; +'''.replace('const uint8_t*=', 'const uint8_t* =') + +MAIN = r''' +int main(){ + using namespace MQTTPublishAdmission; + Heap heap;heap.psram=true;heap.dma_free=15348;heap.dma_largest=9716; + assert(!canPublish(heap)); // Exact screenshot: preserve the DMA reserve. + heap.dma_free+=5*897; // The five CPU-only reassembly buffers move to PSRAM. + assert(canPublish(heap)); // No new 16K internal allocation is needed to write. + heap.dma_free=16384;heap.dma_largest=4096; + assert(canPublish(heap,2048,128,0)&&canPublish(heap,1024,128,1)); + --heap.dma_free;assert(!canPublish(heap));++heap.dma_free; + --heap.dma_largest;assert(!canPublish(heap));++heap.dma_largest; + assert(!canPublish(heap,10240,128,1)); // Large outbox records need more space. + heap.dma_free=24000;heap.dma_largest=11000; + assert(canPublish(heap,10240,128,1)); + assert(!canPublish(heap,SIZE_MAX,1,1)&&!canPublish(heap,1,SIZE_MAX,0)); + heap.psram=false;heap.internal_largest=7999;assert(!canPublish(heap)); + heap.internal_largest=8000;assert(canPublish(heap)); + State state;assert(state.needsSample(0));state.update(0,heap); + assert(!state.needsSample(0)&&!state.needsSample(4999)&&state.needsSample(5000)); + state.update(UINT32_MAX-100,heap); + assert(!state.needsSample(50)&&state.needsSample(5000)); + + MQTTBridge bridge;Client client;bridge._slots[0].client=&client; + mesh::Packet packet;bool eligible=false; + assert(!bridge.publishPacket(&packet,false,eligible)&&eligible); + assert(!bridge.publishRaw(&packet,eligible)&&eligible); + assert(!bridge.publishToSlot(0,"status","{}",2,true,1)); + assert(builds==0&&client.calls==0&&bridge._skipped_publishes==3); + assert(heap_queries==2); // Both denied paths cache the heap walk, even at tick0. + assert(bridge._slots[0].transport_health.recovery_reason== + MQTTTransportHealth::RecoveryReason::None); + dma_free=19833;clock_ms=5000; + assert(bridge.publishPacket(&packet,false,eligible)); + assert(bridge.publishRaw(&packet,eligible)); + assert(bridge.publishToSlot(0,"status","{}",2,true,1)); + assert(builds==2&&client.calls==3&&heap_queries==4); + bridge.eligible=0;assert(!bridge.publishRaw(&packet,eligible)&&!eligible); + assert(builds==2&&client.calls==3); + client.result=-1; + assert(!bridge.publishToSlot(0,"raw","{}",2,false,0)); + assert(!bridge._publish_memory.sampled); // Recheck after a real SDK failure. +} +''' + + +class MqttPublishAdmissionTests(unittest.TestCase): + def test_actual_publish_paths_and_policy(self): + compiler = shutil.which('g++') or shutil.which('clang++') + self.assertIsNotNone(compiler) + source = HARNESS + '\n'.join(method(BRIDGE, signature) for signature in ( + 'bool MQTTBridge::hasPublishHeadroom(', 'bool MQTTBridge::publishToSlot(', + 'bool MQTTBridge::publishPacket(', 'bool MQTTBridge::publishRaw(')) + MAIN + with tempfile.TemporaryDirectory() as directory: + path, binary = Path(directory)/'publish.cpp', Path(directory)/'publish' + path.write_text(source) + result = subprocess.run([compiler,'-std=c++17','-Wall','-Wextra','-Werror', + '-I'+str(ROOT/'src'),str(path),'-o',str(binary)],capture_output=True,text=True) + self.assertEqual(result.returncode,0,result.stderr) + result = subprocess.run([str(binary)],capture_output=True,text=True,timeout=15) + self.assertEqual(result.returncode,0,result.stdout+result.stderr) + + +if __name__=='__main__': + unittest.main() diff --git a/test/test_mqtt_reassembly_buffer.py b/test/test_mqtt_reassembly_buffer.py new file mode 100644 index 000000000..31c1caeab --- /dev/null +++ b/test/test_mqtt_reassembly_buffer.py @@ -0,0 +1,214 @@ +#!/usr/bin/env python3 +"""Run real MQTT allocation/lifetime and multipart delivery with heap failures.""" +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import unittest + +import test_mqtt_publish_ack as ack + +ROOT = Path(__file__).resolve().parents[1] + +MEMORY = r''' +#include +#include +#include "helpers/MQTTConnectionAdmission.h" +enum { MALLOC_CAP_INTERNAL=1, MALLOC_CAP_DMA=2, MALLOC_CAP_SPIRAM=4, MALLOC_CAP_8BIT=8 }; +struct Allocation { size_t bytes; unsigned caps; }; +static std::map allocations; +static std::vector heap_requests; +static bool psram_ok=true, internal_ok=true, ordinary_ok=true; +static size_t dma_free=65536, dma_largest=16384; +static unsigned ordinary_allocs=0, ordinary_frees=0, caps_frees=0; +static int required_destroy_count=0; +static void* ordinary_malloc(size_t bytes) { + ++ordinary_allocs; + if(!ordinary_ok)return nullptr; + void* p=std::malloc(bytes);assert(p);allocations[p]={bytes,0};return p; +} +static void ordinary_free(void* p) { + assert(destroy_calls>=required_destroy_count); + assert(allocations.count(p)&&allocations[p].caps==0); + allocations.erase(p);++ordinary_frees;std::free(p); +} +void* heap_caps_malloc(size_t bytes,unsigned caps) { + heap_requests.push_back({bytes,caps}); + if((caps&MALLOC_CAP_SPIRAM)?!psram_ok:!internal_ok)return nullptr; + if(caps&MALLOC_CAP_INTERNAL){assert(bytes<=dma_free&&bytes<=dma_largest);dma_free-=bytes;} + void* p=std::malloc(bytes);assert(p);allocations[p]={bytes,caps};return p; +} +void heap_caps_free(void* p) { + assert(destroy_calls>=required_destroy_count); + assert(allocations.count(p)&&allocations[p].caps); + if(allocations[p].caps&MALLOC_CAP_INTERNAL)dma_free+=allocations[p].bytes; + allocations.erase(p);++caps_frees;std::free(p); +} +size_t heap_caps_get_free_size(unsigned caps) { + assert(caps==(MALLOC_CAP_INTERNAL|MALLOC_CAP_DMA));return dma_free; +} +size_t heap_caps_get_largest_free_block(unsigned caps) { + assert(caps==(MALLOC_CAP_INTERNAL|MALLOC_CAP_DMA));return dma_largest; +} +#define malloc ordinary_malloc +#define free ordinary_free +''' + +MAIN = r''' +static void configure(PsychicMqttClient& client,int size=896) { + client._mqtt_cfg.uri="mqtt://test"; + client._mqtt_cfg.broker.address.uri="mqtt://test"; + client._mqtt_cfg.buffer_size=size; + client._mqtt_cfg.buffer.size=size; +} +static size_t allocated(){return ordinary_allocs+heap_requests.size();} +static void multipart(PsychicMqttClient& client,const std::string& text) { + char topic[]="test/data"; + const size_t first=text.size()/3,second=2*text.size()/3; + for(size_t part=0;part<3;++part) { + size_t offset=part==0?0:part==1?first:second; + size_t end=part==0?first:part==1?second:text.size(); + esp_mqtt_event_t event{MQTT_EVENT_DATA,0,false}; + event.total_data_len=text.size();event.current_data_offset=offset; + event.data_len=end-offset;event.data=const_cast(text.data()+offset); + event.topic=topic;event.topic_len=sizeof(topic)-1; + esp_mqtt_event_handle_t pointer=&event; + client._onMessage(pointer); + } +} +int main() { + { + PsychicMqttClient client;configure(client); + // A successful PSRAM allocation needs no internal-heap headroom. + dma_free=0;dma_largest=0; + assert(client.connect()==ESP_OK&&client._buffer_capacity==896); + assert(allocations[client._buffer].bytes==897); +#if defined(BOARD_HAS_PSRAM) + assert(allocations[client._buffer].caps==(MALLOC_CAP_SPIRAM|MALLOC_CAP_8BIT)); + assert(ordinary_allocs==0&&heap_requests.size()==1); +#else + assert(allocations[client._buffer].caps==0&&heap_requests.empty()); +#endif + char* retained=client._buffer; + size_t count=allocated(); + unsigned received=0; + std::string expected(896,'Q'); + client.onMessage([&](char* topic,char* payload,int,int,bool){ + assert(!strcmp(topic,"test/data")&&expected==payload);++received; + }); + multipart(client,expected);assert(received==1&&allocated()==count); + multipart(client,std::string(897,'X'));assert(received==1&&allocated()==count); + client.disconnect(); + assert(client._buffer==retained&&allocations.count(retained)&&allocated()==count); + assert(client.connect()==ESP_OK&&client._buffer==retained&&allocated()==count); + expected="after restart";multipart(client,expected);assert(received==2); + // Existing clients retain their original fixed reassembly allocation. + configure(client,3072); + assert(client.connect()==ESP_OK&&client._buffer==retained&&client._buffer_capacity==896); + assert(allocated()==count); + required_destroy_count=destroy_calls+1; // SDK ownership ends before buffer release. + } + assert(allocations.empty()&&destroy_calls==1); +#if defined(BOARD_HAS_PSRAM) + assert(caps_frees==1&&ordinary_frees==0); + psram_ok=false; + // Exact fallback boundary preserves the full DMA reserve after allocation. + dma_free=MQTTConnectionAdmission::kDmaReserveBytes+897;dma_largest=897; + { + PsychicMqttClient client;configure(client); + assert(client.connect()==ESP_OK&&client._buffer_capacity==896); + assert(allocations[client._buffer].caps==(MALLOC_CAP_INTERNAL|MALLOC_CAP_DMA|MALLOC_CAP_8BIT)); + assert(dma_free==MQTTConnectionAdmission::kDmaReserveBytes); + } + assert(allocations.empty()&&caps_frees==2); + for(unsigned scenario=0;scenario<3;++scenario) { + dma_free=MQTTConnectionAdmission::kDmaReserveBytes+(scenario==0?896:897); + dma_largest=scenario==1?896:897; + internal_ok=scenario!=2; + PsychicMqttClient client;configure(client); + const int before=init_calls; + const size_t requests=heap_requests.size(); + assert(client.connect()==ESP_ERR_NO_MEM); + assert(client._buffer==nullptr&&client._buffer_capacity==0&&!client.isInitialized()); + assert(init_calls==before&&allocations.empty()); + assert(heap_requests.size()==requests+(scenario==2?2:1)); + psram_ok=true; + assert(client.connect()==ESP_OK&&client._buffer_capacity==896); + psram_ok=false; + } + internal_ok=true;psram_ok=true; +#else + assert(ordinary_frees==1&&caps_frees==0); + { + ordinary_ok=false; + PsychicMqttClient client;configure(client); + const int before=init_calls; + assert(client.connect()==ESP_ERR_NO_MEM&&!client._buffer&&!client._buffer_capacity); + assert(init_calls==before&&heap_requests.empty()); + ordinary_ok=true; + assert(client.connect()==ESP_OK&&client._buffer_capacity==896); + } +#endif + assert(allocations.empty()); + // Default and explicitly larger desired capacities still include one NUL. + for(int desired:{0,3072}) { + PsychicMqttClient client;configure(client,desired); + const size_t capacity=desired?desired:1024; + assert(client.connect()==ESP_OK&&client._buffer_capacity==capacity); + assert(allocations[client._buffer].bytes==capacity+1); + } + assert(allocations.empty()); +} +''' + + +class MqttReassemblyBufferTest(unittest.TestCase): + def test_real_allocator_lifetime_recovery_and_multipart_capacity(self): + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler) + event = """struct esp_mqtt_event_t { int event_id, msg_id; bool session_present; + int total_data_len=0,data_len=0,current_data_offset=0,topic_len=0,retain=0,qos=0; + char* data=nullptr;char* topic=nullptr;bool dup=false; };""" + mqtt = ack.MQTT_HEADER.replace( + "struct esp_mqtt_event_t { int event_id, msg_id; bool session_present; };", event) + preamble = ack.PREAMBLE.replace( + "void PsychicMqttClient::_onMessage(esp_mqtt_event_handle_t&) { ++receive_events; }", "") + source = preamble + MEMORY + "\n".join(ack.method(ack.ADAPTER, signature) for signature in ( + "PsychicMqttClient::PsychicMqttClient()", "PsychicMqttClient::~PsychicMqttClient()", + "PsychicMqttClient &PsychicMqttClient::onMessage(", "int PsychicMqttClient::subscribe(", + "esp_err_t PsychicMqttClient::applyConfig()", "esp_err_t PsychicMqttClient::connect()", + "void PsychicMqttClient::disconnect()", "void PsychicMqttClient::_onMqttEventStatic(", + "void PsychicMqttClient::_onMqttEvent(", "void PsychicMqttClient::_onConnect(", + "void PsychicMqttClient::_onDisconnect(", "void PsychicMqttClient::_onSubscribe(", + "void PsychicMqttClient::_onUnsubscribe(", "void PsychicMqttClient::_onPublish(", + "size_t PsychicMqttClient::refreshOutboxSize()", + "void PsychicMqttClient::_onMessage(", "bool PsychicMqttClient::_isTopicMatch(", + )) + MAIN + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-buffer-") as directory: + directory = Path(directory) + (directory / "Arduino.h").write_text(ack.ARDUINO_HEADER, encoding="ascii") + (directory / "mqtt_client.h").write_text(mqtt, encoding="ascii") + (directory / "esp_crt_bundle.h").write_text("#pragma once\n", encoding="ascii") + path = directory / "buffer.cpp" + path.write_text(source, encoding="ascii") + for idf in (4, 5): + for psram in (False, True): + with self.subTest(idf=idf, psram=psram): + binary = directory / "buffer" + sanitize = (["-fsanitize=address,undefined", "-fno-omit-frame-pointer", "-fno-pie", "-no-pie"] + if sys.platform.startswith("linux") else []) + command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + "-Wno-unused-function", "-Wno-unused-variable", *sanitize, + "-DARDUINO_ARCH_ESP32=1", f"-DESP_IDF_VERSION_MAJOR={idf}", + *(["-DBOARD_HAS_PSRAM=1"] if psram else []), + "-I", str(directory), "-I", str(ROOT / "lib/PsychicMqttClient/src"), + "-I", str(ROOT / "src"), str(path), "-o", str(binary)] + result = subprocess.run(command, capture_output=True, text=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mqtt_slot_status.py b/test/test_mqtt_slot_status.py new file mode 100644 index 000000000..4e9954950 --- /dev/null +++ b/test/test_mqtt_slot_status.py @@ -0,0 +1,228 @@ +"""Run actual slot identity and status scheduling methods without a broker.""" + +from pathlib import Path +import os +import shutil +import subprocess +import tempfile +import unittest + +from test_mqtt_transport_results import method + + +ROOT = Path(__file__).resolve().parents[1] +BRIDGE = "src/helpers/bridges/MQTTBridge.cpp" +PREAMBLE = r''' +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#ifdef _timezone +#undef _timezone +#endif +#include "helpers/MQTTStatusPublishPolicy.h" +#define MQTT_DEBUG_PRINTLN(...) ((void)0) +constexpr int RUNTIME_MQTT_SLOTS = 6, MALLOC_CAP_INTERNAL = 1; +static uint32_t now_ms = 0; +static unsigned builds = 0; +static bool build_succeeds = true; +uint32_t millis() { return now_ms; } +size_t heap_caps_get_free_size(int) { return 100000; } +struct { uint64_t value = 0xAABBCCDDEEFF; uint64_t getEfuseMac() const { return value; } } ESP; +namespace MQTTMessageBuilder { +void formatIsoTimestampForMqtt(long, long, const char*, char* out, size_t size) { snprintf(out, size, "UTC"); } +template +int buildStatusMessage(int, const char*, const char*, const char*, const char*, + const char*, const char*, const char*, const char*, char* buffer, int size, Args...) { + ++builds; + if (!build_succeeds) return 0; + assert(size > 2); strcpy(buffer, "{}"); return 2; +} +} +void formatRadioInfo(float, float, int, int, char* out, size_t size) { snprintf(out, size, "radio"); } +struct Client { + const char* id = nullptr; + void setClientId(const char* value) { id = value; } // Real adapter retains this pointer. +}; +struct Preset { bool allow_retain = false; }; +struct MQTTBridge { + struct MQTTSlot { + Client* client = nullptr; + char client_id[32] = {}; + bool enabled = false, connected = false; + const Preset* preset = nullptr; + MQTTStatusPublishPolicy::State status_publish; + } _slots[RUNTIME_MQTT_SLOTS]; + Client clients[RUNTIME_MQTT_SLOTS]; + std::atomic _status_publish_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _stop_requested{false}; + struct Observer { bool mqtt_status_enabled = true; } prefs; + Observer* _obs = &prefs; + bool accepted[RUNTIME_MQTT_SLOTS] = {true,true,true,true,true,true}; + bool topic_enabled[RUNTIME_MQTT_SLOTS] = {true,true,true,true,true,true}; + unsigned calls[RUNTIME_MQTT_SLOTS] = {}; + std::function during_enqueue; + static constexpr int STATUS_JSON_BUFFER_SIZE = 1024, MSG_STATUS = 0; + static constexpr uint32_t STATUS_RETRY_INTERVAL = 30000; + uint32_t _status_interval = 300000; + struct { float* freq = nullptr; float* bw = nullptr; uint8_t* sf = nullptr; uint8_t* cr = nullptr; } _node_info; + const char *_timezone = "UTC", *_origin = "node", *_device_id = "ABCDEF"; + const char *_board_model = "G3", *_firmware_version = "test"; + int _json_scratch_doc = 0, _queue_count = 0; + char _json_scratch_buffer[STATUS_JSON_BUFFER_SIZE] = {}; + struct Board { int getBattMilliVolts() const { return 4000; } }; + struct Clock { uint32_t getMillis() const { return now_ms; } }; + struct Dispatcher { + int getErrFlags() const { return 0; } int getTotalAirTime() const { return 0; } + int getReceiveAirTime() const { return 0; } int getNumSentFlood() const { return 0; } + int getNumSentDirect() const { return 0; } int getNumRecvFlood() const { return 0; } + int getNumRecvDirect() const { return 0; } + }; + struct Radio { int getNoiseFloor() const { return -120; } int getPacketsRecvErrors() const { return 0; } }; + Board* _board = nullptr; Clock* _ms = nullptr; Dispatcher* _dispatcher = nullptr; Radio* _radio = nullptr; + void refreshOriginFromPrefs() {} + void getClientVersion(char* out, size_t size) { snprintf(out, size, "test"); } + int repeatStatus() const { return 0; } int collectConnHealth() const { return 0; } + bool buildTopicForSlot(int i, int type, char* out, size_t size) { + assert(type == MSG_STATUS); snprintf(out, size, "status%d", i); return topic_enabled[i]; + } + bool publishToSlot(int i, const char* topic, const char* payload, size_t length, bool retained, uint8_t qos) { + assert(qos == 1 && length == 2 && !strcmp(payload, "{}") && !strncmp(topic, "status", 6)); + assert(retained == (_slots[i].preset ? _slots[i].preset->allow_retain : false)); + ++calls[i]; + if (during_enqueue) { auto callback = during_enqueue; during_enqueue = nullptr; callback(); } + return accepted[i]; + } + void connect(int count) { + for (int i = 0; i < count; ++i) { + _slots[i].enabled = _slots[i].connected = true; _slots[i].client = &clients[i]; + _status_publish_pending[i] = true; + } + } + void setSlotClientId(int); + void maintainStatus(uint32_t); + bool publishStatus(uint8_t); + bool publishStatusToSlot(int); +}; +''' + + +class MqttSlotStatusTests(unittest.TestCase): + def compile_run(self, main): + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler, "a native C++ compiler is required") + production = "\n".join(method(BRIDGE, signature) for signature in ( + "void MQTTBridge::setSlotClientId(", "void MQTTBridge::maintainStatus(", + "bool MQTTBridge::publishStatus(", "bool MQTTBridge::publishStatusToSlot(")) + flags = [] if os.name == "nt" else ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"] + with tempfile.TemporaryDirectory(prefix="mqtt-slot-status-") as directory: + source, binary = Path(directory) / "status.cpp", Path(directory) / "status" + source.write_text(PREAMBLE + production + main) + result = subprocess.run([compiler, "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror", + *flags, "-I", str(ROOT / "src"), str(source), "-o", str(binary)], + capture_output=True, text=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_duplicate_broker_slots_have_stable_distinct_persistent_client_ids(self): + setup = method(BRIDGE, "bool MQTTBridge::setupSlot(") + self.assertLess(setup.index("setSlotClientId(index)"), setup.index("reconnectSlotClient(index)")) + self.compile_run(r''' +int main() { + MQTTBridge bridge; bridge.connect(5); + std::map broker_sessions; + for (int i = 0; i < 5; ++i) { + bridge.setSlotClientId(i); + assert(bridge.clients[i].id == bridge._slots[i].client_id); + assert(strlen(bridge.clients[i].id) <= 23); + for (const char* p = bridge.clients[i].id; *p; ++p) assert(isalnum(static_cast(*p))); + assert(broker_sessions.emplace(bridge.clients[i].id, i).second); + } + assert(broker_sessions.size() == 5); + assert(!strcmp(bridge.clients[0].id, "meshcoreAABBCCDDEEFFs1")); + char saved[32]; strcpy(saved, bridge.clients[1].id); + // Reconfigure/reconnect retains identity and storage despite stack churn. + for (unsigned n = 0; n < 100; ++n) { + volatile char scratch[128]; for (unsigned i = 0; i < sizeof(scratch); ++i) scratch[i] = char(i + n); + bridge.setSlotClientId(1); + assert(!strcmp(bridge.clients[1].id, saved)); + } + assert(broker_sessions.size() == 5 && !bridge._slots[5].enabled); + MQTTBridge next_boot; next_boot.connect(1); next_boot.setSlotClientId(0); + assert(!strcmp(next_boot.clients[0].id, bridge.clients[0].id)); + ESP.value ^= 1; next_boot.setSlotClientId(0); + assert(strcmp(next_boot.clients[0].id, bridge.clients[0].id)); +} +''') + + def test_failed_status_is_retried_per_slot_without_republishing_successful_slots(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; bridge.connect(5); + bridge.accepted[1] = bridge.accepted[3] = false; + now_ms = 0; bridge.maintainStatus(now_ms); + assert(builds == 1); + for (int i = 0; i < 5; ++i) assert(bridge.calls[i] == 1); + assert(bridge._slots[0].status_publish.has_accepted && !bridge._slots[0].status_publish.pending); + assert(bridge._slots[1].status_publish.pending); + // Every 50 ms bridge tick remains bounded under prolonged memory/outbox rejection. + for (now_ms = 50; now_ms < 30000; now_ms += 50) bridge.maintainStatus(now_ms); + assert(builds == 1 && bridge.calls[1] == 1); + now_ms = 30000; bridge.accepted[1] = true; bridge.maintainStatus(now_ms); + assert(builds == 2 && bridge.calls[0] == 1 && bridge.calls[1] == 2 && bridge.calls[3] == 2); + for (now_ms = 30050; now_ms < 60000; now_ms += 50) bridge.maintainStatus(now_ms); + assert(builds == 2); + now_ms = 60000; bridge.accepted[3] = true; bridge.maintainStatus(now_ms); + assert(builds == 3 && bridge.calls[3] == 3 && bridge.calls[1] == 2); + now_ms = 300000; bridge.maintainStatus(now_ms); + assert(bridge.calls[0] == 2 && bridge.calls[2] == 2 && bridge.calls[4] == 2); + assert(bridge.calls[1] == 2 && bridge.calls[3] == 3); + now_ms = 330000; bridge.maintainStatus(now_ms); assert(bridge.calls[1] == 3); + now_ms = 360000; bridge.maintainStatus(now_ms); assert(bridge.calls[3] == 4); +} +''') + + def test_json_failure_wraparound_reconnect_latch_and_modes_do_not_flood(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; bridge.connect(2); bridge.topic_enabled[1] = false; + build_succeeds = false; now_ms = UINT32_MAX - 1000; bridge.maintainStatus(now_ms); + assert(builds == 1 && bridge.calls[0] == 0 && bridge._slots[0].status_publish.pending); + for (unsigned elapsed = 1; elapsed < 30000; elapsed += 50) { + now_ms = uint32_t(UINT32_MAX - 1000 + elapsed); bridge.maintainStatus(now_ms); + } + assert(builds == 1); + build_succeeds = true; now_ms = uint32_t(UINT32_MAX - 1000 + 30000); bridge.maintainStatus(now_ms); + assert(builds == 2 && bridge.calls[0] == 1 && bridge.calls[1] == 0); + bridge._slots[0].connected = false; now_ms += 300000; bridge.maintainStatus(now_ms); + assert(builds == 2); + bridge._slots[0].connected = true; bridge._status_publish_pending[0] = true; + bridge.during_enqueue = [&] { bridge._status_publish_pending[0] = true; }; + bridge.maintainStatus(now_ms); assert(bridge.calls[0] == 2); + assert(bridge._status_publish_pending[0]); + bridge.maintainStatus(now_ms); assert(bridge.calls[0] == 3); + assert(!bridge._status_publish_pending[0]); + bridge.prefs.mqtt_status_enabled = false; + now_ms += 300000; bridge.maintainStatus(now_ms); assert(bridge.calls[0] == 3); + // Preserve the existing on-connect status even when periodic status is off. + bridge._status_publish_pending[0] = true; bridge.maintainStatus(now_ms); assert(bridge.calls[0] == 4); + bridge._stop_requested = true; bridge._status_publish_pending[0] = true; + bridge.maintainStatus(now_ms); assert(bridge.calls[0] == 4 && bridge._status_publish_pending[0]); + MQTTBridge stopping; stopping.connect(5); + stopping.during_enqueue = [&] { stopping._stop_requested = true; }; + stopping.maintainStatus(now_ms); + assert(stopping.calls[0] == 1); + for (int i = 1; i < 5; ++i) assert(stopping.calls[i] == 0); +} +''') + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mqtt_transport_results.py b/test/test_mqtt_transport_results.py index 78575ff73..b8a1b434f 100644 --- a/test/test_mqtt_transport_results.py +++ b/test/test_mqtt_transport_results.py @@ -224,7 +224,10 @@ struct MQTTBridge { std::atomic _network_suspended{false}; bool _network_espnow_restore_pending = false; struct Slot { - struct Client { bool isStarted() const { return false; } }; + struct Client { + bool isStarted() const { return false; } + void serviceOutboxDiagnostics() {} + }; Client* client = nullptr; bool enabled = true, initial_connect_done = false, connected = false; unsigned long last_reconnect_attempt = 0; @@ -282,9 +285,13 @@ struct MQTTBridge { work(); ++reconfigures; if (scenario == 8) _stop_requested = true; } - void publishStatusToSlot(int) { - work(); ++publishes; - if (scenario == 9) _stop_requested = true; + void maintainStatus(uint32_t) { + for (int i = 0; i < RUNTIME_MQTT_SLOTS && !_stop_requested; ++i) { + if (!_status_publish_pending[i]) continue; + _status_publish_pending[i] = false; + work(); ++publishes; + if (scenario == 9) _stop_requested = true; + } } void maintainSlotConnections() { work(); @@ -616,7 +623,8 @@ class MqttTransportResultsTests(unittest.TestCase): source = (ROOT / "src/helpers/bridges/MQTTBridge.cpp").read_text(encoding="utf-8") self.assertEqual(source.count("result = slot.client->connect();"), 1) self.assertEqual(source.count("result = slot.client->reconnect();"), 1) - self.assertEqual(source.count("repeatStatus(), collectConnHealth()"), 2) + # On-connect and periodic status share one snapshot builder. + self.assertEqual(source.count("repeatStatus(), collectConnHealth()"), 1) for signature in ("void MQTTBridge::teardownSlot(", "void MQTTBridge::destroySlotClients("): body = method("src/helpers/bridges/MQTTBridge.cpp", signature) self.assertIn("_slot_attempt_pending[", body) diff --git a/test/test_mqtt_wifi_network_power.py b/test/test_mqtt_wifi_network_power.py index d3c23129f..c9faadc4c 100644 --- a/test/test_mqtt_wifi_network_power.py +++ b/test/test_mqtt_wifi_network_power.py @@ -145,6 +145,7 @@ struct Client { int stops = 0; uint32_t publish_acks = 0; uint32_t getPublishAck() const { return publish_acks; } + void serviceOutboxDiagnostics() {} bool isStarted() const { return started; } bool connected() const { return live; } void disconnect() { diff --git a/test/test_webconfig_json_allocator.py b/test/test_webconfig_json_allocator.py new file mode 100644 index 000000000..cdfd601bd --- /dev/null +++ b/test/test_webconfig_json_allocator.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""Exercise real JSON workspaces and authenticated handlers under allocation failure.""" +from pathlib import Path +import os +import re +import subprocess +import tempfile +import unittest + +import test_webconfig_lost_reply as config_fixture +import test_webconfig_login_throttle as login_fixture +from test_replay_reset_integration import extract_braced + +ROOT = Path(__file__).resolve().parents[1] +SERVER = (ROOT / 'src/helpers/esp32/WebConfigServer.cpp').read_text() +HEADER = (ROOT / 'src/helpers/esp32/WebConfigServer.h').read_text() +ALIASES = HEADER.split('class WebConfigServer {', 1)[1].split('public:', 1)[0] +HEAP = r''' +#pragma once +#include +#include +#include +#include +#include +enum { MALLOC_CAP_INTERNAL=1, MALLOC_CAP_DMA=2, MALLOC_CAP_SPIRAM=4, MALLOC_CAP_8BIT=8 }; +namespace heap_test { +struct Block { size_t size; bool psram; }; +inline std::map blocks; +inline size_t dma_free=65536,dma_largest=32768; +inline bool psram=true; +inline int successful_before_failure=-1; +inline unsigned psram_calls=0,internal_calls=0; +inline void reset() { + assert(blocks.empty()); dma_free=65536;dma_largest=32768; + psram=true;successful_before_failure=-1;psram_calls=internal_calls=0; +} +} +inline void* heap_caps_malloc(size_t size,unsigned caps) { + const bool external=caps&MALLOC_CAP_SPIRAM; + if(external)++heap_test::psram_calls;else ++heap_test::internal_calls; + if((external&&!heap_test::psram)||heap_test::successful_before_failure==0)return nullptr; + if(!external&&(size>heap_test::dma_largest||size>heap_test::dma_free))return nullptr; + void* pointer=malloc(size);if(!pointer)return nullptr; + if(heap_test::successful_before_failure>0)--heap_test::successful_before_failure; + heap_test::blocks[pointer]={size,external}; + if(!external)heap_test::dma_free-=size; + return pointer; +} +inline void heap_caps_free(void* pointer) { + if(!pointer)return; + const auto found=heap_test::blocks.find(pointer);assert(found!=heap_test::blocks.end()); + if(!found->second.psram)heap_test::dma_free+=found->second.size; + heap_test::blocks.erase(found);free(pointer); +} +inline size_t heap_caps_get_free_size(unsigned) { return heap_test::dma_free; } +inline size_t heap_caps_get_largest_free_block(unsigned) { return heap_test::dma_largest; } +inline size_t heap_caps_get_allocated_size(void* pointer) { return heap_test::blocks.at(pointer).size; } +''' +PRELUDE = '#define ESP_PLATFORM 1\n#define BOARD_HAS_PSRAM 1\n#include \n' + + +class WebConfigJsonAllocatorTests(unittest.TestCase): + def compile_run(self, source, *, psram=True): + json_header = next((ROOT / '.pio/libdeps').glob('*/ArduinoJson/src/ArduinoJson.h')) + with tempfile.TemporaryDirectory(prefix='webconfig-json-') as folder: + work = Path(folder) + (work / 'esp_heap_caps.h').write_text(HEAP) + cpp, binary = work / 'test.cpp', work / 'test' + cpp.write_text((PRELUDE if psram else '#include \n') + source) + sanitizers = [] if os.name == 'nt' else [ + '-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie'] + result = subprocess.run(['g++', '-std=c++17', '-Wall', '-Wextra', + *sanitizers, + '-I'+str(work), '-I'+str(ROOT/'src'), '-I'+str(json_header.parent), + str(cpp), '-o', str(binary)], capture_output=True, text=True, timeout=45) + self.assertEqual(result.returncode, 0, result.stdout+result.stderr) + run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=15) + self.assertEqual(run.returncode, 0, run.stdout+run.stderr) + + def test_psram_allocations_realloc_failure_and_shrink_preserve_dma_reserve(self): + self.compile_run(r''' +int main() { + using namespace heap_test; + auto* allocator=mesh::webconfig::JsonAllocator::instance(); + reset(); + auto* first=static_cast(allocator->allocate(4096));assert(first&&internal_calls==0); + memset(first,'x',4096);const auto free_before=dma_free; + auto* larger=static_cast(allocator->reallocate(first,8192)); + assert(larger&&dma_free==free_before&&internal_calls==0); + for(size_t i=0;i<4096;++i)assert(larger[i]=='x'); + psram=false;dma_free=dma_largest=0; + const auto calls=psram_calls; + assert(allocator->reallocate(larger,32)==larger&&psram_calls==calls); + assert(!allocator->reallocate(larger,16384)); + assert(blocks.count(larger)&&larger[4095]=='x'); + assert(!allocator->reallocate(larger,0));assert(blocks.empty()); + reset();psram=false;dma_free=16384+4096;dma_largest=4096; + first=static_cast(allocator->allocate(4096));assert(first&&dma_free==16384); + assert(!allocator->allocate(1));allocator->deallocate(first); + reset();psram=false;dma_free=16384+4096-1;dma_largest=4096; + assert(!allocator->allocate(4096)&&internal_calls==0); + dma_free=65536;dma_largest=4095; + assert(!allocator->allocate(4096)&&internal_calls==0); + reset(); + { mesh::webconfig::DynamicJsonDocument large(6656); + mesh::webconfig::StaticJsonDocument<96> small; + assert(large.capacity()==6656&&small.capacity()==96); + small["state"]="pending"; + for(unsigned i=0;i<200;++i)large.add(i); + assert(!small.overflowed()&&!large.overflowed()&&internal_calls==0); + } + assert(blocks.empty()); +} +''') + + def config(self): + source = config_fixture.LostReplyWebConfigTest.fixture(self).split('int main() {', 1)[0] + source = source.replace('struct WebConfigServer {', 'struct WebConfigServer {'+ALIASES) + start = source.index('void WebConfigServer::handleConfigGet(') + end = source.index('\nvoid WebConfigServer::handleConfigPost(', start) + source = source[:start] + extract_braced(SERVER, 'void WebConfigServer::handleConfigGet(') + source[end:] + source = source.replace('struct AsyncWebServerRequest {', + 'struct AsyncResponseStream : String { void addHeader(const char*,const char*) {} };\n' + 'static void webConfigRegulatorsSnapshot(JsonObject) {}\nstruct AsyncWebServerRequest {') + source = source.replace(' Client socket;', ' AsyncResponseStream response;\n Client socket;') + source = source.replace(' void send(int code,', + ' AsyncResponseStream* beginResponseStream(const char*) { return &response; }\n' + ' void send(AsyncResponseStream* result) { status=200;output=*result; }\n void send(int code,') + source = source.replace(' Callbacks* _cb = &callbacks;', + ' Callbacks* _cb = &callbacks;\n char _wifi_password[65]="saved-secret";\n' + ' unsigned _wifi_power_save=0;\n') + return source + + def test_actual_config_handlers_use_psram_and_reject_no_memory_before_queue_admission(self): + self.compile_run(self.config()+r''' +int main() { + using namespace heap_test; + assert(mesh::esp32::managementPassword::publish("password")); + reset(); + { WebConfigServer server;AsyncWebServerRequest get;server.handleConfigGet(&get); + assert(get.status==200&&psram_calls&&internal_calls==0); + assert(get.output.find("saved-secret")==String::npos); + JsonDocument response;assert(!deserializeJson(response,get.output)); + assert(response["radio"]["freq"].is()); + } + reset();psram=false;dma_free=16384; + { WebConfigServer server;AsyncWebServerRequest get;server.handleConfigGet(&get); + assert(get.status==503&&get.output=="{\"error\":\"not enough memory\"}"); + assert(!internal_calls&&server._batch_state==WebConfigBatch::State::Idle); + } + // Fail every successive allocation, including construction of the accepted + // receipt. A failed JSON workspace cannot publish a partly parsed batch. + bool succeeded=false; + for(int permitted=0;permitted<30&&!succeeded;++permitted) { + reset();dma_free=16384;successful_before_failure=permitted; + { WebConfigServer server;server.callbacks.node.capabilities=WebConfigServer::CAP_LOST_REPLY; + AsyncWebServerRequest post(writeBody("\"yes\""));server.handleConfigPost(&post); + if(post.status==202) { + succeeded=true;assert(server._batch_state==WebConfigBatch::State::Pending); + assert(server._batch_count==1&&!strcmp(server._batch[0].cmd,"set lost.reply yes")); + } else { + assert(post.status==503&&server._batch_state==WebConfigBatch::State::Idle); + assert(server._batch_count==0); + } + assert(!internal_calls); + } + } + assert(succeeded);reset(); + { WebConfigServer server;AsyncWebServerRequest post(writeBody("\"yes\"")); + post.header.cookies="";psram=false;dma_free=0;server.handleConfigPost(&post); + assert(post.status==401&&psram_calls==0&&internal_calls==0); + } +} +''') + + def test_actual_login_no_memory_does_not_count_as_wrong_password(self): + source = login_fixture.WebConfigLoginThrottleTest.handler_fixture(self).split('static const char* kLocked', 1)[0] + source = source.replace('struct WebConfigServer {', 'struct WebConfigServer {'+ALIASES) + self.assertIn('DeserializationError::NoMemory', source) + self.compile_run(source+r''' +int main() { + using namespace heap_test; + reset();WebConfigServer portal; + for(unsigned i=0;i<12;++i) { + AsyncWebServerRequest request;request.body="{\"password\":\"existing-secret\"}"; + request._tempObject=request.body.data();psram=false;dma_free=16384; + portal.handleLogin(&request);assert(request.status==503&&internal_calls==0); + } + psram=true; + AsyncWebServerRequest request;request.body="{\"password\":\"existing-secret\"}"; + request._tempObject=request.body.data();portal.handleLogin(&request); + assert(request.status==200&&portal._session_token[0]); +} +''') + + def test_non_psram_keeps_original_arduinojson_document_types(self): + self.compile_run(r''' +#include +#include +static_assert(std::is_same::value,""); +static_assert(std::is_same,ArduinoJson::StaticJsonDocument<96>>::value,""); +int main() { + mesh::webconfig::DynamicJsonDocument doc(6656);doc["key"]="value"; + auto* allocator=mesh::webconfig::JsonAllocator::instance(); + void* memory=allocator->allocate(16);assert(memory); + memory=allocator->reallocate(memory,32);assert(memory);allocator->deallocate(memory); + return doc.overflowed(); +} +''', psram=False) + + +if __name__ == '__main__': + unittest.main()