Merge branch 'feature/irtt'

Replace the ICMP-only inventory with protocol-aware records, remove Mullvad DNS entries and omit misleading anycast countries. Document IRTT authentication and explicit TOML/UCI reflector configuration, including disabling file loading.

Cargo.lock

@@ -60,6 +60,15 @@ version = "2.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 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"
 version = "3.20.3"
@@ -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"
@@ -156,6 +186,27 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "ctutils"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+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"
@@ -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"
@@ -265,6 +334,31 @@ dependencies = [
  "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"
@@ -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"
@@ -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"
@@ -756,9 +867,9 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.53.2"
+version = "1.53.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044"
+checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e"
 dependencies = [
  "bytes",
  "libc",
@@ -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"

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

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.

crates/uci-derive/src/lib.rs

@@ -9,13 +9,13 @@ use syn::{Data, DeriveInput, Field, Fields, FieldsNamed, LitStr, Token, parse_ma
 
 /// Generates `UciSectionSchema` from a settings struct.
 ///
-/// Every named field is mapped one-to-one:
+/// Every named field without `#[uci(skip)]` is mapped one-to-one:
 ///
 /// ```text
 /// Rust field name -> UCI option name
 /// Rust field name -> config-rs field name
 /// ```
-#[proc_macro_derive(UciSection)]
+#[proc_macro_derive(UciSection, attributes(uci))]
 pub fn derive_uci_section(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);
 
@@ -56,30 +56,33 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
     let struct_name = &input.ident;
     let fields = require_named_fields(input)?;
 
-    let mappings = fields
+    let mut mappings = Vec::with_capacity(fields.named.len());
-        .named
-        .iter()
-        .map(|field| {
-            let field_ident = field
-                .ident
-                .as_ref()
-                .expect("named fields always have identifiers");
 
-            let field_name = field_ident.unraw().to_string();
+    for field in &fields.named {
-            let field_name = LitStr::new(&field_name, field_ident.span());
+        if parse_skip(field)? {
+            continue;
+        }
 
-            quote! {
+        let field_ident = field
-                crate::settings::uci_schema::UciOptionMapping {
+            .ident
-                    uci_option: #field_name,
+            .as_ref()
-                    config_field: #field_name,
+            .expect("named fields always have identifiers");
-                }
+
+        let field_name = field_ident.unraw().to_string();
+        let field_name = LitStr::new(&field_name, field_ident.span());
+
+        mappings.push(quote! {
+            crate::settings::uci_schema::UciOptionMapping {
+                uci_option: #field_name,
+                config_field: #field_name,
             }
-        })
+        });
-        .collect::<Vec<_>>();
+    }
 
     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
@@ -95,6 +98,29 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
     })
 }
 
+fn parse_skip(field: &Field) -> syn::Result<bool> {
+    let mut skip = false;
+
+    for attribute in &field.attrs {
+        if !attribute.path().is_ident("uci") {
+            continue;
+        }
+
+        attribute.parse_nested_meta(|meta| {
+            if !meta.path.is_ident("skip") {
+                return Err(meta.error("expected `skip`"));
+            }
+            if skip {
+                return Err(meta.error("duplicate `skip` attribute"));
+            }
+            skip = true;
+            Ok(())
+        })?;
+    }
+
+    Ok(skip)
+}
+
 fn expand_uci_config(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
     let struct_name = &input.ident;
     let fields = require_named_fields(input)?;
@@ -130,6 +156,7 @@ 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

reflectors-icmp.csv

@@ -1,305 +0,0 @@
-reflector_ip,ip_version,description
-94.140.14.14,4,AdGuard
-94.140.15.15,4,AdGuard
-94.140.14.15,4,AdGuard Family
-94.140.15.16,4,AdGuard Family
-94.140.14.140,4,AdGuard Unfiltered
-94.140.14.141,4,AdGuard Unfiltered
-205.171.3.65,4,CenturyLink/Level3
-205.171.2.65,4,CenturyLink/Level3
-185.228.168.9,4,CleanBrowsing
-185.228.169.9,4,CleanBrowsing
-185.228.168.168,4,CleanBrowsing
-185.228.169.168,4,CleanBrowsing
-185.228.168.10,4,CleanBrowsing
-185.228.169.11,4,CleanBrowsing
-8.26.56.26,4,Comodo
-8.20.247.20,4,Comodo
-84.200.69.80,4,DNS.WATCH
-84.200.70.40,4,DNS.WATCH
-46.227.200.54,4,FAELIX Anycast
-46.227.200.55,4,FAELIX Anycast
-45.33.97.5,4,FreeDNS
-37.235.1.177,4,FreeDNS
-74.82.42.42,4,Hurricane Electric
-194.242.2.2,4,Mullvad
-194.242.2.3,4,Mullvad
-193.19.108.2,4,Mullvad
-193.19.108.3,4,Mullvad
-208.67.222.222,4,OpenDNS
-208.67.220.220,4,OpenDNS
-208.67.222.123,4,OpenDNS Family
-208.67.220.123,4,OpenDNS Family
-208.67.222.2,4,OpenDNS Sbx
-208.67.220.2,4,OpenDNS Sbx
-159.89.120.99,4,OpenNIC
-134.195.4.2,4,OpenNIC
-9.9.9.9,4,Quad9
-9.9.9.10,4,Quad9
-9.9.9.11,4,Quad9
-149.112.112.10,4,Quad9
-149.112.112.11,4,Quad9
-149.112.112.112,4,Quad9
-176.9.93.198,4,dnsforge.de
-176.9.1.117,4,dnsforge.de
-104.236.210.29,4,DNSReactor
-37.235.1.174,4,FreeDNS
-81.218.119.1,4,GreenTeamDNS
-194.8.57.12,4,Nuernberg Internet Exchange (N-IX)
-51.89.30.51,4,PublicDNS.eu
-51.89.93.25,4,PublicDNS.eu
-130.59.31.248,4,SWITCH
-130.59.31.251,4,SWITCH
-185.49.141.37,4,GetDNS
-158.64.1.29,4,Restena Foundation
-89.234.186.112,4,Neutopia
-185.95.218.42,4,Digitale Gesellschaft Schweiz
-185.95.218.43,4,Digitale Gesellschaft Schweiz
-140.238.215.192,4,post-factum
-199.58.81.218,4,CMRG DNS
-146.255.56.98,4,Foundation for Applied Privacy
-185.222.222.222,4,dns.sb
-185.184.222.222,4,dns.sb
-5.1.66.255,4,Freifunk München e.V.
-80.80.80.80,4,Freenom World
-80.80.81.81,4,Freenom World
-80.67.169.12,4,French Data Network (FDN)
-80.67.169.40,4,French Data Network (FDN)
-217.24.255.134,4,Albtelecom Sh.a.
-213.163.127.229,4,Telekomi i Kosoves SH.A.
-85.94.178.198,4,ANDORRA TELECOM SAU
-186.148.128.86,4,ETERNET S.R.L.
-190.151.144.21,4,Lima Video Cable S.A. (Cabletel)
-200.45.184.43,4,Telecom Argentina S.A.
-179.60.232.10,4,WICORP SA
-170.210.83.110,4,Red de Interconexion Universitaria
-190.57.234.194,4,Gigared S.A.
-186.153.180.148,4,Telecom Argentina S.A.
-81.16.8.110,4,Ucom CJSC
-45.133.105.123,4,GNC-Alfa CJSC
-61.8.0.113,4,Telstra Corporation Ltd
-103.224.162.40,4,Ezi-Web
-139.130.4.4,4,Telstra Corporation Ltd
-139.134.5.51,4,Telstra Corporation Ltd
-139.134.2.190,4,Telstra Corporation Ltd
-203.50.2.71,4,Telstra Corporation Ltd
-83.137.41.8,4,nemox.net Informationstechnologie OG
-83.137.41.9,4,nemox.net Informationstechnologie OG
-194.36.144.87,4,netcup GmbH
-188.118.227.21,4,Hutchison Drei Austria GmbH
-83.218.176.140,4,myNet GmbH
-85.132.85.85,4,Delta Telecom Ltd
-45.250.253.222,4,New Generation Internet Services Limited
-103.77.188.18,4,NETSCOPE
-103.77.188.19,4,NETSCOPE
-103.80.1.2,4,SS Online
-103.153.154.2,4,Pabna Cable Vision Dot Net
-114.130.5.5,4,Tire-1 IP Transit Provider of Bangladesh
-114.130.5.6,4,Tire-1 IP Transit Provider of Bangladesh
-103.146.221.20,4,Kloud Technologies Limited
-103.146.221.21,4,Kloud Technologies Limited
-115.127.19.129,4,BRACNet Limited
-115.127.19.137,4,BRACNet Limited
-103.113.200.10,4,National University
-177.131.114.86,4,ACESSOLINE TELECOMUNICACOES LTDA
-177.66.203.10,4,Downup Telecomunicacoes e servico LTDA
-177.184.131.180,4,SOBRALNET SERVICOS E TELECOMUNICACOES LTDA - ME
-186.194.224.82,4,R&R PROVEDOR DE INTERNET LTDA
-186.225.194.29,4,Pannet Serviços On Line Ltda
-189.42.239.34,4,CLARO S.A.
-200.99.138.94,4,TELEFÔNICA BRASIL S.A
-200.99.138.103,4,TELEFÔNICA BRASIL S.A
-200.167.191.114,4,CLARO S.A.
-200.169.8.1,4,Century Telecom Ltda
-141.0.102.202,4,Altibox AS
-62.97.223.62,4,BKK Digitek AS
-217.18.206.12,4,EVRY NORGE AS
-217.18.206.22,4,EVRY NORGE AS
-195.204.130.85,4,GLOBALCONNECT AS
-81.175.38.1,4,GLOBALCONNECT AS
-46.182.19.48,4,Martin Prager trading as NbIServ
-82.96.64.2,4,Probe Networks
-94.247.43.254,4,OpenNIC
-195.10.195.195,4,meerfarbig GmbH & Co. KG
-194.172.160.4,4,Baumann Technologie GmbH
-185.233.106.232,4,netcup GmbH
-185.233.107.4,4,netcup GmbH
-141.1.1.1,4,Vodafone Group PLC
-141.1.27.249,4,Vodafone Group PLC
-176.103.130.130,4,Serveroid LLC
-176.103.130.131,4,Serveroid LLC
-80.243.64.67,4,PJSC Vimpelcom
-195.218.133.88,4,PJSC Vimpelcom
-83.69.120.1,4,PJSC Vimpelcom
-46.16.229.223,4,JSC Elektrosvyaz
-193.183.98.66,4,Prometeus di Daniela Agro
-212.19.96.2,4,Leonet4cloud srl
-217.12.181.97,4,IRIDEOS S.P.A.
-62.149.128.4,4,Aruba S.p.A.
-62.149.132.2,4,Aruba S.p.A.
-85.47.198.82,4,Telecom Italia S.p.A.
-202.136.162.11,4,ASN - NTT SINGAPORE PTE LTD
-51.79.144.245,4,OVH SAS
-61.46.244.211,4,Jupiter Telecommunications Co. Ltd.
-1.33.184.173,4,NTT PC Communications Inc.
-1.33.184.116,4,NTT PC Communications Inc.
-1.33.184.174,4,NTT PC Communications Inc.
-1.33.184.108,4,NTT PC Communications Inc.
-1.33.184.220,4,NTT PC Communications Inc.
-1.33.184.230,4,NTT PC Communications Inc.
-1.33.184.158,4,NTT PC Communications Inc.
-62.212.113.125,4,Nerim SAS
-185.30.211.154,4,ENTREPRISE DECIMA SARL
-87.240.90.128,4,KERTEL SAS
-45.77.61.165,4,The Constant Company LLC
-178.62.57.141,4,DigitalOcean LLC
-194.145.240.6,4,M247 Ltd
-195.27.1.1,4,Vodafone Group PLC
-81.17.66.13,4,South West Communications Group Ltd
-87.117.196.200,4,IOMART CLOUD SERVICES LIMITED
-213.123.251.87,4,British Telecommunications PLC
-81.149.21.254,4,British Telecommunications PLC
-196.21.247.1,4,TENET (The UNINET Project)
-89.140.69.49,4,VODAFONE ONO S.A.
-62.97.123.107,4,COLT Technology Services Group Limited
-194.177.210.210,4,National Infrastructures for Research and Technology S.A.
-168.126.63.1,4,Korea Telecom
-121.254.134.99,4,LG DACOM Corporation
-203.239.131.1,4,ELIMNET INC.
-202.57.32.49,4,A Multihomed ISP Company
-81.17.31.34,4,Private Layer INC
-79.141.82.250,4,NTD SA
-80.254.77.39,4,Monzoon Networks AG
-85.93.212.2,4,Visual Online S.A.
-196.223.158.58,4,MAX NET FOR INTERNET SERVICES
-89.150.195.195,4,Media Network i Halmstad AB
-130.244.126.100,4,TELE2
-87.96.144.179,4,A3 Sverige AB
-146.247.80.144,4,A3 Sverige AB
-46.246.29.69,4,GleSYS AB
-203.210.237.165,4,VNPT Corp
-113.190.253.229,4,VNPT Corp
-119.29.29.29,4,Shenzhen Tencent Computer Systems Company Limited
-209.193.4.7,4,Alaska Communications
-209.112.160.2,4,Alaska Communications
-216.68.4.10,4,Cincinnati Bell ZoomTown
-216.68.5.10,4,Cincinnati Bell ZoomTown
-1.250.0.14,4,Verizon (Newark NJ)
-1.252.0.14,4,Verizon (Reston VA)
-1.252.0.12,4,Verizon (Reston VA)
-8.238.120.14,4,Verizon (Atlanta GA)
-8.238.120.12,4,Verizon (Atlanta GA)
-78.143.192.10,4,Fast (UK)
-201.220.136.186,4,CABLECOLOR S.A.
-190.124.166.8,4,Redes y Telecomunicaciones
-190.64.140.243,4,Administracion Nacional de Telecomunicaciones
-200.40.230.36,4,Administracion Nacional de Telecomunicaciones
-134.249.156.228,4,Kyivstar PJSC
-195.60.70.5,4,VAK Ltd.
-62.140.239.1,4,LLC TRC FIORD
-94.158.153.236,4,Comfo LTD
-109.254.13.190,4,Donbass Electronic Communications Ltd.
-176.67.0.60,4,LLC McLaut-Invest
-193.34.173.206,4,New Information Systems PP
-94.153.241.134,4,Kyivstar PJSC
-216.104.206.136,4,Africa Online Uganda
-154.72.212.38,4,National Information Technology Authority Uganda
-217.174.226.78,4,State Company of Electro Communications Turkmentelecom
-46.45.187.181,4,Radore Veri Merkezi Hizmetleri A.S.
-46.45.187.163,4,Radore Veri Merkezi Hizmetleri A.S.
-46.45.187.184,4,Radore Veri Merkezi Hizmetleri A.S.
-46.45.187.186,4,Radore Veri Merkezi Hizmetleri A.S.
-46.20.159.27,4,DoraTelekom
-93.113.60.84,4,Netinternet Bilisim Teknolojileri AS
-93.113.60.82,4,Netinternet Bilisim Teknolojileri AS
-196.3.132.153,4,Telecommunication Services of Trinidad and Tobago
-196.3.132.154,4,Telecommunication Services of Trinidad and Tobago
-80.248.72.1,4,CAFE Informatique et telecommunications
-117.121.215.102,4,UIH
-1.20.151.94,4,TOT Public Company Limited
-117.121.215.101,4,UIH
-122.154.230.146,4,The Communication Authoity of Thailand CAT
-101.101.101.101,4,Taiwan Network Information Center
-101.102.103.104,4,Taiwan Network Information Center
-113.196.55.130,4,New Century InfoComm Tech Co. Ltd.
-210.242.68.59,4,Data Communication Business Group
-92.60.50.40,4,BONET Systems s.r.o.
-217.73.17.110,4,VNET a.s.
-84.245.69.193,4,SWAN a.s.
-178.254.129.65,4,Drustvo za telekomunikacije Orion telekom doo Beograd-Zemun
-93.86.63.54,4,TELEKOM SRBIJA a.d.
-109.92.22.97,4,TELEKOM SRBIJA a.d.
-197.159.166.81,4,Companhia Santomense de Telecomunicacoes
-208.84.200.28,4,Columbus Communication St. Vincent and the Grenadines Ltd.
-208.94.176.225,4,Karib Cable
-199.116.57.181,4,IP SOLUTIONS INC.
-74.120.24.129,4,IP SOLUTIONS INC.
-74.120.24.81,4,IP SOLUTIONS INC.
-74.120.24.97,4,IP SOLUTIONS INC.
-78.130.38.222,4,NOS COMUNICACOES S.A.
-195.22.25.130,4,Claranet Limited
-213.241.88.98,4,Netia SA
-87.204.28.12,4,Netia SA
-212.160.85.232,4,Orange Polska Spolka Akcyjna
-45.173.200.10,4,INVERSIONES EN TELECOMUNICACIONES DIGITALES S.A.C
-200.10.231.110,4,Centro Nacional de Computacion
-201.222.48.206,4,TECNOLOGIA EN ELECTRONICA E INFORMATICA SOCIEDAD ANONIMA (T.E.I.S.A)
-201.222.50.30,4,TECNOLOGIA EN ELECTRONICA E INFORMATICA SOCIEDAD ANONIMA (T.E.I.S.A)
-201.222.55.246,4,TECNOLOGIA EN ELECTRONICA E INFORMATICA SOCIEDAD ANONIMA (T.E.I.S.A)
-103.209.52.250,4,The Professional Communications Pvt Ltd
-119.63.132.26,4,Transworld Associates (Pvt.) Ltd.
-83.143.8.249,4,MTN NIGERIA Communication limited
-165.98.138.194,4,Amnet Telecomunicaciones S.A.
-203.81.75.37,4,Myanma Posts and Telecommunications
-65.18.114.254,4,Telecom International Myanmar Co. Ltd
-65.18.117.174,4,Telecom International Myanmar Co. Ltd
-69.160.0.86,4,OOREDOO MYANMAR
-103.154.241.252,4,Myanmar Link Telecommunication Ltd
-66.181.167.232,4,The first E-commerce and TriplePlay Service ISP in Mongolia.
-202.21.99.42,4,Mobinet LLC. AS Mobinet Internet Service Provider
-202.131.241.74,4,Mobinet LLC. AS Mobinet Internet Service Provider
-124.158.93.2,4,STXCitinet. Leading Internet & VOIP Service Provider. Ulaanbaatar. Mongolia
-77.89.232.86,4,ORANGE MOLDOVA S.A.
-200.56.224.11,4,Marcatel Com S.A. de C.V.
-201.144.135.211,4,Marcatel Com S.A. de C.V.
-201.134.115.36,4,Uninet S.A. de C.V.
-1.32.58.251,4,TM Net. Internet Service Provider
-202.86.149.20,4,Companhia de Telecomunicacoes de Macau SARL
-202.175.45.2,4,Companhia de Telecomunicacoes de Macau SARL
-202.86.191.50,4,Companhia de Telecomunicacoes de Macau SARL
-80.250.60.241,4,Sia Nano IT
-101.78.12.50,4,Enterprise of Telecommunications Lao
-168.187.161.250,4,Zajil International Telecom Company W.L.L.
-62.12.114.25,4,Angani Limited
-103.134.220.158,4,PT Global Media Pratama Solusindo
-121.100.28.163,4,PT. Sakti Putra Mandiri
-103.28.114.57,4,Lintas Data Prima PT
-202.51.111.149,4,PT iForte Global Internet
-103.23.20.23,4,PT Infinys System Indonesia
-103.98.176.24,4,Universitas PGRI Semarang
-103.98.176.143,4,Universitas PGRI Semarang
-119.235.48.3,4,VAINAVI INDUSTIES LTD. INTERNET SERVICE PROVIDER. INDIA
-103.194.120.33,4,GOLDY DIGINET PRIVATE LIMITED
-103.194.120.3,4,GOLDY DIGINET PRIVATE LIMITED
-103.194.120.129,4,GOLDY DIGINET PRIVATE LIMITED
-223.31.121.171,4,Power Grid Corporation of India Limited
-169.38.73.5,4,SoftLayer Technologies Inc.
-202.86.251.201,4,Artech Infosystems Pvt Ltd
-202.164.44.246,4,Quadrant Televentures Limited
-87.229.99.1,4,SigmaNet Szolgaltato es Kereskedelm Ltd
-91.137.135.75,4,OPC Networks Kft.
-85.90.160.69,4,OPC Networks Kft.
-195.228.219.226,4,Magyar Telekom plc.
-103.6.168.114,4,NewMedia Express Pte Ltd
-45.71.185.100,4,NEDETEL S.A.
-186.149.243.52,4,ALTICE DOMINICANA S.A.
-193.17.47.1,4,CZ.NIC z.s.p.o.
-195.113.144.194,4,CZ.NIC z.s.p.o.
-213.211.50.1,4,itself s.r.o.
-213.211.50.2,4,itself s.r.o.
-185.43.135.1,4,CZ.NIC z.s.p.o.
-90.183.151.107,4,O2 Czech Republic. a.s.
-27.50.20.21,4,PT. Mora Telematika Indonesia
-116.193.190.164,4,PT Cloud Hosting Indonesia
-103.56.148.179,4,PT Beon Intermedia
-8.215.26.147,4,Alibaba.com Singapore E-Commerce Private Limited

reflectors.csv

@@ -0,0 +1,307 @@
+host,country,protocols,irtt_port,irtt_hmac
+94.140.14.14,,icmp-echo|icmp-timestamps,,
+94.140.15.15,,icmp-echo|icmp-timestamps,,
+94.140.14.15,,icmp-echo|icmp-timestamps,,
+94.140.15.16,,icmp-echo|icmp-timestamps,,
+94.140.14.140,,icmp-echo|icmp-timestamps,,
+94.140.14.141,,icmp-echo|icmp-timestamps,,
+205.171.3.65,,icmp-echo|icmp-timestamps,,
+205.171.2.65,,icmp-echo|icmp-timestamps,,
+185.228.168.9,,icmp-echo|icmp-timestamps,,
+185.228.169.9,,icmp-echo|icmp-timestamps,,
+185.228.168.168,,icmp-echo|icmp-timestamps,,
+185.228.169.168,,icmp-echo|icmp-timestamps,,
+185.228.168.10,,icmp-echo|icmp-timestamps,,
+185.228.169.11,,icmp-echo|icmp-timestamps,,
+8.26.56.26,,icmp-echo|icmp-timestamps,,
+8.20.247.20,,icmp-echo|icmp-timestamps,,
+84.200.69.80,DE,icmp-echo|icmp-timestamps,,
+84.200.70.40,DE,icmp-echo|icmp-timestamps,,
+46.227.200.54,GB,icmp-echo|icmp-timestamps,,
+46.227.200.55,,icmp-echo|icmp-timestamps,,
+45.33.97.5,US,icmp-echo|icmp-timestamps,,
+37.235.1.177,AT,icmp-echo|icmp-timestamps,,
+74.82.42.42,,icmp-echo|icmp-timestamps,,
+208.67.222.222,,icmp-echo|icmp-timestamps,,
+208.67.220.220,,icmp-echo|icmp-timestamps,,
+208.67.222.123,,icmp-echo|icmp-timestamps,,
+208.67.220.123,,icmp-echo|icmp-timestamps,,
+208.67.222.2,,icmp-echo|icmp-timestamps,,
+208.67.220.2,,icmp-echo|icmp-timestamps,,
+159.89.120.99,CA,icmp-echo|icmp-timestamps,,
+134.195.4.2,,icmp-echo|icmp-timestamps,,
+9.9.9.9,,icmp-echo|icmp-timestamps,,
+9.9.9.10,,icmp-echo|icmp-timestamps,,
+9.9.9.11,,icmp-echo|icmp-timestamps,,
+149.112.112.10,,icmp-echo|icmp-timestamps,,
+149.112.112.11,,icmp-echo|icmp-timestamps,,
+149.112.112.112,,icmp-echo|icmp-timestamps,,
+176.9.93.198,DE,icmp-echo|icmp-timestamps,,
+176.9.1.117,DE,icmp-echo|icmp-timestamps,,
+104.236.210.29,US,icmp-echo|icmp-timestamps,,
+37.235.1.174,AT,icmp-echo|icmp-timestamps,,
+81.218.119.1,IL,icmp-echo|icmp-timestamps,,
+194.8.57.12,DE,icmp-echo|icmp-timestamps,,
+51.89.30.51,DE,icmp-echo|icmp-timestamps,,
+51.89.93.25,DE,icmp-echo|icmp-timestamps,,
+130.59.31.248,CH,icmp-echo|icmp-timestamps,,
+130.59.31.251,CH,icmp-echo|icmp-timestamps,,
+185.49.141.37,NL,icmp-echo|icmp-timestamps,,
+158.64.1.29,LU,icmp-echo|icmp-timestamps,,
+89.234.186.112,FR,icmp-echo|icmp-timestamps,,
+185.95.218.42,CH,icmp-echo|icmp-timestamps,,
+185.95.218.43,CH,icmp-echo|icmp-timestamps,,
+140.238.215.192,CH,icmp-echo|icmp-timestamps,,
+199.58.81.218,CA,icmp-echo|icmp-timestamps,,
+146.255.56.98,AT,icmp-echo|icmp-timestamps,,
+185.222.222.222,,icmp-echo|icmp-timestamps,,
+185.184.222.222,NL,icmp-echo|icmp-timestamps,,
+5.1.66.255,AT,icmp-echo|icmp-timestamps,,
+80.80.80.80,,icmp-echo|icmp-timestamps,,
+80.80.81.81,,icmp-echo|icmp-timestamps,,
+80.67.169.12,FR,icmp-echo|icmp-timestamps,,
+80.67.169.40,FR,icmp-echo|icmp-timestamps,,
+217.24.255.134,AL,icmp-echo|icmp-timestamps,,
+213.163.127.229,XK,icmp-echo|icmp-timestamps,,
+85.94.178.198,AD,icmp-echo|icmp-timestamps,,
+186.148.128.86,AR,icmp-echo|icmp-timestamps,,
+190.151.144.21,AR,icmp-echo|icmp-timestamps,,
+200.45.184.43,AR,icmp-echo|icmp-timestamps,,
+179.60.232.10,AR,icmp-echo|icmp-timestamps,,
+170.210.83.110,AR,icmp-echo|icmp-timestamps,,
+190.57.234.194,AR,icmp-echo|icmp-timestamps,,
+186.153.180.148,AR,icmp-echo|icmp-timestamps,,
+81.16.8.110,AM,icmp-echo|icmp-timestamps,,
+45.133.105.123,AM,icmp-echo|icmp-timestamps,,
+61.8.0.113,AU,icmp-echo|icmp-timestamps,,
+103.224.162.40,AU,icmp-echo|icmp-timestamps,,
+139.130.4.4,,icmp-echo|icmp-timestamps,,
+139.134.5.51,AU,icmp-echo|icmp-timestamps,,
+139.134.2.190,AU,icmp-echo|icmp-timestamps,,
+203.50.2.71,,icmp-echo|icmp-timestamps,,
+83.137.41.8,AT,icmp-echo|icmp-timestamps,,
+83.137.41.9,AT,icmp-echo|icmp-timestamps,,
+194.36.144.87,DE,icmp-echo|icmp-timestamps,,
+188.118.227.21,AT,icmp-echo|icmp-timestamps,,
+83.218.176.140,AT,icmp-echo|icmp-timestamps,,
+85.132.85.85,AZ,icmp-echo|icmp-timestamps,,
+45.250.253.222,US,icmp-echo|icmp-timestamps,,
+103.77.188.18,BD,icmp-echo|icmp-timestamps,,
+103.77.188.19,BD,icmp-echo|icmp-timestamps,,
+103.80.1.2,BD,icmp-echo|icmp-timestamps,,
+103.153.154.2,BD,icmp-echo|icmp-timestamps,,
+114.130.5.5,BD,icmp-echo|icmp-timestamps,,
+114.130.5.6,BD,icmp-echo|icmp-timestamps,,
+103.146.221.20,BD,icmp-echo|icmp-timestamps,,
+103.146.221.21,BD,icmp-echo|icmp-timestamps,,
+115.127.19.129,BD,icmp-echo|icmp-timestamps,,
+115.127.19.137,BD,icmp-echo|icmp-timestamps,,
+103.113.200.10,BD,icmp-echo|icmp-timestamps,,
+177.131.114.86,BR,icmp-echo|icmp-timestamps,,
+177.66.203.10,BR,icmp-echo|icmp-timestamps,,
+177.184.131.180,BR,icmp-echo|icmp-timestamps,,
+186.194.224.82,BR,icmp-echo|icmp-timestamps,,
+186.225.194.29,BR,icmp-echo|icmp-timestamps,,
+189.42.239.34,BR,icmp-echo|icmp-timestamps,,
+200.99.138.94,BR,icmp-echo|icmp-timestamps,,
+200.99.138.103,BR,icmp-echo|icmp-timestamps,,
+200.167.191.114,BR,icmp-echo|icmp-timestamps,,
+200.169.8.1,BR,icmp-echo|icmp-timestamps,,
+141.0.102.202,NO,icmp-echo|icmp-timestamps,,
+62.97.223.62,NO,icmp-echo|icmp-timestamps,,
+217.18.206.12,NO,icmp-echo|icmp-timestamps,,
+217.18.206.22,NO,icmp-echo|icmp-timestamps,,
+195.204.130.85,NO,icmp-echo|icmp-timestamps,,
+81.175.38.1,NO,icmp-echo|icmp-timestamps,,
+46.182.19.48,DE,icmp-echo|icmp-timestamps,,
+82.96.64.2,DE,icmp-echo|icmp-timestamps,,
+94.247.43.254,DE,icmp-echo|icmp-timestamps,,
+195.10.195.195,DE,icmp-echo|icmp-timestamps,,
+194.172.160.4,DE,icmp-echo|icmp-timestamps,,
+185.233.106.232,DE,icmp-echo|icmp-timestamps,,
+185.233.107.4,DE,icmp-echo|icmp-timestamps,,
+141.1.1.1,,icmp-echo|icmp-timestamps,,
+141.1.27.249,,icmp-echo|icmp-timestamps,,
+176.103.130.130,,icmp-echo|icmp-timestamps,,
+176.103.130.131,,icmp-echo|icmp-timestamps,,
+80.243.64.67,RU,icmp-echo|icmp-timestamps,,
+195.218.133.88,RU,icmp-echo|icmp-timestamps,,
+83.69.120.1,RU,icmp-echo|icmp-timestamps,,
+46.16.229.223,RU,icmp-echo|icmp-timestamps,,
+193.183.98.66,IT,icmp-echo|icmp-timestamps,,
+212.19.96.2,IT,icmp-echo|icmp-timestamps,,
+217.12.181.97,IT,icmp-echo|icmp-timestamps,,
+62.149.128.4,IT,icmp-echo|icmp-timestamps,,
+62.149.132.2,IT,icmp-echo|icmp-timestamps,,
+85.47.198.82,IT,icmp-echo|icmp-timestamps,,
+202.136.162.11,SG,icmp-echo|icmp-timestamps,,
+51.79.144.245,SG,icmp-echo|icmp-timestamps,,
+61.46.244.211,JP,icmp-echo|icmp-timestamps,,
+1.33.184.173,JP,icmp-echo|icmp-timestamps,,
+1.33.184.116,JP,icmp-echo|icmp-timestamps,,
+1.33.184.174,JP,icmp-echo|icmp-timestamps,,
+1.33.184.108,JP,icmp-echo|icmp-timestamps,,
+1.33.184.220,JP,icmp-echo|icmp-timestamps,,
+1.33.184.230,JP,icmp-echo|icmp-timestamps,,
+1.33.184.158,JP,icmp-echo|icmp-timestamps,,
+62.212.113.125,FR,icmp-echo|icmp-timestamps,,
+185.30.211.154,FR,icmp-echo|icmp-timestamps,,
+87.240.90.128,FR,icmp-echo|icmp-timestamps,,
+45.77.61.165,FR,icmp-echo|icmp-timestamps,,
+178.62.57.141,GB,icmp-echo|icmp-timestamps,,
+194.145.240.6,GB,icmp-echo|icmp-timestamps,,
+195.27.1.1,,icmp-echo|icmp-timestamps,,
+81.17.66.13,GB,icmp-echo|icmp-timestamps,,
+87.117.196.200,GB,icmp-echo|icmp-timestamps,,
+213.123.251.87,GB,icmp-echo|icmp-timestamps,,
+81.149.21.254,GB,icmp-echo|icmp-timestamps,,
+196.21.247.1,ZA,icmp-echo|icmp-timestamps,,
+89.140.69.49,ES,icmp-echo|icmp-timestamps,,
+62.97.123.107,ES,icmp-echo|icmp-timestamps,,
+194.177.210.210,GR,icmp-echo|icmp-timestamps,,
+168.126.63.1,KR,icmp-echo|icmp-timestamps,,
+121.254.134.99,KR,icmp-echo|icmp-timestamps,,
+203.239.131.1,KR,icmp-echo|icmp-timestamps,,
+202.57.32.49,PH,icmp-echo|icmp-timestamps,,
+81.17.31.34,CH,icmp-echo|icmp-timestamps,,
+79.141.82.250,CH,icmp-echo|icmp-timestamps,,
+80.254.77.39,CH,icmp-echo|icmp-timestamps,,
+85.93.212.2,LU,icmp-echo|icmp-timestamps,,
+196.223.158.58,SD,icmp-echo|icmp-timestamps,,
+89.150.195.195,HK,icmp-echo|icmp-timestamps,,
+130.244.126.100,SE,icmp-echo|icmp-timestamps,,
+87.96.144.179,SE,icmp-echo|icmp-timestamps,,
+146.247.80.144,SE,icmp-echo|icmp-timestamps,,
+46.246.29.69,SE,icmp-echo|icmp-timestamps,,
+203.210.237.165,VN,icmp-echo|icmp-timestamps,,
+113.190.253.229,VN,icmp-echo|icmp-timestamps,,
+119.29.29.29,,icmp-echo|icmp-timestamps,,
+209.193.4.7,US,icmp-echo|icmp-timestamps,,
+209.112.160.2,US,icmp-echo|icmp-timestamps,,
+216.68.4.10,US,icmp-echo|icmp-timestamps,,
+216.68.5.10,US,icmp-echo|icmp-timestamps,,
+1.250.0.14,KR,icmp-echo|icmp-timestamps,,
+1.252.0.14,KR,icmp-echo|icmp-timestamps,,
+1.252.0.12,KR,icmp-echo|icmp-timestamps,,
+8.238.120.14,US,icmp-echo|icmp-timestamps,,
+8.238.120.12,US,icmp-echo|icmp-timestamps,,
+78.143.192.10,GB,icmp-echo|icmp-timestamps,,
+201.220.136.186,HN,icmp-echo|icmp-timestamps,,
+190.124.166.8,HN,icmp-echo|icmp-timestamps,,
+190.64.140.243,UY,icmp-echo|icmp-timestamps,,
+200.40.230.36,UY,icmp-echo|icmp-timestamps,,
+134.249.156.228,UA,icmp-echo|icmp-timestamps,,
+195.60.70.5,UA,icmp-echo|icmp-timestamps,,
+62.140.239.1,RU,icmp-echo|icmp-timestamps,,
+94.158.153.236,UA,icmp-echo|icmp-timestamps,,
+109.254.13.190,UA,icmp-echo|icmp-timestamps,,
+176.67.0.60,UA,icmp-echo|icmp-timestamps,,
+193.34.173.206,UA,icmp-echo|icmp-timestamps,,
+94.153.241.134,UA,icmp-echo|icmp-timestamps,,
+216.104.206.136,UG,icmp-echo|icmp-timestamps,,
+154.72.212.38,UG,icmp-echo|icmp-timestamps,,
+217.174.226.78,TM,icmp-echo|icmp-timestamps,,
+46.45.187.181,TR,icmp-echo|icmp-timestamps,,
+46.45.187.163,TR,icmp-echo|icmp-timestamps,,
+46.45.187.184,TR,icmp-echo|icmp-timestamps,,
+46.45.187.186,TR,icmp-echo|icmp-timestamps,,
+46.20.159.27,TR,icmp-echo|icmp-timestamps,,
+93.113.60.84,TR,icmp-echo|icmp-timestamps,,
+93.113.60.82,TR,icmp-echo|icmp-timestamps,,
+196.3.132.153,TT,icmp-echo|icmp-timestamps,,
+196.3.132.154,TT,icmp-echo|icmp-timestamps,,
+80.248.72.1,TG,icmp-echo|icmp-timestamps,,
+117.121.215.102,TH,icmp-echo|icmp-timestamps,,
+1.20.151.94,TH,icmp-echo|icmp-timestamps,,
+117.121.215.101,TH,icmp-echo|icmp-timestamps,,
+122.154.230.146,TH,icmp-echo|icmp-timestamps,,
+101.101.101.101,TW,icmp-echo|icmp-timestamps,,
+101.102.103.104,TW,icmp-echo|icmp-timestamps,,
+113.196.55.130,TW,icmp-echo|icmp-timestamps,,
+210.242.68.59,TW,icmp-echo|icmp-timestamps,,
+92.60.50.40,SK,icmp-echo|icmp-timestamps,,
+217.73.17.110,SK,icmp-echo|icmp-timestamps,,
+84.245.69.193,SK,icmp-echo|icmp-timestamps,,
+178.254.129.65,RS,icmp-echo|icmp-timestamps,,
+93.86.63.54,RS,icmp-echo|icmp-timestamps,,
+109.92.22.97,RS,icmp-echo|icmp-timestamps,,
+197.159.166.81,ST,icmp-echo|icmp-timestamps,,
+208.84.200.28,VC,icmp-echo|icmp-timestamps,,
+208.94.176.225,LC,icmp-echo|icmp-timestamps,,
+199.116.57.181,PR,icmp-echo|icmp-timestamps,,
+74.120.24.129,PR,icmp-echo|icmp-timestamps,,
+74.120.24.81,PR,icmp-echo|icmp-timestamps,,
+74.120.24.97,PR,icmp-echo|icmp-timestamps,,
+78.130.38.222,PT,icmp-echo|icmp-timestamps,,
+195.22.25.130,PT,icmp-echo|icmp-timestamps,,
+213.241.88.98,PL,icmp-echo|icmp-timestamps,,
+87.204.28.12,PL,icmp-echo|icmp-timestamps,,
+212.160.85.232,PL,icmp-echo|icmp-timestamps,,
+45.173.200.10,PE,icmp-echo|icmp-timestamps,,
+200.10.231.110,PY,icmp-echo|icmp-timestamps,,
+201.222.48.206,PY,icmp-echo|icmp-timestamps,,
+201.222.50.30,PY,icmp-echo|icmp-timestamps,,
+201.222.55.246,PY,icmp-echo|icmp-timestamps,,
+103.209.52.250,PK,icmp-echo|icmp-timestamps,,
+119.63.132.26,PK,icmp-echo|icmp-timestamps,,
+83.143.8.249,NG,icmp-echo|icmp-timestamps,,
+165.98.138.194,NI,icmp-echo|icmp-timestamps,,
+203.81.75.37,MM,icmp-echo|icmp-timestamps,,
+65.18.114.254,MM,icmp-echo|icmp-timestamps,,
+65.18.117.174,MM,icmp-echo|icmp-timestamps,,
+69.160.0.86,MM,icmp-echo|icmp-timestamps,,
+103.154.241.252,MM,icmp-echo|icmp-timestamps,,
+66.181.167.232,MN,icmp-echo|icmp-timestamps,,
+202.21.99.42,MN,icmp-echo|icmp-timestamps,,
+202.131.241.74,MN,icmp-echo|icmp-timestamps,,
+124.158.93.2,SG,icmp-echo|icmp-timestamps,,
+77.89.232.86,MD,icmp-echo|icmp-timestamps,,
+200.56.224.11,MX,icmp-echo|icmp-timestamps,,
+201.144.135.211,MX,icmp-echo|icmp-timestamps,,
+201.134.115.36,MX,icmp-echo|icmp-timestamps,,
+1.32.58.251,MY,icmp-echo|icmp-timestamps,,
+202.86.149.20,MO,icmp-echo|icmp-timestamps,,
+202.175.45.2,MO,icmp-echo|icmp-timestamps,,
+202.86.191.50,MO,icmp-echo|icmp-timestamps,,
+80.250.60.241,LV,icmp-echo|icmp-timestamps,,
+101.78.12.50,LA,icmp-echo|icmp-timestamps,,
+168.187.161.250,KW,icmp-echo|icmp-timestamps,,
+62.12.114.25,KE,icmp-echo|icmp-timestamps,,
+103.134.220.158,ID,icmp-echo|icmp-timestamps,,
+121.100.28.163,ID,icmp-echo|icmp-timestamps,,
+103.28.114.57,ID,icmp-echo|icmp-timestamps,,
+202.51.111.149,ID,icmp-echo|icmp-timestamps,,
+103.23.20.23,ID,icmp-echo|icmp-timestamps,,
+103.98.176.24,ID,icmp-echo|icmp-timestamps,,
+103.98.176.143,ID,icmp-echo|icmp-timestamps,,
+119.235.48.3,IN,icmp-echo|icmp-timestamps,,
+103.194.120.33,IN,icmp-echo|icmp-timestamps,,
+103.194.120.3,IN,icmp-echo|icmp-timestamps,,
+103.194.120.129,IN,icmp-echo|icmp-timestamps,,
+223.31.121.171,IN,icmp-echo|icmp-timestamps,,
+169.38.73.5,IN,icmp-echo|icmp-timestamps,,
+202.86.251.201,IN,icmp-echo|icmp-timestamps,,
+202.164.44.246,IN,icmp-echo|icmp-timestamps,,
+87.229.99.1,HU,icmp-echo|icmp-timestamps,,
+91.137.135.75,HU,icmp-echo|icmp-timestamps,,
+85.90.160.69,HU,icmp-echo|icmp-timestamps,,
+195.228.219.226,HU,icmp-echo|icmp-timestamps,,
+103.6.168.114,SG,icmp-echo|icmp-timestamps,,
+45.71.185.100,EC,icmp-echo|icmp-timestamps,,
+186.149.243.52,DO,icmp-echo|icmp-timestamps,,
+193.17.47.1,CZ,icmp-echo|icmp-timestamps,,
+195.113.144.194,CZ,icmp-echo|icmp-timestamps,,
+213.211.50.1,CZ,icmp-echo|icmp-timestamps,,
+213.211.50.2,CZ,icmp-echo|icmp-timestamps,,
+185.43.135.1,CZ,icmp-echo|icmp-timestamps,,
+90.183.151.107,CZ,icmp-echo|icmp-timestamps,,
+27.50.20.21,ID,icmp-echo|icmp-timestamps,,
+116.193.190.164,ID,icmp-echo|icmp-timestamps,,
+103.56.148.179,ID,icmp-echo|icmp-timestamps,,
+8.215.26.147,ID,icmp-echo|icmp-timestamps,,
+global.irtt.lochnair.net,,irtt,2112,
+uk-cov1.irtt.lochnair.net,GB,irtt,2112,
+uk-cov2.irtt.lochnair.net,GB,irtt,2112,
+de-fra.irtt.lochnair.net,DE,irtt,2112,
+nl-ams1.irtt.lochnair.net,NL,irtt,2112,
+netperf-eu.bufferbloat.net,SE,irtt,2112,

src/baseliner.rs

@@ -6,13 +6,12 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use crate::SHUTDOWN;
-use crate::measurement::model::{MeasurementEvent, MeasurementObservation};
+use crate::measurement::model::{MeasurementEvent, MeasurementObservation, MeasurementStream};
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
 use flume::{Receiver, Sender};
 use log::{debug, info};
 use std::collections::HashMap;
-use std::net::IpAddr;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 use std::time::Instant;
@@ -72,7 +71,7 @@ impl ReflectorState {
 #[derive(Clone, Debug)]
 pub struct ControlSnapshot {
     pub generated_at: Instant,
-    pub reflectors: HashMap<IpAddr, ReflectorState>,
+    pub reflectors: HashMap<MeasurementStream, ReflectorState>,
 }
 
 pub struct Baseliner {
@@ -121,7 +120,7 @@ fn update_reflector_state(
 }
 
 fn process_reply(
-    reflectors: &mut HashMap<IpAddr, ReflectorState>,
+    reflectors: &mut HashMap<MeasurementStream, ReflectorState>,
     reply: &MeasurementObservation,
     start_time: Instant,
     slow_factor: f64,
@@ -129,7 +128,7 @@ fn process_reply(
 ) -> (ReflectorState, bool) {
     let sample = EwmaStats::from_reply(reply);
     let state = reflectors
-        .entry(reply.stream.peer)
+        .entry(reply.stream)
         .or_insert_with(|| ReflectorState::new(sample, reply.observed_at));
     let anomaly = update_reflector_state(
         state,
@@ -144,7 +143,7 @@ fn process_reply(
 }
 
 fn control_snapshot(
-    reflectors: &HashMap<IpAddr, ReflectorState>,
+    reflectors: &HashMap<MeasurementStream, ReflectorState>,
     generated_at: Instant,
 ) -> ControlSnapshot {
     ControlSnapshot {
@@ -155,7 +154,7 @@ fn control_snapshot(
 
 impl Baseliner {
     pub fn run(&self) -> anyhow::Result<()> {
-        let mut reflectors = HashMap::<IpAddr, ReflectorState>::new();
+        let mut reflectors = HashMap::<MeasurementStream, ReflectorState>::new();
 
         let slow_factor = ewma_factor(
             self.settings.advanced_settings.tick_interval,
@@ -249,6 +248,7 @@ mod tests {
         MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency,
         SignedDuration,
     };
+    use std::net::IpAddr;
     use std::time::Duration;
 
     fn state(
@@ -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),
@@ -434,7 +435,8 @@ mod tests {
         let received_at = start + Duration::from_secs(1);
         let reflector = "192.0.2.1".parse().unwrap();
         let mut reflectors = HashMap::new();
-        reflectors.insert(reflector, state(100.0, 100.0, 100.0, 100.0, start));
+        let stream = reply(reflector, 0.0, 0.0, start).stream;
+        reflectors.insert(stream, state(100.0, 100.0, 100.0, 100.0, start));
 
         let (updated, anomaly) = process_reply(
             &mut reflectors,
@@ -450,7 +452,7 @@ mod tests {
         assert_close(updated.recent.down, 150.0);
         assert_close(updated.recent.up, 200.0);
         assert_eq!(updated.last_receive_at, received_at);
-        assert_close(reflectors[&reflector].recent.up, 200.0);
+        assert_close(reflectors[&stream].recent.up, 200.0);
     }
 
     #[test]
@@ -469,14 +471,15 @@ mod tests {
             0.5,
         );
         let snapshot = control_snapshot(&reflectors, generated_at);
+        let stream = reply(reflector, 0.0, 0.0, received_at).stream;
 
         assert!(!anomaly);
         assert_eq!(snapshot.generated_at, generated_at);
         assert_close(
-            snapshot.reflectors[&reflector].baseline.down,
+            snapshot.reflectors[&stream].baseline.down,
             updated.baseline.down,
         );
-        assert_close(snapshot.reflectors[&reflector].recent.up, updated.recent.up);
+        assert_close(snapshot.reflectors[&stream].recent.up, updated.recent.up);
-        assert_eq!(snapshot.reflectors[&reflector].last_receive_at, received_at);
+        assert_eq!(snapshot.reflectors[&stream].last_receive_at, received_at);
     }
 }

src/controller.rs

@@ -1,100 +1,150 @@
-use std::{collections::HashMap, net::IpAddr, time::Duration};
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    sync::{Arc, RwLock},
+    time::Duration,
+};
 
 use flume::{Receiver, SendError, Sender};
 use thiserror::Error;
 
-use crate::measurement::{
+#[cfg(feature = "irtt")]
-    icmp::IcmpBinding,
+use crate::measurement::irtt::IrttBinding;
-    model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule},
+use crate::{
+    measurement::{
+        icmp::IcmpBinding,
+        model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule},
+    },
+    settings::ReflectorDefinition,
 };
 
 #[derive(Debug, Error)]
 pub enum ControllerError {
     #[error("ICMP channel error")]
     IcmpChannelError(#[from] SendError<Vec<IcmpBinding>>),
+    #[cfg(feature = "irtt")]
+    #[error("IRTT channel error")]
+    IrttChannelError(#[from] SendError<Vec<IrttBinding>>),
 }
 
 pub struct Controller {
-    reflector_rx: Receiver<Vec<IpAddr>>,
+    reflector_rx: Receiver<Vec<SocketAddr>>,
     icmp_binding_tx: Sender<Vec<IcmpBinding>>,
-
+    #[cfg(feature = "irtt")]
+    irtt_binding_tx: Sender<Vec<IrttBinding>>,
+    reflectors: HashMap<SocketAddr, ReflectorDefinition>,
+    active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+    endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
     source: MeasurementSource,
     period: Duration,
-
     next_stream_id: u64,
-    stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>,
+    stream_ids: HashMap<(SocketAddr, MeasurementSource), MeasurementStreamId>,
 }
 
 impl Controller {
     pub fn new(
-        reflector_rx: Receiver<Vec<IpAddr>>,
+        reflector_rx: Receiver<Vec<SocketAddr>>,
         icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+        #[cfg(feature = "irtt")] irtt_binding_tx: Sender<Vec<IrttBinding>>,
+        reflectors: HashMap<SocketAddr, ReflectorDefinition>,
+        active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+        endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
+        source: MeasurementSource,
+        period: Duration,
     ) -> Self {
         Self {
             reflector_rx,
             icmp_binding_tx,
-            source: MeasurementSource::IcmpTimestamp,
+            #[cfg(feature = "irtt")]
-            period: Duration::from_millis(500),
+            irtt_binding_tx,
+            reflectors,
+            active_streams,
+            endpoint_streams,
+            source,
+            period,
             next_stream_id: 0,
             stream_ids: HashMap::new(),
         }
     }
 
-    pub async fn run(mut self, initial_reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
+    pub async fn run(mut self, initial_reflectors: Vec<SocketAddr>) -> Result<(), ControllerError> {
         self.update_reflectors(initial_reflectors).await?;
-
         while let Ok(reflectors) = self.reflector_rx.recv_async().await {
             self.update_reflectors(reflectors).await?;
         }
-
         Ok(())
     }
 
-    fn stream_for(&mut self, peer: IpAddr) -> MeasurementStream {
+    fn stream_for(&mut self, endpoint: SocketAddr) -> MeasurementStream {
-        let key = (peer, self.source);
+        let key = (endpoint, self.source);
-
         let id = *self.stream_ids.entry(key).or_insert_with(|| {
             let id = MeasurementStreamId(self.next_stream_id);
             self.next_stream_id += 1;
             id
         });
-
         MeasurementStream {
             id,
-            peer,
+            peer: endpoint.ip(),
             source: self.source,
         }
     }
 
-    async fn update_reflectors(&mut self, reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
+    async fn update_reflectors(
-        let count = reflectors.len();
+        &mut self,
-
+        endpoints: Vec<SocketAddr>,
-        let bindings = reflectors
+    ) -> Result<(), ControllerError> {
-            .into_iter()
+        let count = endpoints.len();
-            .enumerate()
+        let mut icmp = Vec::new();
-            .map(|(index, peer)| {
+        #[cfg(feature = "irtt")]
-                let stream = self.stream_for(peer);
+        let mut irtt = Vec::new();
-
+        let mut streams = Vec::with_capacity(count);
-                IcmpBinding::new(
+        let mut endpoint_streams = HashMap::new();
-                    stream,
+        for (index, endpoint) in endpoints.into_iter().enumerate() {
-                    ProbeSchedule {
+            let Some(_definition) = self.reflectors.get(&endpoint) else {
-                        period: self.period,
+                continue;
-                        offset: evenly_spaced_offset(self.period, index, count),
+            };
-                    },
+            #[cfg(feature = "irtt")]
+            let (port, hmac) = {
+                (
+                    _definition.irtt_port.unwrap_or(2112),
+                    _definition.irtt_hmac.clone(),
                 )
-            })
+            };
-            .collect();
+            let stream = self.stream_for(endpoint);
-
+            let schedule = ProbeSchedule {
-        self.icmp_binding_tx
+                period: self.period,
-            .send_async(bindings)
+                offset: evenly_spaced_offset(self.period, index, count),
-            .await
+            };
-            .map_err(|e| ControllerError::IcmpChannelError(e))
+            match self.source {
+                MeasurementSource::IcmpEcho | MeasurementSource::IcmpTimestamp => {
+                    icmp.push(IcmpBinding::new(stream, schedule))
+                }
+                #[cfg(feature = "irtt")]
+                MeasurementSource::Irtt => irtt.push(IrttBinding::new(stream, port, hmac)),
+                #[cfg(not(feature = "irtt"))]
+                MeasurementSource::Irtt => unreachable!("IRTT feature is checked before startup"),
+                _ => unreachable!("unsupported reflector measurement source"),
+            }
+            streams.push(stream);
+            endpoint_streams.insert(endpoint, stream);
+        }
+        *self
+            .active_streams
+            .write()
+            .expect("active stream lock poisoned") = streams;
+        *self
+            .endpoint_streams
+            .write()
+            .expect("endpoint stream lock poisoned") = endpoint_streams;
+        self.icmp_binding_tx.send_async(icmp).await?;
+        #[cfg(feature = "irtt")]
+        self.irtt_binding_tx.send_async(irtt).await?;
+        Ok(())
     }
 }
 
 fn evenly_spaced_offset(period: Duration, index: usize, count: usize) -> Duration {
     let nanos = period.as_nanos() * index as u128 / count as u128;
-
     Duration::new(
         (nanos / 1_000_000_000) as u64,
         (nanos % 1_000_000_000) as u32,

src/main.rs

@@ -22,16 +22,19 @@ mod util;
 use crate::baseliner::Baseliner;
 use crate::controller::Controller;
 use crate::measurement::icmp::IcmpEngine;
+#[cfg(feature = "irtt")]
+use crate::measurement::irtt::IrttEngine;
+use crate::measurement::model::{MeasurementSource, MeasurementStream};
 use crate::metrics::{Metric, Metrics, MetricsSender};
 use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations};
 use crate::ratecontroller::Ratecontroller;
 use crate::reflector_selector::ReflectorSelector;
-use crate::settings::Settings;
+use crate::settings::{ReflectorDefinition, ReflectorProtocol, Settings};
 use crate::util::RwLockExt;
 use ::log::{info, warn};
 use flume::RecvTimeoutError;
-use std::net::IpAddr;
+use std::collections::HashMap;
-use std::str::FromStr;
+use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 use std::thread;
@@ -39,8 +42,11 @@ use std::thread::{JoinHandle, sleep};
 use std::time::{Duration, Instant};
 
 struct ReflectorSetup {
-    peers: Arc<RwLock<Vec<IpAddr>>>,
+    peers: Arc<RwLock<Vec<SocketAddr>>>,
-    pool: Vec<IpAddr>,
+    active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+    endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
+    pool: Vec<SocketAddr>,
+    reflectors: HashMap<SocketAddr, ReflectorDefinition>,
     reselection_enabled: bool,
 }
 pub static SHUTDOWN: AtomicBool = AtomicBool::new(false);
@@ -134,29 +140,126 @@ fn restore_shaper<T: TrafficControlBackend>(
     }
 }
 
-fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
+fn resolve_reflectors(
-    let configured = settings.load_reflectors()?;
+    definitions: &[ReflectorDefinition],
-    let configured_count = configured.len();
+    protocol: ReflectorProtocol,
+    countries: &[String],
+    filter_country: bool,
+) -> anyhow::Result<HashMap<SocketAddr, ReflectorDefinition>> {
+    let mut resolved = HashMap::new();
+    for definition in definitions {
+        if !definition.protocols.contains(&protocol) {
+            continue;
+        }
+        if protocol != ReflectorProtocol::Irtt
+            && definition
+                .host
+                .parse::<IpAddr>()
+                .is_ok_and(|ip| ip.is_ipv6())
+        {
+            continue;
+        }
+        if filter_country
+            && !countries.is_empty()
+            && definition.country.as_ref().is_some_and(|country| {
+                !countries
+                    .iter()
+                    .any(|allowed| allowed.eq_ignore_ascii_case(country))
+            })
+        {
+            continue;
+        }
+        let port = if protocol == ReflectorProtocol::Irtt {
+            definition.irtt_port.unwrap_or(2112)
+        } else {
+            0
+        };
+        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;
+            }
+            insert_reflector(&mut resolved, endpoint, definition.clone())?;
+        }
+    }
+    Ok(resolved)
+}
 
-    let default_reflectors = [
+fn insert_reflector(
-        IpAddr::from_str("9.9.9.9")?,
+    reflectors: &mut HashMap<SocketAddr, ReflectorDefinition>,
-        IpAddr::from_str("8.238.120.14")?,
+    endpoint: SocketAddr,
-        IpAddr::from_str("74.82.42.42")?,
+    definition: ReflectorDefinition,
-        IpAddr::from_str("194.242.2.2")?,
+) -> anyhow::Result<()> {
-        IpAddr::from_str("208.67.222.222")?,
+    if let Some(existing) = reflectors.get_mut(&endpoint) {
-        IpAddr::from_str("94.140.14.14")?,
+        let same_country = match (&existing.country, &definition.country) {
-    ];
+            (Some(a), Some(b)) => a.eq_ignore_ascii_case(b),
-
+            _ => true,
-    let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize;
+        };
-    let pool = if reselection_enabled {
+        let same_protocols = existing
-        configured
+            .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}");
+        }
+        if existing.country.is_none() {
+            existing.country = definition.country;
+        }
     } else {
-        Vec::new()
+        reflectors.insert(endpoint, definition);
-    };
+    }
+    Ok(())
+}
+
+fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
+    let protocol = settings.advanced_settings.measurement_type;
+    let configured = resolve_reflectors(
+        &settings.reflectors.entries,
+        protocol,
+        &settings.reflectors.countries,
+        true,
+    )?;
+    let bootstrap = resolve_reflectors(&settings.reflectors.bootstrap, protocol, &[], false)?;
+    let mut reflectors = bootstrap.clone();
+    for (endpoint, definition) in &configured {
+        insert_reflector(&mut reflectors, *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<_>>();
+    initial.sort();
+    if initial.is_empty() {
+        initial = pool
+            .iter()
+            .take(settings.reflectors.num_reflectors as usize)
+            .copied()
+            .collect();
+    }
+    let reselection_enabled = pool.len() > settings.reflectors.num_reflectors as usize;
 
     Ok(ReflectorSetup {
-        peers: Arc::new(RwLock::new(default_reflectors.to_vec())),
+        peers: Arc::new(RwLock::new(initial)),
+        active_streams: Arc::new(RwLock::new(Vec::new())),
+        endpoint_streams: Arc::new(RwLock::new(HashMap::new())),
         pool,
+        reflectors,
         reselection_enabled,
     })
 }
@@ -219,6 +322,12 @@ fn wait_for_exit(error_rx: &flume::Receiver<anyhow::Error>) -> anyhow::Result<()
 }
 
 fn run(settings: &Settings) -> anyhow::Result<()> {
+    #[cfg(not(feature = "irtt"))]
+    if settings.advanced_settings.measurement_type == ReflectorProtocol::Irtt {
+        anyhow::bail!(
+            "IRTT measurement selected, but this binary was built without the `irtt` feature"
+        );
+    }
     let rt = tokio::runtime::Runtime::new()?;
 
     if settings.advanced_settings.dry_run {
@@ -229,13 +338,18 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
 
     let ReflectorSetup {
         peers: reflector_peers,
+        active_streams,
+        endpoint_streams,
         pool: reflector_pool,
+        reflectors,
         reselection_enabled,
     } = setup_reflectors(settings)?;
 
     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();
+    #[cfg(feature = "irtt")]
+    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);
@@ -246,18 +360,45 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         (None, None)
     };
 
-    let controller = Controller::new(reflector_rx, icmp_binding_tx);
+    let source = match settings.advanced_settings.measurement_type {
-    let engine = {
+        ReflectorProtocol::IcmpEcho => MeasurementSource::IcmpEcho,
+        ReflectorProtocol::IcmpTimestamps => MeasurementSource::IcmpTimestamp,
+        ReflectorProtocol::Irtt => MeasurementSource::Irtt,
+    };
+    let controller = Controller::new(
+        reflector_rx,
+        icmp_binding_tx,
+        #[cfg(feature = "irtt")]
+        irtt_binding_tx,
+        reflectors,
+        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();
         IcmpEngine::new(
             start_time,
             &settings.network,
-            measurement_tx,
+            measurement_tx.clone(),
             icmp_binding_rx,
         )?
     };
+    #[cfg(feature = "irtt")]
+    let (irtt_task, irtt_engine) = {
+        let _guard = rt.enter();
+        IrttEngine::new(
+            Duration::from_secs_f64(settings.advanced_settings.tick_interval),
+            &settings.network,
+            measurement_tx,
+            irtt_binding_rx,
+        )?
+    };
 
-    spawn_task(&rt, &error_tx, engine.run());
+    spawn_task(&rt, &error_tx, icmp_engine.run());
+    #[cfg(feature = "irtt")]
+    spawn_task(&rt, &error_tx, irtt_engine.run(irtt_task));
 
     let dropped = Arc::new(AtomicU32::new(0));
 
@@ -329,6 +470,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
                 .expect("reselection snapshot receiver must exist when reselection is enabled"),
             reflector_peers_lock: Arc::clone(&reflector_peers),
             reflector_pool,
+            endpoint_streams,
             trigger_channel: reselect_rx,
             metrics: event_metrics,
             reflector_tx,
@@ -342,7 +484,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     let mut ratecontroller = Ratecontroller::new(
         settings.clone(),
         control_snapshot_rx,
-        reflector_peers,
+        active_streams,
         reselect_tx,
         rate_metrics,
     )?;

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

@@ -0,0 +1,418 @@
+use std::{collections::HashMap, net::SocketAddr};
+
+use flume::{Receiver, Sender};
+use irtt_client::{
+    ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy,
+    managed::{
+        ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask,
+        ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError,
+        ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError,
+        ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetLifecycle, TargetAuth, TargetId,
+    },
+};
+use log::{debug, trace, warn};
+use thiserror::Error;
+use tokio::sync::broadcast::error::RecvError;
+
+use crate::SHUTDOWN;
+use crate::measurement::model::{
+    MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream,
+    MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration,
+};
+use crate::settings::NetworkSettings;
+
+#[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,
+    port: u16,
+    hmac: Option<String>,
+}
+
+impl IrttBinding {
+    pub fn new(stream: MeasurementStream, port: u16, hmac: Option<String>) -> Self {
+        Self { stream, port, hmac }
+    }
+}
+
+pub struct IrttEngine {
+    client: ManagedClientHandle,
+
+    bindings: HashMap<MeasurementStreamId, IrttBinding>,
+    targets: Vec<ManagedTargetConfig>,
+
+    binding_rx: Receiver<Vec<IrttBinding>>,
+    observation_tx: Sender<MeasurementEvent>,
+}
+
+impl IrttEngine {
+    pub fn new(
+        interval: std::time::Duration,
+        settings: &NetworkSettings,
+        observation_tx: Sender<MeasurementEvent>,
+        binding_rx: Receiver<Vec<IrttBinding>>,
+    ) -> Result<(ManagedClientTask, Self), IrttError> {
+        let config = managed_client_config(interval, settings);
+        let (task, handle) = ManagedClient::task(config, Vec::new())?;
+
+        Ok((
+            task,
+            Self {
+                client: handle,
+                bindings: HashMap::new(),
+                targets: Vec::new(),
+                binding_rx,
+                observation_tx,
+            },
+        ))
+    }
+
+    async fn set_bindings(&mut self, bindings: Vec<IrttBinding>) -> Result<(), IrttError> {
+        let targets = bindings
+            .iter()
+            .map(|binding| {
+                let mut target = ManagedTargetConfig::new(
+                    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()) {
+                    target.auth = TargetAuth::Override(irtt_client::Authentication::Hmac(
+                        irtt_client::HmacKey::new(key.as_bytes().to_vec().into_boxed_slice()),
+                    ));
+                }
+                target
+            })
+            .collect::<Vec<_>>();
+
+        self.client.update_targets(targets.clone())?.await?;
+        self.targets = targets;
+
+        self.bindings = bindings
+            .into_iter()
+            .map(|binding| (binding.stream.id, binding))
+            .collect();
+
+        Ok(())
+    }
+
+    pub async fn run(mut self, task: ManagedClientTask) -> Result<(), IrttError> {
+        let mut task = Box::pin(task);
+        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()?;
+        // 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() => {
+                    let bindings = bindings
+                        .map_err(|_| IrttError::BindingChannelClosed)?;
+
+                    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,
+
+                        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,
+                                    rtt,
+                                    ..
+                                } => {
+                                    let Some(stream) = stream else {
+                                        continue;
+                                    };
+
+                                    let observation = MeasurementObservation {
+                                        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),
+                                    };
+
+                                    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);
+                        }
+                    }
+                }
+
+                _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
+                    if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) {
+                        return Ok(());
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn managed_client_config(
+    interval: std::time::Duration,
+    #[allow(unused)] settings: &NetworkSettings,
+) -> ManagedClientConfig {
+    ManagedClientConfig {
+        client: ClientConfig {
+            socket: irtt_client::SocketConfig {
+                #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+                bind_to_device: settings.measurement_bind_device.clone(),
+                #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+                mark: settings.measurement_mark,
+                #[cfg(target_os = "freebsd")]
+                fib: settings.measurement_fib,
+                ..Default::default()
+            },
+            request: irtt_client::SessionRequest {
+                duration: None,
+                interval,
+                ..Default::default()
+            },
+            open: OpenPolicy {
+                negotiation: NegotiationPolicy::Strict,
+                ..Default::default()
+            },
+            ..Default::default()
+        },
+        completion: ManagedCompletionPolicy::ExplicitStop,
+        ..Default::default()
+    }
+}
+
+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;

src/measurement/model.rs

@@ -20,7 +20,7 @@ pub enum MeasurementSource {
 impl Display for MeasurementSource {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            MeasurementSource::IcmpEcho => write!(f, "icmp"),
+            MeasurementSource::IcmpEcho => write!(f, "icmp-echo"),
             MeasurementSource::IcmpTimestamp => write!(f, "icmp-timestamps"),
             MeasurementSource::Irtt => write!(f, "irtt"),
             MeasurementSource::TcpSeqAck => write!(f, "tcp-seqack"),
@@ -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/ratecontroller.rs

@@ -7,6 +7,7 @@
 
 use crate::SHUTDOWN;
 use crate::baseliner::ControlSnapshot;
+use crate::measurement::model::MeasurementStream;
 use crate::metrics::{Metric, MetricsSender};
 use crate::platform::{
     InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend,
@@ -19,7 +20,6 @@ use flume::{Receiver, Sender};
 use log::{debug, info, warn};
 use std::fs::File;
 use std::io::Write;
-use std::net::IpAddr;
 use std::sync::atomic::Ordering;
 use std::thread::sleep;
 use std::time::{Duration, Instant};
@@ -100,7 +100,7 @@ fn drain_latest_snapshot(
 
 fn deltas_from_snapshot(
     snapshot: Option<&ControlSnapshot>,
-    reflectors: &[IpAddr],
+    reflectors: &[MeasurementStream],
     now: Instant,
     tick_interval: f64,
 ) -> (Vec<f64>, Vec<f64>) {
@@ -123,8 +123,8 @@ fn deltas_from_snapshot(
             up.push(up_delta);
 
             debug!(
-                "Reflector: {} down_delay: {} up_delay: {}",
+                "Reflector: {} ({:?}) down_delay: {} up_delay: {}",
-                reflector, down_delta, up_delta
+                reflector.peer, reflector.source, down_delta, up_delta
             );
         }
     }
@@ -139,7 +139,7 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
     settings: Settings,
     control_snapshot: Option<ControlSnapshot>,
     control_snapshot_rx: Receiver<ControlSnapshot>,
-    reflectors_lock: ArcRwLock<Vec<IpAddr>>,
+    reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
     reselect_trigger: Sender<bool>,
     state_dl: State<T::Handle>,
     state_ul: State<T::Handle>,
@@ -153,7 +153,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     fn new_with_backends(
         settings: Settings,
         control_snapshot_rx: Receiver<ControlSnapshot>,
-        reflectors_lock: ArcRwLock<Vec<IpAddr>>,
+        reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
         reselect_trigger: Sender<bool>,
         metrics: MetricsSender,
         mut stats_provider: S,
@@ -491,12 +491,17 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         self.state_dl.deltas = down_deltas;
         self.state_ul.deltas = up_deltas;
 
-        if self.state_dl.deltas.len() < 5 || self.state_ul.deltas.len() < 5 {
+        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
+        {
             // trigger reselection
             warn!(
-                "Not enough delta values (D: {}, U: {}, need 5), triggering reselection",
+                "Not enough delta values (D: {}, U: {}, need {}), triggering reselection",
                 self.state_dl.deltas.len(),
-                self.state_ul.deltas.len()
+                self.state_ul.deltas.len(),
+                minimum
             );
             let _ = self.reselect_trigger.try_send(true);
         }
@@ -509,7 +514,7 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
     pub fn new(
         settings: Settings,
         control_snapshot_rx: Receiver<ControlSnapshot>,
-        reflectors_lock: ArcRwLock<Vec<IpAddr>>,
+        reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
         reselect_trigger: Sender<bool>,
         metrics: MetricsSender,
     ) -> anyhow::Result<Self> {
@@ -530,7 +535,17 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
 mod tests {
     use super::*;
     use crate::baseliner::{EwmaStats, ReflectorState};
+    use crate::measurement::model::{MeasurementSource, MeasurementStreamId};
     use std::collections::HashMap;
+    use std::net::IpAddr;
+
+    fn stream(id: u64, peer: IpAddr) -> MeasurementStream {
+        MeasurementStream {
+            id: MeasurementStreamId(id),
+            peer,
+            source: MeasurementSource::IcmpTimestamp,
+        }
+    }
 
     #[test]
     fn fixed_seed_reproduces_initial_safe_rates() {
@@ -564,7 +579,7 @@ mod tests {
     }
 
     fn snapshot(generated_at: Instant, marker: f64) -> ControlSnapshot {
-        let reflector = "192.0.2.1".parse().unwrap();
+        let reflector = stream(0, "192.0.2.1".parse().unwrap());
         ControlSnapshot {
             generated_at,
             reflectors: HashMap::from([(
@@ -590,7 +605,9 @@ mod tests {
         assert!(rx.is_empty());
         assert_eq!(latest.generated_at, newest_at);
         assert_eq!(
-            latest.reflectors[&"192.0.2.1".parse().unwrap()].recent.down,
+            latest.reflectors[&stream(0, "192.0.2.1".parse().unwrap())]
+                .recent
+                .down,
             2.0
         );
     }
@@ -598,10 +615,10 @@ mod tests {
     #[test]
     fn calculates_sorted_deltas_from_active_fresh_snapshot_state() {
         let now = Instant::now();
-        let first: IpAddr = "192.0.2.1".parse().unwrap();
+        let first = stream(0, "192.0.2.1".parse().unwrap());
-        let second: IpAddr = "192.0.2.2".parse().unwrap();
+        let second = stream(1, "192.0.2.2".parse().unwrap());
-        let stale: IpAddr = "192.0.2.3".parse().unwrap();
+        let stale = stream(2, "192.0.2.3".parse().unwrap());
-        let inactive: IpAddr = "192.0.2.4".parse().unwrap();
+        let inactive = stream(3, "192.0.2.4".parse().unwrap());
         let snapshot = ControlSnapshot {
             generated_at: now,
             reflectors: HashMap::from([

src/reflector_selector.rs

@@ -7,12 +7,13 @@
 
 use crate::SHUTDOWN;
 use crate::baseliner::ControlSnapshot;
+use crate::measurement::model::MeasurementStream;
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
 use crate::util::RwLockExt;
 use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender};
 use log::{debug, info};
-use std::net::IpAddr;
+use std::net::SocketAddr;
 use std::sync::atomic::Ordering;
 use std::sync::{Arc, RwLock};
 use std::thread::sleep;
@@ -98,18 +99,25 @@ fn receive_snapshots_until(
     }
 }
 
-fn recent_rtt_for_peer(snapshot: Option<&ControlSnapshot>, peer: &IpAddr) -> Option<u64> {
+fn recent_rtt_for_peer(
+    snapshot: Option<&ControlSnapshot>,
+    stream: Option<&MeasurementStream>,
+    now: Instant,
+    max_age: Duration,
+) -> Option<u64> {
     snapshot
-        .and_then(|snapshot| snapshot.reflectors.get(peer))
+        .and_then(|snapshot| stream.and_then(|stream| snapshot.reflectors.get(stream)))
+        .filter(|state| now.saturating_duration_since(state.last_receive_at) < max_age)
         .map(|state| (state.recent.down + state.recent.up) as u64)
 }
 
 pub struct ReflectorSelector {
     pub settings: Settings,
     pub snapshot_rx: Receiver<ControlSnapshot>,
-    pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
+    pub reflector_peers_lock: Arc<RwLock<Vec<SocketAddr>>>,
-    pub reflector_tx: Sender<Vec<IpAddr>>,
+    pub reflector_tx: Sender<Vec<SocketAddr>>,
-    pub reflector_pool: Vec<IpAddr>,
+    pub reflector_pool: Vec<SocketAddr>,
+    pub endpoint_streams: Arc<RwLock<std::collections::HashMap<SocketAddr, MeasurementStream>>>,
     pub trigger_channel: Receiver<bool>,
     pub metrics: MetricsSender,
 }
@@ -159,14 +167,13 @@ impl ReflectorSelector {
 
             // After 40 reselections, slow down to every 15 minutes
             if reselection_count > 40 {
-                selector_sleep_time = Duration::new(
+                selector_sleep_time =
-                    self.settings.advanced_settings.peer_reselection_time * 60,
+                    Duration::new(self.settings.reflectors.peer_reselection_time * 60, 0);
-                    0,
-                );
             }
 
-            let mut next_peers: Vec<IpAddr> = Vec::new();
+            let mut next_peers: Vec<SocketAddr> = Vec::new();
             let mut reflectors_peers = self.reflector_peers_lock.write_anyhow()?;
+            let current_peers = reflectors_peers.clone();
 
             // Include all current peers
             for reflector in reflectors_peers.iter() {
@@ -174,6 +181,9 @@ impl ReflectorSelector {
                 next_peers.push(*reflector);
             }
 
+            if self.reflector_pool.is_empty() {
+                continue;
+            }
             for _ in 0..20 {
                 let next_candidate =
                     &self.reflector_pool[fastrand::usize(..self.reflector_pool.len())];
@@ -207,7 +217,13 @@ impl ReflectorSelector {
             let mut candidates = Vec::new();
 
             for peer in next_peers {
-                if let Some(rtt) = recent_rtt_for_peer(latest_snapshot.as_ref(), &peer) {
+                let stream = self.endpoint_streams.read_anyhow()?.get(&peer).copied();
+                if let Some(rtt) = recent_rtt_for_peer(
+                    latest_snapshot.as_ref(),
+                    stream.as_ref(),
+                    Instant::now(),
+                    Duration::from_secs_f64(self.settings.advanced_settings.tick_interval * 2.0),
+                ) {
                     candidates.push((peer, rtt));
                     info!("Candidate reflector: {} RTT: {}", peer, rtt);
                 } else {
@@ -222,7 +238,7 @@ impl ReflectorSelector {
             candidates.sort_by(|a, b| a.1.cmp(&b.1));
 
             // Now we will just limit the candidates down to 2 * num_reflectors
-            let mut num_reflectors = self.settings.advanced_settings.num_reflectors;
+            let num_reflectors = self.settings.reflectors.num_reflectors;
             let candidate_pool_num = (2 * num_reflectors) as usize;
             candidates.truncate(candidate_pool_num);
 
@@ -236,10 +252,21 @@ impl ReflectorSelector {
                 candidates.swap(i, j);
             }
 
-            if (candidates.len() as u8) < num_reflectors {
+            if candidates.is_empty() {
-                num_reflectors = candidates.len() as u8;
+                info!("No usable reflectors, retaining current set");
+                self.reflector_tx.send(current_peers.clone())?;
+                *reflectors_peers = current_peers;
+                continue;
             }
 
+            if (candidates.len() as u8) < num_reflectors {
+                info!(
+                    "Only {} usable reflectors, selecting available candidates",
+                    candidates.len()
+                );
+            }
+            let num_reflectors = num_reflectors.min(candidates.len() as u8);
+
             let mut new_peers = Vec::new();
             for i in 0..num_reflectors {
                 let peer = candidates[i as usize].0;
@@ -248,7 +275,7 @@ impl ReflectorSelector {
                 self.metrics.send(Metric::Event {
                     name: "reflector_selected",
                     reason: "",
-                    reflector: Some(peer),
+                    reflector: Some(peer.ip()),
                     tags: &[],
                 });
             }
@@ -263,7 +290,9 @@ impl ReflectorSelector {
 mod tests {
     use super::*;
     use crate::baseliner::{EwmaStats, ReflectorState};
+    use crate::measurement::model::MeasurementSource;
     use std::collections::HashMap;
+    use std::net::IpAddr;
     use std::thread;
 
     fn snapshot(generated_at: Instant) -> ControlSnapshot {
@@ -276,7 +305,11 @@ mod tests {
     fn snapshot_with_rtt(generated_at: Instant, peer: IpAddr, rtt: f64) -> ControlSnapshot {
         let mut reflectors = HashMap::new();
         reflectors.insert(
-            peer,
+            crate::measurement::model::MeasurementStream {
+                id: crate::measurement::model::MeasurementStreamId(0),
+                peer,
+                source: MeasurementSource::IcmpTimestamp,
+            },
             ReflectorState {
                 baseline: EwmaStats {
                     down: rtt / 2.0,
@@ -376,7 +409,7 @@ mod tests {
     #[test]
     fn baseline_wait_retains_snapshot_for_evaluation() {
         let (snapshot_tx, snapshot_rx) = flume::unbounded();
-        let peer = "192.0.2.1".parse().unwrap();
+        let peer: IpAddr = "192.0.2.1".parse().unwrap();
         let generated_at = Instant::now();
         snapshot_tx
             .send(snapshot_with_rtt(generated_at, peer, 20.0))
@@ -397,7 +430,12 @@ mod tests {
         );
 
         assert_eq!(
-            recent_rtt_for_peer(latest_snapshot.as_ref(), &peer),
+            recent_rtt_for_peer(
+                latest_snapshot.as_ref(),
+                latest_snapshot.as_ref().unwrap().reflectors.keys().next(),
+                generated_at + Duration::from_secs(1),
+                Duration::from_secs(2)
+            ),
             Some(42)
         );
     }

src/settings/mod.rs

@@ -2,12 +2,10 @@
 //
 // 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};
 
@@ -17,6 +15,7 @@ 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;
 
@@ -47,7 +46,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 +56,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,
@@ -78,28 +99,32 @@ where
     deserializer.deserialize_any(FlexiBoolVisitor)
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash)]
-pub enum MeasurementType {
+#[serde(rename_all = "kebab-case")]
-    Icmp = 1,
+pub enum ReflectorProtocol {
+    IcmpEcho = 1,
     IcmpTimestamps,
+    Irtt,
 }
 
-impl Display for MeasurementType {
+impl Display for ReflectorProtocol {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            MeasurementType::Icmp => write!(f, "icmp"),
+            ReflectorProtocol::IcmpEcho => write!(f, "icmp-echo"),
-            MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"),
+            ReflectorProtocol::IcmpTimestamps => write!(f, "icmp-timestamps"),
+            ReflectorProtocol::Irtt => write!(f, "irtt"),
         }
     }
 }
 
-impl FromStr for MeasurementType {
+impl FromStr for ReflectorProtocol {
     type Err = ConfigError;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s.to_lowercase().as_str() {
-            "icmp" => Ok(MeasurementType::Icmp),
+            "icmp-echo" => Ok(ReflectorProtocol::IcmpEcho),
-            "icmp-timestamps" | "icmp_timestamps" => Ok(MeasurementType::IcmpTimestamps),
+            "icmp-timestamps" => Ok(ReflectorProtocol::IcmpTimestamps),
+            "irtt" => Ok(ReflectorProtocol::Irtt),
             _ => Err(ConfigError::Message(format!(
                 "invalid measurement type: {s}"
             ))),
@@ -127,12 +152,23 @@ impl FromStr for ObservabilityProtocol {
     }
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Clone, Deserialize, UciSection)]
-struct ReflectorRecord {
+pub(crate) struct ReflectorDefinition {
-    reflector_ip: IpAddr,
+    pub(crate) host: String,
-    ip_version: u8,
+    pub(crate) country: Option<String>,
+    pub(crate) protocols: Vec<ReflectorProtocol>,
+    pub(crate) irtt_port: Option<u16>,
     #[allow(dead_code)]
-    description: String,
+    pub(crate) irtt_hmac: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct ReflectorRow {
+    host: String,
+    country: Option<String>,
+    protocols: String,
+    irtt_port: Option<u16>,
+    irtt_hmac: Option<String>,
 }
 
 #[derive(Clone, Debug, Deserialize, UciConfig)]
@@ -141,6 +177,10 @@ pub(crate) struct Settings {
     #[uci(section)]
     pub(crate) network: NetworkSettings,
 
+    #[serde(default)]
+    #[uci(section)]
+    pub(crate) reflectors: ReflectorSettings,
+
     #[serde(default)]
     #[uci(section)]
     pub(crate) output: OutputSettings,
@@ -159,55 +199,41 @@ 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(
             config::File::with_name("/etc/sqm-autorate-rust/config.toml").required(false),
         );
 
-        let builder = builder.add_source(Environment::with_prefix("SQMA").separator("__"));
+        let builder = builder.add_source(
-        let built_config = builder.build()?;
+            Environment::with_prefix("SQMA")
+                .separator("__")
+                .try_parsing(true)
+                .list_separator(",")
+                .with_list_parse_key("reflectors.files")
+                .with_list_parse_key("reflectors.countries"),
+        );
+        let mut settings: Self = builder.build()?.try_deserialize()?;
 
-        built_config.try_deserialize()
+        settings.reflectors.load_files()?;
-    }
 
-    pub fn load_reflectors(&self) -> anyhow::Result<Vec<IpAddr>> {
+        Ok(settings)
-        let mut reader = csv::ReaderBuilder::new()
-            .has_headers(true)
-            .trim(csv::Trim::All)
-            .from_path(&self.advanced_settings.reflector_list_file)
-            .with_context(|| {
-                format!(
-                    "failed to open reflector list: {}",
-                    self.advanced_settings.reflector_list_file
-                )
-            })?;
-
-        let mut reflectors = Vec::with_capacity(50);
-
-        for record in reader.deserialize::<ReflectorRecord>() {
-            let record = record.with_context(|| {
-                format!(
-                    "failed to parse reflector list: {}",
-                    self.advanced_settings.reflector_list_file
-                )
-            })?;
-
-            match (record.ip_version, record.reflector_ip) {
-                (4, IpAddr::V4(_)) => {}
-                (6, address @ IpAddr::V6(_)) => {
-                    bail!("reflector {address} is IPv6; only IPv4 reflectors are supported");
-                }
-                (version, address) => {
-                    bail!("reflector {address} has inconsistent IP version {version}");
-                }
-            }
-
-            reflectors.push(record.reflector_ip);
-        }
-
-        Ok(reflectors)
     }
 }
 
@@ -240,6 +266,100 @@ 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>,
+
+    #[uci(skip)]
+    pub(crate) entries: Vec<ReflectorDefinition>,
+
+    #[uci(skip)]
+    pub(crate) bootstrap: Vec<ReflectorDefinition>,
+
+    pub(crate) num_reflectors: u8,
+    pub(crate) peer_reselection_time: u64,
+}
+
+impl ReflectorSettings {
+    fn load_files(&mut self) -> Result<(), ConfigError> {
+        for path in &self.files {
+            let mut reader = csv::Reader::from_path(path)
+                .map_err(|error| ConfigError::Message(error.to_string()))?;
+            for row in reader.deserialize::<ReflectorRow>() {
+                let row = row.map_err(|error| ConfigError::Message(error.to_string()))?;
+                let protocols = row
+                    .protocols
+                    .split(['|', ';'])
+                    .map(str::trim)
+                    .filter(|protocol| !protocol.is_empty())
+                    .map(ReflectorProtocol::from_str)
+                    .collect::<Result<Vec<_>, _>>()?;
+                if protocols.is_empty() {
+                    return Err(ConfigError::Message(format!(
+                        "reflector row in {path} has no protocols"
+                    )));
+                }
+                self.entries.push(ReflectorDefinition {
+                    host: row.host,
+                    country: row.country,
+                    protocols,
+                    irtt_port: row.irtt_port,
+                    irtt_hmac: row.irtt_hmac,
+                });
+            }
+        }
+
+        Ok(())
+    }
+}
+
+impl Default for ReflectorSettings {
+    fn default() -> Self {
+        let icmp = |host: &str| ReflectorDefinition {
+            host: host.to_owned(),
+            country: None,
+            protocols: vec![
+                ReflectorProtocol::IcmpEcho,
+                ReflectorProtocol::IcmpTimestamps,
+            ],
+            irtt_port: None,
+            irtt_hmac: None,
+        };
+
+        let irtt = |host: &str, country: Option<&str>| ReflectorDefinition {
+            host: host.to_owned(),
+            country: country.map(str::to_owned),
+            protocols: vec![ReflectorProtocol::Irtt],
+            irtt_port: Some(2112),
+            irtt_hmac: None,
+        };
+
+        Self {
+            files: vec!["/etc/sqm-autorate-rust/reflectors.csv".into()],
+            countries: Vec::new(),
+            entries: Vec::new(),
+            bootstrap: vec![
+                icmp("9.9.9.9"),
+                icmp("8.238.120.14"),
+                icmp("74.82.42.42"),
+                icmp("194.242.2.2"),
+                icmp("208.67.222.222"),
+                icmp("94.140.14.14"),
+                irtt("global.irtt.lochnair.net", None),
+                irtt("uk-cov1.irtt.lochnair.net", Some("GB")),
+                irtt("uk-cov2.irtt.lochnair.net", Some("GB")),
+                irtt("de-fra.irtt.lochnair.net", Some("DE")),
+                irtt("nl-ams1.irtt.lochnair.net", Some("NL")),
+            ],
+            num_reflectors: 5,
+            peer_reselection_time: 15,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Deserialize, UciSection)]
 #[serde(default)]
 pub(crate) struct OutputSettings {
@@ -309,10 +429,7 @@ pub(crate) struct AdvancedSettings {
     pub(crate) high_load_level: f64,
     pub(crate) min_change_interval: f64,
     #[serde(deserialize_with = "deserialize_from_str")]
-    pub(crate) measurement_type: MeasurementType,
+    pub(crate) measurement_type: ReflectorProtocol,
-    pub(crate) num_reflectors: u8,
-    pub(crate) peer_reselection_time: u64,
-    pub(crate) reflector_list_file: String,
     pub(crate) speed_hist_size: u32,
     pub(crate) tick_interval: f64,
     pub(crate) upload_delay_ms: f64,
@@ -326,10 +443,7 @@ impl Default for AdvancedSettings {
             download_delay_ms: 15.0,
             high_load_level: 0.8,
             min_change_interval: 0.5,
-            measurement_type: MeasurementType::IcmpTimestamps,
+            measurement_type: ReflectorProtocol::IcmpTimestamps,
-            num_reflectors: 5,
-            peer_reselection_time: 15,
-            reflector_list_file: "/etc/sqm-autorate/reflectors-icmp.csv".to_owned(),
             speed_hist_size: 100,
             tick_interval: 0.5,
             upload_delay_ms: 15.0,
@@ -358,3 +472,24 @@ 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/uci.rs

@@ -1,6 +1,6 @@
 use std::path::{Path, PathBuf};
 
-use crate::settings::uci_schema::{UciConfigSchema, UciSectionMapping};
+use crate::settings::uci_schema::{UciConfigSchema, UciRepeatedSectionMapping, UciSectionMapping};
 use config::{ConfigError as SourceError, Map, Source, Value, ValueKind};
 
 use rust_uci::config::{
@@ -18,6 +18,7 @@ pub(crate) struct UciSource {
     packages: PackageSelection,
     required: bool,
     schema: &'static [UciSectionMapping],
+    repeated_schema: &'static [UciRepeatedSectionMapping],
     config_dir: Option<PathBuf>,
     save_dir: Option<PathBuf>,
 }
@@ -28,6 +29,7 @@ impl UciSource {
             packages,
             required: true,
             schema: &[],
+            repeated_schema: &[],
             config_dir: None,
             save_dir: None,
         }
@@ -65,6 +67,17 @@ impl UciSource {
         self
     }
 
+    /// Registers section types collected as arrays of tables at their config paths.
+    /// Absent section types leave the destination unset for Serde/source defaults.
+    #[allow(dead_code)]
+    pub(crate) fn with_repeated_schema(
+        mut self,
+        schema: &'static [UciRepeatedSectionMapping],
+    ) -> Self {
+        self.repeated_schema = schema;
+        self
+    }
+
     #[allow(dead_code)]
     pub(crate) fn with_directories(
         mut self,
@@ -145,6 +158,7 @@ impl UciSource {
         package_name: &str,
         package: &UciPackage,
         values: &mut Map<String, Value>,
+        repeated_values: &mut Map<String, Vec<Value>>,
     ) -> Result<(), SourceError> {
         let sections = package.sections().map_err(|e| {
             SourceError::Message(format!(
@@ -154,6 +168,12 @@ impl UciSource {
         })?;
 
         for section in sections {
+            let repeated_mapping = self
+                .repeated_schema
+                .iter()
+                .find(|mapping| mapping.uci_section == section.type_());
+            let mut section_values = Map::new();
+
             let selector = section.selector().ok_or_else(|| {
                 SourceError::Message(format!(
                     "loaded UCI section of type {:?} in package \
@@ -181,9 +201,21 @@ impl UciSource {
 
                 let source_key = format!("{package_name}.{config_relative_key}");
 
-                let destination_key = self
+                let repeated_option = repeated_mapping.and_then(|mapping| {
-                    .mapped_config_key(&selector, option_name)
+                    mapping
-                    .unwrap_or(source_key);
+                        .options
+                        .iter()
+                        .find(|option| option.uci_option == option_name)
+                });
+
+                let destination_key = if let Some(mapping) = repeated_option {
+                    mapping.config_field.to_owned()
+                } else if repeated_mapping.is_some() {
+                    option_name.to_owned()
+                } else {
+                    self.mapped_config_key(&selector, option_name)
+                        .unwrap_or(source_key)
+                };
 
                 let Some(value) = option.get().map_err(|e| {
                     SourceError::Message(format!(
@@ -210,13 +242,27 @@ impl UciSource {
                     }
                 };
 
-                if values.insert(destination_key.clone(), value).is_some() {
+                let destination = if repeated_mapping.is_some() {
+                    &mut section_values
+                } else {
+                    &mut *values
+                };
+
+                if destination.insert(destination_key.clone(), value).is_some() {
                     return Err(SourceError::Message(format!(
                         "duplicate UCI destination key \
                          {destination_key:?} while mapping {uci_key:?}"
                     )));
                 }
             }
+
+            if let Some(mapping) = repeated_mapping {
+                let origin = format!("UCI section {package_name}.{selector}");
+                repeated_values
+                    .entry(mapping.config_path.to_owned())
+                    .or_default()
+                    .push(Value::new(Some(&origin), ValueKind::Table(section_values)));
+            }
         }
 
         Ok(())
@@ -272,6 +318,7 @@ impl Source for UciSource {
         };
 
         let mut values = Map::new();
+        let mut repeated_values = Map::new();
 
         for package in packages {
             let package_name = package
@@ -279,7 +326,22 @@ impl Source for UciSource {
                 .map_err(|e| SourceError::Message(format!("failed to read UCI package name: {e}")))?
                 .to_owned();
 
-            self.collect_package(&package_name, &package, &mut values)?;
+            self.collect_package(&package_name, &package, &mut values, &mut repeated_values)?;
+        }
+
+        for (config_path, sections) in repeated_values {
+            let origin = "UCI repeated sections".to_owned();
+            if values
+                .insert(
+                    config_path.clone(),
+                    Value::new(Some(&origin), ValueKind::Array(sections)),
+                )
+                .is_some()
+            {
+                return Err(SourceError::Message(format!(
+                    "duplicate UCI destination key {config_path:?} while mapping repeated sections"
+                )));
+            }
         }
 
         Ok(values)

src/settings/uci_schema.rs

@@ -19,6 +19,18 @@ pub(crate) struct UciSectionMapping {
     pub(crate) options: &'static [UciOptionMapping],
 }
 
+#[cfg(all(feature = "uci", unix))]
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct UciRepeatedSectionMapping {
+    /// UCI section type, for both named and anonymous sections.
+    pub(crate) uci_section: &'static str,
+
+    /// Config-rs / Serde path receiving an array of section tables.
+    pub(crate) config_path: &'static str,
+
+    pub(crate) options: &'static [UciOptionMapping],
+}
+
 pub(crate) trait UciSectionSchema {
     const UCI_OPTIONS: &'static [UciOptionMapping];
 }

src/settings/validation.rs

@@ -3,6 +3,8 @@
 // SPDX-License-Identifier: MPL-2.0
 //
 
+use crate::settings::ReflectorSettings;
+
 use super::{AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings, Settings};
 use std::error::Error;
 use std::fmt::{self, Display, Formatter};
@@ -88,6 +90,7 @@ impl Settings {
         let mut report = ValidationReport::default();
 
         self.network.validate(&mut report);
+        self.reflectors.validate(&mut report);
         self.output.validate(&mut report);
         self.observability.validate(&mut report);
         self.advanced_settings.validate(&mut report);
@@ -192,6 +195,60 @@ impl NetworkSettings {
     }
 }
 
+impl ReflectorSettings {
+    fn validate(&self, report: &mut ValidationReport) {
+        if self.num_reflectors == 0 {
+            report.error("reflectors.num_reflectors", "must be greater than 0");
+        }
+
+        if self.peer_reselection_time == 0 {
+            report.error("reflectors.peer_reselection_time", "must be greater than 0");
+        }
+
+        if self.files.is_empty() && self.entries.is_empty() {
+            report.error(
+                "reflectors",
+                "at least one reflector file or inline entry is required",
+            );
+        }
+
+        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",
+                    format!("entry {index} must have at least one measurement protocol"),
+                );
+            }
+        }
+
+        for (index, file) in self.files.iter().enumerate() {
+            if file.trim().is_empty() {
+                report.error(
+                    "reflectors.files",
+                    format!("entry {index} must not be empty"),
+                );
+            }
+        }
+
+        for (index, country) in self.countries.iter().enumerate() {
+            if country.len() != 2 || !country.bytes().all(|c| c.is_ascii_alphabetic()) {
+                report.error(
+                    "reflectors.countries",
+                    format!(
+                        "entry {index} ({country:?}) must be an ISO 3166-1 alpha-2 country code"
+                    ),
+                );
+            }
+        }
+    }
+}
+
 impl OutputSettings {
     fn validate(&self, report: &mut ValidationReport) {
         if self.suppress_statistics {
@@ -298,24 +355,6 @@ impl AdvancedSettings {
             self.tick_interval,
         );
 
-        if self.num_reflectors < 5 {
-            report.error(
-                "advanced_settings.num_reflectors",
-                format!("must be at least 5, got {}", self.num_reflectors,),
-            );
-        }
-
-        if self.peer_reselection_time == 0 {
-            report.error(
-                "advanced_settings.peer_reselection_time",
-                "must be greater than 0",
-            );
-        }
-
-        if self.reflector_list_file.trim().is_empty() {
-            report.error("advanced_settings.reflector_list_file", "must not be empty");
-        }
-
         if self.speed_hist_size == 0 {
             report.error(
                 "advanced_settings.speed_hist_size",

src/time.rs

@@ -14,16 +14,11 @@ 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,
@@ -56,10 +51,6 @@ 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,20 +3,10 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use anyhow::anyhow;
-use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
+use std::sync::{Arc, 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>>;