fix(probe): preserve measurement through wrap and network outages

src/main.rs

@@ -202,19 +202,22 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
     ];
 
     let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize;
-    let pool = if reselection_enabled {
+    let (peers, pool) = if reselection_enabled {
-        configured
+        (default_reflectors.to_vec(), configured)
+    } else if configured.is_empty() {
+        (default_reflectors.to_vec(), Vec::new())
     } else {
-        Vec::new()
+        (configured, Vec::new())
     };
+    let active_count = peers
+        .len()
+        .max(settings.advanced_settings.num_reflectors as usize);
 
     Ok(ReflectorSetup {
-        peers: Arc::new(RwLock::new(default_reflectors.to_vec())),
+        peers: Arc::new(RwLock::new(peers)),
         pool,
         reselection_enabled,
-        active_count: default_reflectors
+        active_count,
-            .len()
-            .max(settings.advanced_settings.num_reflectors as usize),
     })
 }
 
@@ -476,3 +479,41 @@ fn main() -> anyhow::Result<()> {
 
     run(&settings)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::settings::{
+        AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings,
+    };
+
+    #[test]
+    fn small_configured_reflector_list_is_used() {
+        let path = std::env::temp_dir().join(format!("sqma-reflectors-{}.csv", process::id()));
+        std::fs::write(
+            &path,
+            "reflector_ip,ip_version,description\n192.0.2.1,4,test\n192.0.2.2,4,test\n192.0.2.3,4,test\n192.0.2.4,4,test\n192.0.2.5,4,test\n",
+        ).unwrap();
+        let settings = Settings {
+            network: NetworkSettings {
+                download_interface: "ifb0".into(),
+                upload_interface: "eth0".into(),
+                download_base_kbits: 10_000.0,
+                download_min_percent: 20.0,
+                upload_base_kbits: 5_000.0,
+                upload_min_percent: 20.0,
+            },
+            output: OutputSettings::default(),
+            observability: ObservabilitySettings::default(),
+            advanced_settings: AdvancedSettings {
+                reflector_list_file: path.to_string_lossy().into_owned(),
+                ..AdvancedSettings::default()
+            },
+        };
+        let setup = setup_reflectors(&settings).unwrap();
+        let configured = settings.load_reflectors().unwrap();
+        std::fs::remove_file(path).unwrap();
+        assert!(!setup.reselection_enabled);
+        assert_eq!(*setup.peers.read().unwrap(), configured);
+    }
+}

src/pinger.rs

@@ -74,6 +74,8 @@ pub trait PingListener {
         ping_metrics: MetricsSender,
     ) -> anyhow::Result<()> {
         let socket = &mut open_socket(type_)?;
+        socket.set_timeout(Some(Duration::from_millis(500)));
+        let mut last_receive_warning = None;
 
         loop {
             if SHUTDOWN.load(Ordering::Relaxed) {
@@ -82,7 +84,24 @@ pub trait PingListener {
             }
             let (pkt, sender) = match socket.rcv_from() {
                 Ok(val) => val,
-                Err(_) => continue,
+                Err(error) => {
+                    if matches!(
+                        error.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+                    ) {
+                        continue;
+                    }
+                    if error.raw_os_error().is_some() {
+                        if last_receive_warning
+                            .is_none_or(|at: Instant| at.elapsed() >= Duration::from_secs(30))
+                        {
+                            warn!("ICMP receive failed: {error}");
+                            last_receive_warning = Some(Instant::now());
+                        }
+                        thread::sleep(Duration::from_millis(200));
+                    }
+                    continue;
+                }
             };
 
             let addr: IpAddr = sender.as_socket().unwrap().ip();
@@ -158,7 +177,8 @@ pub trait PingSender {
         let mut socket = open_socket(type_)?;
 
         let mut seq: u16 = 0;
-        let tick_duration_ms: u16 = (tick_interval * 1000.0) as u16;
+        let tick_duration = Duration::from_secs_f64(tick_interval);
+        let mut last_send_warning = None;
 
         loop {
             if SHUTDOWN.load(Ordering::Relaxed) {
@@ -172,7 +192,7 @@ pub trait PingSender {
             drop(reflectors_unlocked);
 
             if reflectors.is_empty() {
-                thread::sleep(Duration::from_millis(tick_duration_ms as u64));
+                thread::sleep(tick_duration);
                 continue;
             }
 
@@ -180,13 +200,12 @@ pub trait PingSender {
                 .lock_anyhow()?
                 .retain(|_, probe| probe.sent_at.elapsed() < INFLIGHT_PROBE_TTL);
 
-            let sleep_duration =
+            let sleep_duration = Duration::from_secs_f64(tick_interval / reflectors.len() as f64);
-                Duration::from_millis((tick_duration_ms / reflectors.len() as u16) as u64);
 
             for reflector in reflectors.iter() {
                 let addr: Ipv4Addr = match reflector {
                     IpAddr::V4(ipv4) => *ipv4,
-                    IpAddr::V6(_) => unimplemented!(),
+                    IpAddr::V6(_) => continue,
                 };
 
                 let (packet, originate_timestamp) = self.craft_packet(id, seq);
@@ -201,7 +220,25 @@ pub trait PingSender {
                 );
                 if let Err(e) = socket.send_to(addr, packet) {
                     inflight.lock_anyhow()?.remove(&key);
-                    return Err(e.into());
+                    if matches!(
+                        e.raw_os_error(),
+                        Some(
+                            libc::ENETUNREACH
+                                | libc::EHOSTUNREACH
+                                | libc::ENETDOWN
+                                | libc::ENOBUFS
+                                | libc::EADDRNOTAVAIL
+                        )
+                    ) {
+                        if last_send_warning
+                            .is_none_or(|at: Instant| at.elapsed() >= Duration::from_secs(30))
+                        {
+                            warn!("ICMP send failed during network outage: {e}");
+                            last_send_warning = Some(Instant::now());
+                        }
+                    } else {
+                        return Err(e.into());
+                    }
                 }
                 thread::sleep(sleep_duration);
             }

src/pinger_icmp_ts.rs

@@ -19,6 +19,20 @@ pub struct PingerICMPTimestampListener {}
 
 pub struct PingerICMPTimestampSender {}
 
+const MS_PER_DAY: i64 = 86_400_000;
+
+// RFC 792 timestamps wrap at midnight UTC; real one-way delays are below half a day.
+fn timestamp_delta(later: i64, earlier: i64) -> i64 {
+    let delta = later - earlier;
+    if delta < -MS_PER_DAY / 2 {
+        delta + MS_PER_DAY
+    } else if delta > MS_PER_DAY / 2 {
+        delta - MS_PER_DAY
+    } else {
+        delta
+    }
+}
+
 impl PingListener for PingerICMPTimestampListener {
     // Result: RTT, down time, up time
     fn parse_packet(
@@ -46,12 +60,19 @@ impl PingListener for PingerICMPTimestampListener {
                         });
                     }
 
+                    // The high bit marks a timestamp that is not milliseconds since midnight UTC.
+                    if receive & 0x8000_0000 != 0 || transmit & 0x8000_0000 != 0 {
+                        return Err(PingError::InvalidPacket(
+                            "non-standard RFC 792 timestamp".into(),
+                        ));
+                    }
+
                     let time_now = Time::new(ClockId::Realtime);
                     let time_since_midnight = time_now.get_time_since_midnight();
 
-                    let rtt = (time_since_midnight - originate as i64) as f64;
+                    let rtt = timestamp_delta(time_since_midnight, originate as i64) as f64;
-                    let dl_time = (time_since_midnight - transmit as i64) as f64;
+                    let dl_time = timestamp_delta(time_since_midnight, transmit as i64) as f64;
-                    let ul_time = (receive as i64 - originate as i64) as f64;
+                    let ul_time = timestamp_delta(receive as i64, originate as i64) as f64;
 
                     Ok(PingReply {
                         reflector,
@@ -88,3 +109,27 @@ impl PingSender for PingerICMPTimestampSender {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use icmp_socket2::packet::WithTimestampReply;
+
+    #[test]
+    fn timestamps_wrap_at_midnight_in_both_directions() {
+        assert_eq!(timestamp_delta(20, 86_399_900), 120);
+        assert_eq!(timestamp_delta(86_399_900, 20), -120);
+    }
+
+    #[test]
+    fn non_standard_rfc_792_timestamps_are_rejected() {
+        let packet = Icmpv4Packet::with_timestamp_reply(7, 1, 10, 0x8000_0001, 20).unwrap();
+        let result = PingerICMPTimestampListener {}.parse_packet(
+            7,
+            "192.0.2.1".parse().unwrap(),
+            MeasurementType::IcmpTimestamps,
+            packet,
+        );
+        assert!(matches!(result, Err(PingError::InvalidPacket(_))));
+    }
+}

src/reflector_selector.rs

@@ -166,6 +166,7 @@ impl ReflectorSelector {
 
             let mut next_peers: Vec<IpAddr> = Vec::new();
             let mut reflectors_peers = self.reflector_peers_lock.write_anyhow()?;
+            let previous_peers = reflectors_peers.clone();
 
             // Include all current peers
             for reflector in reflectors_peers.iter() {
@@ -251,6 +252,11 @@ impl ReflectorSelector {
                 });
             }
 
+            if new_peers.is_empty() {
+                log::warn!("Reselection found no responsive peers; retaining the current peers");
+                *reflectors_peers = previous_peers;
+                continue;
+            }
             *reflectors_peers = new_peers;
         }
     }

src/settings/validation.rs

@@ -282,11 +282,17 @@ impl AdvancedSettings {
             self.min_change_interval,
         );
 
-        validate_greater_than_zero_f64(
+        let tick_valid = validate_greater_than_zero_f64(
             report,
             "advanced_settings.tick_interval",
             self.tick_interval,
         );
+        if tick_valid && self.tick_interval < 0.01 {
+            report.error(
+                "advanced_settings.tick_interval",
+                "must be at least 0.01 seconds",
+            );
+        }
 
         if self.num_reflectors < 5 {
             report.error(