feat(reflectors): refactor reflector configuration

Cargo.lock

@@ -60,6 +60,15 @@ version = "2.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
 
+[[package]]
+name = "block-buffer"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa"
+dependencies = [
+ "hybrid-array",
+]
+
 [[package]]
 name = "bumpalo"
 version = "3.20.3"
@@ -123,6 +132,12 @@ dependencies = [
  "cc",
 ]
 
+[[package]]
+name = "cmov"
+version = "0.5.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
+
 [[package]]
 name = "config"
 version = "0.15.27"
@@ -135,6 +150,21 @@ dependencies = [
  "winnow",
 ]
 
+[[package]]
+name = "const-oid"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
+
+[[package]]
+name = "crypto-common"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453"
+dependencies = [
+ "hybrid-array",
+]
+
 [[package]]
 name = "csv"
 version = "1.4.0"
@@ -156,6 +186,27 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "ctutils"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574"
+dependencies = [
+ "cmov",
+]
+
+[[package]]
+name = "digest"
+version = "0.11.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
+dependencies = [
+ "block-buffer",
+ "const-oid",
+ "crypto-common",
+ "ctutils",
+]
+
 [[package]]
 name = "either"
 version = "1.19.0"
@@ -242,6 +293,24 @@ version = "0.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
 
+[[package]]
+name = "hmac"
+version = "0.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f"
+dependencies = [
+ "digest",
+]
+
+[[package]]
+name = "hybrid-array"
+version = "0.4.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17"
+dependencies = [
+ "typenum",
+]
+
 [[package]]
 name = "icmp-socket2"
 version = "0.6.4"
@@ -265,6 +334,31 @@ dependencies = [
  "hashbrown",
 ]
 
+[[package]]
+name = "irtt-client"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a59c24704bd94f19b84b2b0190aa9163efede499f8cf4eb02922443cfee08339"
+dependencies = [
+ "irtt-proto",
+ "nix",
+ "socket2",
+ "thiserror",
+ "tokio",
+]
+
+[[package]]
+name = "irtt-proto"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a56ff307983b8f6477724e08cf726c5e24975b9affeb853d0830d91c60026ab2"
+dependencies = [
+ "hmac",
+ "md-5",
+ "subtle",
+ "thiserror",
+]
+
 [[package]]
 name = "itertools"
 version = "0.13.0"
@@ -337,6 +431,16 @@ version = "0.4.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
 
+[[package]]
+name = "md-5"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
+dependencies = [
+ "cfg-if",
+ "digest",
+]
+
 [[package]]
 name = "memchr"
 version = "2.8.3"
@@ -679,6 +783,7 @@ dependencies = [
  "fastrand",
  "flume",
  "icmp-socket2",
+ "irtt-client",
  "libc",
  "log",
  "netlink-bindings",
@@ -697,6 +802,12 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85ec932792e62b2d3936163eacbd2e09984453bcd114f665c1e1694e36ff59c9"
 
+[[package]]
+name = "subtle"
+version = "2.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
+
 [[package]]
 name = "syn"
 version = "2.0.119"
@@ -836,6 +947,12 @@ dependencies = [
  "toml",
 ]
 
+[[package]]
+name = "typenum"
+version = "1.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
+
 [[package]]
 name = "uci-derive"
 version = "0.1.0"

Cargo.toml

@@ -25,6 +25,7 @@ csv = "1.4.0"
 fastrand = "2"
 flume = { version = "0.12.0",  default-features = false, features = ["async", "eventual-fairness", "select"] }
 icmp-socket2 = { version = "0.6.4", features = ["tokio"] }
+irtt-client = { version = "0.6.0", optional = true, features = ["ancillary", "tokio"] }
 libc = "0.2"
 rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
 serde = { version = "1", features = ["derive"] }
@@ -44,7 +45,8 @@ netlink-socket2 = { version = "0.3.5" }
 rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "6835397f4f4ba8b69e75e727105186865c357417", optional = true, default-features = false }
 
 [features]
-default = ["toml", "uci-vendored"]
+default = ["irtt", "toml", "uci-vendored"]
+irtt = ["dep:irtt-client"]
 toml = ["config/toml"]
 uci = ["dep:rust-uci"]
 uci-vendored = ["uci", "rust-uci/vendored"]

crates/uci-derive/src/lib.rs

@@ -9,13 +9,13 @@ use syn::{Data, DeriveInput, Field, Fields, FieldsNamed, LitStr, Token, parse_ma
 
 /// Generates `UciSectionSchema` from a settings struct.
 ///
-/// Every named field is mapped one-to-one:
+/// Every named field without `#[uci(skip)]` is mapped one-to-one:
 ///
 /// ```text
 /// Rust field name -> UCI option name
 /// Rust field name -> config-rs field name
 /// ```
-#[proc_macro_derive(UciSection)]
+#[proc_macro_derive(UciSection, attributes(uci))]
 pub fn derive_uci_section(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);
 
@@ -56,10 +56,13 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
     let struct_name = &input.ident;
     let fields = require_named_fields(input)?;
 
-    let mappings = fields
+    let mut mappings = Vec::with_capacity(fields.named.len());
-        .named
+
-        .iter()
+    for field in &fields.named {
-        .map(|field| {
+        if parse_skip(field)? {
+            continue;
+        }
+
         let field_ident = field
             .ident
             .as_ref()
@@ -68,14 +71,13 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
         let field_name = field_ident.unraw().to_string();
         let field_name = LitStr::new(&field_name, field_ident.span());
 
-            quote! {
+        mappings.push(quote! {
             crate::settings::uci_schema::UciOptionMapping {
                 uci_option: #field_name,
                 config_field: #field_name,
             }
+        });
     }
-        })
-        .collect::<Vec<_>>();
 
     let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl();
 
@@ -95,6 +97,29 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
     })
 }
 
+fn parse_skip(field: &Field) -> syn::Result<bool> {
+    let mut skip = false;
+
+    for attribute in &field.attrs {
+        if !attribute.path().is_ident("uci") {
+            continue;
+        }
+
+        attribute.parse_nested_meta(|meta| {
+            if !meta.path.is_ident("skip") {
+                return Err(meta.error("expected `skip`"));
+            }
+            if skip {
+                return Err(meta.error("duplicate `skip` attribute"));
+            }
+            skip = true;
+            Ok(())
+        })?;
+    }
+
+    Ok(skip)
+}
+
 fn expand_uci_config(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
     let struct_name = &input.ident;
     let fields = require_named_fields(input)?;

src/main.rs

@@ -22,6 +22,7 @@ mod util;
 use crate::baseliner::Baseliner;
 use crate::controller::Controller;
 use crate::measurement::icmp::IcmpEngine;
+use crate::measurement::irtt::IrttEngine;
 use crate::metrics::{Metric, Metrics, MetricsSender};
 use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations};
 use crate::ratecontroller::Ratecontroller;
@@ -147,7 +148,7 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
         IpAddr::from_str("94.140.14.14")?,
     ];
 
-    let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize;
+    let reselection_enabled = configured_count > settings.reflectors.num_reflectors as usize;
     let pool = if reselection_enabled {
         configured
     } else {
@@ -236,6 +237,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded();
     let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
     let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded();
+    let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded();
     let (measurement_tx, measurement_rx) = flume::unbounded();
     let (reflector_tx, reflector_rx) = flume::unbounded();
     let (reselect_tx, reselect_rx) = flume::bounded(1);
@@ -247,17 +249,22 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     };
 
     let controller = Controller::new(reflector_rx, icmp_binding_tx);
-    let engine = {
+    let icmp_engine = {
         let _guard = rt.enter();
         IcmpEngine::new(
             start_time,
             &settings.network,
-            measurement_tx,
+            measurement_tx.clone(),
             icmp_binding_rx,
         )?
     };
+    let (_irtt_task, irtt_engine) = {
+        let _guard = rt.enter();
+        IrttEngine::new(start_time, measurement_tx, irtt_binding_rx)?
+    };
 
-    spawn_task(&rt, &error_tx, engine.run());
+    spawn_task(&rt, &error_tx, icmp_engine.run());
+    spawn_task(&rt, &error_tx, irtt_engine.run());
 
     let dropped = Arc::new(AtomicU32::new(0));
 

src/measurement/irtt.rs

@@ -0,0 +1,343 @@
+use std::{collections::HashMap, net::SocketAddr, time::Instant};
+
+use flume::{Receiver, Sender};
+use irtt_client::{
+    ClientConfig, ClientEvent,
+    managed::{
+        ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask,
+        ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError,
+        ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent,
+        ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason,
+        TargetId,
+    },
+};
+use log::{debug, trace, warn};
+use thiserror::Error;
+use tokio::sync::broadcast::error::RecvError;
+
+use crate::measurement::model::{
+    MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream,
+    MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule, SignedDuration,
+};
+
+#[derive(Debug, Error)]
+pub enum IrttError {
+    #[error("failed to build IRTT config")]
+    InvalidConfig(#[from] ManagedConfigError),
+
+    #[error("event channel closed")]
+    EventChannelClosed,
+
+    #[error("binding channel closed")]
+    BindingChannelClosed,
+
+    #[error("error while reciving on channel")]
+    ChannelRecvError(#[from] RecvError),
+
+    #[error("unable to subscribe to irtt events")]
+    SubscriptionError(#[from] ManagedSubscribeError),
+
+    #[error("failed to submit IRTT target update")]
+    CommandError(#[from] ManagedCommandError),
+
+    #[error("failed to apply IRTT target update")]
+    CommandApplyError(#[from] ManagedCommandApplyError),
+
+    #[error("IRTT managed client failed: {0:?}")]
+    ManagedFailure(ManagedEndReason),
+
+    #[error("IRTT managed task was abandoned")]
+    ManagedAbandoned,
+}
+
+pub struct IrttBinding {
+    stream: MeasurementStream,
+    schedule: ProbeSchedule,
+    port: u16,
+}
+
+impl IrttBinding {
+    pub fn new(stream: MeasurementStream, schedule: ProbeSchedule, port: u16) -> Self {
+        Self {
+            stream,
+            schedule,
+            port,
+        }
+    }
+}
+
+pub struct IrttEngine {
+    client: ManagedClientHandle,
+
+    bindings: HashMap<MeasurementStreamId, IrttBinding>,
+
+    origin: Instant,
+
+    binding_rx: Receiver<Vec<IrttBinding>>,
+    observation_tx: Sender<MeasurementEvent>,
+}
+
+impl IrttEngine {
+    pub fn new(
+        origin: Instant,
+        observation_tx: Sender<MeasurementEvent>,
+        binding_rx: Receiver<Vec<IrttBinding>>,
+    ) -> Result<(ManagedClientTask, Self), IrttError> {
+        let config = ManagedClientConfig {
+            client: ClientConfig {
+                request: irtt_client::SessionRequest {
+                    duration: None,
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            completion: ManagedCompletionPolicy::ExplicitStop,
+            ..Default::default()
+        };
+        let (task, handle) = ManagedClient::task(config, Vec::new())?;
+
+        Ok((
+            task,
+            Self {
+                client: handle,
+                bindings: HashMap::new(),
+                origin,
+                binding_rx,
+                observation_tx,
+            },
+        ))
+    }
+
+    async fn set_bindings(&mut self, bindings: Vec<IrttBinding>) -> Result<(), IrttError> {
+        let targets = bindings
+            .iter()
+            .map(|binding| {
+                ManagedTargetConfig::new(
+                    TargetId::new(binding.stream.id.0.to_string()),
+                    SocketAddr::new(binding.stream.peer, binding.port).to_string(),
+                )
+            })
+            .collect();
+
+        self.client.update_targets(targets)?.await?;
+
+        self.bindings = bindings
+            .into_iter()
+            .map(|binding| (binding.stream.id, binding))
+            .collect();
+
+        Ok(())
+    }
+
+    pub async fn run(mut self) -> Result<(), IrttError> {
+        let mut events = self.client.subscribe()?;
+
+        loop {
+            tokio::select! {
+                bindings = self.binding_rx.recv_async() => {
+                    let bindings = bindings
+                        .map_err(|_| IrttError::BindingChannelClosed)?;
+
+                    self.set_bindings(bindings).await?;
+                }
+
+                event = events.recv() => {
+                    let event = match event {
+                        Ok(event) => event,
+
+                        Err(RecvError::Lagged(n)) => {
+                            warn!("IRTT event receiver dropped {n} events");
+                            continue;
+                        }
+
+                        Err(RecvError::Closed) => {
+                            let status = self.client.status();
+
+                            match status.lifecycle {
+                                ManagedLifecycle::Completed => return Ok(()),
+
+                                ManagedLifecycle::Failed => {
+                                    if let Some(outcome) = &status.final_outcome {
+                                        return Err(IrttError::ManagedFailure(
+                                            outcome.end_reason.clone()
+                                        ));
+                                    }
+
+                                    return Err(IrttError::ManagedAbandoned);
+                                }
+
+                                _ => return Err(IrttError::ManagedAbandoned),
+                            }
+                        }
+                    };
+
+                    match event {
+                        ManagedEvent::Started => {
+                            debug!("IRTT managed client started");
+                        }
+
+                        ManagedEvent::TargetStateChanged {
+                            target,
+                            lifecycle,
+                        } => {
+                            debug!("IRTT target {target:?}: {lifecycle:?}");
+                        }
+
+                        ManagedEvent::Client { target, event } => {
+                            let stream = target.id.as_str()
+                                .parse::<u64>()
+                                .ok()
+                                .and_then(|id| {
+                                    self.bindings.get(&MeasurementStreamId(id))
+                                })
+                                .map(|binding| binding.stream);
+
+                            match event {
+                                ClientEvent::SessionStarted(session) => {
+                                    debug!(
+                                        "IRTT session started for {}: {}",
+                                        target.id,
+                                        session.remote
+                                    );
+                                }
+
+                                ClientEvent::NoTestCompleted(outcome) => {
+                                    debug!(
+                                        "IRTT no-test completed for {}: {}",
+                                        target.id,
+                                        outcome.remote
+                                    );
+                                }
+
+                                ClientEvent::SessionClosed { .. } => {
+                                    debug!("IRTT session closed for {}", target.id);
+                                }
+
+                                ClientEvent::EchoSent { seq, .. } => {
+                                    trace!("IRTT {} sent probe {seq}", target.id);
+                                }
+
+                                ClientEvent::EchoReply {
+                                    sent_at,
+                                    received_at,
+                                    one_way,
+                                    ..
+                                } => {
+                                    let Some(stream) = stream else {
+                                        continue;
+                                    };
+
+                                    let observation = MeasurementObservation {
+                                        stream,
+                                        started_at: sent_at.mono,
+                                        observed_at: received_at.mono,
+                                        one_way: one_way.and_then(map_one_way),
+                                    };
+
+                                    self.observation_tx
+                                        .send_async(MeasurementEvent::Observation(observation))
+                                        .await
+                                        .map_err(|_| IrttError::EventChannelClosed)?;
+                                }
+
+                                ClientEvent::EchoLoss {
+                                    sent_at,
+                                    timeout_at,
+                                    ..
+                                } => {
+                                    let Some(stream) = stream else {
+                                        continue;
+                                    };
+
+                                    let loss = MeasurementLoss {
+                                        stream,
+                                        started_at: sent_at.mono,
+                                        detected_at: timeout_at,
+                                    };
+
+                                    self.observation_tx
+                                        .send_async(MeasurementEvent::Loss(loss))
+                                        .await
+                                        .map_err(|_| IrttError::EventChannelClosed)?;
+                                }
+
+                                ClientEvent::DuplicateReply { seq, .. } => {
+                                    trace!(
+                                        "IRTT {} duplicate reply {seq}",
+                                        target.id
+                                    );
+                                }
+
+                                ClientEvent::LateReply { seq, .. } => {
+                                    trace!(
+                                        "IRTT {} late reply {seq}",
+                                        target.id
+                                    );
+                                }
+
+                                ClientEvent::Warning { kind, message, .. } => {
+                                    debug!(
+                                        "IRTT {} warning ({kind:?}): {message}",
+                                        target.id
+                                    );
+                                }
+                            }
+                        }
+
+                        ManagedEvent::TargetFinished { outcome } => {
+                            match &outcome.end_reason {
+                                ManagedTargetEndReason::Failed(failure) => {
+                                    warn!(
+                                        "IRTT target {} failed: {failure:?}",
+                                        outcome.target.id
+                                    );
+                                }
+
+                                reason => {
+                                    debug!(
+                                        "IRTT target {} finished: {reason:?}",
+                                        outcome.target.id
+                                    );
+                                }
+                            }
+                        }
+
+                        ManagedEvent::Stopping => {
+                            debug!("IRTT managed client stopping");
+                        }
+
+                        ManagedEvent::Completed { outcome } => {
+                            debug!(
+                                "IRTT managed client completed: {:?}",
+                                outcome.end_reason
+                            );
+                            return Ok(());
+                        }
+
+                        ManagedEvent::Failed { outcome } => {
+                            return Err(IrttError::ManagedFailure(
+                                outcome.end_reason.clone()
+                            ));
+                        }
+
+                        ManagedEvent::Abandoned => {
+                            return Err(IrttError::ManagedAbandoned);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn map_one_way(sample: irtt_client::OneWayDelaySample) -> Option<OneWayLatency> {
+    let uplink = i64::try_from(sample.client_to_server?.as_nanos()).ok()?;
+
+    let downlink = i64::try_from(sample.server_to_client?.as_nanos()).ok()?;
+
+    Some(OneWayLatency {
+        uplink: SignedDuration::from_nanos(uplink),
+        downlink: SignedDuration::from_nanos(downlink),
+        clock: OneWayClock::Uncalibrated,
+    })
+}

src/measurement/mod.rs

@@ -1,2 +1,4 @@
 pub(crate) mod icmp;
+#[cfg(feature = "irtt")]
+pub(crate) mod irtt;
 pub(crate) mod model;

src/reflector_selector.rs

@@ -159,10 +159,8 @@ impl ReflectorSelector {
 
             // After 40 reselections, slow down to every 15 minutes
             if reselection_count > 40 {
-                selector_sleep_time = Duration::new(
+                selector_sleep_time =
-                    self.settings.advanced_settings.peer_reselection_time * 60,
+                    Duration::new(self.settings.reflectors.peer_reselection_time * 60, 0);
-                    0,
-                );
             }
 
             let mut next_peers: Vec<IpAddr> = Vec::new();
@@ -222,7 +220,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.advanced_settings.num_reflectors;
+            let mut num_reflectors = self.settings.reflectors.num_reflectors;
             let candidate_pool_num = (2 * num_reflectors) as usize;
             candidates.truncate(candidate_pool_num);
 

src/settings/mod.rs

@@ -78,28 +78,32 @@ where
     deserializer.deserialize_any(FlexiBoolVisitor)
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash)]
-pub enum MeasurementType {
+#[serde(rename_all = "kebab-case")]
-    Icmp = 1,
+pub enum ReflectorProtocol {
+    IcmpEcho = 1,
     IcmpTimestamps,
+    Irtt,
 }
 
-impl Display for MeasurementType {
+impl Display for ReflectorProtocol {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            MeasurementType::Icmp => write!(f, "icmp"),
+            ReflectorProtocol::IcmpEcho => write!(f, "icmp"),
-            MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"),
+            ReflectorProtocol::IcmpTimestamps => write!(f, "icmp-timestamps"),
+            ReflectorProtocol::Irtt => write!(f, "irtt"),
         }
     }
 }
 
-impl FromStr for MeasurementType {
+impl FromStr for ReflectorProtocol {
     type Err = ConfigError;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s.to_lowercase().as_str() {
-            "icmp" => Ok(MeasurementType::Icmp),
+            "icmp" => Ok(ReflectorProtocol::IcmpEcho),
-            "icmp-timestamps" | "icmp_timestamps" => Ok(MeasurementType::IcmpTimestamps),
+            "icmp-timestamps" | "icmp_timestamps" => Ok(ReflectorProtocol::IcmpTimestamps),
+            "irtt" => Ok(ReflectorProtocol::Irtt),
             _ => Err(ConfigError::Message(format!(
                 "invalid measurement type: {s}"
             ))),
@@ -127,12 +131,13 @@ impl FromStr for ObservabilityProtocol {
     }
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Clone, Deserialize, UciSection)]
-struct ReflectorRecord {
+pub(crate) struct ReflectorDefinition {
-    reflector_ip: IpAddr,
+    pub(crate) host: String,
-    ip_version: u8,
+    pub(crate) country: Option<String>,
-    #[allow(dead_code)]
+    pub(crate) protocols: Vec<ReflectorProtocol>,
-    description: String,
+    pub(crate) irtt_port: Option<u16>,
+    pub(crate) irtt_hmac: Option<String>,
 }
 
 #[derive(Clone, Debug, Deserialize, UciConfig)]
@@ -141,6 +146,10 @@ pub(crate) struct Settings {
     #[uci(section)]
     pub(crate) network: NetworkSettings,
 
+    #[serde(default)]
+    #[uci(section)]
+    pub(crate) reflectors: ReflectorSettings,
+
     #[serde(default)]
     #[uci(section)]
     pub(crate) output: OutputSettings,
@@ -173,40 +182,8 @@ impl Settings {
     }
 
     pub fn load_reflectors(&self) -> anyhow::Result<Vec<IpAddr>> {
-        let mut reader = csv::ReaderBuilder::new()
-            .has_headers(true)
-            .trim(csv::Trim::All)
-            .from_path(&self.advanced_settings.reflector_list_file)
-            .with_context(|| {
-                format!(
-                    "failed to open reflector list: {}",
-                    self.advanced_settings.reflector_list_file
-                )
-            })?;
-
         let mut reflectors = Vec::with_capacity(50);
 
-        for record in reader.deserialize::<ReflectorRecord>() {
-            let record = record.with_context(|| {
-                format!(
-                    "failed to parse reflector list: {}",
-                    self.advanced_settings.reflector_list_file
-                )
-            })?;
-
-            match (record.ip_version, record.reflector_ip) {
-                (4, IpAddr::V4(_)) => {}
-                (6, address @ IpAddr::V6(_)) => {
-                    bail!("reflector {address} is IPv6; only IPv4 reflectors are supported");
-                }
-                (version, address) => {
-                    bail!("reflector {address} has inconsistent IP version {version}");
-                }
-            }
-
-            reflectors.push(record.reflector_ip);
-        }
-
         Ok(reflectors)
     }
 }
@@ -240,6 +217,35 @@ impl NetworkSettings {
     }
 }
 
+#[derive(Clone, Debug, Deserialize, UciSection)]
+#[serde(default)]
+pub(crate) struct ReflectorSettings {
+    pub(crate) files: Vec<String>,
+    pub(crate) countries: Vec<String>,
+
+    #[uci(skip)]
+    pub(crate) entries: Vec<ReflectorDefinition>,
+
+    #[uci(skip)]
+    pub(crate) bootstrap: Vec<ReflectorDefinition>,
+
+    pub(crate) num_reflectors: u8,
+    pub(crate) peer_reselection_time: u64,
+}
+
+impl Default for ReflectorSettings {
+    fn default() -> Self {
+        Self {
+            files: vec!["/etc/sqm-autorate-rust/reflectors.csv".into()],
+            countries: Vec::new(),
+            entries: Vec::new(),
+            bootstrap: Vec::new(),
+            num_reflectors: 5,
+            peer_reselection_time: 15,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Deserialize, UciSection)]
 #[serde(default)]
 pub(crate) struct OutputSettings {
@@ -309,10 +315,7 @@ pub(crate) struct AdvancedSettings {
     pub(crate) high_load_level: f64,
     pub(crate) min_change_interval: f64,
     #[serde(deserialize_with = "deserialize_from_str")]
-    pub(crate) measurement_type: MeasurementType,
+    pub(crate) measurement_type: ReflectorProtocol,
-    pub(crate) num_reflectors: u8,
-    pub(crate) peer_reselection_time: u64,
-    pub(crate) reflector_list_file: String,
     pub(crate) speed_hist_size: u32,
     pub(crate) tick_interval: f64,
     pub(crate) upload_delay_ms: f64,
@@ -326,10 +329,7 @@ impl Default for AdvancedSettings {
             download_delay_ms: 15.0,
             high_load_level: 0.8,
             min_change_interval: 0.5,
-            measurement_type: MeasurementType::IcmpTimestamps,
+            measurement_type: ReflectorProtocol::IcmpTimestamps,
-            num_reflectors: 5,
-            peer_reselection_time: 15,
-            reflector_list_file: "/etc/sqm-autorate/reflectors-icmp.csv".to_owned(),
             speed_hist_size: 100,
             tick_interval: 0.5,
             upload_delay_ms: 15.0,

src/settings/uci.rs

@@ -1,6 +1,6 @@
 use std::path::{Path, PathBuf};
 
-use crate::settings::uci_schema::{UciConfigSchema, UciSectionMapping};
+use crate::settings::uci_schema::{UciConfigSchema, UciRepeatedSectionMapping, UciSectionMapping};
 use config::{ConfigError as SourceError, Map, Source, Value, ValueKind};
 
 use rust_uci::config::{
@@ -18,6 +18,7 @@ pub(crate) struct UciSource {
     packages: PackageSelection,
     required: bool,
     schema: &'static [UciSectionMapping],
+    repeated_schema: &'static [UciRepeatedSectionMapping],
     config_dir: Option<PathBuf>,
     save_dir: Option<PathBuf>,
 }
@@ -28,6 +29,7 @@ impl UciSource {
             packages,
             required: true,
             schema: &[],
+            repeated_schema: &[],
             config_dir: None,
             save_dir: None,
         }
@@ -65,6 +67,17 @@ impl UciSource {
         self
     }
 
+    /// Registers section types collected as arrays of tables at their config paths.
+    /// Absent section types leave the destination unset for Serde/source defaults.
+    #[allow(dead_code)]
+    pub(crate) fn with_repeated_schema(
+        mut self,
+        schema: &'static [UciRepeatedSectionMapping],
+    ) -> Self {
+        self.repeated_schema = schema;
+        self
+    }
+
     #[allow(dead_code)]
     pub(crate) fn with_directories(
         mut self,
@@ -145,6 +158,7 @@ impl UciSource {
         package_name: &str,
         package: &UciPackage,
         values: &mut Map<String, Value>,
+        repeated_values: &mut Map<String, Vec<Value>>,
     ) -> Result<(), SourceError> {
         let sections = package.sections().map_err(|e| {
             SourceError::Message(format!(
@@ -154,6 +168,12 @@ impl UciSource {
         })?;
 
         for section in sections {
+            let repeated_mapping = self
+                .repeated_schema
+                .iter()
+                .find(|mapping| mapping.uci_section == section.type_());
+            let mut section_values = Map::new();
+
             let selector = section.selector().ok_or_else(|| {
                 SourceError::Message(format!(
                     "loaded UCI section of type {:?} in package \
@@ -181,9 +201,21 @@ impl UciSource {
 
                 let source_key = format!("{package_name}.{config_relative_key}");
 
-                let destination_key = self
+                let repeated_option = repeated_mapping.and_then(|mapping| {
-                    .mapped_config_key(&selector, option_name)
+                    mapping
-                    .unwrap_or(source_key);
+                        .options
+                        .iter()
+                        .find(|option| option.uci_option == option_name)
+                });
+
+                let destination_key = if let Some(mapping) = repeated_option {
+                    mapping.config_field.to_owned()
+                } else if repeated_mapping.is_some() {
+                    option_name.to_owned()
+                } else {
+                    self.mapped_config_key(&selector, option_name)
+                        .unwrap_or(source_key)
+                };
 
                 let Some(value) = option.get().map_err(|e| {
                     SourceError::Message(format!(
@@ -210,13 +242,27 @@ impl UciSource {
                     }
                 };
 
-                if values.insert(destination_key.clone(), value).is_some() {
+                let destination = if repeated_mapping.is_some() {
+                    &mut section_values
+                } else {
+                    &mut *values
+                };
+
+                if destination.insert(destination_key.clone(), value).is_some() {
                     return Err(SourceError::Message(format!(
                         "duplicate UCI destination key \
                          {destination_key:?} while mapping {uci_key:?}"
                     )));
                 }
             }
+
+            if let Some(mapping) = repeated_mapping {
+                let origin = format!("UCI section {package_name}.{selector}");
+                repeated_values
+                    .entry(mapping.config_path.to_owned())
+                    .or_default()
+                    .push(Value::new(Some(&origin), ValueKind::Table(section_values)));
+            }
         }
 
         Ok(())
@@ -272,6 +318,7 @@ impl Source for UciSource {
         };
 
         let mut values = Map::new();
+        let mut repeated_values = Map::new();
 
         for package in packages {
             let package_name = package
@@ -279,7 +326,22 @@ impl Source for UciSource {
                 .map_err(|e| SourceError::Message(format!("failed to read UCI package name: {e}")))?
                 .to_owned();
 
-            self.collect_package(&package_name, &package, &mut values)?;
+            self.collect_package(&package_name, &package, &mut values, &mut repeated_values)?;
+        }
+
+        for (config_path, sections) in repeated_values {
+            let origin = "UCI repeated sections".to_owned();
+            if values
+                .insert(
+                    config_path.clone(),
+                    Value::new(Some(&origin), ValueKind::Array(sections)),
+                )
+                .is_some()
+            {
+                return Err(SourceError::Message(format!(
+                    "duplicate UCI destination key {config_path:?} while mapping repeated sections"
+                )));
+            }
         }
 
         Ok(values)

src/settings/uci_schema.rs

@@ -19,6 +19,18 @@ pub(crate) struct UciSectionMapping {
     pub(crate) options: &'static [UciOptionMapping],
 }
 
+#[cfg(all(feature = "uci", unix))]
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct UciRepeatedSectionMapping {
+    /// UCI section type, for both named and anonymous sections.
+    pub(crate) uci_section: &'static str,
+
+    /// Config-rs / Serde path receiving an array of section tables.
+    pub(crate) config_path: &'static str,
+
+    pub(crate) options: &'static [UciOptionMapping],
+}
+
 pub(crate) trait UciSectionSchema {
     const UCI_OPTIONS: &'static [UciOptionMapping];
 }

src/settings/validation.rs

@@ -3,6 +3,8 @@
 // SPDX-License-Identifier: MPL-2.0
 //
 
+use crate::settings::ReflectorSettings;
+
 use super::{AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings, Settings};
 use std::error::Error;
 use std::fmt::{self, Display, Formatter};
@@ -88,6 +90,7 @@ impl Settings {
         let mut report = ValidationReport::default();
 
         self.network.validate(&mut report);
+        self.reflectors.validate(&mut report);
         self.output.validate(&mut report);
         self.observability.validate(&mut report);
         self.advanced_settings.validate(&mut report);
@@ -192,6 +195,45 @@ impl NetworkSettings {
     }
 }
 
+impl ReflectorSettings {
+    fn validate(&self, report: &mut ValidationReport) {
+        if self.num_reflectors == 0 {
+            report.error("reflectors.num_reflectors", "must be greater than 0");
+        }
+
+        if self.peer_reselection_time == 0 {
+            report.error("reflectors.peer_reselection_time", "must be greater than 0");
+        }
+
+        if self.files.is_empty() && self.entries.is_empty() {
+            report.error(
+                "reflectors",
+                "at least one reflector file or inline entry is required",
+            );
+        }
+
+        for (index, file) in self.files.iter().enumerate() {
+            if file.trim().is_empty() {
+                report.error(
+                    "reflectors.files",
+                    format!("entry {index} must not be empty"),
+                );
+            }
+        }
+
+        for (index, country) in self.countries.iter().enumerate() {
+            if country.len() != 2 || !country.bytes().all(|c| c.is_ascii_alphabetic()) {
+                report.error(
+                    "reflectors.countries",
+                    format!(
+                        "entry {index} ({country:?}) must be an ISO 3166-1 alpha-2 country code"
+                    ),
+                );
+            }
+        }
+    }
+}
+
 impl OutputSettings {
     fn validate(&self, report: &mut ValidationReport) {
         if self.suppress_statistics {
@@ -298,24 +340,6 @@ impl AdvancedSettings {
             self.tick_interval,
         );
 
-        if self.num_reflectors < 5 {
-            report.error(
-                "advanced_settings.num_reflectors",
-                format!("must be at least 5, got {}", self.num_reflectors,),
-            );
-        }
-
-        if self.peer_reselection_time == 0 {
-            report.error(
-                "advanced_settings.peer_reselection_time",
-                "must be greater than 0",
-            );
-        }
-
-        if self.reflector_list_file.trim().is_empty() {
-            report.error("advanced_settings.reflector_list_file", "must not be empty");
-        }
-
         if self.speed_hist_size == 0 {
             report.error(
                 "advanced_settings.speed_hist_size",