feat(measurement): refine measurement setup and IRTT engine behavior

src/controller.rs

@@ -50,6 +50,7 @@ impl Controller {
         active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
         endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
         source: MeasurementSource,
+        period: Duration,
     ) -> Self {
         Self {
             reflector_rx,
@@ -60,7 +61,7 @@ impl Controller {
             active_streams,
             endpoint_streams,
             source,
-            period: Duration::from_millis(500),
+            period,
             next_stream_id: 0,
             stream_ids: HashMap::new(),
         }

src/main.rs

@@ -174,16 +174,55 @@ fn resolve_reflectors(
         } else {
             0
         };
-        for endpoint in (definition.host.as_str(), port).to_socket_addrs()? {
+        let endpoints = match (definition.host.as_str(), port).to_socket_addrs() {
+            Ok(endpoints) => endpoints,
+            Err(error) => {
+                warn!("Failed to resolve reflector {}: {error}", definition.host);
+                continue;
+            }
+        };
+        for endpoint in endpoints {
             if protocol != ReflectorProtocol::Irtt && endpoint.is_ipv6() {
                 continue;
             }
-            resolved.insert(endpoint, definition.clone());
+            insert_reflector(&mut resolved, endpoint, definition.clone())?;
         }
     }
     Ok(resolved)
 }
 
+fn insert_reflector(
+    reflectors: &mut HashMap<SocketAddr, ReflectorDefinition>,
+    endpoint: SocketAddr,
+    definition: ReflectorDefinition,
+) -> anyhow::Result<()> {
+    if let Some(existing) = reflectors.get(&endpoint) {
+        let same_country = match (&existing.country, &definition.country) {
+            (Some(a), Some(b)) => a.eq_ignore_ascii_case(b),
+            (None, None) => true,
+            _ => false,
+        };
+        let same_protocols = existing
+            .protocols
+            .iter()
+            .all(|p| definition.protocols.contains(p))
+            && definition
+                .protocols
+                .iter()
+                .all(|p| existing.protocols.contains(p));
+        if !same_country
+            || !same_protocols
+            || existing.irtt_port.unwrap_or(2112) != definition.irtt_port.unwrap_or(2112)
+            || existing.irtt_hmac != definition.irtt_hmac
+        {
+            anyhow::bail!("conflicting reflector definitions for resolved endpoint {endpoint}");
+        }
+    } else {
+        reflectors.insert(endpoint, definition);
+    }
+    Ok(())
+}
+
 fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
     let protocol = settings.advanced_settings.measurement_type;
     let configured = resolve_reflectors(
@@ -194,11 +233,12 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
     )?;
     let bootstrap = resolve_reflectors(&settings.reflectors.bootstrap, protocol, &[], false)?;
     let mut reflectors = bootstrap.clone();
-    reflectors.extend(
+    for (endpoint, definition) in &configured {
-        configured
+        insert_reflector(&mut reflectors, *endpoint, definition.clone())?;
-            .iter()
+    }
-            .map(|(endpoint, definition)| (*endpoint, definition.clone())),
+    if reflectors.is_empty() {
-    );
+        anyhow::bail!("no usable reflectors for measurement protocol {protocol}");
+    }
     let mut pool = configured.keys().copied().collect::<Vec<_>>();
     pool.sort();
     let mut initial = bootstrap.keys().copied().collect::<Vec<_>>();
@@ -332,6 +372,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         Arc::clone(&active_streams),
         Arc::clone(&endpoint_streams),
         source,
+        Duration::from_secs_f64(settings.advanced_settings.tick_interval),
     );
     let icmp_engine = {
         let _guard = rt.enter();

src/measurement/irtt.rs

@@ -5,10 +5,9 @@ use irtt_client::{
     ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy,
     managed::{
         ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask,
-        ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError,
+        ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError,
-        ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent,
+        ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError,
-        ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason,
+        ManagedTargetConfig, ManagedTargetEndReason, TargetAuth, TargetId,
-        TargetAuth, TargetId,
     },
 };
 use log::{debug, trace, warn};
@@ -139,17 +138,34 @@ impl IrttEngine {
 
     pub async fn run(mut self, task: ManagedClientTask) -> Result<(), IrttError> {
         let mut task = Box::pin(task);
-        let mut events = self.client.subscribe()?;
+        let client = self.client.clone();
 
+        // Keep driving the client even while the event loop awaits command receipts.
+        let result = tokio::select! {
+            biased;
+            outcome = task.as_mut() => {
+                return match outcome.end_reason {
+                    ManagedEndReason::StopRequested => Ok(()),
+                    end_reason => Err(IrttError::ManagedFailure(end_reason)),
+                };
+            }
+            result = self.run_events() => result,
+        };
+
+        // Every event-loop exit stops and joins the driver, including channel errors.
+        let stop = client.stop();
+        let (outcome, _) = tokio::join!(task, stop);
+        result?;
+        match outcome.end_reason {
+            ManagedEndReason::StopRequested => Ok(()),
+            end_reason => Err(IrttError::ManagedFailure(end_reason)),
+        }
+    }
+
+    async fn run_events(&mut self) -> Result<(), IrttError> {
+        let mut events = self.client.subscribe()?;
         loop {
             tokio::select! {
-                outcome = task.as_mut() => {
-                    return match outcome.end_reason {
-                        ManagedEndReason::StopRequested => Ok(()),
-                        end_reason => Err(IrttError::ManagedFailure(end_reason)),
-                    };
-                }
-
                 bindings = self.binding_rx.recv_async() => {
                     let bindings = bindings
                         .map_err(|_| IrttError::BindingChannelClosed)?;
@@ -344,22 +360,7 @@ impl IrttEngine {
 
                 _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
                     if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) {
-                        let stop = self.client.stop();
+                        return Ok(());
-                        tokio::pin!(stop);
-                        tokio::select! {
-                            _ = &mut stop => {},
-                            outcome = task.as_mut() => {
-                                return match outcome.end_reason {
-                                    ManagedEndReason::StopRequested => Ok(()),
-                                    end_reason => Err(IrttError::ManagedFailure(end_reason)),
-                                };
-                            }
-                        }
-                        let outcome = task.await;
-                        return match outcome.end_reason {
-                            ManagedEndReason::StopRequested => Ok(()),
-                            end_reason => Err(IrttError::ManagedFailure(end_reason)),
-                        };
                     }
                 }
             }

src/settings/mod.rs

@@ -178,7 +178,22 @@ impl Settings {
         let builder = Config::builder();
 
         #[cfg(all(feature = "uci", unix))]
-        let builder = builder.add_source(UciSource::from_schema::<Self>().required(false));
+        let builder = builder.add_source(
+            UciSource::from_schema::<Self>()
+                .with_repeated_schema(&[
+                    uci_schema::UciRepeatedSectionMapping {
+                        uci_section: "reflector",
+                        config_path: "reflectors.entries",
+                        options: <ReflectorDefinition as uci_schema::UciSectionSchema>::UCI_OPTIONS,
+                    },
+                    uci_schema::UciRepeatedSectionMapping {
+                        uci_section: "bootstrap_reflector",
+                        config_path: "reflectors.bootstrap",
+                        options: <ReflectorDefinition as uci_schema::UciSectionSchema>::UCI_OPTIONS,
+                    },
+                ])
+                .required(false),
+        );
 
         #[cfg(feature = "toml")]
         let builder = builder.add_source(