Start radio liveness watchdog when loop service begins

src/Dispatcher.cpp

@@ -48,6 +48,7 @@ void Dispatcher::setRadioAvailable(bool available) {
   if (radio_available == available) return;
 
   radio_available = available;
+  radio_nonrx_timer_armed = false;
   if (!dispatcher_started || !radio_available) return;
 
   const unsigned long now = _ms->getMillis();
@@ -78,6 +79,7 @@ void Dispatcher::begin() {
   outbound_cancellation = OutboundCancellation::None;
   ota_tx_airtime = 0;
   radio_nonrx_start = _ms->getMillis();
+  radio_nonrx_timer_armed = false;
 
   duty_cycle_window_ms = getDutyCycleWindowMs();
   float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
@@ -368,6 +370,13 @@ void Dispatcher::loop() {
   if (_radio->isCarrierWaveActive()) return;
 
   const unsigned long now = _ms->getMillis();
+  if (!radio_nonrx_timer_armed) {
+    // Application setup can outlast the watchdog after begin() leaves the
+    // radio in standby. Start observing when loop service actually begins,
+    // giving checkRecv() below its first normal opportunity to enter RX.
+    radio_nonrx_start = now;
+    radio_nonrx_timer_armed = true;
+  }
   // check for radio 'stuck' in mode other than Rx
   bool is_recv = _radio->isInRecvMode();
   if (is_recv != prev_isrecv_mode) {

src/Dispatcher.h

@@ -315,6 +315,7 @@ class Dispatcher {
   unsigned long next_floor_calib_time, next_agc_reset_time;
   int   armed_agc_reset_interval;
   bool  agc_reset_armed;
+  bool  radio_nonrx_timer_armed;
   bool  prev_isrecv_mode;
   bool  radio_available;
   bool  dispatcher_started;
@@ -392,6 +393,7 @@ protected:
     agc_reset_armed = false;
     _err_flags = 0;
     radio_nonrx_start = 0;
+    radio_nonrx_timer_armed = false;
     prev_isrecv_mode = true;
     radio_available = true;
     dispatcher_started = false;

test/fixtures/dispatcher/startup_watchdog_test.cpp

@@ -0,0 +1,212 @@
+#include <Dispatcher.h>
+#include <helpers/StaticPoolPacketManager.h>
+#include <cassert>
+#include <cstdio>
+#include <cstring>
+#include <vector>
+
+class TestClock : public mesh::MillisecondClock {
+ public:
+  unsigned long now = 0;
+  unsigned long getMillis() override { return now; }
+};
+
+// RadioLib begins in standby and first attempts RX from recvRaw().
+class TestRadio : public mesh::Radio {
+ public:
+  bool in_rx = false;
+  bool receive_succeeds = true;
+  bool recovery_succeeds = false;
+  bool carrier = false;
+  unsigned begins = 0;
+  unsigned receive_polls = 0;
+  std::vector<bool> recoveries;
+  std::vector<unsigned> polls_before_recovery;
+
+  void begin() override { ++begins; in_rx = false; }
+  int recvRaw(uint8_t*, int) override {
+    ++receive_polls;
+    if (receive_succeeds) in_rx = true;
+    return 0;
+  }
+  uint32_t getEstAirtimeFor(int) override { return 1; }
+  float packetScore(float, int) override { return 0; }
+  bool startSendRaw(const uint8_t*, int) override { return false; }
+  bool isSendComplete() override { return false; }
+  void onSendFinished() override {}
+  bool isInRecvMode() const override { return in_rx; }
+  bool isCarrierWaveActive() const override { return carrier; }
+  bool recoverRadio(bool hard) override {
+    recoveries.push_back(hard);
+    polls_before_recovery.push_back(receive_polls);
+    if (recovery_succeeds) in_rx = true;
+    return recovery_succeeds;
+  }
+};
+
+class TestDispatcher : public mesh::Dispatcher {
+ public:
+  TestDispatcher(TestRadio& radio, TestClock& clock, StaticPoolPacketManager& manager)
+      : Dispatcher(radio, clock, manager) {}
+  mesh::DispatcherAction onRecvPacket(mesh::Packet*) override { return ACTION_RELEASE; }
+  using Dispatcher::setRadioAvailable;
+};
+
+struct Fixture {
+  TestClock clock;
+  TestRadio radio;
+  StaticPoolPacketManager manager{4};
+  TestDispatcher dispatcher{radio, clock, manager};
+
+  void beginAt(unsigned long now) { clock.now = now; dispatcher.begin(); }
+  void loopAt(unsigned long now) { clock.now = now; dispatcher.loop(); }
+  void expectHealthy() {
+    assert(dispatcher.getErrFlags() == 0);
+    assert(radio.recoveries.empty());
+  }
+};
+
+static void delayedSetup() {
+  Fixture f;
+  f.beginAt(1000);
+  // Preferences, identity storage and Bluetooth setup run before loop().
+  f.loopAt(31000);
+  f.expectHealthy();
+  assert(f.radio.begins == 1 && f.radio.receive_polls == 1 && f.radio.in_rx);
+  f.loopAt(51000);
+  f.expectHealthy();
+}
+
+static void failedInitialReceive() {
+  Fixture f;
+  f.radio.receive_succeeds = false;
+  f.beginAt(1000);
+  f.loopAt(31000);
+  f.expectHealthy();
+  assert(f.radio.receive_polls == 1);
+  f.loopAt(38999);
+  f.loopAt(39000);
+  f.expectHealthy();
+  f.loopAt(39001);
+  assert(f.dispatcher.getErrFlags() == ERR_EVENT_STARTRX_TIMEOUT);
+  assert(f.radio.recoveries.size() == 1 && !f.radio.recoveries[0]);
+  assert(f.radio.polls_before_recovery[0] == 3);
+  f.loopAt(39001);
+  f.loopAt(47001);
+  assert(f.radio.recoveries.size() == 1);
+  f.loopAt(47002);
+  assert(f.radio.recoveries.size() == 2);
+#ifdef RADIO_LIVENESS_SOFT_ONLY
+  assert(!f.radio.recoveries[1]);
+#else
+  assert(f.radio.recoveries[1]);
+#endif
+}
+
+static void laterReceiveLoss() {
+  Fixture f;
+  f.beginAt(1000);
+  f.loopAt(1000);
+  f.loopAt(1001);  // Observe the successful initial RX transition.
+  f.radio.in_rx = false;
+  f.radio.receive_succeeds = false;
+  f.loopAt(31000);
+  f.loopAt(39000);
+  f.expectHealthy();
+  f.radio.recovery_succeeds = true;
+  f.loopAt(39001);
+  assert(f.radio.recoveries.size() == 1 && !f.radio.recoveries[0]);
+  f.loopAt(39002);  // Observe recovery reaching RX, clearing escalation.
+  f.radio.in_rx = false;
+  f.radio.recovery_succeeds = false;
+  f.loopAt(51000);
+  f.loopAt(59000);
+  assert(f.radio.recoveries.size() == 1);
+  f.loopAt(59001);
+  assert(f.radio.recoveries.size() == 2 && !f.radio.recoveries[1]);
+}
+
+static void delayedReactivation() {
+  Fixture f;
+  f.radio.receive_succeeds = false;
+  f.beginAt(1000);
+  f.loopAt(1000);
+  f.loopAt(9001);
+  assert(f.radio.recoveries.size() == 1);
+  f.dispatcher.setRadioAvailable(false);
+  const unsigned polls = f.radio.receive_polls;
+  f.loopAt(100000);
+  assert(f.radio.receive_polls == polls && f.radio.recoveries.size() == 1);
+  f.dispatcher.resetStats();
+  f.dispatcher.setRadioAvailable(true);
+  assert(f.radio.begins == 2);
+  f.loopAt(130000);
+  assert(f.dispatcher.getErrFlags() == 0 && f.radio.recoveries.size() == 1);
+  assert(f.radio.receive_polls == polls + 1);
+  f.loopAt(138000);
+  assert(f.radio.recoveries.size() == 1);
+  f.loopAt(138001);
+  assert(f.radio.recoveries.size() == 2 && !f.radio.recoveries[1]);
+}
+
+static void repeatedBegin() {
+  Fixture f;
+  f.radio.receive_succeeds = false;
+  f.beginAt(0);
+  f.loopAt(0);  // Millis zero must be a valid timer origin.
+  f.loopAt(8000);
+  f.expectHealthy();
+  f.loopAt(8001);
+  assert(f.radio.recoveries.size() == 1);
+  f.beginAt(100000);
+  f.loopAt(130000);
+  assert(f.dispatcher.getErrFlags() == 0 && f.radio.recoveries.size() == 1);
+  f.loopAt(138000);
+  assert(f.radio.recoveries.size() == 1);
+  f.loopAt(138001);
+  assert(f.radio.recoveries.size() == 2 && !f.radio.recoveries[1]);
+}
+
+static void startupWithoutRadio() {
+  Fixture f;
+  f.dispatcher.setRadioAvailable(false);
+  f.beginAt(0);
+  f.loopAt(31000);
+  f.expectHealthy();
+  assert(f.radio.begins == 0 && f.radio.receive_polls == 0);
+  f.dispatcher.setRadioAvailable(true);
+  f.loopAt(61000);
+  f.expectHealthy();
+  assert(f.radio.begins == 1 && f.radio.receive_polls == 1 && f.radio.in_rx);
+}
+
+static void startupWithCarrier() {
+  Fixture f;
+  f.radio.carrier = true;
+  f.radio.receive_succeeds = false;
+  f.beginAt(0);
+  f.loopAt(31000);
+  f.expectHealthy();
+  assert(f.radio.receive_polls == 0);
+  f.radio.carrier = false;
+  f.loopAt(61000);
+  f.expectHealthy();
+  assert(f.radio.receive_polls == 1);
+  f.loopAt(69000);
+  f.expectHealthy();
+  f.loopAt(69001);
+  assert(f.radio.recoveries.size() == 1 && !f.radio.recoveries[0]);
+}
+
+int main(int argc, char** argv) {
+  assert(argc == 2);
+  if (!strcmp(argv[1], "delayed_setup")) delayedSetup();
+  else if (!strcmp(argv[1], "failed_initial_receive")) failedInitialReceive();
+  else if (!strcmp(argv[1], "later_receive_loss")) laterReceiveLoss();
+  else if (!strcmp(argv[1], "delayed_reactivation")) delayedReactivation();
+  else if (!strcmp(argv[1], "repeated_begin")) repeatedBegin();
+  else if (!strcmp(argv[1], "startup_without_radio")) startupWithoutRadio();
+  else if (!strcmp(argv[1], "startup_with_carrier")) startupWithCarrier();
+  else assert(false);
+  std::printf("%s passed\n", argv[1]);
+}

test/test_dispatcher_startup_watchdog.py

@@ -0,0 +1,53 @@
+"""Exercise the production Dispatcher's startup and non-RX recovery timing."""
+from pathlib import Path
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class DispatcherStartupWatchdogTest(unittest.TestCase):
+    def run_scenarios(self, soft_only):
+        compiler = shutil.which('g++') or shutil.which('clang++')
+        self.assertIsNotNone(compiler)
+        flags = ['-std=c++17', '-Wall', '-Wextra']
+        if os.name != 'nt':
+            flags += ['-fsanitize=address,undefined', '-fno-omit-frame-pointer',
+                      '-fno-pie', '-no-pie']
+        if soft_only:
+            flags += ['-DRADIO_LIVENESS_SOFT_ONLY']
+        with tempfile.TemporaryDirectory() as work:
+            exe = Path(work) / 'dispatcher_watchdog.exe'
+            result = subprocess.run([compiler, *flags,
+                '-I', str(ROOT / 'test/mocks'), '-I', str(ROOT / 'src'),
+                str(ROOT / 'src/Dispatcher.cpp'), str(ROOT / 'src/Packet.cpp'),
+                str(ROOT / 'src/helpers/StaticPoolPacketManager.cpp'),
+                str(ROOT / 'test/fixtures/dispatcher/startup_watchdog_test.cpp'),
+                '-o', str(exe)], capture_output=True, text=True)
+            self.assertEqual(result.returncode, 0, result.stderr[-14000:])
+            env = os.environ.copy()
+            # Production pools live for the whole firmware boot and have no
+            # destructor. Keep bounds/UAF/UB checks without teardown leak checks.
+            if os.name != 'nt':
+                env['ASAN_OPTIONS'] = env.get('ASAN_OPTIONS', '') + ':detect_leaks=0'
+            for scenario in ('delayed_setup', 'failed_initial_receive',
+                             'later_receive_loss', 'delayed_reactivation',
+                             'repeated_begin', 'startup_without_radio',
+                             'startup_with_carrier'):
+                with self.subTest(scenario=scenario, soft_only=soft_only):
+                    run = subprocess.run([str(exe), scenario], capture_output=True,
+                                         text=True, env=env)
+                    self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+
+    def test_soft_and_hard_recovery(self):
+        self.run_scenarios(soft_only=False)
+
+    def test_soft_only_recovery(self):
+        self.run_scenarios(soft_only=True)
+
+
+if __name__ == '__main__':
+    unittest.main()