fix(irtt): retry terminal targets without restarting healthy sessions

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",

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/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/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::{
@@ -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 {
-        Self { stream, port, hmac }
+    pub fn new(
+        stream: MeasurementStream,
+        schedule: ProbeSchedule,
+        port: u16,
+        hmac: Option<String>,
+    ) -> Self {
+        Self {
+            stream,
+            schedule,
+            port,
+            hmac,
+        }
     }
 }
 
@@ -69,12 +83,15 @@ pub struct IrttEngine {
     bindings: HashMap<MeasurementStreamId, IrttBinding>,
     targets: Vec<ManagedTargetConfig>,
 
+    origin: Instant,
+
     binding_rx: Receiver<Vec<IrttBinding>>,
     observation_tx: Sender<MeasurementEvent>,
 }
 
 impl IrttEngine {
     pub fn new(
+        origin: Instant,
         interval: std::time::Duration,
         settings: &NetworkSettings,
         observation_tx: Sender<MeasurementEvent>,
@@ -89,6 +106,7 @@ impl IrttEngine {
                 client: handle,
                 bindings: HashMap::new(),
                 targets: Vec::new(),
+                origin,
                 binding_rx,
                 observation_tx,
             },

src/ratecontroller.rs

@@ -492,10 +492,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         self.state_ul.deltas = up_deltas;
 
         let minimum = usize::from(self.settings.reflectors.num_reflectors).min(reflectors.len());
-        if reflectors.is_empty()
-            || self.state_dl.deltas.len() < minimum
-            || self.state_ul.deltas.len() < minimum
-        {
+        if 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;
 
@@ -493,3 +494,4 @@ where
         )),
     }
 }
+

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>>;