feat(measurement): add IRTT engine

Cargo.lock

@@ -4,9 +4,9 @@ version = 4
 
 [[package]]
 name = "aho-corasick"
-version = "1.1.4"
+version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
+checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
 dependencies = [
  "memchr",
 ]
@@ -56,9 +56,18 @@ dependencies = [
 
 [[package]]
 name = "bitflags"
-version = "2.13.1"
+version = "2.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da"
+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"
@@ -74,9 +83,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
 
 [[package]]
 name = "cc"
-version = "1.3.0"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8"
+checksum = "f74872d07caf508b30a21f6836e7d7016a2eaf7d9ff4f48deaa58cd8a0407630"
 dependencies = [
  "find-msvc-tools",
  "shlex 2.0.1",
@@ -93,9 +102,9 @@ dependencies = [
 
 [[package]]
 name = "cfg-if"
-version = "1.0.4"
+version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
+checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
 
 [[package]]
 name = "cfg_aliases"
@@ -105,9 +114,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"
 
 [[package]]
 name = "clang-sys"
-version = "1.8.1"
+version = "1.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4"
+checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a"
 dependencies = [
  "glob",
  "libc",
@@ -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"
@@ -157,10 +187,31 @@ dependencies = [
 ]
 
 [[package]]
-name = "either"
+name = "ctutils"
-version = "1.16.0"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e"
+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"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be"
 
 [[package]]
 name = "equivalent"
@@ -189,9 +240,9 @@ dependencies = [
 
 [[package]]
 name = "find-msvc-tools"
-version = "0.1.9"
+version = "0.1.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
+checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484"
 
 [[package]]
 name = "flume"
@@ -213,9 +264,9 @@ checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
 
 [[package]]
 name = "futures-sink"
-version = "0.3.32"
+version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
+checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d"
 
 [[package]]
 name = "getrandom"
@@ -232,9 +283,9 @@ dependencies = [
 
 [[package]]
 name = "glob"
-version = "0.3.3"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
+checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b"
 
 [[package]]
 name = "hashbrown"
@@ -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"
@@ -257,14 +326,39 @@ dependencies = [
 
 [[package]]
 name = "indexmap"
-version = "2.14.0"
+version = "2.14.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
 dependencies = [
  "equivalent",
  "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"
@@ -282,9 +376,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
 
 [[package]]
 name = "js-sys"
-version = "0.3.103"
+version = "0.3.106"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
+checksum = "7883d941dae510fb2d978fc3fe018c71c9e2892fd38854de3e8b92c2e5ad9cc5"
 dependencies = [
  "cfg-if",
  "wasm-bindgen",
@@ -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"
@@ -507,9 +611,9 @@ dependencies = [
 
 [[package]]
 name = "regex-automata"
-version = "0.4.16"
+version = "0.4.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad"
+checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -615,9 +719,9 @@ dependencies = [
 
 [[package]]
 name = "serde_spanned"
-version = "1.1.1"
+version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26"
+checksum = "4e7523beb55eece201a2356bee0bbca0d1ab466c14c07703b2e0ee6d42cb0c2c"
 dependencies = [
  "serde_core",
 ]
@@ -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"
@@ -784,9 +895,9 @@ dependencies = [
 
 [[package]]
 name = "toml"
-version = "1.1.3+spec-1.1.0"
+version = "1.1.7+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c"
+checksum = "7874cd34dd99040fc24e30923aa64a60a81ff3ddfe2b8129ccc84ba24949deed"
 dependencies = [
  "indexmap",
  "serde_core",
@@ -799,27 +910,27 @@ dependencies = [
 
 [[package]]
 name = "toml_datetime"
-version = "1.1.1+spec-1.1.0"
+version = "1.1.2+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7"
+checksum = "2b86d767906c6c42421dcba507eb9d203e779497710a47782a224bb871653053"
 dependencies = [
  "serde_core",
 ]
 
 [[package]]
 name = "toml_parser"
-version = "1.1.2+spec-1.1.0"
+version = "1.1.4+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526"
+checksum = "bed810ee007d8e7f9cf20b4774e7198142fa27186e0f805871da7f3c39eaf333"
 dependencies = [
  "winnow",
 ]
 
 [[package]]
 name = "toml_writer"
-version = "1.1.2+spec-1.1.0"
+version = "1.1.3+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2"
+checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770"
 
 [[package]]
 name = "trybuild"
@@ -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"
@@ -848,9 +965,9 @@ dependencies = [
 
 [[package]]
 name = "unicode-ident"
-version = "1.0.24"
+version = "1.0.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
+checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
 
 [[package]]
 name = "wasi"
@@ -860,9 +977,9 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
+checksum = "9bb54f33acc68fd454578d9820b0bde1a1a3d17aa17bb7b6595806d02886d409"
 dependencies = [
  "cfg-if",
  "once_cell",
@@ -873,9 +990,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
+checksum = "2e29d0c35b16e224a7eeb5cd2d25e3e1968fbd65604117b44d3b789d00ee8535"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -883,22 +1000,22 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
+checksum = "6f501a8bc3719dba86ef8ae4728879c08001bea749eb1333ac5b91e040e2a6b7"
 dependencies = [
  "bumpalo",
  "proc-macro2",
  "quote",
- "syn 2.0.119",
+ "syn 3.0.6",
  "wasm-bindgen-shared",
 ]
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
+checksum = "23f0c9c52aa7cd7d77769a4cfe2a9adb1b331f489a41d912ce14513d5ab995c6"
 dependencies = [
  "unicode-ident",
 ]

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.3" }
 rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "fcfedd204fb76928ef617bf9631b041c816e03d4", 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"]

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