fix(reflectors): merge missing country metadata without startup conflicts

Deduplicate compatible definitions and reject conflicting metadata or authentication within each list and across bootstrap and pool entries. Treat hostname aliases, protocol ordering, country-code case, and omitted default ports consistently.

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"
@@ -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"]

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,26 +56,28 @@ 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();
 
@@ -95,6 +97,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)?;

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,311 @@
+host,country,protocols,irtt_port,irtt_hmac
+94.140.14.14,CY,icmp-echo|icmp-timestamps,,
+94.140.15.15,CY,icmp-echo|icmp-timestamps,,
+94.140.14.15,CY,icmp-echo|icmp-timestamps,,
+94.140.15.16,CY,icmp-echo|icmp-timestamps,,
+94.140.14.140,CY,icmp-echo|icmp-timestamps,,
+94.140.14.141,CY,icmp-echo|icmp-timestamps,,
+205.171.3.65,US,icmp-echo|icmp-timestamps,,
+205.171.2.65,US,icmp-echo|icmp-timestamps,,
+185.228.168.9,US,icmp-echo|icmp-timestamps,,
+185.228.169.9,US,icmp-echo|icmp-timestamps,,
+185.228.168.168,US,icmp-echo|icmp-timestamps,,
+185.228.169.168,US,icmp-echo|icmp-timestamps,,
+185.228.168.10,US,icmp-echo|icmp-timestamps,,
+185.228.169.11,US,icmp-echo|icmp-timestamps,,
+8.26.56.26,US,icmp-echo|icmp-timestamps,,
+8.20.247.20,US,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,GB,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,US,icmp-echo|icmp-timestamps,,
+194.242.2.2,US,icmp-echo|icmp-timestamps,,
+194.242.2.3,US,icmp-echo|icmp-timestamps,,
+193.19.108.2,US,icmp-echo|icmp-timestamps,,
+193.19.108.3,US,icmp-echo|icmp-timestamps,,
+208.67.222.222,US,icmp-echo|icmp-timestamps,,
+208.67.220.220,US,icmp-echo|icmp-timestamps,,
+208.67.222.123,US,icmp-echo|icmp-timestamps,,
+208.67.220.123,US,icmp-echo|icmp-timestamps,,
+208.67.222.2,US,icmp-echo|icmp-timestamps,,
+208.67.220.2,US,icmp-echo|icmp-timestamps,,
+159.89.120.99,CA,icmp-echo|icmp-timestamps,,
+134.195.4.2,US,icmp-echo|icmp-timestamps,,
+9.9.9.9,US,icmp-echo|icmp-timestamps,,
+9.9.9.10,US,icmp-echo|icmp-timestamps,,
+9.9.9.11,US,icmp-echo|icmp-timestamps,,
+149.112.112.10,US,icmp-echo|icmp-timestamps,,
+149.112.112.11,US,icmp-echo|icmp-timestamps,,
+149.112.112.112,US,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,DE,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,SM,icmp-echo|icmp-timestamps,,
+80.80.81.81,SM,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,AU,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,AU,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,GB,icmp-echo|icmp-timestamps,,
+141.1.27.249,GB,icmp-echo|icmp-timestamps,,
+176.103.130.130,CY,icmp-echo|icmp-timestamps,,
+176.103.130.131,CY,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,GB,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,CN,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,SI,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(
@@ -434,7 +434,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 +451,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 +470,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,152 @@
-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, schedule, 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,46 @@ 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(
+            start_time,
+            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 +471,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 +485,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/irtt.rs

@@ -0,0 +1,414 @@
+use std::{collections::HashMap, net::SocketAddr, time::Instant};
+
+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, 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, ProbeSchedule, 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,
+    /// Retained for future hybrid-mode scheduling. IRTT currently uses the
+    /// global interval configured on its managed client.
+    #[allow(dead_code)]
+    schedule: ProbeSchedule,
+    port: u16,
+    hmac: Option<String>,
+}
+
+impl IrttBinding {
+    pub fn new(
+        stream: MeasurementStream,
+        schedule: ProbeSchedule,
+        port: u16,
+        hmac: Option<String>,
+    ) -> Self {
+        Self {
+            stream,
+            schedule,
+            port,
+            hmac,
+        }
+    }
+}
+
+pub struct IrttEngine {
+    client: ManagedClientHandle,
+
+    bindings: HashMap<MeasurementStreamId, IrttBinding>,
+
+    origin: Instant,
+
+    binding_rx: Receiver<Vec<IrttBinding>>,
+    observation_tx: Sender<MeasurementEvent>,
+}
+
+impl IrttEngine {
+    pub fn new(
+        origin: Instant,
+        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(),
+                origin,
+                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 {
+                    target.auth = TargetAuth::Override(irtt_client::Authentication::Hmac(
+                        irtt_client::HmacKey::new(key.as_bytes().to_vec().into_boxed_slice()),
+                    ));
+                }
+                target
+            })
+            .collect();
+
+        self.client.update_targets(targets)?.await?;
+
+        self.bindings = bindings
+            .into_iter()
+            .map(|binding| (binding.stream.id, binding))
+            .collect();
+
+        Ok(())
+    }
+
+    pub async fn run(mut self, 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()?;
+        loop {
+            tokio::select! {
+                bindings = self.binding_rx.recv_async() => {
+                    let bindings = bindings
+                        .map_err(|_| IrttError::BindingChannelClosed)?;
+
+                    self.set_bindings(bindings).await?;
+                }
+
+                event = events.recv() => {
+                    let event = match event {
+                        Ok(event) => event,
+
+                        Err(RecvError::Lagged(n)) => {
+                            warn!("IRTT event receiver dropped {n} events");
+                            continue;
+                        }
+
+                        Err(RecvError::Closed) => {
+                            let status = self.client.status();
+
+                            match status.lifecycle {
+                                ManagedLifecycle::Completed => return Ok(()),
+
+                                ManagedLifecycle::Failed => {
+                                    if let Some(outcome) = &status.final_outcome {
+                                        return Err(IrttError::ManagedFailure(
+                                            outcome.end_reason.clone()
+                                        ));
+                                    }
+
+                                    return Err(IrttError::ManagedAbandoned);
+                                }
+
+                                _ => return Err(IrttError::ManagedAbandoned),
+                            }
+                        }
+                    };
+
+                    match event {
+                        ManagedEvent::Started => {
+                            debug!("IRTT managed client started");
+                        }
+
+                        ManagedEvent::TargetStateChanged {
+                            target,
+                            lifecycle,
+                        } => {
+                            debug!("IRTT target {target:?}: {lifecycle:?}");
+                        }
+
+                        ManagedEvent::Client { target, event } => {
+                            let stream = target.id.as_str()
+                                .parse::<u64>()
+                                .ok()
+                                .and_then(|id| {
+                                    self.bindings.get(&MeasurementStreamId(id))
+                                })
+                                .map(|binding| binding.stream);
+
+                            match event {
+                                ClientEvent::SessionStarted(session) => {
+                                    debug!(
+                                        "IRTT session started for {}: {}",
+                                        target.id,
+                                        session.remote
+                                    );
+                                }
+
+                                ClientEvent::NoTestCompleted(outcome) => {
+                                    debug!(
+                                        "IRTT no-test completed for {}: {}",
+                                        target.id,
+                                        outcome.remote
+                                    );
+                                }
+
+                                ClientEvent::SessionClosed { .. } => {
+                                    debug!("IRTT session closed for {}", target.id);
+                                }
+
+                                ClientEvent::EchoSent { seq, .. } => {
+                                    trace!("IRTT {} sent probe {seq}", target.id);
+                                }
+
+                                ClientEvent::EchoReply {
+                                    sent_at,
+                                    received_at,
+                                    one_way,
+                                    ..
+                                } => {
+                                    let Some(stream) = stream else {
+                                        continue;
+                                    };
+
+                                    let observation = MeasurementObservation {
+                                        stream,
+                                        started_at: sent_at.mono,
+                                        observed_at: received_at.mono,
+                                        one_way: one_way.and_then(map_one_way),
+                                    };
+
+                                    self.observation_tx
+                                        .send_async(MeasurementEvent::Observation(observation))
+                                        .await
+                                        .map_err(|_| IrttError::EventChannelClosed)?;
+                                }
+
+                                ClientEvent::EchoLoss {
+                                    sent_at,
+                                    timeout_at,
+                                    ..
+                                } => {
+                                    let Some(stream) = stream else {
+                                        continue;
+                                    };
+
+                                    let loss = MeasurementLoss {
+                                        stream,
+                                        started_at: sent_at.mono,
+                                        detected_at: timeout_at,
+                                    };
+
+                                    self.observation_tx
+                                        .send_async(MeasurementEvent::Loss(loss))
+                                        .await
+                                        .map_err(|_| IrttError::EventChannelClosed)?;
+                                }
+
+                                ClientEvent::DuplicateReply { seq, .. } => {
+                                    trace!(
+                                        "IRTT {} duplicate reply {seq}",
+                                        target.id
+                                    );
+                                }
+
+                                ClientEvent::LateReply { seq, .. } => {
+                                    trace!(
+                                        "IRTT {} late reply {seq}",
+                                        target.id
+                                    );
+                                }
+
+                                ClientEvent::Warning { kind, message, .. } => {
+                                    debug!(
+                                        "IRTT {} warning ({kind:?}): {message}",
+                                        target.id
+                                    );
+                                }
+                            }
+                        }
+
+                        ManagedEvent::TargetFinished { outcome } => {
+                            match &outcome.end_reason {
+                                ManagedTargetEndReason::Failed(failure) => {
+                                    warn!(
+                                        "IRTT target {} failed: {failure:?}",
+                                        outcome.target.id
+                                    );
+                                }
+
+                                reason => {
+                                    debug!(
+                                        "IRTT target {} finished: {reason:?}",
+                                        outcome.target.id
+                                    );
+                                }
+                            }
+                        }
+
+                        ManagedEvent::Stopping => {
+                            debug!("IRTT managed client stopping");
+                        }
+
+                        ManagedEvent::Completed { outcome } => {
+                            debug!(
+                                "IRTT managed client completed: {:?}",
+                                outcome.end_reason
+                            );
+                            return Ok(());
+                        }
+
+                        ManagedEvent::Failed { outcome } => {
+                            return Err(IrttError::ManagedFailure(
+                                outcome.end_reason.clone()
+                            ));
+                        }
+
+                        ManagedEvent::Abandoned => {
+                            return Err(IrttError::ManagedAbandoned);
+                        }
+                    }
+                }
+
+                _ = 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"),

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,14 @@ 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);
+        if 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 +511,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 +532,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 +576,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 +602,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 +612,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);
 
@@ -237,7 +253,14 @@ impl ReflectorSelector {
             }
 
             if (candidates.len() as u8) < num_reflectors {
-                num_reflectors = candidates.len() as u8;
+                info!(
+                    "Only {} usable reflectors, retaining current set of {}",
+                    candidates.len(),
+                    num_reflectors
+                );
+                self.reflector_tx.send(current_peers.clone())?;
+                *reflectors_peers = current_peers;
+                continue;
             }
 
             let mut new_peers = Vec::new();
@@ -248,7 +271,7 @@ impl ReflectorSelector {
                 self.metrics.send(Metric::Event {
                     name: "reflector_selected",
                     reason: "",
-                    reflector: Some(peer),
+                    reflector: Some(peer.ip()),
                     tags: &[],
                 });
             }
@@ -263,7 +286,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 +301,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 +405,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 +426,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

@@ -78,28 +78,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 +131,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 +156,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,7 +178,22 @@ impl Settings {
         let builder = Config::builder();
 
         #[cfg(all(feature = "uci", unix))]
-        let builder = builder.add_source(UciSource::from_schema::<Self>().required(false));
+        let builder = builder.add_source(
+            UciSource::from_schema::<Self>()
+                .with_repeated_schema(&[
+                    uci_schema::UciRepeatedSectionMapping {
+                        uci_section: "reflector",
+                        config_path: "reflectors.entries",
+                        options: <ReflectorDefinition as uci_schema::UciSectionSchema>::UCI_OPTIONS,
+                    },
+                    uci_schema::UciRepeatedSectionMapping {
+                        uci_section: "bootstrap_reflector",
+                        config_path: "reflectors.bootstrap",
+                        options: <ReflectorDefinition as uci_schema::UciSectionSchema>::UCI_OPTIONS,
+                    },
+                ])
+                .required(false),
+        );
 
         #[cfg(feature = "toml")]
         let builder = builder.add_source(
@@ -167,47 +201,11 @@ impl Settings {
         );
 
         let builder = builder.add_source(Environment::with_prefix("SQMA").separator("__"));
-        let built_config = builder.build()?;
+        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 +238,99 @@ impl NetworkSettings {
     }
 }
 
+#[derive(Clone, Debug, Deserialize, UciSection)]
+#[serde(default)]
+pub(crate) struct ReflectorSettings {
+    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 +400,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 +414,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,

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,45 @@ 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, 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 +340,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",