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

README.md

@@ -57,3 +57,23 @@ 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.

src/baseliner.rs

@@ -289,6 +289,7 @@ 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/measurement/icmp.rs

@@ -326,6 +326,7 @@ impl IcmpEngine {
                     stream: probe.stream,
                     started_at: probe.sent_at,
                     observed_at,
+                    round_trip: None,
                     one_way: None,
                 })
             }
@@ -385,6 +386,7 @@ 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

@@ -7,7 +7,7 @@ use irtt_client::{
         ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask,
         ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError,
         ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError,
-        ManagedTargetConfig, ManagedTargetEndReason, TargetAuth, TargetId,
+        ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetLifecycle, TargetAuth, TargetId,
     },
 };
 use log::{debug, trace, warn};
@@ -81,6 +81,7 @@ pub struct IrttEngine {
     client: ManagedClientHandle,
 
     bindings: HashMap<MeasurementStreamId, IrttBinding>,
+    targets: Vec<ManagedTargetConfig>,
 
     origin: Instant,
 
@@ -104,6 +105,7 @@ impl IrttEngine {
             Self {
                 client: handle,
                 bindings: HashMap::new(),
+                targets: Vec::new(),
                 origin,
                 binding_rx,
                 observation_tx,
@@ -119,16 +121,17 @@ 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 {
+                if let Some(key) = binding.hmac.as_ref().filter(|key| !key.trim().is_empty()) {
                     target.auth = TargetAuth::Override(irtt_client::Authentication::Hmac(
                         irtt_client::HmacKey::new(key.as_bytes().to_vec().into_boxed_slice()),
                     ));
                 }
                 target
             })
-            .collect();
+            .collect::<Vec<_>>();
 
-        self.client.update_targets(targets)?.await?;
+        self.client.update_targets(targets.clone())?.await?;
+        self.targets = targets;
 
         self.bindings = bindings
             .into_iter()
@@ -166,6 +169,11 @@ 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() => {
@@ -175,6 +183,17 @@ 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,
@@ -255,6 +274,7 @@ impl IrttEngine {
                                     sent_at,
                                     received_at,
                                     one_way,
+                                    rtt,
                                     ..
                                 } => {
                                     let Some(stream) = stream else {
@@ -265,6 +285,8 @@ 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,6 +121,9 @@ 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>,
 }
@@ -131,7 +134,10 @@ impl MeasurementObservation {
     }
 
     pub fn rtt_ms(&self) -> f64 {
-        self.rtt().as_secs_f64() * 1000.0
+        self.round_trip
+            .map(SignedDuration::as_secs_f64)
+            .unwrap_or_else(|| self.rtt().as_secs_f64())
+            * 1000.0
     }
 }
 

src/settings/mod.rs

@@ -47,7 +47,7 @@ where
         type Value = bool;
 
         fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-            formatter.write_str("a boolean or a UCI-compatible boolean string")
+            formatter.write_str("a boolean, integer 0 or 1, or a UCI-compatible boolean string")
         }
 
         fn visit_bool<E>(self, value: bool) -> Result<Self::Value, E>
@@ -57,6 +57,28 @@ 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,
@@ -248,6 +270,7 @@ 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>,
 
@@ -450,3 +473,25 @@ 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",
+        )),
+    }
+}
+