fix(observer): keep heap walks and OTA update off the loop

Sample snapshot heap at 1 Hz, finish ota update after the queued
check, and use a task mutex for mailbox copies so multi-kilobyte
preference images no longer disable interrupts.

MQTT_OWNERSHIP.md

@@ -18,7 +18,7 @@ Current implementation: `feat/observer-reliability`, based on `5c0da89117cdddcaf
 
 Core affinity is not ownership: the SDK MQTT task and bridge worker may both run on Core 0 and still require synchronization. The queue and snapshot publishers above run in task context, not from an ISR.
 
-`ObserverMailbox`, `ObserverConfigMailbox` and `ObserverAsyncJob` use bounded copy critical sections on ESP32 and a mutex in host tests. No allocation, formatting, flash write, SDK call or network operation runs under their lock. `MQTTPrefs` is 2,880 bytes in the current layout; the bridge retains two copies (pending and applied). The portal retains another copy while its object exists. These costs need to be included in non-PSRAM heap measurements.
+`ObserverMailbox`, `ObserverConfigMailbox`, `MqttEventChannel` and `ObserverAsyncJob` share `ObserverLock`: a statically allocated FreeRTOS mutex on ESP32 and `std::mutex` in host tests. Every holder is a task, never an ISR, so the lock does not disable interrupts; priority inheritance covers a preempted holder. No allocation, formatting, flash write, SDK call or network operation runs under it. The worker's runtime snapshot publishes every pass, but its heap fields come from a once-per-second sample so the heap walk is not paid per iteration. `MQTTPrefs` is 2,880 bytes in the current layout; the bridge retains two copies (pending and applied). The portal retains another copy while its object exists. These costs need to be included in non-PSRAM heap measurements.
 
 ## Event and configuration ordering
 
@@ -44,7 +44,7 @@ Client handles can be reused across compatible configuration changes, but TLS se
 
 `get mqtt.ntp.diag` queues a probe and returns immediately. Repeat it to retrieve the completed result, identified by request ID and age. Results are cached for 30 seconds; after expiry, the next call starts a new probe. The probe never sets the clock. Forced primary NTP synchronization also returns immediately, and shares the completion-based retry scheduler with ordinary synchronization.
 
-`ota check` follows the same request/result pattern, caching results for 60 seconds. A queued or running check cannot authorize flashing. If `ota update` starts a check, repeat the command after completion to request the update. Actual flashing still uses the existing deferred stop/barrier path and re-fetches the manifest with certificate verification. The check remains advisory.
+`ota check` follows the same request/result pattern, caching results for 60 seconds. A queued or running check cannot authorize flashing. If `ota update` has to start a check, the app loop arms `OtaUpdateFollowUp`, polls `MainBoard::otaCheckInProgress()`, and schedules the ordinary deferred flash itself once the cached dry run reports an applicable build; a refusal or a check that does not settle within two minutes is reported on the console and alert channel instead. Actual flashing still uses the existing deferred stop/barrier path and re-fetches the manifest with certificate verification. The check remains advisory.
 
 The portal terminal uses the same commands and replies; repeat a queued diagnostic command to poll its job. A completed portal command batch means its CLI commands returned, not that an asynchronous probe finished.
 

docs/cli_commands.md

@@ -1379,4 +1379,4 @@ Ethernet support is available on RAK4631 boards with a RAK13800 (W5100S) Etherne
 
 ### Observer OTA checks
 
-`ota check` queues a manifest check and returns immediately. Repeat it for the result, which is cached for 60 seconds and includes an ID and age. If `ota update` returns a queued/running check, repeat `ota update` after the check finishes to request flashing. A pending check never starts an update. Actual updates retain the deferred bridge-stop barrier and revalidate the manifest before flashing.
+`ota check` queues a manifest check and returns immediately. Repeat it for the result, which is cached for 60 seconds and includes an ID and age. `ota update` with no fresh check queues one and replies `OTA check #N queued; update starts automatically if a newer build applies`: the node waits for that check (up to two minutes) and then schedules the flash itself, or reports `OTA not started: <reason>` on the serial console and the alert channel. A pending check never starts an update. Actual updates retain the deferred bridge-stop barrier and revalidate the manifest before flashing.

examples/simple_repeater/MyMesh.cpp

@@ -1776,6 +1776,7 @@ void MyMesh::loop() {
   }
 
 #if defined(WITH_MQTT_BRIDGE) && defined(OTA_MANIFEST_BASE)
+  pollDeferredOtaCheck();
   if (_ota_update_at && millisHasNowPassed(_ota_update_at)) { // deferred `ota update`
     _ota_update_at = 0;                                       // clear timer
     // The "Beginning update..." reply has now gone out. Free the bridge for heap

examples/simple_repeater/MyMesh.h

@@ -29,6 +29,7 @@
 #include "helpers/bridges/MQTTBridge.h"
 #define WITH_BRIDGE
 #include "helpers/esp32/WebConfigServer.h"   // defines WITH_WEBCONFIG on ESP32
+#include "helpers/OtaUpdateFollowUp.h"
 #endif
 
 #ifdef WITH_SNMP
@@ -126,6 +127,9 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
   CayenneLPP telemetry;
   unsigned long set_radio_at, revert_radio_at;
   unsigned long _ota_update_at = 0;  // deferred `ota update` fire time (0 = none scheduled)
+#ifdef WITH_MQTT_BRIDGE
+  OtaUpdateFollowUp _ota_follow_up;  // `ota update` waiting on a queued manifest check
+#endif
   float pending_freq;
   float pending_bw;
   uint8_t pending_sf;
@@ -489,6 +493,44 @@ public:
     return true;
   }
 
+#if defined(WITH_MQTT_BRIDGE)
+  bool beginDeferredOtaUpdateAfterCheck() override {
+    _ota_follow_up.arm(millis());
+    return true;
+  }
+
+  // Poll the queued manifest check that an `ota update` is waiting on. Runs on
+  // the loop task; the board's check task owns the HTTP work.
+  void pollDeferredOtaCheck() {
+    if (!_ota_follow_up.armed()) return;
+    char ota_reply[160] = "";
+    mesh::MainBoard* board = _cli.getBoard();
+    const OtaUpdateFollowUp::Action action = _ota_follow_up.poll(millis(), board->otaCheckInProgress(), [&] {
+      if (board->otaFromManifest(getFirmwareVer(), true, ota_reply)) return OtaUpdateFollowUp::Check::Applicable;
+      return board->otaCheckInProgress() ? OtaUpdateFollowUp::Check::Requeued
+                                         : OtaUpdateFollowUp::Check::NotApplicable;
+    });
+    char msg[160];
+    switch (action) {
+      case OtaUpdateFollowUp::Action::Start:
+        Serial.print("OTA: check complete - "); Serial.println(ota_reply);
+        beginDeferredOtaUpdate();
+        break;
+      case OtaUpdateFollowUp::Action::Refuse:
+        Serial.print("OTA: not started - "); Serial.println(ota_reply);
+        snprintf(msg, sizeof(msg), "OTA not started: %s", ota_reply);
+        otaAlert(msg);
+        break;
+      case OtaUpdateFollowUp::Action::Timeout:
+        Serial.println("OTA: manifest check did not complete; update cancelled");
+        otaAlert("OTA not started: manifest check did not complete");
+        break;
+      default:
+        break;
+    }
+  }
+#endif
+
   int getQueueSize() override {
     return bridge ? bridge->getQueueSize() : 0;
   }

examples/simple_room_server/MyMesh.cpp

@@ -1644,6 +1644,7 @@ void MyMesh::loop() {
 #endif
 
 #if defined(WITH_MQTT_BRIDGE) && defined(OTA_MANIFEST_BASE)
+  pollDeferredOtaCheck();
   if (_ota_update_at && millisHasNowPassed(_ota_update_at)) { // deferred `ota update`
     _ota_update_at = 0;                                       // clear timer
     // The "Beginning update..." reply has now been queued.  Flush it before OTA

examples/simple_room_server/MyMesh.h

@@ -34,6 +34,7 @@
 #include "helpers/bridges/MQTTBridge.h"
 #define WITH_BRIDGE
 #include "helpers/esp32/WebConfigServer.h"   // defines WITH_WEBCONFIG on ESP32
+#include "helpers/OtaUpdateFollowUp.h"
 #endif
 
 #ifdef WITH_SNMP
@@ -143,6 +144,9 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
   TransportKey default_scope;
   unsigned long set_radio_at, revert_radio_at;
   unsigned long _ota_update_at = 0;  // deferred `ota update` fire time (0 = none scheduled)
+#ifdef WITH_MQTT_BRIDGE
+  OtaUpdateFollowUp _ota_follow_up;  // `ota update` waiting on a queued manifest check
+#endif
   float pending_freq;
   float pending_bw;
   uint8_t pending_sf;
@@ -479,6 +483,46 @@ public:
     return true;
   }
 
+#if defined(WITH_MQTT_BRIDGE)
+  bool beginDeferredOtaUpdateAfterCheck() override {
+    _ota_follow_up.arm(millis());
+    return true;
+  }
+
+  // Poll the queued manifest check that an `ota update` is waiting on. Runs on
+  // the loop task; the board's check task owns the HTTP work.
+  void pollDeferredOtaCheck() {
+    if (!_ota_follow_up.armed()) return;
+    char ota_reply[160] = "";
+    mesh::MainBoard* board = _cli.getBoard();
+    const OtaUpdateFollowUp::Action action = _ota_follow_up.poll(millis(), board->otaCheckInProgress(), [&] {
+      if (board->otaFromManifest(getFirmwareVer(), true, ota_reply)) return OtaUpdateFollowUp::Check::Applicable;
+      return board->otaCheckInProgress() ? OtaUpdateFollowUp::Check::Requeued
+                                         : OtaUpdateFollowUp::Check::NotApplicable;
+    });
+    char msg[160];
+    switch (action) {
+      case OtaUpdateFollowUp::Action::Start:
+        Serial.print("OTA: check complete - "); Serial.println(ota_reply);
+        beginDeferredOtaUpdate();
+        break;
+      case OtaUpdateFollowUp::Action::Refuse:
+        Serial.print("OTA: not started - "); Serial.println(ota_reply);
+        snprintf(msg, sizeof(msg), "OTA not started: %s", ota_reply);
+        if (_cli.getObserverPrefs() && _cli.getObserverPrefs()->alert_enabled) _alerter.sendText(msg);
+        break;
+      case OtaUpdateFollowUp::Action::Timeout:
+        Serial.println("OTA: manifest check did not complete; update cancelled");
+        if (_cli.getObserverPrefs() && _cli.getObserverPrefs()->alert_enabled) {
+          _alerter.sendText("OTA not started: manifest check did not complete");
+        }
+        break;
+      default:
+        break;
+    }
+  }
+#endif
+
   int getQueueSize() override {
     return bridge ? bridge->getQueueSize() : 0;
   }

src/MeshCore.h

@@ -76,6 +76,9 @@ public:
   // current_ver is the running firmware version string (used to skip if already up to date); when
   // dry_run is true the build is only reported, not flashed. Observer (ESP32+WiFi) builds only.
   virtual bool otaFromManifest(const char* current_ver, bool dry_run, char reply[]) { return false; }
+  // True while a queued or running manifest check owns the dry-run result, so a
+  // caller that wants the outcome knows to poll rather than treat "false" as "no update".
+  virtual bool otaCheckInProgress() const { return false; }
 
   // Power management interface (boards with power management override these)
   virtual bool isPwrMgtInitialised() const { return false; }

src/helpers/CommonCLI.h

@@ -340,6 +340,13 @@ public:
     return false;
   };
 
+  // `ota update` arrived while the manifest check was still queued. The app loop
+  // waits for that check and then arms beginDeferredOtaUpdate() itself if a
+  // newer build applies. Returns true if the follow-up was armed.
+  virtual bool beginDeferredOtaUpdateAfterCheck() {
+    return false;
+  };
+
   virtual int getQueueSize() {
     return 0; // no op by default
   };

src/helpers/CommonCLI_Observer.cpp

@@ -1203,10 +1203,9 @@ bool CommonCLI::handleObserverCommand(uint32_t sender_timestamp, char* command,
       snprintf(reply, 160, "ERR: %s not connected",
                activeNetworkLink().mediumName());
     } else if (memcmp(command, "ota check", 9) == 0) {
-      // Check is synchronous so its result lands in this reply, and runs with the
-      // MQTT bridge UP: the slim per-variant manifest is tiny, so the fetch only
-      // costs a single TLS handshake (no large JSON doc) — which fits alongside
-      // the live MQTT sessions even on no-PSRAM boards. No bridge bounce needed.
+      // Queues a Core-0 HTTP fetch and returns immediately. Repeat the command
+      // for the cached result. The MQTT bridge stays up: the slim per-variant
+      // manifest is tiny enough to fetch alongside live sessions.
       _board->otaFromManifest(_callbacks->getFirmwareVer(), true, reply);
     } else {
       // `ota update`: cheap pre-check first (plain HTTP, bridge stays up). Only
@@ -1244,6 +1243,18 @@ bool CommonCLI::handleObserverCommand(uint32_t sender_timestamp, char* command,
         } else {
           strcpy(reply, "ERR: online OTA not available");
         }
+      } else if (_board->otaCheckInProgress()) {
+        // reply holds "OTA check #N queued|running; repeat ...". Keep the ID and
+        // let the app loop pick the result up instead of asking the operator to.
+        char head[48];
+        const char* tail = strchr(reply, ';');
+        const size_t len = tail ? (size_t)(tail - reply) : strlen(reply);
+        snprintf(head, sizeof(head), "%.*s", (int)(len < sizeof(head) - 1 ? len : sizeof(head) - 1), reply);
+        if (_callbacks->beginDeferredOtaUpdateAfterCheck()) {
+          snprintf(reply, 160, "%s; update starts automatically if a newer build applies", head);
+        } else {
+          snprintf(reply, 160, "%s; repeat 'ota update' for result", head);
+        }
       }
     }
 #else

src/helpers/ESP32Board.cpp

@@ -306,6 +306,12 @@ static void ota_task_entry(void* param) {
   vTaskDelete(nullptr);
 }
 
+bool ESP32Board::otaCheckInProgress() const {
+  using Check = ObserverAsyncJob<OtaCheckResult>;
+  const auto check = s_ota_check.read();
+  return check.state == Check::State::Queued || check.state == Check::State::Running;
+}
+
 bool ESP32Board::otaFromManifest(const char* current_ver, bool dry_run, char reply[]) {
   using Check = ObserverAsyncJob<OtaCheckResult>;
   const auto check = s_ota_check.read();
@@ -587,6 +593,7 @@ bool ESP32Board::otaFromManifest(const char* current_ver, bool dry_run, char rep
   strcpy(reply, "ERR: not supported");
   return false;
 }
+bool ESP32Board::otaCheckInProgress() const { return false; }
 #endif  // WITH_MQTT_BRIDGE
 
 void ESP32Board::powerOff() {

src/helpers/ESP32Board.h

@@ -162,6 +162,7 @@ public:
   bool isOTAUpdateInProgress() const override;
   void maintainOTAUpdate(uint32_t now_ms) override;
   bool otaFromManifest(const char* current_ver, bool dry_run, char reply[]) override;
+  bool otaCheckInProgress() const override;
   // Heavy body (TLS + JSON / HTTPUpdate). Runs in a dedicated large-stack task
   // spawned by otaFromManifest() — public only so that task entry point can call
   // it; not meant to be invoked directly.

src/helpers/ObserverMailbox.h

@@ -3,31 +3,44 @@
 #include <type_traits>
 #ifdef ESP_PLATFORM
 #include <freertos/FreeRTOS.h>
-#include <freertos/portmacro.h>
+#include <freertos/semphr.h>
 #else
 #include <mutex>
 #endif
 
-// Only bounded value copies belong under this lock, never allocation or I/O.
+// Task-context lock for bounded value copies. Every holder is a FreeRTOS task
+// (loop, bridge worker, SDK event task, AsyncTCP), never an ISR, so a mutex is
+// the right primitive: a critical section would disable interrupts on the radio
+// core for the whole copy, and some of these records are kilobytes. Priority
+// inheritance covers a preempted holder. Never hold this across allocation,
+// I/O, flash writes or SDK calls.
 class ObserverLock {
  public:
+  ObserverLock() {
+#ifdef ESP_PLATFORM
+    _mux = xSemaphoreCreateMutexStatic(&_storage);
+#endif
+  }
+  ObserverLock(const ObserverLock&) = delete;
+  ObserverLock& operator=(const ObserverLock&) = delete;
   void lock() {
 #ifdef ESP_PLATFORM
-    portENTER_CRITICAL(&_mux);
+    xSemaphoreTake(_mux, portMAX_DELAY);
 #else
     _mux.lock();
 #endif
   }
   void unlock() {
 #ifdef ESP_PLATFORM
-    portEXIT_CRITICAL(&_mux);
+    xSemaphoreGive(_mux);
 #else
     _mux.unlock();
 #endif
   }
  private:
 #ifdef ESP_PLATFORM
-  portMUX_TYPE _mux = portMUX_INITIALIZER_UNLOCKED;
+  StaticSemaphore_t _storage;
+  SemaphoreHandle_t _mux;
 #else
   std::mutex _mux;
 #endif

src/helpers/OtaUpdateFollowUp.h

@@ -0,0 +1,42 @@
+#pragma once
+#include <stdint.h>
+
+// `ota update` was issued while the manifest check was still queued or running.
+// The app loop polls the board until that check settles, then either arms the
+// ordinary deferred flash or reports why it will not run. Bounded so a check
+// that never completes cannot leave an update armed for the rest of the boot.
+class OtaUpdateFollowUp {
+ public:
+  enum class Action : uint8_t { None, Start, Refuse, Timeout };
+  enum class Check : uint8_t { Applicable, NotApplicable, Requeued };
+  static constexpr uint32_t kTimeoutMs = 120000;
+
+  bool armed() const { return _armed; }
+  void arm(uint32_t now) {
+    _armed = true;
+    _deadline = now + kTimeoutMs;
+  }
+  void clear() { _armed = false; }
+
+  // dry_run runs only once the board reports no check in progress. Requeued
+  // means the cached result had already expired and the board started another
+  // check; stay armed and wait for that one.
+  template<class DryRun>
+  Action poll(uint32_t now, bool check_in_progress, DryRun dry_run) {
+    if (!_armed) return Action::None;
+    if ((int32_t)(now - _deadline) >= 0) {
+      _armed = false;
+      return Action::Timeout;
+    }
+    if (check_in_progress) return Action::None;
+    switch (dry_run()) {
+      case Check::Applicable: _armed = false; return Action::Start;
+      case Check::Requeued: return Action::None;
+      default: _armed = false; return Action::Refuse;
+    }
+  }
+
+ private:
+  bool _armed = false;
+  uint32_t _deadline = 0;
+};

src/helpers/bridges/MQTTBridge.cpp

@@ -241,8 +241,14 @@ void MQTTBridge::publishRuntimeSnapshot() {
   snapshot.skipped = _skipped_publishes;
   snapshot.filtered = _filtered_packets.load(std::memory_order_relaxed);
   snapshot.event_overflows = _slot_events.overflows();
-  snapshot.free_heap = ESP.getFreeHeap();
-  snapshot.max_heap = ESP.getMaxAllocHeap();
+  if (!_heap_sampled || (uint32_t)(snapshot.sampled_ms - _heap_sample_ms) >= 1000) {
+    _heap_sampled = true;
+    _heap_sample_ms = snapshot.sampled_ms;
+    _heap_sample_free = ESP.getFreeHeap();
+    _heap_sample_max = ESP.getMaxAllocHeap();
+  }
+  snapshot.free_heap = _heap_sample_free;
+  snapshot.max_heap = _heap_sample_max;
 #if defined(WITH_MQTT_NEIGHBORS)
   snapshot.neighbors_phase = _neighbors_phase.load(std::memory_order_relaxed);
   snapshot.neighbors_result = _neighbors_last_result.load(std::memory_order_relaxed);
@@ -884,6 +890,7 @@ void MQTTBridge::begin() {
   }
 
   _ntp_schedule = NtpSchedule();
+  _heap_sampled = false;
   _slots_setup_done = false;
   _ntp_force_requested.store(false);
 

src/helpers/bridges/MQTTBridge.h

@@ -232,6 +232,11 @@ private:
   };
   static ObserverMailbox<RuntimeSnapshot> _published;
   uint32_t _snapshot_sequence = 0;
+  // Snapshots publish every worker pass; the heap walk behind getMaxAllocHeap()
+  // is sampled at most once per second and reused between samples.
+  uint32_t _heap_sample_ms = 0;
+  uint32_t _heap_sample_free = 0, _heap_sample_max = 0;
+  bool _heap_sampled = false;
   void publishRuntimeSnapshot();
   bool mayEditStartupMetadata();
 

test/test_observer_handoffs/test_observer_handoffs.cpp

@@ -8,6 +8,7 @@
 #include "helpers/MQTTSlotEvents.h"
 #include "helpers/ObserverPreferenceCommit.h"
 #include "helpers/ObserverSlotStatsJson.h"
+#include "helpers/OtaUpdateFollowUp.h"
 
 
 struct Image { uint32_t sequence = 0; uint32_t fields[64]{}; };
@@ -320,6 +321,40 @@ TEST(ObserverHandoffs, StatsReserveClosingBracketsAtEveryBufferBoundary) {
   }
 }
 
+TEST(ObserverHandoffs, UpdateFollowUpWaitsForTheCheckThenStartsOnce) {
+  using F = OtaUpdateFollowUp;
+  F follow;
+  int dry_runs = 0;
+  auto applicable = [&] { ++dry_runs; return F::Check::Applicable; };
+  EXPECT_EQ(F::Action::None, follow.poll(0, false, applicable));  // never armed
+  follow.arm(1000);
+  EXPECT_EQ(F::Action::None, follow.poll(1500, true, applicable));   // still checking
+  EXPECT_EQ(0, dry_runs);
+  EXPECT_EQ(F::Action::Start, follow.poll(2000, false, applicable));
+  EXPECT_EQ(1, dry_runs);
+  EXPECT_FALSE(follow.armed());
+  EXPECT_EQ(F::Action::None, follow.poll(2001, false, applicable));  // consumed
+  EXPECT_EQ(1, dry_runs);
+}
+
+TEST(ObserverHandoffs, UpdateFollowUpRefusesRequeuesAndTimesOut) {
+  using F = OtaUpdateFollowUp;
+  F follow;
+  follow.arm(0);
+  EXPECT_EQ(F::Action::Refuse, follow.poll(10, false, [] { return F::Check::NotApplicable; }));
+  EXPECT_FALSE(follow.armed());
+
+  follow.arm(0);
+  EXPECT_EQ(F::Action::None, follow.poll(10, false, [] { return F::Check::Requeued; }));
+  EXPECT_TRUE(follow.armed());  // the cache had expired; wait for the new check
+  EXPECT_EQ(F::Action::Start, follow.poll(20, false, [] { return F::Check::Applicable; }));
+
+  follow.arm(0xffffff00);  // deadline wraps past zero
+  EXPECT_EQ(F::Action::None, follow.poll(0xffffff10, true, [] { return F::Check::Applicable; }));
+  EXPECT_EQ(F::Action::Timeout, follow.poll(0xffffff00 + F::kTimeoutMs, true, [] { return F::Check::Applicable; }));
+  EXPECT_FALSE(follow.armed());
+}
+
 int main(int argc, char** argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();

webui/index.html

@@ -1507,7 +1507,7 @@ var CLI_VERBS=[
   ["password ","Change the admin password {new-password}"],
   ["alert test","Send a test alert on the configured channel"],
   ["ota check","Queue update check; repeat for result (does not flash)"],
-  ["ota update","Download and flash the newer build, then reboot"],
+  ["ota update","Check, then download and flash a newer build and reboot"],
   // `start ota` is absent: it binds port 80, which this portal is already using.
   ["start webconfig","Start this portal on the LAN"],
   ["start webconfig ap","Start this portal on its own setup AP"],