fix(reflectors): retain current peers when no candidates respond

Restore the pre-probe selection when evaluation finds no usable candidates. Preserve master’s existing selection of smaller candidate sets.

src/ratecontroller.rs

@@ -491,12 +491,18 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         self.state_dl.deltas = down_deltas;
         self.state_ul.deltas = up_deltas;
 
-        if self.state_dl.deltas.len() < 5 || self.state_ul.deltas.len() < 5 {
+        let minimum =
+            usize::from(self.settings.advanced_settings.num_reflectors).min(reflectors.len());
+        if reflectors.is_empty()
+            || self.state_dl.deltas.len() < minimum
+            || self.state_ul.deltas.len() < minimum
+        {
             // trigger reselection
             warn!(
-                "Not enough delta values (D: {}, U: {}, need 5), triggering reselection",
+                "Not enough delta values (D: {}, U: {}, need {}), triggering reselection",
                 self.state_dl.deltas.len(),
-                self.state_ul.deltas.len()
+                self.state_ul.deltas.len(),
+                minimum
             );
             let _ = self.reselect_trigger.try_send(true);
         }

src/reflector_selector.rs

@@ -167,6 +167,7 @@ impl ReflectorSelector {
 
             let mut next_peers: Vec<IpAddr> = Vec::new();
             let mut reflectors_peers = self.reflector_peers_lock.write_anyhow()?;
+            let current_peers = reflectors_peers.clone();
 
             // Include all current peers
             for reflector in reflectors_peers.iter() {
@@ -236,6 +237,13 @@ impl ReflectorSelector {
                 candidates.swap(i, j);
             }
 
+            if candidates.is_empty() {
+                info!("No usable reflectors, retaining current set");
+                self.reflector_tx.send(current_peers.clone())?;
+                *reflectors_peers = current_peers;
+                continue;
+            }
+
             if (candidates.len() as u8) < num_reflectors {
                 num_reflectors = candidates.len() as u8;
             }

src/time.rs

@@ -14,16 +14,11 @@ pub trait Clock {
 }
 
 pub enum Realtime {}
-pub enum Monotonic {}
 
 impl Clock for Realtime {
     const ID: ClockId = ClockId::Realtime;
 }
 
-impl Clock for Monotonic {
-    const ID: ClockId = ClockId::Monotonic;
-}
-
 pub struct ClockSample<C> {
     time_s: u64,
     time_ns: u64,
@@ -56,10 +51,6 @@ impl<C: Clock> ClockSample<C> {
     pub fn as_secs_f64(&self) -> f64 {
         self.time_s as f64 + self.time_ns as f64 / 1_000_000_000.0
     }
-
-    pub fn to_milliseconds(&self) -> u64 {
-        (self.time_s * 1000) + (self.time_ns / 1000000)
-    }
 }
 
 impl ClockSample<Realtime> {

src/util.rs

@@ -3,20 +3,10 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use anyhow::anyhow;
-use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
+use std::sync::{Arc, RwLock, RwLockReadGuard, RwLockWriteGuard};
 
 pub type ArcRwLock<T> = Arc<RwLock<T>>;
 
-pub trait MutexExt<T> {
-    fn lock_anyhow(&self) -> anyhow::Result<MutexGuard<'_, T>>;
-}
-
-impl<T> MutexExt<T> for Mutex<T> {
-    fn lock_anyhow(&self) -> anyhow::Result<MutexGuard<'_, T>> {
-        self.lock().map_err(|e| anyhow!("mutex poisoned: {e}"))
-    }
-}
-
 pub trait RwLockExt<T> {
     fn read_anyhow(&self) -> anyhow::Result<RwLockReadGuard<'_, T>>;
     fn write_anyhow(&self) -> anyhow::Result<RwLockWriteGuard<'_, T>>;