diff --git a/src/main.rs b/src/main.rs index ad6c000..9cb7758 100644 --- a/src/main.rs +++ b/src/main.rs @@ -202,19 +202,22 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result { ]; let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize; - let pool = if reselection_enabled { - configured + let (peers, pool) = if reselection_enabled { + (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 - .len() - .max(settings.advanced_settings.num_reflectors as usize), + active_count, }) } @@ -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); + } +} diff --git a/src/pinger.rs b/src/pinger.rs index 54e83a6..2436c18 100644 --- a/src/pinger.rs +++ b/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 = - Duration::from_millis((tick_duration_ms / reflectors.len() as u16) as u64); + let sleep_duration = Duration::from_secs_f64(tick_interval / reflectors.len() as f64); 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); } diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs index c34cbf1..e61b1a0 100644 --- a/src/pinger_icmp_ts.rs +++ b/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 dl_time = (time_since_midnight - transmit as i64) as f64; - let ul_time = (receive as i64 - originate as i64) as f64; + let rtt = timestamp_delta(time_since_midnight, originate as i64) as f64; + let dl_time = timestamp_delta(time_since_midnight, transmit 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(_)))); + } +} diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index ab78889..5b7a9ce 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -166,6 +166,7 @@ impl ReflectorSelector { let mut next_peers: Vec = 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; } } diff --git a/src/settings/validation.rs b/src/settings/validation.rs index 231c85d..c84d317 100644 --- a/src/settings/validation.rs +++ b/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(