From 2e1a1410cec71a55837dd9d2e105f28673e6100b Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 00:25:37 -0700 Subject: [PATCH] feat(mqtt): cooperative MQTT shutdown + OTA teardown barrier (Phase 5) Wire the Phase 4 MQTTLifecycle state machine into MQTTBridge to replace the blind vTaskDelete teardown that could kill the MQTT task mid-mbedTLS and then free client buffers on a corrupted heap (the observed OTA teardown panic). - MQTTBridge owns a MQTTLifecycle::Coordinator driven only by the loop task (Core 1) from begin()/end(); a nested LifecycleOps binds the host-tested Ops spec to FreeRTOS/PsychicMqttClient. - end() requests a cooperative stop; the MQTT task (Core 0) tears down its own clients where the mbedTLS contexts live, acks via _stop_acked, and self-terminates. end() waits (bounded) for the ack, then frees queue/buffers. - Bounded stop timeout -> reviewed fallback (force kill + Core-1 teardown) sets a dirty latch that withholds OTA flashing. - begin() gains an idempotent double-call guard and syncs the Coordinator to Running. - OTA teardown barrier: simple_repeater's deferred flash aborts/resumes unless end() reported a clean stop (canFlashAfterStop()). Scope: minimal cooperative-shutdown unit. The volatile NTP/reconfigure handshake replacement and the plain-data snapshot / consumer repointing (MQTT_OWNERSHIP.md sections 1-3) are deferred. MQTT_STOP_TIMEOUT_MS is a Phase-0 placeholder pending on-hardware characterization. Native suite green (incl. test_mqtt_lifecycle); both observer firmware smoke builds compile. Not yet hardware-validated (Phase 7 gate). --- MQTT_OWNERSHIP.md | 45 ++++-- STABILITY_TESTABILITY_HANDOFF.md | 51 +++++-- examples/simple_repeater/MyMesh.cpp | 9 +- src/helpers/bridges/MQTTBridge.cpp | 226 +++++++++++++++++++++------- src/helpers/bridges/MQTTBridge.h | 37 +++++ 5 files changed, 292 insertions(+), 76 deletions(-) diff --git a/MQTT_OWNERSHIP.md b/MQTT_OWNERSHIP.md index d1a81d79..79298905 100644 --- a/MQTT_OWNERSHIP.md +++ b/MQTT_OWNERSHIP.md @@ -7,11 +7,14 @@ for each. It is paired with the fork-owned lifecycle test seam (`src/helpers/MQTTLifecycle.h`, `test/test_mqtt_lifecycle/`). **Status:** ownership model documented and the lifecycle/teardown test seam -landed. The production rewiring it describes (publishing a plain-data snapshot, -replacing the `volatile` handshakes, cooperative shutdown) is **deferred to -Phase 5** — this document is the plan Phase 5 executes, not a description of -current behavior. Line references are against the tree at the time of writing -and should be re-verified before editing. +landed (Phase 4). **Phase 5 (branch `phase5/cooperative-mqtt-shutdown`) has now +implemented the cooperative shutdown and the OTA barrier — hazard §4 below.** +Still **deferred** (carried to Phase 5b / Phase 6): publishing a plain-data +snapshot and repointing the §1/§2 consumers, and replacing the §3 `volatile` +handshakes with a command channel. This document is the plan; §1–§3 still +describe current behavior, while §4 is now resolved (see its note). Line +references are against the tree at the time of writing and should be re-verified +before editing. ## Execution contexts @@ -98,15 +101,31 @@ clears/processes and writes a "done" flag last, Core 1 spins: The two blocking waiters (`requestForcedNtpSync` `:3298`, `ntpDiag` `:3344`) spin on Core 1 while `end()` could tear down the singleton/task concurrently. -### 4. Abrupt teardown, no start guard +### 4. Abrupt teardown, no start guard — RESOLVED in Phase 5 -- `end()` uses `vTaskDelete(_mqtt_task_handle)` (`MQTTBridge.cpp:862`) — the task - is killed wherever it is (possibly mid-`_slots[]` mutation, or inside - mbedTLS); slot/client cleanup then runs *after* deletion on the caller's - context (`:902-905`). This is the OTA teardown heap-panic path. -- `begin()` (`:626`) has **no double-call guard** — a second `begin()` re-runs - allocation and `xTaskCreatePinnedToCore`, leaking the prior queue/task. -- Lifecycle state is a single `_initialized` bool; there is no state enum. +Original hazard (retained for context): `end()` used +`vTaskDelete(_mqtt_task_handle)` to kill the task wherever it was (possibly +mid-`_slots[]` mutation or inside mbedTLS), then ran slot/client cleanup *after* +deletion on the caller's context — the OTA teardown heap-panic path. `begin()` +had no double-call guard, and lifecycle state was a single `_initialized` bool. + +**Phase 5 resolution** (branch `phase5/cooperative-mqtt-shutdown`): + +- `end()` now requests a cooperative stop through `MQTTLifecycle::Coordinator`. + The MQTT task (Core 0) tears down its own clients where the mbedTLS contexts + live, acknowledges via `_stop_acked`, and self-terminates; `end()` waits for + the ack before freeing the queue/buffers. The blind `vTaskDelete` survives + only as the bounded-timeout fallback, which sets a dirty latch that withholds + OTA flashing (`canFlashAfterStop()`). +- `begin()` has an idempotent double-call guard and drives the Coordinator to + `Running`; the lifecycle state now lives in the tested state machine, not a + bare bool. +- The OTA barrier gates `simple_repeater`'s deferred flash on a clean stop. + +Not hardware-validated yet (Phase 7); `MQTT_STOP_TIMEOUT_MS` is a Phase-0 +placeholder. Note the residual §1/§2 instance-pointer reads remain deferred, so +consumers can still (as before) touch a torn-down bridge — that is unchanged by +Phase 5 and tracked above. ## Target primitives (Phase 5) diff --git a/STABILITY_TESTABILITY_HANDOFF.md b/STABILITY_TESTABILITY_HANDOFF.md index e0a0f8fc..09a0b214 100644 --- a/STABILITY_TESTABILITY_HANDOFF.md +++ b/STABILITY_TESTABILITY_HANDOFF.md @@ -26,10 +26,10 @@ index. | 1 | PR CI smoke builds + ArduinoJson pin enforcement | Done (build-size gate and ASan/UBSan still pending) | | 2 | PSRAM restart resource symmetry | Done | | 3 | MQTT preference migration fixtures | Done (filesystem adapter still lives in `CommonCLI`) | -| 0 | Pre-change lifecycle characterization | Not started — deterministic behavior encoded in Phase 4; hardware items pending | +| 0 | Pre-change lifecycle characterization | Deterministic behavior encoded in Phase 4/5; hardware items pending (stop-timeout value is a placeholder) | | 4 | Ownership and teardown test seams | Seams + ownership doc + teardown tests done; production rewiring deferred to Phase 5 | -| 5 | Cooperative MQTT shutdown | Not started (fixes the OTA teardown panic) | -| — | OTA teardown barrier | Not started — release-critical | +| 5 | Cooperative MQTT shutdown | Minimal cooperative `end()` + `begin()` guard + OTA barrier implemented on branch `phase5/cooperative-mqtt-shutdown` (native green, firmware smoke build green); NOT hardware-validated. Volatile-handshake replacement + snapshot-consumer repointing deferred | +| — | OTA teardown barrier | Implemented — flash gated on a clean MQTT stop in `simple_repeater`; not hardware-validated | | 6 | Request/queue/connection/publication integration tests | Not started | | 7 | Uptime, memory, and fault-injection gates | Not started | @@ -387,15 +387,44 @@ to bridge teardown, OTA sequencing, MQTT client lifetime, or task ownership. ### Phase 5: Implement cooperative MQTT shutdown -**Status: Not started.** Verified premise: `end()` stops the MQTT task with a -direct `vTaskDelete` mid-operation and no handshake, and lifecycle state is a -single `_initialized` bool (no state enum). Note also that `begin()` is not -currently guarded against a double-call — it re-creates the queue and task, -leaking the prior ones — so fold that guard into the idempotent-start requirement -below rather than leaving it to caller discipline. +**Status: Minimal cooperative shutdown implemented on branch +`phase5/cooperative-mqtt-shutdown` (scope: "the smallest change that fixes the +OTA teardown panic as one reviewable unit"). Native suite green; the non-PSRAM +observer firmware smoke build compiles. NOT yet hardware-validated — that is the +Phase 7 gate — and the stop timeout is a Phase-0 placeholder (see below).** -With Phase 0 behavior recorded and Phase 4 tests in place, replace direct task -deletion with an explicit lifecycle such as: +What landed (wiring the Phase 4 `MQTTLifecycle` state machine into the bridge): + +- `MQTTBridge` owns a `MQTTLifecycle::Coordinator` driven **only** by the loop + task (Core 1) from `begin()`/`end()`. A nested `LifecycleOps` binds the pure, + host-tested `Ops` spec to FreeRTOS/PsychicMqttClient. +- `end()` no longer blind-`vTaskDelete`s. It requests a cooperative stop; the + MQTT task (Core 0) sees a new `volatile _stop_requested`, tears down its own + clients **on Core 0 where the mbedTLS contexts live**, sets `_stop_acked` + last, and self-terminates. `end()` waits (bounded) for the ack, then frees the + queue/buffers. This removes the "kill the task mid-mbedTLS, then free client + buffers on a corrupted heap" teardown path. +- Bounded stop timeout → reviewed fallback: on timeout the task is force-killed + and torn down on Core 1 (the old behavior), but a **dirty latch** is set so + `canFlashAfterStop()` is false and OTA flashing is withheld. +- `begin()` has a double-call guard and syncs the Coordinator to `Running`. +- OTA teardown barrier: `simple_repeater`'s deferred-OTA fire site aborts and + resumes the bridge unless the preceding `end()` reported a clean stop. + +Deferred (kept out of this reviewable unit; carried to a Phase 5b / Phase 6): +replacing the `volatile` NTP/reconfigure handshakes with a command channel, and +publishing a plain-data status snapshot to repoint the §1/§2 consumers in +`MQTT_OWNERSHIP.md` (the `AlertReporter`/`buildStatsJson` instance-pointer reads +that can still touch a torn-down bridge). Those are not required to fix the OTA +panic and would enlarge the merge-sensitive diff. + +**Phase 0 dependency still open:** `MQTT_STOP_TIMEOUT_MS` is a conservative 8 s +placeholder. It must be characterized against real mbedTLS teardown over `wss` +with a down broker (Phase 0) before this ships; it is a single named constant so +that change is trivial. + +The original plan of record follows. Replace direct task deletion with an +explicit lifecycle such as: `Stopped -> Starting -> Running -> StopRequested -> Stopping -> Stopped` diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index e2467c1b..a99d4de9 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -1569,7 +1569,14 @@ void MyMesh::loop() { Serial.println("OTA: starting update"); setBridgeState(false); char ota_reply[160]; - if (!_cli.getBoard()->otaFromManifest(getFirmwareVer(), false, ota_reply)) { + // OTA teardown barrier (Phase 5): only flash after a CLEAN MQTT shutdown. + // A timed-out/forced stop leaves mbedTLS/heap ownership uncertain — writing + // firmware then is the observed teardown heap-panic path — so abort and + // resume the bridge instead of flashing under uncertain ownership. + if (bridge && !bridge->canFlashAfterStop()) { + Serial.println("OTA: aborted, MQTT stop did not complete cleanly - resuming bridge"); + setBridgeState(true); + } else if (!_cli.getBoard()->otaFromManifest(getFirmwareVer(), false, ota_reply)) { Serial.print("OTA: aborted, resuming bridge - "); Serial.println(ota_reply); setBridgeState(true); } diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 53f89fb5..af241ee5 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -473,6 +473,15 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) } } +// Bounded cooperative-stop timeout for end() (see MQTTLifecycle::Coordinator). +// Phase 0 TODO: characterize on hardware — mbedTLS teardown over wss with a +// DOWN broker — before this ships. The placeholder is chosen conservatively: +// long enough for a real cooperative client teardown (per-slot disconnect + the +// existing 50 ms settle delays across RUNTIME_MQTT_SLOTS), short enough to bound +// the loop-task stall and the OTA/restart wait. A stop that exceeds this is +// treated as dirty: the task is force-killed and OTA flashing is withheld. +static const uint32_t MQTT_STOP_TIMEOUT_MS = 8000; + // --------------------------------------------------------------------------- // Constructor // --------------------------------------------------------------------------- @@ -514,6 +523,9 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg #else , _queue_head(0), _queue_tail(0) #endif + // Cooperative lifecycle: _lifecycle_ops must be constructed before + // _lifecycle (declaration order guarantees this) so the reference binds. + , _lifecycle_ops(this), _lifecycle(_lifecycle_ops, MQTT_STOP_TIMEOUT_MS) { // Initialize default values strncpy(_origin, "MeshCore-Repeater", sizeof(_origin) - 1); @@ -626,6 +638,14 @@ void MQTTBridge::releaseRuntimeBuffers() { void MQTTBridge::begin() { MQTT_DEBUG_PRINTLN("Initializing MQTT Bridge..."); + // Idempotent start (Phase 5): a second begin() on an already-running bridge + // would re-run allocation and re-create the task, leaking the previous + // queue/task. Guard here instead of relying on caller discipline. + if (_initialized) { + MQTT_DEBUG_PRINTLN("MQTT Bridge already running — begin() ignored"); + return; + } + // PSRAM diagnostic - helps debug memory fragmentation on boards with external RAM #ifdef BOARD_HAS_PSRAM { @@ -805,6 +825,11 @@ void MQTTBridge::begin() { // causes resets on some boards (e.g. Heltec V4) when the task runs from PSRAM stack. _mqtt_task_stack = nullptr; _mqtt_task_handle = nullptr; + // Clear the cooperative-stop handshake before the new task starts reading it. + // deliverStop() leaves _stop_requested latched true after a stop cycle, so a + // restart must reset it or the fresh task would self-terminate immediately. + _stop_requested = false; + _stop_acked = false; BaseType_t create_result = xTaskCreatePinnedToCore( mqttTask, "MQTTBridge", @@ -845,6 +870,14 @@ void MQTTBridge::begin() { // instead of churning ~40 KB of internal heap per cycle. initSlotClients(); + // Sync the lifecycle Coordinator to Running now that all resources exist and + // the task is created. Driven only on the success path: the failure rollbacks + // above already free what they acquired and leave the bridge Stopped, so we + // must not also fire the state machine's release effect there (double free). + // A fresh start also clears any dirty-stop latch (re-enabling OTA flashing). + _lifecycle.requestStart(); // Stopped -> Starting (startTask() is a no-op here) + _lifecycle.onTaskStarted(); // Starting -> Running + _initialized = true; s_mqtt_bridge_instance = this; MQTT_DEBUG_PRINTLN("MQTT Bridge initialized"); @@ -855,65 +888,140 @@ void MQTTBridge::begin() { // --------------------------------------------------------------------------- void MQTTBridge::end() { MQTT_DEBUG_PRINTLN("Stopping MQTT Bridge..."); + + // Idempotent stop: nothing to tear down if we never started (or already stopped). + if (!_initialized) { + MQTT_DEBUG_PRINTLN("MQTT Bridge already stopped — end() ignored"); + return; + } + + // Stop new diagnostic reads through the singleton before teardown begins. s_mqtt_bridge_instance = nullptr; - #ifdef ESP_PLATFORM - // Delete FreeRTOS task first (it will clean up WiFi/MQTT connections) - if (_mqtt_task_handle != nullptr) { - vTaskDelete(_mqtt_task_handle); - _mqtt_task_handle = nullptr; - } - // Free PSRAM task stack - psram_free(_mqtt_task_stack); - _mqtt_task_stack = nullptr; + // Cooperative shutdown (Phase 5). Request the stop, then let the lifecycle + // Coordinator drive it. On ESP32 the MQTT task (Core 0) tears down its own + // clients where the mbedTLS contexts live and acknowledges via _stop_acked; + // the queue/buffer release happens inside LifecycleOps::releaseResources() + // once the Coordinator reaches Stopped (clean ack OR the reviewed timeout + // fallback). This replaces the former blind vTaskDelete that could kill the + // task mid-mbedTLS and then free client buffers on a corrupted heap. + _lifecycle.requestStop(); // Running -> StopRequested; deliverStop() sets _stop_requested - // Clean up queued packets from FreeRTOS queue - // Packets are value-copied in the queue, so no external pointers to clean up. - if (_packet_queue_handle != nullptr) { - QueuedPacket queued; - while (xQueueReceive(_packet_queue_handle, &queued, 0) == pdTRUE) { - _queue_count--; +#ifdef ESP_PLATFORM + // Wait (bounded) for the task to acknowledge. tick() synthesizes the timeout + // fallback if the task never acks. Checking the ack first each iteration means + // a stop that completes right as the timeout expires is still treated as clean. + while (_lifecycle.isStopInProgress()) { + if (_stop_acked) { + _lifecycle.onTaskStopped(); // StopRequested -> Stopped (clean): releaseResources() + break; } - vQueueDelete(_packet_queue_handle); - _packet_queue_handle = nullptr; + _lifecycle.tick(); // may fire StopTimedOut -> Stopped (dirty): releaseResources() + if (!_lifecycle.isStopInProgress()) break; + vTaskDelay(pdMS_TO_TICKS(20)); } - #if defined(BOARD_HAS_PSRAM) - psram_free(_packet_queue_storage); - #endif - _packet_queue_storage = nullptr; - - #else - // Clean up queued packet references - // Packets are value-copied in the queue, so no external pointers to clean up. - for (int i = 0; i < _queue_count; i++) { - int index = (_queue_head + i) % MAX_QUEUE_SIZE; - memset(&_packet_queue[index], 0, sizeof(QueuedPacket)); - } - - _queue_count = 0; - _queue_head = 0; - _queue_tail = 0; - memset(_packet_queue, 0, sizeof(_packet_queue)); - #endif - - // Disconnect and delete persistent MQTT clients. teardownSlot() intentionally - // only disconnects; destruction happens here so the mbedTLS contexts survive - // the reconfigure/reconnect hot path. - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - teardownSlot(i); - } - destroySlotClients(); - - // Timezone is inline class storage (_timezone_storage) since Phase 3 of - // the MQTT memory-defrag work — nothing to delete. _timezone always - // points at &_timezone_storage and stays valid for the bridge lifetime. - - releaseRuntimeBuffers(); - // JSON documents are StaticJsonDocument inline members — no heap allocation to free. +#else + // Non-ESP32: the bridge runs cooperatively in loop(); there is no separate + // task to signal. Drive straight to a clean Stopped and let releaseResources() + // perform the (unchanged) synchronous teardown. + _stop_acked = true; + _lifecycle.onTaskStopped(); +#endif + // Timezone is inline class storage (_timezone_storage) — nothing to delete. + // JSON documents are StaticJsonDocument inline members — no heap to free. _initialized = false; _slots_setup_done = false; // Reset so deferred setup runs again on next begin() - MQTT_DEBUG_PRINTLN("MQTT Bridge stopped"); + MQTT_DEBUG_PRINTLN("MQTT Bridge stopped (%s)", + _lifecycle.stopTimedOut() ? "forced/timeout — OTA blocked" : "clean"); +} + +// --------------------------------------------------------------------------- +// LifecycleOps — binds MQTTLifecycle::Ops (the pure, host-tested spec) to the +// FreeRTOS / PsychicMqttClient runtime. Every method runs on the loop task +// (Core 1): the Coordinator that calls them is driven only from begin()/end(). +// --------------------------------------------------------------------------- +uint32_t MQTTBridge::LifecycleOps::nowMs() { + return (uint32_t)millis(); +} + +void MQTTBridge::LifecycleOps::startTask() { + // No-op: begin() owns task/queue/buffer creation and its rollback paths. The + // Coordinator is synced to Running there via requestStart()/onTaskStarted(). +} + +void MQTTBridge::LifecycleOps::deliverStop() { + // Clear any stale ack before raising the request (same ordering as the NTP + // handshake: clear the done-flag, then set the request). The MQTT task polls + // _stop_requested at the top of mqttTaskLoop(). + _b->_stop_acked = false; + _b->_stop_requested = true; +} + +void MQTTBridge::LifecycleOps::releaseResources() { + MQTTBridge* b = _b; +#ifdef ESP_PLATFORM + // stopTimedOut() is set before this effect fires (Coordinator::dispatch), so + // it reliably distinguishes a clean ack from the timeout fallback. + const bool dirty = b->_lifecycle.stopTimedOut(); + if (dirty && !b->_stop_acked) { + // Reviewed fallback: the task never acknowledged (likely wedged in mbedTLS). + // Force-kill it and tear down clients here on Core 1 — the pre-cooperative + // behavior — accepting the heap risk. The dirty latch keeps OTA flashing + // blocked (canFlashAfterStop() == false) so firmware is never written after + // this path. + if (b->_mqtt_task_handle != nullptr) { + vTaskDelete(b->_mqtt_task_handle); + } + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) b->teardownSlot(i); + b->destroySlotClients(); + } + // Clean path (or a task that acked right at the deadline): the MQTT task + // already disconnected/deleted its clients on Core 0 and self-terminated, so + // we must NOT touch slots here (that would be a cross-core double-delete). + // Just drop our handle reference; FreeRTOS reclaims the self-deleted task's + // dynamically-allocated stack/TCB in the idle task. + b->_mqtt_task_handle = nullptr; + + // Free the PSRAM task stack (nullptr for dynamic tasks — no-op). + psram_free(b->_mqtt_task_stack); + b->_mqtt_task_stack = nullptr; + + // Drain and delete the FreeRTOS packet queue (value-copied packets, no + // external pointers to clean up). Safe on Core 1: not a TLS resource. + if (b->_packet_queue_handle != nullptr) { + QueuedPacket queued; + while (xQueueReceive(b->_packet_queue_handle, &queued, 0) == pdTRUE) { + b->_queue_count--; + } + vQueueDelete(b->_packet_queue_handle); + b->_packet_queue_handle = nullptr; + } + #if defined(BOARD_HAS_PSRAM) + psram_free(b->_packet_queue_storage); + #endif + b->_packet_queue_storage = nullptr; +#else + // Non-ESP32 circular buffer + synchronous client teardown (unchanged behavior). + for (int i = 0; i < b->_queue_count; i++) { + int index = (b->_queue_head + i) % MAX_QUEUE_SIZE; + memset(&b->_packet_queue[index], 0, sizeof(QueuedPacket)); + } + b->_queue_count = 0; + b->_queue_head = 0; + b->_queue_tail = 0; + memset(b->_packet_queue, 0, sizeof(b->_packet_queue)); + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) b->teardownSlot(i); + b->destroySlotClients(); +#endif + + b->releaseRuntimeBuffers(); +} + +void MQTTBridge::LifecycleOps::onStopComplete(bool clean) { + MQTT_DEBUG_PRINTLN("MQTT stop %s", clean + ? "acknowledged (clean)" + : "TIMED OUT (dirty; OTA flashing withheld)"); } // --------------------------------------------------------------------------- @@ -1000,6 +1108,22 @@ void MQTTBridge::mqttTaskLoop() { static unsigned long last_agent_log = 0; #endif while (true) { + // Cooperative stop (Phase 5). end() on the loop task (Core 1) set this flag. + // Tear down our own clients HERE on Core 0 — where the mbedTLS/transport + // state lives — instead of letting Core 1 free them after a blind + // vTaskDelete. Acknowledge LAST so end() only frees the queue/buffers once + // this teardown has completed, then self-terminate via the mqttTask() + // trampoline (vTaskDelete(nullptr)). + if (_stop_requested) { + MQTT_DEBUG_PRINTLN("MQTT task: cooperative stop — tearing down clients on Core 0"); + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + teardownSlot(i); + } + destroySlotClients(); + _stop_acked = true; // release semantics: set only after teardown is done + return; + } + #ifdef MQTT_MEMORY_DEBUG // #region agent log unsigned long now_loop = millis(); diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 6893d3c3..a5d405fd 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -9,6 +9,7 @@ #include #include "helpers/JWTHelper.h" #include "helpers/MQTTPresets.h" +#include "helpers/MQTTLifecycle.h" #ifdef WITH_SNMP class MeshSNMPAgent; // Forward declaration @@ -221,6 +222,16 @@ private: NtpDiagResult _ntp_diag_results[kMaxNtpServers]; int _ntp_diag_count; + // Cooperative-shutdown handshake (Phase 5). The loop task (Core 1) raises + // _stop_requested through the lifecycle Coordinator; the MQTT task (Core 0) + // sees it, tears down its own clients on Core 0 (where the mbedTLS contexts + // live), sets _stop_acked LAST, and self-terminates. end() waits for the ack + // before freeing the queue/buffers. Plain volatile matches the existing + // NTP/reconfigure handshake idiom above; replacing all of these with a command + // channel / task notifications is explicitly deferred (see MQTT_OWNERSHIP.md). + volatile bool _stop_requested = false; + volatile bool _stop_acked = false; + // Timezone handling. // _timezone_storage is inline class storage (zero heap) that is reconfigured // via setRules() whenever the preferred timezone string changes. _timezone @@ -382,6 +393,27 @@ private: void allocateRuntimeBuffers(); void releaseRuntimeBuffers(); + // --- Cooperative lifecycle (Phase 5) --------------------------------------- + // The pure state machine, bounded stop timeout, and OTA barrier live in + // src/helpers/MQTTLifecycle.h and are host-tested by test/test_mqtt_lifecycle/. + // This nested Ops binds that spec to FreeRTOS/PsychicMqttClient. The + // Coordinator is owned and driven ONLY by the loop task (Core 1) from + // begin()/end(); the MQTT task (Core 0) communicates solely through the + // _stop_requested/_stop_acked flags above. Methods are defined in the .cpp. + class LifecycleOps : public MQTTLifecycle::Ops { + public: + explicit LifecycleOps(MQTTBridge* bridge) : _b(bridge) {} + uint32_t nowMs() override; + void startTask() override; + void deliverStop() override; + void releaseResources() override; + void onStopComplete(bool clean) override; + private: + MQTTBridge* _b; + }; + LifecycleOps _lifecycle_ops; + MQTTLifecycle::Coordinator _lifecycle; + // Observer config (MQTT/WiFi/timezone/SNMP/alert), persisted to /mqtt_prefs. // _prefs (held by BridgeBase) still provides upstream fields (freq/sf/node_name…). MQTTPrefs* _obs = nullptr; @@ -430,6 +462,11 @@ public: int getConnectedBrokers() const; int getQueueSize() const; bool isReady() const; + /** True only after a CLEAN cooperative stop — end() received the MQTT task's + * acknowledgment within the timeout. A timed-out/forced stop returns false so + * OTA flashing is withheld until a clean start/stop cycle. Mirrors + * MQTTLifecycle::mayBeginFlash(); read on the loop task (Core 1). */ + bool canFlashAfterStop() const { return _lifecycle.mayBeginFlash(); } static unsigned long getWifiConnectedAtMillis();