fix(client): bound select-phase watchdog stalls

Author: nightowlnerd

crates/slipstream-client/src/runtime.rs

@@ -77,6 +77,8 @@ const HEALTH_LOG_INTERVAL_US: u64 = 300_000_000;
 const WATCHDOG_STALE_SECS: u64 = 15;
 const WATCHDOG_CHECK_INTERVAL: Duration = Duration::from_secs(3);
 const WATCHDOG_ABORT_STRIKES: u32 = 3;
+const WATCHDOG_SELECT_RECONNECT_SECS: u64 = 30;
+const WATCHDOG_SELECT_ABORT_SECS: u64 = 180;
 const ACTIVE_PATH_LOSS_RECONNECT_STREAMS: usize = 32;
 
 /// Watchdog that runs on a separate OS thread (not tokio) to detect when the
@@ -85,11 +87,12 @@ const ACTIVE_PATH_LOSS_RECONNECT_STREAMS: usize = 32;
 /// the watchdog emits warnings and only aborts in non-select phases.
 ///
 /// `PHASE_SELECT` may legitimately appear stale under host scheduler jitter,
-/// so aborting there creates false outages.
+/// so we first request reconnect; abort is only a last-resort cap.
 struct Watchdog {
     heartbeat: Arc<AtomicU64>,
     phase: Arc<AtomicU32>,
     alive: Arc<AtomicBool>,
+    select_reconnect_requested: Arc<AtomicBool>,
     _handle: std::thread::JoinHandle<()>,
 }
 
@@ -125,9 +128,11 @@ impl Watchdog {
         let heartbeat = Arc::new(AtomicU64::new(0));
         let phase = Arc::new(AtomicU32::new(0));
         let alive = Arc::new(AtomicBool::new(true));
+        let select_reconnect_requested = Arc::new(AtomicBool::new(false));
         let hb = Arc::clone(&heartbeat);
         let ph = Arc::clone(&phase);
         let al = Arc::clone(&alive);
+        let reconnect_flag = Arc::clone(&select_reconnect_requested);
         let handle = std::thread::Builder::new()
             .name("watchdog".into())
             .spawn(move || {
@@ -173,8 +178,26 @@ impl Watchdog {
                         let stuck_phase = ph.load(Ordering::Relaxed);
                         let phase_name = phase_name(stuck_phase);
                         if stuck_phase == PHASE_SELECT {
+                            if stale_us >= WATCHDOG_SELECT_RECONNECT_SECS * 1_000_000 {
+                                let first_request = !reconnect_flag.swap(true, Ordering::Relaxed);
+                                if first_request {
+                                    eprintln!(
+                                        "WATCHDOG: select phase stale for {:.1}s; requesting reconnect",
+                                        stale_us as f64 / 1_000_000.0,
+                                    );
+                                }
+                            }
+                            if stale_us >= WATCHDOG_SELECT_ABORT_SECS * 1_000_000 {
+                                eprintln!(
+                                    "WATCHDOG: select phase stalled for {:.1}s (phase {} / {}), aborting as last resort",
+                                    stale_us as f64 / 1_000_000.0,
+                                    stuck_phase,
+                                    phase_name,
+                                );
+                                std::process::abort();
+                            }
                             eprintln!(
-                                "WATCHDOG: stale heartbeat {:.1}s at phase {} ({}), strikes={}, waiting (select phase never aborts)",
+                                "WATCHDOG: stale heartbeat {:.1}s at phase {} ({}), strikes={}, waiting (select phase uses reconnect-first policy)",
                                 stale_us as f64 / 1_000_000.0,
                                 stuck_phase,
                                 phase_name,
@@ -209,6 +232,7 @@ impl Watchdog {
             heartbeat,
             phase,
             alive,
+            select_reconnect_requested,
             _handle: handle,
         }
     }
@@ -221,6 +245,11 @@ impl Watchdog {
     fn set_phase(&self, p: u32) {
         self.phase.store(p, Ordering::Relaxed);
     }
+
+    fn take_select_reconnect_requested(&self) -> bool {
+        self.select_reconnect_requested
+            .swap(false, Ordering::Relaxed)
+    }
 }
 
 impl Drop for Watchdog {
@@ -467,6 +496,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result<i32, ClientError> {
             watchdog.pet();
             watchdog.set_phase(PHASE_DRAIN_COMMANDS);
             let current_time = unsafe { picoquic_current_time() };
+            if watchdog.take_select_reconnect_requested() {
+                warn!(
+                    "watchdog requested reconnect after select-phase stall; recycling connection"
+                );
+                break;
+            }
             // Only process application commands after the QUIC handshake
             // completes.  During reconnect the acceptor may queue NewStream
             // commands while the connection is still in initial state;
@@ -694,6 +729,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result<i32, ClientError> {
                 );
             }
             watchdog.pet();
+            if watchdog.take_select_reconnect_requested() {
+                warn!(
+                    "watchdog requested reconnect after select-phase stall; recycling connection"
+                );
+                break;
+            }
 
             watchdog.set_phase(PHASE_POST_DRAIN);
             if unsafe { (*state_ptr).is_ready() } {