refactor(sync): Replace unwrap with anyhow-based locking

src/baseliner.rs

@@ -44,8 +44,8 @@ impl Baseliner {
         loop {
             let time_data = self.stats_receiver.recv()?;
 
-            let mut owd_baseline_map = self.owd_baseline.lock().unwrap();
-            let mut owd_recent_map = self.owd_recent.lock().unwrap();
+            let mut owd_baseline_map = self.owd_baseline.lock_anyhow()?;
+            let mut owd_recent_map = self.owd_recent.lock_anyhow()?;
 
             let owd_baseline_new = ReflectorStats {
                 down_ewma: time_data.down_time,

src/pinger.rs

@@ -64,7 +64,7 @@ pub trait PingListener {
 
             let addr: IpAddr = sender.as_socket().unwrap().ip();
 
-            let reflectors = reflectors_lock.read().unwrap();
+            let reflectors = reflectors_lock.read_anyhow()?;
             if !reflectors.contains(&addr) {
                 continue;
             }
@@ -100,9 +100,17 @@ pub trait PingSender {
         let tick_duration_ms: u16 = (tick_interval * 1000.0) as u16;
 
         loop {
-            let reflectors_unlocked = reflectors_lock.read().unwrap();
+            // Clone the reflectors vec and drop the read lock immediately —
+            // holding it for the entire tick would starve the reflector selector's write lock.
+            let reflectors_unlocked = reflectors_lock.read_anyhow()?;
             let reflectors = reflectors_unlocked.clone();
             drop(reflectors_unlocked);
+
+            if reflectors.is_empty() {
+                thread::sleep(Duration::from_millis(tick_duration_ms as u64));
+                continue;
+            }
+
             let sleep_duration =
                 Duration::from_millis((tick_duration_ms / reflectors.len() as u16) as u64);
 

src/ratecontroller.rs

@@ -196,9 +196,9 @@ impl Ratecontroller {
         state_ul.deltas.clear();
 
         let now_t = Instant::now();
-        let owd_baseline = self.owd_baseline.lock().unwrap();
-        let owd_recent = self.owd_recent.lock().unwrap();
-        let reflectors = self.reflectors_lock.read().unwrap();
+        let owd_baseline = self.owd_baseline.lock_anyhow()?;
+        let owd_recent = self.owd_recent.lock_anyhow()?;
+        let reflectors = self.reflectors_lock.read_anyhow()?;
 
         for reflector in reflectors.iter() {
             // only consider this data if it's less than 2 * tick_duration seconds old

src/reflector_selector.rs

@@ -45,7 +45,7 @@ impl ReflectorSelector {
             }
 
             let mut next_peers: Vec<IpAddr> = Vec::new();
-            let mut reflectors_peers = self.reflector_peers_lock.write().unwrap();
+            let mut reflectors_peers = self.reflector_peers_lock.write_anyhow()?;
 
             // Include all current peers
             for reflector in reflectors_peers.iter() {
@@ -62,7 +62,8 @@ impl ReflectorSelector {
                 next_peers.push(*next_candidate);
             }
 
-            // Put all the pool members back into the peers for some re-baselining...
+            // Clone next_peers because we need it again after the baseline sleep
+            // to iterate over candidates for RTT measurement.
             *reflectors_peers = next_peers.clone();
 
             // Drop the MutexGuard explicitly, as Rust won't unlock the mutex by default
@@ -74,10 +75,10 @@ impl ReflectorSelector {
             sleep(baseline_sleep_time);
 
             // Re-acquire the lock when we wake up again
-            reflectors_peers = self.reflector_peers_lock.write().unwrap();
+            reflectors_peers = self.reflector_peers_lock.write_anyhow()?;
 
             let mut candidates = Vec::new();
-            let owd_recent = self.owd_recent.lock().unwrap();
+            let owd_recent = self.owd_recent.lock_anyhow()?;
 
             for peer in next_peers {
                 if owd_recent.contains_key(&peer) {