Preserve stalled USB recovery after a cancelled watchdog reboot

Restart the retry interval without treating queue cleanup as completed USB transmission. Cover cancellation journals, fresh recovery deadlines, real progress and rollover, with historical policy and runtime negative controls.

src/helpers/UsbLoggingWatchdogPolicy.h

@@ -111,7 +111,8 @@ class UsbLoggingWatchdogPolicy {
   void restartInterval() {
     _stage = 0;
     _fault_since = _now;
-    _need_progress = false;
+    // Restarting after a veto is not evidence of transport completion.
+    // Only update() clears a latched stall after real TX progress.
   }
   uint8_t backoffStep() const { return _step; }
   uint8_t stage() const { return _stage; }

test/fixtures/usb_logging_watchdog/test_usb_logging_watchdog_runtime.cpp

@@ -528,6 +528,124 @@ static void check_real_progress_not_probe_or_purge_and_master_reenable() {
   assert(recovery_calls.size() == attempts + 1);
 }
 
+static void check_cancelled_stall_requires_progress_and_restarts_full_recovery() {
+  unsigned scenarios = 0;
+#if defined(ENABLE_OTA)
+  const unsigned veto_kinds = 2;
+#else
+  const unsigned veto_kinds = 1;
+#endif
+  for (uint32_t origin : {0U, UINT32_MAX - Policy::BASE_MS / 2}) {
+    for (unsigned veto = 0; veto < veto_kinds; ++veto) {
+      for (bool completes_after_cancel : {false, true}) {
+        Fixture fixture(1, 0, 0, 0, origin);
+#if defined(ENABLE_OTA)
+        mesh::ota::OtaContext ota;
+        if (veto == 1) mesh::ota::test_ota_context = &ota;
+#endif
+        observed = {true, true, true, true, 7};
+        assert(!fixture.run(0));
+        assert(!fixture.run(Policy::STALL_OBSERVE_MS));
+        assert(mesh::usbLoggingStatus().stalled);
+        assert(!fixture.run(Policy::EARLY_RECOVERY_MS));
+        assert(!fixture.run(Policy::EARLY_RECOVERY_MS + Policy::RECOVERY_GRACE_MS));
+        assert((recovery_calls == std::vector<uint8_t>{1, 2}));
+        // Backend cleanup can remove demand without completing a USB TX.
+        observed.pending = false;
+        assert(!fixture.run(Policy::EARLY_RECOVERY_MS + Policy::RECOVERY_GRACE_MS + 1));
+        assert(mesh::usbLoggingStatus().stalled && mesh::usbLoggingStatus().stage == 2);
+        fixture.fs.after_final_rename = [&] {
+#if defined(ENABLE_OTA)
+          if (veto == 1) ota.apply_pending = true;
+          else
+#endif
+          fixture.safe = false;
+        };
+        const uint32_t cancelled_at = Policy::BASE_MS;
+        assert(!fixture.run(cancelled_at));
+        auto status = mesh::usbLoggingStatus();
+        assert(status.persistence_ready && status.backoff_step == 1 && status.stage == 0);
+        assert(status.reboot_count == 1 && status.recovery_count == 2);
+        assert(status.stalled && !status.logger_active);
+        assert(status.last_event.action == Event::REBOOT_CANCELLED
+            && status.last_event.reasons == Event::TX_STALLED && !status.last_event.persisted);
+        auto request = status.last_event;
+        request.action = Event::REBOOT_REQUESTED;
+        --request.sequence;
+        request.persisted = true;
+        assert(fixture.fs.files.at("/usb_wdg") == image(1, 1, 2, 1, &request));
+        const auto writes = fixture.fs.write_opens;
+        assert(!fixture.run(cancelled_at + 1));
+        assert(fixture.fs.write_opens == writes); // Still unsafe: cancellation remains pending.
+        assert(!mesh::usbLoggingStatus().last_event.persisted);
+        fixture.fs.after_final_rename = {};
+        fixture.safe = true;
+#if defined(ENABLE_OTA)
+        ota.apply_pending = false;
+#endif
+        assert(!fixture.run(cancelled_at + 2));
+        assert(mesh::usbLoggingStatus().last_event.persisted);
+        assert(fixture.fs.files.at("/usb_wdg") == current_image(1, 1, 2, 1));
+        assert(fixture.fs.write_opens == writes + 1);
+        assert(!fixture.run(cancelled_at + 1000));
+        status = mesh::usbLoggingStatus();
+        assert(status.stalled && status.stage == 0 && status.inactive_seconds == 1
+            && "cancelled purge must remain faulty without TX completion");
+        assert(status.backoff_step == 1 && !status.logger_active);
+        assert(!observed.pending && observed.tx_progress == 7);
+
+        if (completes_after_cancel) {
+          // Unlike a purge, a downstream completion is genuine health evidence.
+          const uint32_t healthy_at = cancelled_at + 1001;
+          ++observed.tx_progress;
+          assert(!fixture.run(healthy_at));
+          status = mesh::usbLoggingStatus();
+          assert(!status.stalled && status.stage == 0 && status.inactive_seconds == 0);
+          assert(status.logger_active && status.backoff_step == 1);
+          assert(!fixture.run(healthy_at + Policy::HEALTHY_RESET_MS - 1));
+          assert(mesh::usbLoggingStatus().backoff_step == 1);
+          assert(!fixture.run(healthy_at + Policy::HEALTHY_RESET_MS));
+          assert(mesh::usbLoggingStatus().backoff_step == 0);
+          assert((recovery_calls == std::vector<uint8_t>{1, 2}));
+          assert(fixture.fs.files.at("/usb_wdg") == current_image(1, 0, 2, 1));
+        } else {
+          assert(!fixture.run(cancelled_at + Policy::EARLY_RECOVERY_MS - 1));
+          assert((recovery_calls == std::vector<uint8_t>{1, 2}));
+          assert(!fixture.run(cancelled_at + Policy::EARLY_RECOVERY_MS));
+          assert((recovery_calls == std::vector<uint8_t>{1, 2, 1}));
+          assert(mesh::usbLoggingStatus().stage == 1);
+          assert(!fixture.run(cancelled_at + Policy::EARLY_RECOVERY_MS + Policy::RECOVERY_GRACE_MS - 1));
+          assert(recovery_calls.size() == 3);
+          assert(!fixture.run(cancelled_at + Policy::EARLY_RECOVERY_MS + Policy::RECOVERY_GRACE_MS));
+          assert((recovery_calls == std::vector<uint8_t>{1, 2, 1, 2}));
+          assert(mesh::usbLoggingStatus().stage == 2);
+          assert(!fixture.run(cancelled_at + Policy::HEALTHY_RESET_MS));
+          status = mesh::usbLoggingStatus();
+          assert(status.stalled && !status.logger_active && status.backoff_step == 1
+              && status.recovery_count == 4 && status.reboot_count == 1
+              && "purged stalled TX must not reset its durable retry tier as healthy");
+          assert(!observed.pending && observed.tx_progress == 7);
+          const uint32_t retry_at = cancelled_at + Policy::intervalMs(1);
+          assert(!fixture.run(retry_at - 1));
+          assert(fixture.run(retry_at));
+          status = mesh::usbLoggingStatus();
+          assert(status.stage == 3 && status.backoff_step == 2
+              && status.recovery_count == 4 && status.reboot_count == 2);
+          assert(status.last_event.action == Event::REBOOT_REQUESTED
+              && status.last_event.reasons == Event::TX_STALLED && status.last_event.persisted);
+          assert(fixture.fs.files.at("/usb_wdg") == current_image(1, 2, 4, 2));
+        }
+#if defined(ENABLE_OTA)
+        mesh::ota::test_ota_context = nullptr;
+#endif
+        ++scenarios;
+      }
+    }
+  }
+  assert(scenarios == 4 * veto_kinds);
+  std::printf("%u cancelled-stall recovery timelines passed\n", scenarios);
+}
+
 static void check_healthy_reset_is_continuous_safe_and_saved_once() {
   Fixture fixture(1, 5);
   observed = {true, true, true, false, 0};
@@ -1184,6 +1302,7 @@ int main() {
   check_ownership_deferred_and_no_work_when_inactive();
   check_reboot_commit_failure_and_postcommit_veto();
   check_real_progress_not_probe_or_purge_and_master_reenable();
+  check_cancelled_stall_requires_progress_and_restarts_full_recovery();
   check_healthy_reset_is_continuous_safe_and_saved_once();
   check_auto_promotion_continuity_and_failed_write();
   check_auto_seen_reader_usb_only_and_probe_is_not_health();

test/test_usb_logging_watchdog.py

@@ -326,6 +326,65 @@ static void check_pre_recovery_purge_cannot_erase_a_stall() {
   assert(!policy.stalled() && policy.stage() == 0);
 }
 
+static void check_restarted_interval_preserves_completion_requirement() {
+  for (uint8_t step : {uint8_t(0), uint8_t(1), uint8_t(8)})
+  for (uint32_t origin : {0U, UINT32_MAX - 360100U})
+  for (bool purged : {false, true}) {
+    Policy policy;
+    Clock clock(origin);
+    auto observation = pending();
+    observation.tx_progress = UINT32_MAX;
+    policy.begin(clock.now(), step);
+    assert(policy.update(clock.now(), true, observation) == Action::None);
+    clock.advance(Policy::EARLY_RECOVERY_MS);
+    assert(policy.update(clock.now(), true, observation) == Action::SoftRecovery);
+    policy.actionAttempted(Action::SoftRecovery);
+    clock.advance(Policy::RECOVERY_GRACE_MS);
+    assert(policy.update(clock.now(), true, observation) == Action::Reenumerate);
+    policy.actionAttempted(Action::Reenumerate);
+    observation.pending = !purged;
+    clock.advance(Policy::intervalMs(step) - clock.elapsed);
+    assert(policy.update(clock.now(), true, observation) == Action::Reboot);
+    const uint8_t next = Policy::nextStep(step);
+    policy.backoffCommitted(next);
+    policy.restartInterval(); // Durable request was cancelled by its owner.
+    assert(policy.stalled() && "restart must retain unresolved USB stall");
+    assert(policy.stage() == 0 && policy.inactiveSeconds() == 0);
+    const uint64_t restart = clock.elapsed;
+    clock.advance(Policy::EARLY_RECOVERY_MS - 1);
+    assert(policy.update(clock.now(), true, observation) == Action::None);
+    assert(policy.stalled() && policy.inactiveSeconds() == 299);
+    clock.advance(1);
+    assert(policy.update(clock.now(), true, observation) == Action::SoftRecovery);
+    policy.actionAttempted(Action::SoftRecovery);
+    clock.advance(Policy::RECOVERY_GRACE_MS - 1);
+    assert(policy.update(clock.now(), true, observation) == Action::None);
+    clock.advance(1);
+    assert(policy.update(clock.now(), true, observation) == Action::Reenumerate);
+    policy.actionAttempted(Action::Reenumerate);
+    clock.advance(Policy::HEALTHY_RESET_MS - (clock.elapsed - restart));
+    assert(policy.update(clock.now(), true, observation) == Action::None);
+    assert(policy.stalled() && policy.backoffStep() == next && policy.stage() == 2);
+    clock.advance(Policy::intervalMs(next) - (clock.elapsed - restart) - 1);
+    assert(policy.update(clock.now(), true, observation) == Action::None);
+    clock.advance(1);
+    assert(policy.update(clock.now(), true, observation) == Action::Reboot);
+    assert(policy.inactiveSeconds() == Policy::intervalMs(next) / 1000);
+
+    // Real completion, including counter wrap, still restores health and can
+    // reset the durable tier after the normal continuous healthy interval.
+    ++observation.tx_progress;
+    observation.pending = false;
+    assert(policy.update(clock.now(), true, observation) == Action::None);
+    policy.restartInterval();
+    assert(!policy.stalled() && policy.stage() == 0 && policy.inactiveSeconds() == 0);
+    clock.advance(Policy::HEALTHY_RESET_MS - 1);
+    assert(policy.update(clock.now(), true, observation) == Action::None);
+    clock.advance(1);
+    assert(policy.update(clock.now(), true, observation) == Action::ResetBackoff);
+  }
+}
+
 static void check_auto_qualification_is_continuous() {
   AutoArm policy;
   policy.begin(100);
@@ -464,6 +523,7 @@ int main() {
   check_millis_wrap_long_run_and_old_samples();
   check_pending_resets_after_real_progress();
   check_pre_recovery_purge_cannot_erase_a_stall();
+  check_restarted_interval_preserves_completion_requirement();
   check_auto_qualification_is_continuous();
   check_auto_clock_wrap_and_saturation();
   check_auto_stall_observation_does_not_arm_recovery();
@@ -496,6 +556,36 @@ class UsbLoggingWatchdogTest(unittest.TestCase):
             self.assertEqual(tested.returncode, 0, tested.stderr)
             self.assertIn('USB watchdog policy timelines passed', tested.stdout)
 
+    def test_old_restart_clear_has_intended_semantic_failure(self):
+        compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++')
+        if compiler is None:
+            self.skipTest('a host C++17 compiler is required')
+        header = (ROOT / 'src/helpers/UsbLoggingWatchdogPolicy.h').read_text(encoding='ascii')
+        restart = '    _stage = 0;\n    _fault_since = _now;\n'
+        self.assertEqual(header.count(restart), 1)
+        legacy = header.replace(restart, restart + '    _need_progress = false;\n', 1)
+        sanitizer_flags = [] if os.name == 'nt' else [
+            '-fsanitize=address,undefined', '-fno-sanitize-recover=all',
+            '-fno-pie', '-no-pie']
+        with tempfile.TemporaryDirectory(prefix='usb-watchdog-restart-negative-') as directory:
+            work = Path(directory)
+            (work / 'helpers').mkdir()
+            (work / 'helpers/UsbLoggingWatchdogPolicy.h').write_text(legacy, encoding='ascii')
+            cpp = work / 'test.cpp'
+            binary = work / ('test.exe' if os.name == 'nt' else 'test')
+            cpp.write_text(HARNESS, encoding='ascii')
+            built = subprocess.run([
+                compiler, '-std=c++17', '-Wall', '-Wextra', '-Werror',
+                *sanitizer_flags, '-I' + str(work), '-I' + str(ROOT / 'src'),
+                '-I' + str(ROOT / 'src/helpers'), str(cpp), '-o', str(binary)],
+                capture_output=True, text=True, timeout=60)
+            self.assertEqual(built.returncode, 0, built.stderr)
+            tested = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+            self.assertNotEqual(tested.returncode, 0)
+            self.assertIn('restart must retain unresolved USB stall', tested.stderr)
+            self.assertNotIn('runtime error:', tested.stderr)
+            self.assertNotIn('AddressSanitizer:', tested.stderr)
+
 
 if __name__ == '__main__':
     unittest.main()

test/test_usb_logging_watchdog_runtime.py

@@ -128,6 +128,12 @@ int main() {
             self.assertEqual(tested.returncode, 0, tested.stderr)
 
     def test_production_runtime_and_persistence(self):
+        self.run_runtime()
+
+    def test_old_cancelled_stall_clear_has_intended_semantic_failure(self):
+        self.run_runtime(legacy=True)
+
+    def run_runtime(self, legacy=False):
         compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++')
         if compiler is None:
             self.skipTest('a host C++17 compiler is required')
@@ -140,26 +146,46 @@ int main() {
                 with self.subTest(ota_enabled=ota):
                     binary = Path(directory) / ('ota' + str(ota) + '.exe')
                     compile_source = source
-                    if ota:
-                        # Copy this translation unit byte-for-byte. Its quoted
-                        # OTA include can then use an isolated dependency fake,
-                        # while every tested production statement is unchanged.
-                        compile_source = Path(directory) / 'UsbLoggingWatchdog.cpp'
+                    if ota or legacy:
+                        # Copy this translation unit byte-for-byte so quoted
+                        # includes can use the isolated OTA fake or negative
+                        # policy header. Production statements stay unchanged.
+                        shadow = Path(directory) / 'helpers'
+                        shadow.mkdir(exist_ok=True)
+                        compile_source = shadow / 'UsbLoggingWatchdog.cpp'
                         compile_source.write_bytes(source.read_bytes())
+                        if legacy:
+                            header = (ROOT / 'src/helpers/UsbLoggingWatchdogPolicy.h').read_text(
+                                encoding='ascii')
+                            restart = '    _stage = 0;\n    _fault_since = _now;\n'
+                            self.assertEqual(header.count(restart), 1)
+                            (shadow / 'UsbLoggingWatchdogPolicy.h').write_text(
+                                header.replace(restart, restart + '    _need_progress = false;\n', 1),
+                                encoding='ascii')
                     built = subprocess.run([
                         compiler, '-std=c++17', '-Wall', '-Wextra', '-Werror', '-pthread',
                         *sanitizer_flags, '-DARDUINO', '-DESP32', '-DESP32_PLATFORM',
                         '-DARDUINO_USB_MODE=0', '-DARDUINO_USB_CDC_ON_BOOT=1',
                         *(['-DENABLE_OTA'] if ota else []),
+                        '-I' + directory,
                         '-I' + str(FIXTURE / 'mocks'),
                         '-I' + str(ROOT / 'src' / 'helpers'), '-I' + str(ROOT / 'src'),
                         str(compile_source), str(FIXTURE / 'test_usb_logging_watchdog_runtime.cpp'),
                         '-o', str(binary)], capture_output=True, text=True, timeout=60)
                     self.assertEqual(built.returncode, 0, built.stderr)
                     tested = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
-                    self.assertEqual(tested.returncode, 0, tested.stderr)
-                    self.assertIn('USB watchdog runtime and persistence passed', tested.stdout)
-                    self.assertIn('14 USB watchdog setter durability scenarios passed', tested.stdout)
+                    if legacy:
+                        self.assertNotEqual(tested.returncode, 0)
+                        self.assertIn('cancelled purge must remain faulty without TX completion',
+                                      tested.stderr)
+                        self.assertNotIn('runtime error:', tested.stderr)
+                        self.assertNotIn('AddressSanitizer:', tested.stderr)
+                    else:
+                        self.assertEqual(tested.returncode, 0, tested.stderr)
+                        self.assertIn('USB watchdog runtime and persistence passed', tested.stdout)
+                        self.assertIn('14 USB watchdog setter durability scenarios passed', tested.stdout)
+                        self.assertIn(f'{8 if ota else 4} cancelled-stall recovery timelines passed',
+                                      tested.stdout)
 
     def test_unsupported_transport_has_no_recovery_or_storage_work(self):
         compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++')