feat(measurement): integrate IRTT with full configuration and per-stream tracking

Extend reflector configuration to support IRTT-specific parameters (ports, HMAC) and track per-stream measurement state, enabling dynamic selection and authentication for IRTT probes.

src/baseliner.rs

@@ -6,13 +6,12 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use crate::SHUTDOWN;
-use crate::measurement::model::{MeasurementEvent, MeasurementObservation};
+use crate::measurement::model::{MeasurementEvent, MeasurementObservation, MeasurementStream};
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
 use flume::{Receiver, Sender};
 use log::{debug, info};
 use std::collections::HashMap;
-use std::net::IpAddr;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 use std::time::Instant;
@@ -72,7 +71,7 @@ impl ReflectorState {
 #[derive(Clone, Debug)]
 pub struct ControlSnapshot {
     pub generated_at: Instant,
-    pub reflectors: HashMap<IpAddr, ReflectorState>,
+    pub reflectors: HashMap<MeasurementStream, ReflectorState>,
 }
 
 pub struct Baseliner {
@@ -121,7 +120,7 @@ fn update_reflector_state(
 }
 
 fn process_reply(
-    reflectors: &mut HashMap<IpAddr, ReflectorState>,
+    reflectors: &mut HashMap<MeasurementStream, ReflectorState>,
     reply: &MeasurementObservation,
     start_time: Instant,
     slow_factor: f64,
@@ -129,7 +128,7 @@ fn process_reply(
 ) -> (ReflectorState, bool) {
     let sample = EwmaStats::from_reply(reply);
     let state = reflectors
-        .entry(reply.stream.peer)
+        .entry(reply.stream)
         .or_insert_with(|| ReflectorState::new(sample, reply.observed_at));
     let anomaly = update_reflector_state(
         state,
@@ -144,7 +143,7 @@ fn process_reply(
 }
 
 fn control_snapshot(
-    reflectors: &HashMap<IpAddr, ReflectorState>,
+    reflectors: &HashMap<MeasurementStream, ReflectorState>,
     generated_at: Instant,
 ) -> ControlSnapshot {
     ControlSnapshot {
@@ -155,7 +154,7 @@ fn control_snapshot(
 
 impl Baseliner {
     pub fn run(&self) -> anyhow::Result<()> {
-        let mut reflectors = HashMap::<IpAddr, ReflectorState>::new();
+        let mut reflectors = HashMap::<MeasurementStream, ReflectorState>::new();
 
         let slow_factor = ewma_factor(
             self.settings.advanced_settings.tick_interval,
@@ -249,6 +248,7 @@ mod tests {
         MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency,
         SignedDuration,
     };
+    use std::net::IpAddr;
     use std::time::Duration;
 
     fn state(
@@ -434,7 +434,8 @@ mod tests {
         let received_at = start + Duration::from_secs(1);
         let reflector = "192.0.2.1".parse().unwrap();
         let mut reflectors = HashMap::new();
-        reflectors.insert(reflector, state(100.0, 100.0, 100.0, 100.0, start));
+        let stream = reply(reflector, 0.0, 0.0, start).stream;
+        reflectors.insert(stream, state(100.0, 100.0, 100.0, 100.0, start));
 
         let (updated, anomaly) = process_reply(
             &mut reflectors,
@@ -450,7 +451,7 @@ mod tests {
         assert_close(updated.recent.down, 150.0);
         assert_close(updated.recent.up, 200.0);
         assert_eq!(updated.last_receive_at, received_at);
-        assert_close(reflectors[&reflector].recent.up, 200.0);
+        assert_close(reflectors[&stream].recent.up, 200.0);
     }
 
     #[test]
@@ -469,14 +470,34 @@ mod tests {
             0.5,
         );
         let snapshot = control_snapshot(&reflectors, generated_at);
+        let stream = reply(reflector, 0.0, 0.0, received_at).stream;
 
         assert!(!anomaly);
         assert_eq!(snapshot.generated_at, generated_at);
         assert_close(
-            snapshot.reflectors[&reflector].baseline.down,
+            snapshot.reflectors[&stream].baseline.down,
             updated.baseline.down,
         );
-        assert_close(snapshot.reflectors[&reflector].recent.up, updated.recent.up);
-        assert_eq!(snapshot.reflectors[&reflector].last_receive_at, received_at);
+        assert_close(snapshot.reflectors[&stream].recent.up, updated.recent.up);
+        assert_eq!(snapshot.reflectors[&stream].last_receive_at, received_at);
+    }
+
+    #[test]
+    fn keeps_protocol_streams_for_the_same_peer_independent() {
+        let start = Instant::now();
+        let peer = "192.0.2.1".parse().unwrap();
+        let mut reflectors = HashMap::new();
+        let echo = reply(peer, 10.0, 10.0, start);
+        let mut irtt = echo;
+        irtt.stream.id = MeasurementStreamId(1);
+        irtt.stream.source = MeasurementSource::Irtt;
+
+        process_reply(&mut reflectors, &echo, start, 0.9, 0.5);
+        process_reply(&mut reflectors, &irtt, start, 0.9, 0.5);
+
+        let snapshot = control_snapshot(&reflectors, start);
+        assert_eq!(snapshot.reflectors.len(), 2);
+        assert!(snapshot.reflectors.contains_key(&echo.stream));
+        assert!(snapshot.reflectors.contains_key(&irtt.stream));
     }
 }

src/controller.rs

@@ -1,102 +1,167 @@
-use std::{collections::HashMap, net::IpAddr, time::Duration};
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    sync::{Arc, RwLock},
+    time::Duration,
+};
 
 use flume::{Receiver, SendError, Sender};
 use thiserror::Error;
 
-use crate::measurement::{
+use crate::{
+    measurement::{
         icmp::IcmpBinding,
+        irtt::IrttBinding,
         model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule},
+    },
+    settings::ReflectorDefinition,
 };
 
 #[derive(Debug, Error)]
 pub enum ControllerError {
     #[error("ICMP channel error")]
     IcmpChannelError(#[from] SendError<Vec<IcmpBinding>>),
+    #[error("IRTT channel error")]
+    IrttChannelError(#[from] SendError<Vec<IrttBinding>>),
 }
 
 pub struct Controller {
-    reflector_rx: Receiver<Vec<IpAddr>>,
+    reflector_rx: Receiver<Vec<SocketAddr>>,
     icmp_binding_tx: Sender<Vec<IcmpBinding>>,
-
+    irtt_binding_tx: Sender<Vec<IrttBinding>>,
+    reflectors: HashMap<SocketAddr, ReflectorDefinition>,
+    active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+    endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
     source: MeasurementSource,
     period: Duration,
-
     next_stream_id: u64,
-    stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>,
+    stream_ids: HashMap<(SocketAddr, MeasurementSource), MeasurementStreamId>,
 }
 
 impl Controller {
     pub fn new(
-        reflector_rx: Receiver<Vec<IpAddr>>,
+        reflector_rx: Receiver<Vec<SocketAddr>>,
         icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+        irtt_binding_tx: Sender<Vec<IrttBinding>>,
+        reflectors: HashMap<SocketAddr, ReflectorDefinition>,
+        active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+        endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
+        source: MeasurementSource,
     ) -> Self {
         Self {
             reflector_rx,
             icmp_binding_tx,
-            source: MeasurementSource::IcmpTimestamp,
+            irtt_binding_tx,
+            reflectors,
+            active_streams,
+            endpoint_streams,
+            source,
             period: Duration::from_millis(500),
             next_stream_id: 0,
             stream_ids: HashMap::new(),
         }
     }
 
-    pub async fn run(mut self, initial_reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
+    pub async fn run(mut self, initial_reflectors: Vec<SocketAddr>) -> Result<(), ControllerError> {
         self.update_reflectors(initial_reflectors).await?;
-
         while let Ok(reflectors) = self.reflector_rx.recv_async().await {
             self.update_reflectors(reflectors).await?;
         }
-
         Ok(())
     }
 
-    fn stream_for(&mut self, peer: IpAddr) -> MeasurementStream {
-        let key = (peer, self.source);
-
+    fn stream_for(&mut self, endpoint: SocketAddr) -> MeasurementStream {
+        let key = (endpoint, self.source);
         let id = *self.stream_ids.entry(key).or_insert_with(|| {
             let id = MeasurementStreamId(self.next_stream_id);
             self.next_stream_id += 1;
             id
         });
-
         MeasurementStream {
             id,
-            peer,
+            peer: endpoint.ip(),
             source: self.source,
         }
     }
 
-    async fn update_reflectors(&mut self, reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
-        let count = reflectors.len();
-
-        let bindings = reflectors
-            .into_iter()
-            .enumerate()
-            .map(|(index, peer)| {
-                let stream = self.stream_for(peer);
-
-                IcmpBinding::new(
-                    stream,
-                    ProbeSchedule {
+    async fn update_reflectors(
+        &mut self,
+        endpoints: Vec<SocketAddr>,
+    ) -> Result<(), ControllerError> {
+        let count = endpoints.len();
+        let mut icmp = Vec::new();
+        let mut irtt = Vec::new();
+        let mut streams = Vec::with_capacity(count);
+        let mut endpoint_streams = HashMap::new();
+        for (index, endpoint) in endpoints.into_iter().enumerate() {
+            let Some((port, hmac)) = self.reflectors.get(&endpoint).map(|definition| {
+                (
+                    definition.irtt_port.unwrap_or(2112),
+                    definition.irtt_hmac.clone(),
+                )
+            }) else {
+                continue;
+            };
+            let stream = self.stream_for(endpoint);
+            let schedule = ProbeSchedule {
                 period: self.period,
                 offset: evenly_spaced_offset(self.period, index, count),
-                    },
-                )
-            })
-            .collect();
-
-        self.icmp_binding_tx
-            .send_async(bindings)
-            .await
-            .map_err(|e| ControllerError::IcmpChannelError(e))
+            };
+            match self.source {
+                MeasurementSource::IcmpEcho | MeasurementSource::IcmpTimestamp => {
+                    icmp.push(IcmpBinding::new(stream, schedule))
+                }
+                MeasurementSource::Irtt => {
+                    irtt.push(IrttBinding::new(stream, schedule, port, hmac))
+                }
+                _ => unreachable!("unsupported reflector measurement source"),
+            }
+            streams.push(stream);
+            endpoint_streams.insert(endpoint, stream);
+        }
+        *self
+            .active_streams
+            .write()
+            .expect("active stream lock poisoned") = streams;
+        *self
+            .endpoint_streams
+            .write()
+            .expect("endpoint stream lock poisoned") = endpoint_streams;
+        self.icmp_binding_tx.send_async(icmp).await?;
+        self.irtt_binding_tx.send_async(irtt).await?;
+        Ok(())
     }
 }
 
 fn evenly_spaced_offset(period: Duration, index: usize, count: usize) -> Duration {
     let nanos = period.as_nanos() * index as u128 / count as u128;
-
     Duration::new(
         (nanos / 1_000_000_000) as u64,
         (nanos % 1_000_000_000) as u32,
     )
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn stream_ids_are_stable_for_reselected_endpoints() {
+        let (_reflector_tx, reflector_rx) = flume::unbounded();
+        let (icmp_tx, _icmp_rx) = flume::unbounded();
+        let (irtt_tx, _irtt_rx) = flume::unbounded();
+        let endpoint = "192.0.2.1:0".parse().unwrap();
+        let mut controller = Controller::new(
+            reflector_rx,
+            icmp_tx,
+            irtt_tx,
+            HashMap::new(),
+            Arc::new(RwLock::new(Vec::new())),
+            Arc::new(RwLock::new(HashMap::new())),
+            MeasurementSource::IcmpTimestamp,
+        );
+
+        let first = controller.stream_for(endpoint);
+        assert_eq!(controller.stream_for(endpoint), first);
+    }
+}

src/main.rs

@@ -23,16 +23,17 @@ use crate::baseliner::Baseliner;
 use crate::controller::Controller;
 use crate::measurement::icmp::IcmpEngine;
 use crate::measurement::irtt::IrttEngine;
+use crate::measurement::model::{MeasurementSource, MeasurementStream};
 use crate::metrics::{Metric, Metrics, MetricsSender};
 use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations};
 use crate::ratecontroller::Ratecontroller;
 use crate::reflector_selector::ReflectorSelector;
-use crate::settings::Settings;
+use crate::settings::{ReflectorDefinition, ReflectorProtocol, Settings};
 use crate::util::RwLockExt;
 use ::log::{info, warn};
 use flume::RecvTimeoutError;
-use std::net::IpAddr;
-use std::str::FromStr;
+use std::collections::HashMap;
+use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 use std::thread;
@@ -40,8 +41,11 @@ use std::thread::{JoinHandle, sleep};
 use std::time::{Duration, Instant};
 
 struct ReflectorSetup {
-    peers: Arc<RwLock<Vec<IpAddr>>>,
-    pool: Vec<IpAddr>,
+    peers: Arc<RwLock<Vec<SocketAddr>>>,
+    active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+    endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
+    pool: Vec<SocketAddr>,
+    reflectors: HashMap<SocketAddr, ReflectorDefinition>,
     reselection_enabled: bool,
 }
 pub static SHUTDOWN: AtomicBool = AtomicBool::new(false);
@@ -135,29 +139,84 @@ fn restore_shaper<T: TrafficControlBackend>(
     }
 }
 
-fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
-    let configured = settings.load_reflectors()?;
-    let configured_count = configured.len();
-
-    let default_reflectors = [
-        IpAddr::from_str("9.9.9.9")?,
-        IpAddr::from_str("8.238.120.14")?,
-        IpAddr::from_str("74.82.42.42")?,
-        IpAddr::from_str("194.242.2.2")?,
-        IpAddr::from_str("208.67.222.222")?,
-        IpAddr::from_str("94.140.14.14")?,
-    ];
-
-    let reselection_enabled = configured_count > settings.reflectors.num_reflectors as usize;
-    let pool = if reselection_enabled {
-        configured
+fn resolve_reflectors(
+    definitions: &[ReflectorDefinition],
+    protocol: ReflectorProtocol,
+    countries: &[String],
+    filter_country: bool,
+) -> anyhow::Result<HashMap<SocketAddr, ReflectorDefinition>> {
+    let mut resolved = HashMap::new();
+    for definition in definitions {
+        if !definition.protocols.contains(&protocol) {
+            continue;
+        }
+        if protocol != ReflectorProtocol::Irtt
+            && definition
+                .host
+                .parse::<IpAddr>()
+                .is_ok_and(|ip| ip.is_ipv6())
+        {
+            continue;
+        }
+        if filter_country
+            && !countries.is_empty()
+            && !definition.country.as_ref().is_some_and(|country| {
+                countries
+                    .iter()
+                    .any(|allowed| allowed.eq_ignore_ascii_case(country))
+            })
+        {
+            continue;
+        }
+        let port = if protocol == ReflectorProtocol::Irtt {
+            definition.irtt_port.unwrap_or(2112)
         } else {
-        Vec::new()
+            0
         };
+        for endpoint in (definition.host.as_str(), port).to_socket_addrs()? {
+            if protocol != ReflectorProtocol::Irtt && endpoint.is_ipv6() {
+                continue;
+            }
+            resolved.insert(endpoint, definition.clone());
+        }
+    }
+    Ok(resolved)
+}
+
+fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
+    let protocol = settings.advanced_settings.measurement_type;
+    let configured = resolve_reflectors(
+        &settings.reflectors.entries,
+        protocol,
+        &settings.reflectors.countries,
+        true,
+    )?;
+    let bootstrap = resolve_reflectors(&settings.reflectors.bootstrap, protocol, &[], false)?;
+    let mut reflectors = bootstrap.clone();
+    reflectors.extend(
+        configured
+            .iter()
+            .map(|(endpoint, definition)| (*endpoint, definition.clone())),
+    );
+    let mut pool = configured.keys().copied().collect::<Vec<_>>();
+    pool.sort();
+    let mut initial = bootstrap.keys().copied().collect::<Vec<_>>();
+    initial.sort();
+    if initial.is_empty() {
+        initial = pool
+            .iter()
+            .take(settings.reflectors.num_reflectors as usize)
+            .copied()
+            .collect();
+    }
+    let reselection_enabled = pool.len() > settings.reflectors.num_reflectors as usize;
 
     Ok(ReflectorSetup {
-        peers: Arc::new(RwLock::new(default_reflectors.to_vec())),
+        peers: Arc::new(RwLock::new(initial)),
+        active_streams: Arc::new(RwLock::new(Vec::new())),
+        endpoint_streams: Arc::new(RwLock::new(HashMap::new())),
         pool,
+        reflectors,
         reselection_enabled,
     })
 }
@@ -230,7 +289,10 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
 
     let ReflectorSetup {
         peers: reflector_peers,
+        active_streams,
+        endpoint_streams,
         pool: reflector_pool,
+        reflectors,
         reselection_enabled,
     } = setup_reflectors(settings)?;
 
@@ -248,7 +310,20 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         (None, None)
     };
 
-    let controller = Controller::new(reflector_rx, icmp_binding_tx);
+    let source = match settings.advanced_settings.measurement_type {
+        ReflectorProtocol::IcmpEcho => MeasurementSource::IcmpEcho,
+        ReflectorProtocol::IcmpTimestamps => MeasurementSource::IcmpTimestamp,
+        ReflectorProtocol::Irtt => MeasurementSource::Irtt,
+    };
+    let controller = Controller::new(
+        reflector_rx,
+        icmp_binding_tx,
+        irtt_binding_tx,
+        reflectors,
+        Arc::clone(&active_streams),
+        Arc::clone(&endpoint_streams),
+        source,
+    );
     let icmp_engine = {
         let _guard = rt.enter();
         IcmpEngine::new(
@@ -336,6 +411,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
                 .expect("reselection snapshot receiver must exist when reselection is enabled"),
             reflector_peers_lock: Arc::clone(&reflector_peers),
             reflector_pool,
+            endpoint_streams,
             trigger_channel: reselect_rx,
             metrics: event_metrics,
             reflector_tx,
@@ -349,7 +425,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     let mut ratecontroller = Ratecontroller::new(
         settings.clone(),
         control_snapshot_rx,
-        reflector_peers,
+        active_streams,
         reselect_tx,
         rate_metrics,
     )?;
@@ -404,3 +480,71 @@ fn main() -> anyhow::Result<()> {
 
     run(&settings)
 }
+
+#[cfg(test)]
+mod reflector_tests {
+    use super::*;
+
+    fn definition(
+        host: &str,
+        country: Option<&str>,
+        protocols: Vec<ReflectorProtocol>,
+    ) -> ReflectorDefinition {
+        ReflectorDefinition {
+            host: host.to_owned(),
+            country: country.map(str::to_owned),
+            protocols,
+            irtt_port: Some(3011),
+            irtt_hmac: Some("key".to_owned()),
+        }
+    }
+
+    #[test]
+    fn resolves_protocol_and_country_filtered_ipv4_pool() {
+        let definitions = vec![
+            definition(
+                "127.0.0.1",
+                Some("GB"),
+                vec![ReflectorProtocol::IcmpTimestamps],
+            ),
+            definition(
+                "127.0.0.2",
+                Some("DE"),
+                vec![ReflectorProtocol::IcmpTimestamps],
+            ),
+            definition("::1", Some("GB"), vec![ReflectorProtocol::IcmpTimestamps]),
+            definition("127.0.0.3", Some("GB"), vec![ReflectorProtocol::Irtt]),
+        ];
+        let resolved = resolve_reflectors(
+            &definitions,
+            ReflectorProtocol::IcmpTimestamps,
+            &["gb".to_owned()],
+            true,
+        )
+        .unwrap();
+
+        assert_eq!(resolved.len(), 1);
+        assert_eq!(
+            resolved.keys().next().unwrap().ip(),
+            "127.0.0.1".parse::<IpAddr>().unwrap()
+        );
+    }
+
+    #[test]
+    fn irtt_resolution_keeps_ipv6_and_definition_configuration() {
+        let definition = definition("::1", Some("GB"), vec![ReflectorProtocol::Irtt]);
+        let resolved = resolve_reflectors(
+            &[definition],
+            ReflectorProtocol::Irtt,
+            &["GB".to_owned()],
+            true,
+        )
+        .unwrap();
+        let (endpoint, definition) = resolved.iter().next().unwrap();
+
+        assert!(endpoint.is_ipv6());
+        assert_eq!(endpoint.port(), 3011);
+        assert_eq!(definition.irtt_port, Some(3011));
+        assert_eq!(definition.irtt_hmac.as_deref(), Some("key"));
+    }
+}

src/measurement/irtt.rs

@@ -8,7 +8,7 @@ use irtt_client::{
         ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError,
         ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent,
         ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason,
-        TargetId,
+        TargetAuth, TargetId,
     },
 };
 use log::{debug, trace, warn};
@@ -54,14 +54,21 @@ pub struct IrttBinding {
     stream: MeasurementStream,
     schedule: ProbeSchedule,
     port: u16,
+    hmac: Option<String>,
 }
 
 impl IrttBinding {
-    pub fn new(stream: MeasurementStream, schedule: ProbeSchedule, port: u16) -> Self {
+    pub fn new(
+        stream: MeasurementStream,
+        schedule: ProbeSchedule,
+        port: u16,
+        hmac: Option<String>,
+    ) -> Self {
         Self {
             stream,
             schedule,
             port,
+            hmac,
         }
     }
 }
@@ -112,10 +119,16 @@ impl IrttEngine {
         let targets = bindings
             .iter()
             .map(|binding| {
-                ManagedTargetConfig::new(
+                let mut target = ManagedTargetConfig::new(
                     TargetId::new(binding.stream.id.0.to_string()),
                     SocketAddr::new(binding.stream.peer, binding.port).to_string(),
-                )
+                );
+                if let Some(key) = &binding.hmac {
+                    target.auth = TargetAuth::Override(irtt_client::Authentication::Hmac(
+                        irtt_client::HmacKey::new(key.as_bytes().to_vec().into_boxed_slice()),
+                    ));
+                }
+                target
             })
             .collect();
 

src/ratecontroller.rs

@@ -7,6 +7,7 @@
 
 use crate::SHUTDOWN;
 use crate::baseliner::ControlSnapshot;
+use crate::measurement::model::MeasurementStream;
 use crate::metrics::{Metric, MetricsSender};
 use crate::platform::{
     InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend,
@@ -19,7 +20,6 @@ use flume::{Receiver, Sender};
 use log::{debug, info, warn};
 use std::fs::File;
 use std::io::Write;
-use std::net::IpAddr;
 use std::sync::atomic::Ordering;
 use std::thread::sleep;
 use std::time::{Duration, Instant};
@@ -100,7 +100,7 @@ fn drain_latest_snapshot(
 
 fn deltas_from_snapshot(
     snapshot: Option<&ControlSnapshot>,
-    reflectors: &[IpAddr],
+    reflectors: &[MeasurementStream],
     now: Instant,
     tick_interval: f64,
 ) -> (Vec<f64>, Vec<f64>) {
@@ -123,8 +123,8 @@ fn deltas_from_snapshot(
             up.push(up_delta);
 
             debug!(
-                "Reflector: {} down_delay: {} up_delay: {}",
-                reflector, down_delta, up_delta
+                "Reflector: {} ({:?}) down_delay: {} up_delay: {}",
+                reflector.peer, reflector.source, down_delta, up_delta
             );
         }
     }
@@ -139,7 +139,7 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
     settings: Settings,
     control_snapshot: Option<ControlSnapshot>,
     control_snapshot_rx: Receiver<ControlSnapshot>,
-    reflectors_lock: ArcRwLock<Vec<IpAddr>>,
+    reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
     reselect_trigger: Sender<bool>,
     state_dl: State<T::Handle>,
     state_ul: State<T::Handle>,
@@ -153,7 +153,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     fn new_with_backends(
         settings: Settings,
         control_snapshot_rx: Receiver<ControlSnapshot>,
-        reflectors_lock: ArcRwLock<Vec<IpAddr>>,
+        reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
         reselect_trigger: Sender<bool>,
         metrics: MetricsSender,
         mut stats_provider: S,
@@ -509,7 +509,7 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
     pub fn new(
         settings: Settings,
         control_snapshot_rx: Receiver<ControlSnapshot>,
-        reflectors_lock: ArcRwLock<Vec<IpAddr>>,
+        reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
         reselect_trigger: Sender<bool>,
         metrics: MetricsSender,
     ) -> anyhow::Result<Self> {
@@ -530,7 +530,17 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
 mod tests {
     use super::*;
     use crate::baseliner::{EwmaStats, ReflectorState};
+    use crate::measurement::model::{MeasurementSource, MeasurementStreamId};
     use std::collections::HashMap;
+    use std::net::IpAddr;
+
+    fn stream(id: u64, peer: IpAddr) -> MeasurementStream {
+        MeasurementStream {
+            id: MeasurementStreamId(id),
+            peer,
+            source: MeasurementSource::IcmpTimestamp,
+        }
+    }
 
     #[test]
     fn fixed_seed_reproduces_initial_safe_rates() {
@@ -564,7 +574,7 @@ mod tests {
     }
 
     fn snapshot(generated_at: Instant, marker: f64) -> ControlSnapshot {
-        let reflector = "192.0.2.1".parse().unwrap();
+        let reflector = stream(0, "192.0.2.1".parse().unwrap());
         ControlSnapshot {
             generated_at,
             reflectors: HashMap::from([(
@@ -590,7 +600,9 @@ mod tests {
         assert!(rx.is_empty());
         assert_eq!(latest.generated_at, newest_at);
         assert_eq!(
-            latest.reflectors[&"192.0.2.1".parse().unwrap()].recent.down,
+            latest.reflectors[&stream(0, "192.0.2.1".parse().unwrap())]
+                .recent
+                .down,
             2.0
         );
     }
@@ -598,10 +610,10 @@ mod tests {
     #[test]
     fn calculates_sorted_deltas_from_active_fresh_snapshot_state() {
         let now = Instant::now();
-        let first: IpAddr = "192.0.2.1".parse().unwrap();
-        let second: IpAddr = "192.0.2.2".parse().unwrap();
-        let stale: IpAddr = "192.0.2.3".parse().unwrap();
-        let inactive: IpAddr = "192.0.2.4".parse().unwrap();
+        let first = stream(0, "192.0.2.1".parse().unwrap());
+        let second = stream(1, "192.0.2.2".parse().unwrap());
+        let stale = stream(2, "192.0.2.3".parse().unwrap());
+        let inactive = stream(3, "192.0.2.4".parse().unwrap());
         let snapshot = ControlSnapshot {
             generated_at: now,
             reflectors: HashMap::from([

src/reflector_selector.rs

@@ -7,12 +7,13 @@
 
 use crate::SHUTDOWN;
 use crate::baseliner::ControlSnapshot;
+use crate::measurement::model::MeasurementStream;
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
 use crate::util::RwLockExt;
 use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender};
 use log::{debug, info};
-use std::net::IpAddr;
+use std::net::SocketAddr;
 use std::sync::atomic::Ordering;
 use std::sync::{Arc, RwLock};
 use std::thread::sleep;
@@ -98,18 +99,25 @@ fn receive_snapshots_until(
     }
 }
 
-fn recent_rtt_for_peer(snapshot: Option<&ControlSnapshot>, peer: &IpAddr) -> Option<u64> {
+fn recent_rtt_for_peer(
+    snapshot: Option<&ControlSnapshot>,
+    stream: Option<&MeasurementStream>,
+    now: Instant,
+    max_age: Duration,
+) -> Option<u64> {
     snapshot
-        .and_then(|snapshot| snapshot.reflectors.get(peer))
+        .and_then(|snapshot| stream.and_then(|stream| snapshot.reflectors.get(stream)))
+        .filter(|state| now.saturating_duration_since(state.last_receive_at) < max_age)
         .map(|state| (state.recent.down + state.recent.up) as u64)
 }
 
 pub struct ReflectorSelector {
     pub settings: Settings,
     pub snapshot_rx: Receiver<ControlSnapshot>,
-    pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
-    pub reflector_tx: Sender<Vec<IpAddr>>,
-    pub reflector_pool: Vec<IpAddr>,
+    pub reflector_peers_lock: Arc<RwLock<Vec<SocketAddr>>>,
+    pub reflector_tx: Sender<Vec<SocketAddr>>,
+    pub reflector_pool: Vec<SocketAddr>,
+    pub endpoint_streams: Arc<RwLock<std::collections::HashMap<SocketAddr, MeasurementStream>>>,
     pub trigger_channel: Receiver<bool>,
     pub metrics: MetricsSender,
 }
@@ -163,8 +171,9 @@ impl ReflectorSelector {
                     Duration::new(self.settings.reflectors.peer_reselection_time * 60, 0);
             }
 
-            let mut next_peers: Vec<IpAddr> = Vec::new();
+            let mut next_peers: Vec<SocketAddr> = 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() {
@@ -172,6 +181,9 @@ impl ReflectorSelector {
                 next_peers.push(*reflector);
             }
 
+            if self.reflector_pool.is_empty() {
+                continue;
+            }
             for _ in 0..20 {
                 let next_candidate =
                     &self.reflector_pool[fastrand::usize(..self.reflector_pool.len())];
@@ -205,7 +217,13 @@ impl ReflectorSelector {
             let mut candidates = Vec::new();
 
             for peer in next_peers {
-                if let Some(rtt) = recent_rtt_for_peer(latest_snapshot.as_ref(), &peer) {
+                let stream = self.endpoint_streams.read_anyhow()?.get(&peer).copied();
+                if let Some(rtt) = recent_rtt_for_peer(
+                    latest_snapshot.as_ref(),
+                    stream.as_ref(),
+                    Instant::now(),
+                    Duration::from_secs_f64(self.settings.advanced_settings.tick_interval * 2.0),
+                ) {
                     candidates.push((peer, rtt));
                     info!("Candidate reflector: {} RTT: {}", peer, rtt);
                 } else {
@@ -220,7 +238,7 @@ impl ReflectorSelector {
             candidates.sort_by(|a, b| a.1.cmp(&b.1));
 
             // Now we will just limit the candidates down to 2 * num_reflectors
-            let mut num_reflectors = self.settings.reflectors.num_reflectors;
+            let num_reflectors = self.settings.reflectors.num_reflectors;
             let candidate_pool_num = (2 * num_reflectors) as usize;
             candidates.truncate(candidate_pool_num);
 
@@ -235,7 +253,14 @@ impl ReflectorSelector {
             }
 
             if (candidates.len() as u8) < num_reflectors {
-                num_reflectors = candidates.len() as u8;
+                info!(
+                    "Only {} usable reflectors, retaining current set of {}",
+                    candidates.len(),
+                    num_reflectors
+                );
+                self.reflector_tx.send(current_peers.clone())?;
+                *reflectors_peers = current_peers;
+                continue;
             }
 
             let mut new_peers = Vec::new();
@@ -246,7 +271,7 @@ impl ReflectorSelector {
                 self.metrics.send(Metric::Event {
                     name: "reflector_selected",
                     reason: "",
-                    reflector: Some(peer),
+                    reflector: Some(peer.ip()),
                     tags: &[],
                 });
             }
@@ -261,7 +286,9 @@ impl ReflectorSelector {
 mod tests {
     use super::*;
     use crate::baseliner::{EwmaStats, ReflectorState};
+    use crate::measurement::model::MeasurementSource;
     use std::collections::HashMap;
+    use std::net::IpAddr;
     use std::thread;
 
     fn snapshot(generated_at: Instant) -> ControlSnapshot {
@@ -274,7 +301,11 @@ mod tests {
     fn snapshot_with_rtt(generated_at: Instant, peer: IpAddr, rtt: f64) -> ControlSnapshot {
         let mut reflectors = HashMap::new();
         reflectors.insert(
+            crate::measurement::model::MeasurementStream {
+                id: crate::measurement::model::MeasurementStreamId(0),
                 peer,
+                source: MeasurementSource::IcmpTimestamp,
+            },
             ReflectorState {
                 baseline: EwmaStats {
                     down: rtt / 2.0,
@@ -374,7 +405,7 @@ mod tests {
     #[test]
     fn baseline_wait_retains_snapshot_for_evaluation() {
         let (snapshot_tx, snapshot_rx) = flume::unbounded();
-        let peer = "192.0.2.1".parse().unwrap();
+        let peer: IpAddr = "192.0.2.1".parse().unwrap();
         let generated_at = Instant::now();
         snapshot_tx
             .send(snapshot_with_rtt(generated_at, peer, 20.0))
@@ -395,7 +426,12 @@ mod tests {
         );
 
         assert_eq!(
-            recent_rtt_for_peer(latest_snapshot.as_ref(), &peer),
+            recent_rtt_for_peer(
+                latest_snapshot.as_ref(),
+                latest_snapshot.as_ref().unwrap().reflectors.keys().next(),
+                generated_at + Duration::from_secs(1),
+                Duration::from_secs(2)
+            ),
             Some(42)
         );
     }

src/settings/mod.rs

@@ -140,6 +140,16 @@ pub(crate) struct ReflectorDefinition {
     pub(crate) irtt_hmac: Option<String>,
 }
 
+#[derive(Debug, Deserialize)]
+struct ReflectorRow {
+    #[serde(alias = "reflector_ip")]
+    host: String,
+    country: Option<String>,
+    protocols: Option<String>,
+    irtt_port: Option<u16>,
+    irtt_hmac: Option<String>,
+}
+
 #[derive(Clone, Debug, Deserialize, UciConfig)]
 #[uci(package = "sqm-autorate-rust")]
 pub(crate) struct Settings {
@@ -176,15 +186,11 @@ impl Settings {
         );
 
         let builder = builder.add_source(Environment::with_prefix("SQMA").separator("__"));
-        let built_config = builder.build()?;
+        let mut settings: Self = builder.build()?.try_deserialize()?;
 
-        built_config.try_deserialize()
-    }
+        settings.reflectors.load_files()?;
 
-    pub fn load_reflectors(&self) -> anyhow::Result<Vec<IpAddr>> {
-        let mut reflectors = Vec::with_capacity(50);
-
-        Ok(reflectors)
+        Ok(settings)
     }
 }
 
@@ -233,19 +239,117 @@ pub(crate) struct ReflectorSettings {
     pub(crate) peer_reselection_time: u64,
 }
 
+impl ReflectorSettings {
+    fn load_files(&mut self) -> Result<(), ConfigError> {
+        for path in &self.files {
+            let mut reader = csv::Reader::from_path(path)
+                .map_err(|error| ConfigError::Message(error.to_string()))?;
+            for row in reader.deserialize::<ReflectorRow>() {
+                let row = row.map_err(|error| ConfigError::Message(error.to_string()))?;
+                let protocols = match row.protocols {
+                    Some(protocols) => protocols
+                        .split(['|', ';'])
+                        .map(str::trim)
+                        .filter(|protocol| !protocol.is_empty())
+                        .map(ReflectorProtocol::from_str)
+                        .collect::<Result<Vec<_>, _>>()?,
+                    None => vec![
+                        ReflectorProtocol::IcmpEcho,
+                        ReflectorProtocol::IcmpTimestamps,
+                    ],
+                };
+                self.entries.push(ReflectorDefinition {
+                    host: row.host,
+                    country: row.country,
+                    protocols,
+                    irtt_port: row.irtt_port,
+                    irtt_hmac: row.irtt_hmac,
+                });
+            }
+        }
+
+        Ok(())
+    }
+}
+
 impl Default for ReflectorSettings {
     fn default() -> Self {
+        let icmp = |host: &str| ReflectorDefinition {
+            host: host.to_owned(),
+            country: None,
+            protocols: vec![
+                ReflectorProtocol::IcmpEcho,
+                ReflectorProtocol::IcmpTimestamps,
+            ],
+            irtt_port: None,
+            irtt_hmac: None,
+        };
+
+        let irtt = |host: &str, country: Option<&str>| ReflectorDefinition {
+            host: host.to_owned(),
+            country: country.map(str::to_owned),
+            protocols: vec![ReflectorProtocol::Irtt],
+            irtt_port: Some(2112),
+            irtt_hmac: None,
+        };
+
         Self {
             files: vec!["/etc/sqm-autorate-rust/reflectors.csv".into()],
             countries: Vec::new(),
             entries: Vec::new(),
-            bootstrap: Vec::new(),
+            bootstrap: vec![
+                icmp("9.9.9.9"),
+                icmp("8.238.120.14"),
+                icmp("74.82.42.42"),
+                icmp("194.242.2.2"),
+                icmp("208.67.222.222"),
+                icmp("94.140.14.14"),
+                irtt("global.irtt.lochnair.net", None),
+                irtt("uk-cov1.irtt.lochnair.net", Some("GB")),
+                irtt("uk-cov2.irtt.lochnair.net", Some("GB")),
+                irtt("de-fra.irtt.lochnair.net", Some("DE")),
+                irtt("nl-ams1.irtt.lochnair.net", Some("NL")),
+            ],
             num_reflectors: 5,
             peer_reselection_time: 15,
         }
     }
 }
 
+#[cfg(test)]
+mod reflector_tests {
+    use super::*;
+
+    #[test]
+    fn loads_legacy_and_protocol_aware_reflector_csv_rows() {
+        let path = std::env::temp_dir().join(format!("reflectors-{}.csv", std::process::id()));
+        std::fs::write(
+            &path,
+            "reflector_ip,ip_version,description,country,protocols,irtt_port,irtt_hmac\n192.0.2.1,4,legacy,,,,\nreflector.example,,secure,GB,irtt,3011,secret\n",
+        )
+        .unwrap();
+        let mut settings = ReflectorSettings {
+            files: vec![path.to_string_lossy().into_owned()],
+            ..ReflectorSettings::default()
+        };
+
+        settings.load_files().unwrap();
+        let _ = std::fs::remove_file(path);
+
+        assert_eq!(settings.entries.len(), 2);
+        assert_eq!(
+            settings.entries[0].protocols,
+            vec![
+                ReflectorProtocol::IcmpEcho,
+                ReflectorProtocol::IcmpTimestamps
+            ]
+        );
+        assert_eq!(settings.entries[1].protocols, vec![ReflectorProtocol::Irtt]);
+        assert_eq!(settings.entries[1].irtt_port, Some(3011));
+        assert_eq!(settings.entries[1].irtt_hmac.as_deref(), Some("secret"));
+    }
+}
+
 #[derive(Clone, Debug, Deserialize, UciSection)]
 #[serde(default)]
 pub(crate) struct OutputSettings {