fix(reflectors): retain responsive peers on partial reselection

Deduplicate compatible definitions and reject conflicting metadata or authentication within each list and across bootstrap and pool entries. Treat hostname aliases, protocol ordering, country-code case, and omitted default ports consistently.

.github/workflows/rust-clippy.yml

@@ -50,7 +50,7 @@ jobs:
           echo "exit_code=${clippy_status}" >> "$GITHUB_OUTPUT"
           exit 0
       - name: Upload Clippy SARIF
-        uses: github/codeql-action/upload-sarif@6f2ac81f5cd81fe351a197ba0b00d91d03cd2aee # v4
+        uses: github/codeql-action/upload-sarif@24c54180a607b1449ed407dd24f251e4e9147c8d # v4
         if: always()
         with:
           sarif_file: rust-clippy-results.sarif

Cargo.lock

@@ -867,9 +867,9 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.53.3"
+version = "1.53.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e"
+checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044"
 dependencies = [
  "bytes",
  "libc",

README.md

@@ -57,23 +57,3 @@ You can also build an OpenWrt package to install in your images with the SDK:
    # Copy the generated .ipk from bin/packages/... to your router
    opkg install sqm-autorate-rust_*.ipk
    ```
-
-## Reflector files and inline UCI reflectors
-
-Inline `config reflector` sections supplement the default
-`/etc/sqm-autorate-rust/reflectors.csv`. To use only inline reflectors, explicitly
-disable file loading:
-
-```uci
-config reflectors
-    option files 'none'
-
-config reflector
-    option host 'global.irtt.lochnair.net'
-    list protocols 'irtt'
-```
-
-Use `list files '/path/to/reflectors.csv'` instead of `option files 'none'` to
-combine CSV files with inline reflectors. Omitting `files` keeps the default CSV;
-UCI discards empty string options, so `option files ''` does not disable it.
-Absent or blank `irtt_hmac` values use unauthenticated IRTT sessions.

crates/uci-derive/src/lib.rs

@@ -82,7 +82,6 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
     let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl();
 
     Ok(quote! {
-        #[cfg(all(feature = "uci", unix))]
         #[automatically_derived]
         impl #impl_generics
             crate::settings::uci_schema::UciSectionSchema
@@ -156,7 +155,6 @@ fn expand_uci_config(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStrea
     let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl();
 
     Ok(quote! {
-        #[cfg(all(feature = "uci", unix))]
         #[automatically_derived]
         impl #impl_generics
             crate::settings::uci_schema::UciConfigSchema

src/baseliner.rs

@@ -289,7 +289,6 @@ mod tests {
             },
             started_at: received_at,
             observed_at: received_at,
-            round_trip: None,
             one_way: Some(OneWayLatency {
                 uplink: SignedDuration::from_millis(up_time as i64),
                 downlink: SignedDuration::from_millis(down_time as i64),

src/controller.rs

@@ -120,7 +120,9 @@ impl Controller {
                     icmp.push(IcmpBinding::new(stream, schedule))
                 }
                 #[cfg(feature = "irtt")]
-                MeasurementSource::Irtt => irtt.push(IrttBinding::new(stream, port, hmac)),
+                MeasurementSource::Irtt => {
+                    irtt.push(IrttBinding::new(stream, schedule, port, hmac))
+                }
                 #[cfg(not(feature = "irtt"))]
                 MeasurementSource::Irtt => unreachable!("IRTT feature is checked before startup"),
                 _ => unreachable!("unsupported reflector measurement source"),

src/main.rs

@@ -389,6 +389,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     let (irtt_task, irtt_engine) = {
         let _guard = rt.enter();
         IrttEngine::new(
+            start_time,
             Duration::from_secs_f64(settings.advanced_settings.tick_interval),
             &settings.network,
             measurement_tx,

src/measurement/icmp.rs

@@ -326,7 +326,6 @@ impl IcmpEngine {
                     stream: probe.stream,
                     started_at: probe.sent_at,
                     observed_at,
-                    round_trip: None,
                     one_way: None,
                 })
             }
@@ -386,7 +385,6 @@ impl IcmpEngine {
                     stream: probe.stream,
                     started_at: probe.sent_at,
                     observed_at,
-                    round_trip: None,
                     one_way: Some(OneWayLatency {
                         uplink: SignedDuration::from_millis(uplink_ms),
                         downlink: SignedDuration::from_millis(downlink_ms),

src/measurement/irtt.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, net::SocketAddr};
+use std::{collections::HashMap, net::SocketAddr, time::Instant};
 
 use flume::{Receiver, Sender};
 use irtt_client::{
@@ -7,7 +7,7 @@ use irtt_client::{
         ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask,
         ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError,
         ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError,
-        ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetLifecycle, TargetAuth, TargetId,
+        ManagedTargetConfig, ManagedTargetEndReason, TargetAuth, TargetId,
     },
 };
 use log::{debug, trace, warn};
@@ -17,7 +17,7 @@ use tokio::sync::broadcast::error::RecvError;
 use crate::SHUTDOWN;
 use crate::measurement::model::{
     MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream,
-    MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration,
+    MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule, SignedDuration,
 };
 use crate::settings::NetworkSettings;
 
@@ -53,13 +53,27 @@ pub enum IrttError {
 
 pub struct IrttBinding {
     stream: MeasurementStream,
+    /// Retained for future hybrid-mode scheduling. IRTT currently uses the
+    /// global interval configured on its managed client.
+    #[allow(dead_code)]
+    schedule: ProbeSchedule,
     port: u16,
     hmac: Option<String>,
 }
 
 impl IrttBinding {
-    pub fn new(stream: MeasurementStream, port: u16, hmac: Option<String>) -> Self {
+    pub fn new(
-        Self { stream, port, hmac }
+        stream: MeasurementStream,
+        schedule: ProbeSchedule,
+        port: u16,
+        hmac: Option<String>,
+    ) -> Self {
+        Self {
+            stream,
+            schedule,
+            port,
+            hmac,
+        }
     }
 }
 
@@ -67,7 +81,8 @@ pub struct IrttEngine {
     client: ManagedClientHandle,
 
     bindings: HashMap<MeasurementStreamId, IrttBinding>,
-    targets: Vec<ManagedTargetConfig>,
+
+    origin: Instant,
 
     binding_rx: Receiver<Vec<IrttBinding>>,
     observation_tx: Sender<MeasurementEvent>,
@@ -75,6 +90,7 @@ pub struct IrttEngine {
 
 impl IrttEngine {
     pub fn new(
+        origin: Instant,
         interval: std::time::Duration,
         settings: &NetworkSettings,
         observation_tx: Sender<MeasurementEvent>,
@@ -88,7 +104,7 @@ impl IrttEngine {
             Self {
                 client: handle,
                 bindings: HashMap::new(),
-                targets: Vec::new(),
+                origin,
                 binding_rx,
                 observation_tx,
             },
@@ -103,17 +119,16 @@ impl IrttEngine {
                     TargetId::new(binding.stream.id.0.to_string()),
                     SocketAddr::new(binding.stream.peer, binding.port).to_string(),
                 );
-                if let Some(key) = binding.hmac.as_ref().filter(|key| !key.trim().is_empty()) {
+                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::<Vec<_>>();
+            .collect();
 
-        self.client.update_targets(targets.clone())?.await?;
+        self.client.update_targets(targets)?.await?;
-        self.targets = targets;
 
         self.bindings = bindings
             .into_iter()
@@ -151,11 +166,6 @@ impl IrttEngine {
 
     async fn run_events(&mut self) -> Result<(), IrttError> {
         let mut events = self.client.subscribe()?;
-        // Durable status catches failures even if their presentation events were lost.
-        // Resubmitting the full desired set restarts terminal targets only.
-        let mut retry = tokio::time::interval(std::time::Duration::from_secs(5));
-        retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-        retry.tick().await;
         loop {
             tokio::select! {
                 bindings = self.binding_rx.recv_async() => {
@@ -165,17 +175,6 @@ impl IrttEngine {
                     self.set_bindings(bindings).await?;
                 }
 
-                _ = retry.tick() => {
-                    if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) {
-                        return Ok(());
-                    }
-                    if self.client.status().targets.iter().any(|target| {
-                        target.desired && target.lifecycle == ManagedTargetLifecycle::Terminal
-                    }) {
-                        self.client.update_targets(self.targets.clone())?.await?;
-                    }
-                }
-
                 event = events.recv() => {
                     let event = match event {
                         Ok(event) => event,
@@ -256,7 +255,6 @@ impl IrttEngine {
                                     sent_at,
                                     received_at,
                                     one_way,
-                                    rtt,
                                     ..
                                 } => {
                                     let Some(stream) = stream else {
@@ -267,8 +265,6 @@ impl IrttEngine {
                                         stream,
                                         started_at: sent_at.mono,
                                         observed_at: received_at.mono,
-                                        round_trip: i64::try_from(rtt.effective.as_nanos())
-                                            .ok().map(SignedDuration::from_nanos),
                                         one_way: one_way.and_then(map_one_way),
                                     };
 

src/measurement/model.rs

@@ -121,9 +121,6 @@ pub struct MeasurementObservation {
     /// Local monotonic time at which the result was observed.
     pub observed_at: Instant,
 
-    /// Measured round-trip latency when it differs from elapsed local time.
-    pub round_trip: Option<SignedDuration>,
-
     /// Directional timing, when the measurement mechanism can provide it.
     pub one_way: Option<OneWayLatency>,
 }
@@ -134,10 +131,7 @@ impl MeasurementObservation {
     }
 
     pub fn rtt_ms(&self) -> f64 {
-        self.round_trip
+        self.rtt().as_secs_f64() * 1000.0
-            .map(SignedDuration::as_secs_f64)
-            .unwrap_or_else(|| self.rtt().as_secs_f64())
-            * 1000.0
     }
 }
 

src/ratecontroller.rs

@@ -491,11 +491,8 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         self.state_dl.deltas = down_deltas;
         self.state_ul.deltas = up_deltas;
 
-        let minimum = usize::from(self.settings.reflectors.num_reflectors).min(reflectors.len());
+        let minimum = usize::from(self.settings.reflectors.num_reflectors);
-        if reflectors.is_empty()
+        if self.state_dl.deltas.len() < minimum || self.state_ul.deltas.len() < minimum {
-            || self.state_dl.deltas.len() < minimum
-            || self.state_ul.deltas.len() < minimum
-        {
             // trigger reselection
             warn!(
                 "Not enough delta values (D: {}, U: {}, need {}), triggering reselection",

src/settings/mod.rs

@@ -2,10 +2,12 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
+use anyhow::{Context, bail};
 use config::{Config, ConfigError, Environment};
 use log::Level;
 use serde::{Deserialize, Deserializer, de};
 use std::fmt::{self, Display};
+use std::net::IpAddr;
 use std::str::FromStr;
 use uci_derive::{UciConfig, UciSection};
 
@@ -15,7 +17,6 @@ use crate::settings::uci::UciSource;
 #[cfg(all(feature = "uci", unix))]
 pub mod uci;
 
-#[cfg(all(feature = "uci", unix))]
 pub(crate) mod uci_schema;
 mod validation;
 
@@ -46,7 +47,7 @@ where
         type Value = bool;
 
         fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-            formatter.write_str("a boolean, integer 0 or 1, or a UCI-compatible boolean string")
+            formatter.write_str("a boolean or a UCI-compatible boolean string")
         }
 
         fn visit_bool<E>(self, value: bool) -> Result<Self::Value, E>
@@ -56,28 +57,6 @@ where
             Ok(value)
         }
 
-        fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E>
-        where
-            E: de::Error,
-        {
-            match value {
-                0 => Ok(false),
-                1 => Ok(true),
-                _ => Err(E::custom("boolean integer must be 0 or 1")),
-            }
-        }
-
-        fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E>
-        where
-            E: de::Error,
-        {
-            match value {
-                0 => Ok(false),
-                1 => Ok(true),
-                _ => Err(E::custom("boolean integer must be 0 or 1")),
-            }
-        }
-
         fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
         where
             E: de::Error,
@@ -269,7 +248,6 @@ impl NetworkSettings {
 #[derive(Clone, Debug, Deserialize, UciSection)]
 #[serde(default)]
 pub(crate) struct ReflectorSettings {
-    #[serde(deserialize_with = "deserialize_reflector_files")]
     pub(crate) files: Vec<String>,
     pub(crate) countries: Vec<String>,
 
@@ -472,24 +450,3 @@ where
     let value = String::deserialize(deserializer)?;
     value.parse().map_err(de::Error::custom)
 }
-
-// UCI drops empty string options; `option files 'none'` disables file loading.
-fn deserialize_reflector_files<'de, D>(deserializer: D) -> Result<Vec<String>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    #[derive(Deserialize)]
-    #[serde(untagged)]
-    enum Files {
-        List(Vec<String>),
-        String(String),
-    }
-
-    match Files::deserialize(deserializer)? {
-        Files::List(files) => Ok(files),
-        Files::String(value) if value.trim().is_empty() || value.trim() == "none" => Ok(Vec::new()),
-        Files::String(_) => Err(de::Error::custom(
-            "reflector files must be a list, an empty string, or none",
-        )),
-    }
-}

src/settings/validation.rs

@@ -213,12 +213,6 @@ impl ReflectorSettings {
         }
 
         for (index, reflector) in self.entries.iter().enumerate() {
-            if reflector.host.trim().is_empty() {
-                report.error(
-                    "reflectors.entries",
-                    format!("entry {index} must have a nonempty host"),
-                );
-            }
             if reflector.protocols.is_empty() {
                 report.error(
                     "reflectors.entries",

src/time.rs

@@ -14,11 +14,16 @@ pub trait Clock {
 }
 
 pub enum Realtime {}
+pub enum Monotonic {}
 
 impl Clock for Realtime {
     const ID: ClockId = ClockId::Realtime;
 }
 
+impl Clock for Monotonic {
+    const ID: ClockId = ClockId::Monotonic;
+}
+
 pub struct ClockSample<C> {
     time_s: u64,
     time_ns: u64,
@@ -51,6 +56,10 @@ impl<C: Clock> ClockSample<C> {
     pub fn as_secs_f64(&self) -> f64 {
         self.time_s as f64 + self.time_ns as f64 / 1_000_000_000.0
     }
+
+    pub fn to_milliseconds(&self) -> u64 {
+        (self.time_s * 1000) + (self.time_ns / 1000000)
+    }
 }
 
 impl ClockSample<Realtime> {

src/util.rs

@@ -3,10 +3,20 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use anyhow::anyhow;
-use std::sync::{Arc, RwLock, RwLockReadGuard, RwLockWriteGuard};
+use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
 
 pub type ArcRwLock<T> = Arc<RwLock<T>>;
 
+pub trait MutexExt<T> {
+    fn lock_anyhow(&self) -> anyhow::Result<MutexGuard<'_, T>>;
+}
+
+impl<T> MutexExt<T> for Mutex<T> {
+    fn lock_anyhow(&self) -> anyhow::Result<MutexGuard<'_, T>> {
+        self.lock().map_err(|e| anyhow!("mutex poisoned: {e}"))
+    }
+}
+
 pub trait RwLockExt<T> {
     fn read_anyhow(&self) -> anyhow::Result<RwLockReadGuard<'_, T>>;
     fn write_anyhow(&self) -> anyhow::Result<RwLockWriteGuard<'_, T>>;