fix(dispatcher): pick radio-watchdog activity by age, not largest timestamp

The watchdog took max(last_recv, last_irq, last_tx) of three raw millis()
timestamps. Around the 32-bit wrap a transmit stamped in the upper half of the
range is numerically larger than every fresh post-wrap receive, so it kept
winning for up to ~24 days: the watchdog measured silence from that stale
transmit, concluded the radio had gone deaf, and reset a radio that was
receiving normally — once per watchdog interval, until the old value stopped
winning.

Extract the decision into RadioWatchdog (pure, host-testable). It compares each
timestamp only against its own previous value, so any change is fresh activity,
and stamps it against a monotonic 64-bit clock extended from millis(). Evidence
older than one 32-bit cycle now stays old instead of aliasing back to "recent",
which the previous unsigned-subtraction age could not do either.

Covered in test/test_radio_watchdog: pre-wrap TX with continuous post-wrap RX,
genuine silence across the wrap, silence past a full millis() cycle, no activity
since boot, disabled watchdog, TX-only traffic, and a stalled mesh loop.

src/Dispatcher.cpp

@@ -29,6 +29,9 @@ void Dispatcher::begin() {
 
   _radio->begin();
   prev_isrecv_mode = _radio->isInRecvMode();
+#ifdef WITH_MQTT_BRIDGE
+  radio_watchdog.reset((uint32_t)_ms->getMillis());
+#endif
 }
 
 float Dispatcher::getAirtimeBudgetFactor() const {
@@ -94,23 +97,20 @@ void Dispatcher::loop() {
   // MQTTPrefs radio_watchdog_minutes setting.
 #ifdef WITH_MQTT_BRIDGE
   {
-    const uint32_t watchdog_ms = getRadioWatchdogMillis();
-    if (watchdog_ms > 0) {
-      unsigned long last_recv = _radio->getLastRecvMillis();
-      unsigned long last_irq  = _radio->getLastRadioInterruptMillis();
-      unsigned long last_active = (last_recv > last_irq ? last_recv : last_irq);
-      if (last_radio_active_ms > last_active) last_active = last_radio_active_ms;
-      if (is_recv && last_active > 0) {
-        unsigned long silent_ms = _ms->getMillis() - last_active;
-        unsigned long since_recovery = _ms->getMillis() - last_watchdog_recovery;
-        if (silent_ms > watchdog_ms && since_recovery > watchdog_ms) {
-          _err_flags |= ERR_EVENT_RADIO_WATCHDOG;
-          MESH_DEBUG_PRINTLN("Radio watchdog: silent %lu ms, state=%d, recovering", silent_ms, _radio->getRadioState());
-          _radio->idle();
-          _radio->startRecv();
-          last_watchdog_recovery = _ms->getMillis();
-        }
-      }
+    // Selecting the most recent event by age rather than by largest timestamp is
+    // what keeps this correct across the millis() wrap — see RadioWatchdog.
+    RadioWatchdogDecision wd = radio_watchdog.update(
+        (uint32_t)_ms->getMillis(), is_recv, getRadioWatchdogMillis(),
+        (uint32_t)_radio->getLastRecvMillis(),
+        (uint32_t)_radio->getLastRadioInterruptMillis(),
+        (uint32_t)last_radio_active_ms);
+    if (wd.recover) {
+      _err_flags |= ERR_EVENT_RADIO_WATCHDOG;
+      MESH_DEBUG_PRINTLN("Radio watchdog: silent %lu ms, state=%d, recovering",
+                         (unsigned long)wd.silent_ms, _radio->getRadioState());
+      _radio->idle();
+      _radio->startRecv();
+      radio_watchdog.noteRecovery();
     }
   }
 #endif // WITH_MQTT_BRIDGE (radio watchdog)

src/Dispatcher.h

@@ -5,6 +5,9 @@
 #include <Packet.h>
 #include <Utils.h>
 #include <string.h>
+#ifdef WITH_MQTT_BRIDGE
+  #include "helpers/RadioWatchdog.h"
+#endif
 
 namespace mesh {
 
@@ -135,8 +138,10 @@ typedef uint32_t  DispatcherAction;
 class Dispatcher {
   Packet* outbound;  // current outbound packet
   unsigned long outbound_expiry, outbound_start, total_air_time, rx_air_time;
-  unsigned long last_watchdog_recovery;
   unsigned long last_radio_active_ms;   // updated on any TX or RX event; used by watchdog
+#ifdef WITH_MQTT_BRIDGE
+  RadioWatchdog radio_watchdog;
+#endif
   unsigned long next_tx_time;
   unsigned long cad_busy_start;
   unsigned long radio_nonrx_start;
@@ -171,7 +176,6 @@ protected:
     tx_budget_ms = 0;
     last_budget_update = 0;
     duty_cycle_window_ms = 3600000;
-    last_watchdog_recovery = 0;
     last_radio_active_ms = 0;
   }
 

src/helpers/RadioWatchdog.h

@@ -0,0 +1,116 @@
+#pragma once
+
+#include <stdint.h>
+
+// Decides when the observer radio watchdog should kick a radio that is parked in
+// RX but hearing nothing (see Dispatcher::loop()).
+//
+// Pure logic: no Arduino, radio or mesh headers. The caller feeds the raw 32-bit
+// activity timestamps it can see (last receive, last radio interrupt, last
+// successful transmit) plus its millis() counter.
+//
+// Why this is not a max() of those three timestamps: millis() wraps every ~49.7
+// days, and a pre-wrap timestamp is numerically larger than every fresh
+// post-wrap one. Picking the largest therefore latches onto an old transmit and
+// reports the radio silent while it is receiving continuously, resetting a
+// healthy radio once per watchdog interval until that timestamp stops winning.
+//
+// Instead the caller's timestamps are only ever compared with their own previous
+// values: any change is fresh activity, stamped against a monotonic 64-bit clock
+// extended from millis(). Nothing downstream has a rollover case, and evidence
+// older than one 32-bit cycle stays old instead of aliasing back to "recent".
+//
+// How far a timestamp may run behind the previous one and still count as an
+// out-of-order reading rather than a very long forward gap (same rule as
+// RadioActivityWindow).
+#define RADIO_WATCHDOG_BACKSTEP_TOLERANCE_MS 5000UL
+
+struct RadioWatchdogDecision {
+  bool     recover;      // trip now: idle + restart receive
+  bool     measurable;   // false until some activity has been observed
+  uint64_t silent_ms;    // age of the most recent activity (0 when !measurable)
+};
+
+class RadioWatchdog {
+public:
+  RadioWatchdog() { reset(0); }
+
+  void reset(uint32_t now_ms) {
+    _now_ms = now_ms;
+    _last_input_ms = now_ms;
+    _seeded = false;
+    _has_activity = false;
+    _last_recv = _last_irq = _last_tx = 0;
+    _last_activity_ms = now_ms;
+    _recovered = false;
+    _last_recovery_ms = 0;
+  }
+
+  // Call once per loop with the current activity timestamps. `in_recv_mode` and
+  // `watchdog_ms` (0 disables) gate the decision; the silence measurement is
+  // updated either way so a disabled or transmitting radio does not accumulate
+  // phantom silence.
+  RadioWatchdogDecision update(uint32_t now_ms, bool in_recv_mode, uint32_t watchdog_ms,
+                               uint32_t last_recv, uint32_t last_irq, uint32_t last_tx) {
+    tick(now_ms);
+
+    if (!_seeded) {
+      // First observation: the timestamps carry activity from before this
+      // tracker existed, so date it to now rather than trusting a raw value we
+      // cannot place on our own clock. Worst case that delays the first
+      // possible trip by one watchdog interval.
+      _seeded = true;
+      _has_activity = (last_recv != 0 || last_irq != 0 || last_tx != 0);
+      _last_activity_ms = _now_ms;
+    } else if (last_recv != _last_recv || last_irq != _last_irq || last_tx != _last_tx) {
+      _has_activity = true;
+      _last_activity_ms = _now_ms;
+    }
+    _last_recv = last_recv;
+    _last_irq = last_irq;
+    _last_tx = last_tx;
+
+    RadioWatchdogDecision d;
+    d.recover = false;
+    d.measurable = _has_activity;
+    d.silent_ms = _has_activity ? (_now_ms - _last_activity_ms) : 0;
+
+    if (!_has_activity || !in_recv_mode || watchdog_ms == 0) return d;
+    if (d.silent_ms <= watchdog_ms) return d;
+    // One recovery per watchdog interval: a radio that stays silent through a
+    // reset must not be reset every loop.
+    if (_recovered && (_now_ms - _last_recovery_ms) <= watchdog_ms) return d;
+
+    d.recover = true;
+    return d;
+  }
+
+  // Call after acting on a decision with recover set.
+  void noteRecovery() {
+    _recovered = true;
+    _last_recovery_ms = _now_ms;
+  }
+
+  uint64_t nowMs() const { return _now_ms; }
+
+private:
+  // Extends the caller's 32-bit millis() to a monotonic 64-bit clock.
+  void tick(uint32_t now_ms) {
+    uint32_t delta = now_ms - _last_input_ms;
+    if (delta > 0x80000000u &&
+        (uint32_t)(_last_input_ms - now_ms) <= RADIO_WATCHDOG_BACKSTEP_TOLERANCE_MS) {
+      return;   // out-of-order reading: no time has passed
+    }
+    _now_ms += delta;
+    _last_input_ms = now_ms;
+  }
+
+  uint64_t _now_ms;
+  uint32_t _last_input_ms;
+  bool     _seeded;
+  bool     _has_activity;
+  uint32_t _last_recv, _last_irq, _last_tx;
+  uint64_t _last_activity_ms;
+  bool     _recovered;
+  uint64_t _last_recovery_ms;
+};

test/test_radio_watchdog/test_radio_watchdog.cpp

@@ -0,0 +1,204 @@
+#include "helpers/RadioWatchdog.h"
+
+#include <gtest/gtest.h>
+
+namespace {
+
+const uint32_t WATCHDOG_MS = 300000UL;   // RADIO_WATCHDOG_MS default: 5 minutes
+
+// Mirrors Dispatcher::loop(): the radio's three timestamps are sampled every
+// pass, and a recovery is acknowledged as soon as it is decided.
+struct Radio {
+  uint32_t last_recv = 0;
+  uint32_t last_irq = 0;
+  uint32_t last_tx = 0;
+};
+
+RadioWatchdogDecision run(RadioWatchdog& w, const Radio& r, uint32_t now,
+                          bool in_recv_mode = true, uint32_t watchdog_ms = WATCHDOG_MS) {
+  RadioWatchdogDecision d = w.update(now, in_recv_mode, watchdog_ms,
+                                     r.last_recv, r.last_irq, r.last_tx);
+  if (d.recover) w.noteRecovery();
+  return d;
+}
+
+} // namespace
+
+TEST(RadioWatchdog, NoActivitySinceBootNeverTrips) {
+  RadioWatchdog w;
+  w.reset(0);
+  Radio r;   // radio has never received, interrupted or transmitted
+
+  for (uint32_t t = 0; t <= 4 * WATCHDOG_MS; t += 10000) {
+    RadioWatchdogDecision d = run(w, r, t);
+    EXPECT_FALSE(d.recover) << "tripped at t=" << t;
+    EXPECT_FALSE(d.measurable);
+  }
+}
+
+TEST(RadioWatchdog, ContinuousReceiveNeverTrips) {
+  RadioWatchdog w;
+  w.reset(0);
+  Radio r;
+
+  for (uint32_t t = 1000; t <= 3 * WATCHDOG_MS; t += 1000) {
+    r.last_recv = t;
+    r.last_irq = t;
+    RadioWatchdogDecision d = run(w, r, t);
+    EXPECT_FALSE(d.recover) << "tripped at t=" << t;
+    EXPECT_LE(d.silent_ms, 1000u);
+  }
+}
+
+TEST(RadioWatchdog, SilentRadioTripsOncePerInterval) {
+  RadioWatchdog w;
+  w.reset(0);
+  Radio r;
+  r.last_recv = 1000;
+  run(w, r, 1000);
+
+  int trips = 0;
+  uint32_t first_trip = 0;
+  for (uint32_t t = 2000; t <= 1000000; t += 1000) {
+    if (run(w, r, t).recover) {
+      if (trips == 0) first_trip = t;
+      trips++;
+    }
+  }
+
+  EXPECT_EQ(302000u, first_trip);           // strictly past the interval, from the last receive
+  EXPECT_EQ(3, trips);                      // ~1 M ms / 300 s, never per-loop
+}
+
+// F13: a transmit timestamp from just before the millis() wrap is numerically
+// larger than every fresh post-wrap receive. Selecting the largest timestamp
+// reported ~24 days of silence and reset a radio that was receiving normally.
+TEST(RadioWatchdog, PreWrapTransmitDoesNotSilenceFreshPostWrapReceives) {
+  const uint32_t WRAP = 0xFFFFFFFFu;
+  const uint32_t TX_AT = 0x80000000u;       // upper half: beats any post-wrap value
+
+  RadioWatchdog w;
+  w.reset(TX_AT - 60000);
+  Radio r;
+  r.last_recv = TX_AT - 60000;
+  r.last_irq = TX_AT - 60000;
+  run(w, r, TX_AT - 60000);
+
+  r.last_tx = TX_AT;
+  run(w, r, TX_AT);
+
+  // Receive steadily right up to the wrap...
+  for (uint32_t t = TX_AT + 1000; t < WRAP - 1000; t += 1000) {
+    r.last_recv = t;
+    r.last_irq = t;
+    ASSERT_FALSE(run(w, r, t).recover) << "tripped before wrap at t=" << t;
+  }
+
+  // ...and straight through it. The stale transmit is still the numerically
+  // largest of the three timestamps for the next ~24 days.
+  for (uint32_t t = 1000; t < 2 * WATCHDOG_MS; t += 1000) {
+    r.last_recv = t;
+    r.last_irq = t;
+    RadioWatchdogDecision d = run(w, r, t);
+    ASSERT_FALSE(d.recover) << "tripped after wrap at t=" << t;
+    ASSERT_LE(d.silent_ms, 1000u);
+  }
+}
+
+// The wrap must not hide a genuinely silent radio either.
+TEST(RadioWatchdog, SilenceAcrossTheWrapIsStillDetected) {
+  const uint32_t LAST_RX = 0xFFFF0000u;     // ~65 s before the wrap
+
+  RadioWatchdog w;
+  w.reset(LAST_RX - 1000);
+  Radio r;
+  r.last_recv = LAST_RX;
+  r.last_irq = LAST_RX;
+  run(w, r, LAST_RX);
+
+  bool tripped = false;
+  uint64_t silent_at_trip = 0;
+  for (uint32_t step = 1000; step <= WATCHDOG_MS + 5000; step += 1000) {
+    uint32_t t = LAST_RX + step;            // wraps on its own
+    RadioWatchdogDecision d = run(w, r, t);
+    if (d.recover) {
+      tripped = true;
+      silent_at_trip = d.silent_ms;
+      break;
+    }
+  }
+
+  EXPECT_TRUE(tripped);
+  EXPECT_GT(silent_at_trip, (uint64_t)WATCHDOG_MS);
+}
+
+// Uptime past 2^32 ms must not alias a long silence back to "recent".
+TEST(RadioWatchdog, SilenceBeyondOneMillisCycleKeepsGrowing) {
+  RadioWatchdog w;
+  w.reset(0);
+  Radio r;
+  r.last_recv = 1000;
+  run(w, r, 1000);
+
+  // Walk a full cycle in 1-minute steps; the raw timestamp aliases back to its
+  // starting value, the tracked age does not.
+  uint64_t last_silent = 0;
+  for (uint64_t step = 60000; step <= 0x100000000ull + 600000ull; step += 60000) {
+    uint32_t t = (uint32_t)((1000ull + step) & 0xFFFFFFFFull);
+    RadioWatchdogDecision d = w.update(t, true, WATCHDOG_MS, r.last_recv, r.last_irq, r.last_tx);
+    if (d.recover) w.noteRecovery();
+    ASSERT_GT(d.silent_ms, last_silent);
+    last_silent = d.silent_ms;
+  }
+
+  EXPECT_GT(last_silent, 0x100000000ull);
+}
+
+TEST(RadioWatchdog, DisabledWatchdogAndNonRecvModeNeverTrip) {
+  RadioWatchdog w;
+  w.reset(0);
+  Radio r;
+  r.last_recv = 1000;
+  run(w, r, 1000);
+
+  for (uint32_t t = 2000; t <= 1000000; t += 10000) {
+    EXPECT_FALSE(run(w, r, t, /*in_recv_mode=*/true, /*watchdog_ms=*/0).recover);
+    EXPECT_FALSE(run(w, r, t, /*in_recv_mode=*/false).recover);
+  }
+}
+
+// Transmit-only traffic still counts as a live radio.
+TEST(RadioWatchdog, TransmitActivityRefreshesSilence) {
+  RadioWatchdog w;
+  w.reset(0);
+  Radio r;
+  r.last_recv = 1000;
+  run(w, r, 1000);
+
+  for (uint32_t t = 100000; t <= 900000; t += 100000) {
+    r.last_tx = t;
+    EXPECT_FALSE(run(w, r, t).recover) << "tripped at t=" << t;
+  }
+}
+
+// A blocked mesh loop (e.g. a 30 s NTP probe) collapses several events into one
+// observation. That may only shorten measured silence, never lengthen it.
+TEST(RadioWatchdog, LoopStallDoesNotManufactureSilence) {
+  RadioWatchdog w;
+  w.reset(0);
+  Radio r;
+  r.last_recv = 1000;
+  run(w, r, 1000);
+
+  r.last_recv = 20000;   // arrived during the stall
+  r.last_irq = 20000;
+  RadioWatchdogDecision d = run(w, r, 31000);
+
+  EXPECT_FALSE(d.recover);
+  EXPECT_EQ(0u, d.silent_ms);
+}
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}