diff --git a/src/helpers/MQTTRuntimeBufferLifecycle.h b/src/helpers/MQTTRuntimeBufferLifecycle.h new file mode 100644 index 00000000..bb061a9e --- /dev/null +++ b/src/helpers/MQTTRuntimeBufferLifecycle.h @@ -0,0 +1,23 @@ +#pragma once + +#include + +// Small ownership helpers for MQTT runtime buffers. They intentionally keep +// each buffer independent: a failed allocation leaves that buffer null (so its +// caller can use its stack fallback) without discarding the other buffers. +namespace MQTTRuntimeBufferLifecycle { + +template +inline void* allocateIfMissing(void* buffer, size_t size, Allocator allocate) { + return buffer != nullptr ? buffer : allocate(size); +} + +template +inline void* release(void* buffer, Deallocator deallocate) { + if (buffer != nullptr) { + deallocate(buffer); + } + return nullptr; +} + +} // namespace MQTTRuntimeBufferLifecycle diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index fb62b79f..53f89fb5 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -2,6 +2,7 @@ #include "../MQTTConnectionPolicy.h" #include "../MQTTMessageBuilder.h" #include "../MQTTPacketQueuePolicy.h" +#include "../MQTTRuntimeBufferLifecycle.h" #include "../MQTTTopicRouter.h" #include "../TxtDataHelpers.h" #include @@ -488,7 +489,13 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg // so we must pass rules here. _timezone_storage(TimeChangeRule{"UTC", Last, Sun, Mar, 0, 0}, TimeChangeRule{"UTC", Last, Sun, Mar, 0, 0}), _timezone(&_timezone_storage), +#if defined(BOARD_HAS_PSRAM) + _last_raw_data(nullptr), +#endif _last_raw_len(0), _last_snr(0), _last_rssi(0), _last_raw_timestamp(0), +#if defined(BOARD_HAS_PSRAM) + _publish_json_buffer(nullptr), _status_json_buffer(nullptr), +#endif _identity(identity), _cached_has_connected_slots(false), _last_memory_check(0), _skipped_publishes(0), @@ -567,20 +574,52 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg #endif #endif - // On PSRAM boards, allocate raw radio buffer and JSON char buffers in PSRAM to preserve - // internal heap. On non-PSRAM boards these are inline arrays in the class object — - // no separate allocation needed. - #if defined(BOARD_HAS_PSRAM) - _last_raw_data = (uint8_t*)psram_malloc(LAST_RAW_DATA_SIZE); - _publish_json_buffer = (char*)psram_malloc(PUBLISH_JSON_BUFFER_SIZE); - _status_json_buffer = (char*)psram_malloc(STATUS_JSON_BUFFER_SIZE); - #else + // Non-PSRAM boards keep the raw cache inline for the bridge lifetime. + // PSRAM boards allocate their runtime buffers in begin(), after PSRAM has + // been probed/initialized, and release them in end(). + #if !defined(BOARD_HAS_PSRAM) memset(_last_raw_data, 0, sizeof(_last_raw_data)); #endif // JSON document scratch space is now a StaticJsonDocument inline class member — // no heap allocation needed; reused via doc.clear() on every publish. } +void MQTTBridge::allocateRuntimeBuffers() { + #if defined(BOARD_HAS_PSRAM) + // Keep each allocation independent. A nullptr is deliberately retained on + // failure: status/packet publish paths already use stack fallbacks, and the + // next begin() will retry only the missing buffer. + _last_raw_data = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( + _last_raw_data, LAST_RAW_DATA_SIZE, psram_malloc)); + _publish_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( + _publish_json_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc)); + _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( + _status_json_buffer, STATUS_JSON_BUFFER_SIZE, psram_malloc)); + MQTT_DEBUG_PRINTLN("Runtime buffers: raw=%s publish=%s status=%s", + _last_raw_data ? "PSRAM" : "unavailable", + _publish_json_buffer ? "PSRAM" : "stack fallback", + _status_json_buffer ? "PSRAM" : "stack fallback"); + #endif +} + +void MQTTBridge::releaseRuntimeBuffers() { + #if defined(BOARD_HAS_PSRAM) + _last_raw_data = static_cast(MQTTRuntimeBufferLifecycle::release( + _last_raw_data, psram_free)); + _publish_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( + _publish_json_buffer, psram_free)); + _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( + _status_json_buffer, psram_free)); + #endif + + // Never pair a newly allocated raw buffer with metadata from a prior bridge + // run. This also makes non-PSRAM restarts discard their stale raw cache. + _last_raw_len = 0; + _last_snr = 0; + _last_rssi = 0; + _last_raw_timestamp = 0; +} + // --------------------------------------------------------------------------- // begin() // --------------------------------------------------------------------------- @@ -629,6 +668,10 @@ void MQTTBridge::begin() { return; } + // These are begin()/end()-scoped on PSRAM targets. Allocation happens after + // the PSRAM probe above so a late psramInit() has taken effect. + allocateRuntimeBuffers(); + refreshOriginFromPrefs(); strncpy(_iata, _obs->mqtt_iata, sizeof(_iata) - 1); @@ -743,6 +786,7 @@ void MQTTBridge::begin() { psram_free(_packet_queue_storage); #endif _packet_queue_storage = nullptr; + releaseRuntimeBuffers(); return; } @@ -781,6 +825,7 @@ void MQTTBridge::begin() { psram_free(_packet_queue_storage); #endif _packet_queue_storage = nullptr; + releaseRuntimeBuffers(); return; } @@ -863,13 +908,8 @@ void MQTTBridge::end() { // the MQTT memory-defrag work — nothing to delete. _timezone always // points at &_timezone_storage and stays valid for the bridge lifetime. - // Free PSRAM-backed buffers (non-PSRAM builds use inline class arrays — no free needed) - #if defined(BOARD_HAS_PSRAM) - psram_free(_last_raw_data); _last_raw_data = nullptr; - psram_free(_publish_json_buffer); _publish_json_buffer = nullptr; - psram_free(_status_json_buffer); _status_json_buffer = nullptr; - #endif - // JSON documents are now StaticJsonDocument inline members — no heap allocation to free. + releaseRuntimeBuffers(); + // JSON documents are StaticJsonDocument inline members — no heap allocation to free. _initialized = false; _slots_setup_done = false; // Reset so deferred setup runs again on next begin() diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 7e8c0466..6893d3c3 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -376,6 +376,11 @@ private: void getClientVersion(char* buffer, size_t buffer_size) const; void logMemoryStatus(); void refreshOriginFromPrefs(); + // begin()/end()-scoped PSRAM buffers. Each allocation is independent so a + // transient heap shortage degrades to the existing stack fallback instead + // of making the bridge unusable. + void allocateRuntimeBuffers(); + void releaseRuntimeBuffers(); // Observer config (MQTT/WiFi/timezone/SNMP/alert), persisted to /mqtt_prefs. // _prefs (held by BridgeBase) still provides upstream fields (freq/sf/node_name…). diff --git a/test/README.md b/test/README.md index 1be24625..e60a3879 100644 --- a/test/README.md +++ b/test/README.md @@ -30,6 +30,7 @@ does not reflect the GoogleTest count — run the built binary directly | `test_mqtt_topic_router` | `src/helpers/MQTTTopicRouter.h` | complete preset/custom topic-routing contract; MeshRank packets-only behavior; required identifiers; invalid inputs/slots; exact buffer boundaries | | `test_mqtt_connection_policy` | `src/helpers/MQTTConnectionPolicy.h` | reconnect guard/backoff/stagger and breaker transitions; stable reset; JWT lifetime/renewal policy; exact timing boundaries and 32-bit `millis()` rollover | | `test_mqtt_packet_queue_policy` | `src/helpers/MQTTPacketQueuePolicy.h` | queue-full eviction; stale-disconnect flush; adaptive drain limits; bounded QoS0 retries; exact timing boundaries and 32-bit `millis()` rollover | +| `test_mqtt_runtime_buffer_lifecycle` | `src/helpers/MQTTRuntimeBufferLifecycle.h` | idempotent allocation/release; partial-allocation degradation; retry of only missing buffers | | `test_mqtt_payload_builder` | `src/helpers/MQTTPayloadBuilder.cpp` | status/packet/raw JSON contracts; optional fields; escaping; RX metrics and path; score handling; exact buffer bounds; maximum representative payloads | | `test_utils` | `src/Utils.cpp` | `Utils::toHex` (upstream) | diff --git a/test/test_mqtt_runtime_buffer_lifecycle/test_mqtt_runtime_buffer_lifecycle.cpp b/test/test_mqtt_runtime_buffer_lifecycle/test_mqtt_runtime_buffer_lifecycle.cpp new file mode 100644 index 00000000..959a501d --- /dev/null +++ b/test/test_mqtt_runtime_buffer_lifecycle/test_mqtt_runtime_buffer_lifecycle.cpp @@ -0,0 +1,70 @@ +#include + +#include +#include + +#include "helpers/MQTTRuntimeBufferLifecycle.h" + +namespace RuntimeBuffers = MQTTRuntimeBufferLifecycle; + +TEST(MQTTRuntimeBufferLifecycle, PartialAllocationKeepsOtherBuffersAndRetriesOnlyMissing) { + std::vector allocation_sizes; + int allocation_attempt = 0; + const auto allocate = [&allocation_sizes, &allocation_attempt](size_t size) -> void* { + allocation_sizes.push_back(size); + allocation_attempt++; + if (allocation_attempt == 2) { + return nullptr; + } + return std::malloc(size); + }; + + void* raw = nullptr; + void* publish = nullptr; + void* status = nullptr; + + raw = RuntimeBuffers::allocateIfMissing(raw, 256, allocate); + publish = RuntimeBuffers::allocateIfMissing(publish, 2048, allocate); + status = RuntimeBuffers::allocateIfMissing(status, 768, allocate); + + ASSERT_NE(nullptr, raw); + EXPECT_EQ(nullptr, publish); + ASSERT_NE(nullptr, status); + ASSERT_EQ(3U, allocation_sizes.size()); + void* const initial_raw = raw; + void* const initial_status = status; + + raw = RuntimeBuffers::allocateIfMissing(raw, 256, allocate); + publish = RuntimeBuffers::allocateIfMissing(publish, 2048, allocate); + status = RuntimeBuffers::allocateIfMissing(status, 768, allocate); + + EXPECT_EQ(4U, allocation_sizes.size()); + EXPECT_EQ(2048U, allocation_sizes.back()); + EXPECT_EQ(initial_raw, raw); + EXPECT_EQ(initial_status, status); + EXPECT_NE(nullptr, publish); + + int releases = 0; + const auto release = [&releases](void* allocation) { + releases++; + std::free(allocation); + }; + raw = RuntimeBuffers::release(raw, release); + publish = RuntimeBuffers::release(publish, release); + status = RuntimeBuffers::release(status, release); + + EXPECT_EQ(nullptr, raw); + EXPECT_EQ(nullptr, publish); + EXPECT_EQ(nullptr, status); + EXPECT_EQ(3, releases); + + raw = RuntimeBuffers::release(raw, release); + publish = RuntimeBuffers::release(publish, release); + status = RuntimeBuffers::release(status, release); + EXPECT_EQ(3, releases); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}