diff --git a/Cargo.lock b/Cargo.lock index d0be125b..7ddb2ff1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -151,6 +151,7 @@ name = "aeronet_iroh" version = "0.21.0" dependencies = [ "aeronet_io", + "aeronet_iroh", "aeronet_tokio_runtime", "bevy", "bevy_app", @@ -161,6 +162,8 @@ dependencies = [ "derive_more", "futures", "iroh", + "iroh-relay", + "tokio", "tracing", ] @@ -3035,6 +3038,20 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "dashmap" +version = "6.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6361d5c062261c78a176addb82d4c821ae42bed6089de0e12603cd25de2059c" +dependencies = [ + "cfg-if", + "crossbeam-utils", + "hashbrown 0.14.5", + "lock_api", + "once_cell", + "parking_lot_core", +] + [[package]] name = "dasp_sample" version = "0.11.0" @@ -4125,6 +4142,12 @@ dependencies = [ "byteorder", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" + [[package]] name = "hashbrown" version = "0.15.5" @@ -4785,12 +4808,20 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "291065721ad7c477b972e581bbc528df031dc8eb5e39fe1ff3300ae5dfb157ef" dependencies = [ + "http-body-util", + "hyper", + "hyper-util", "iroh-metrics-derive", "itoa", "n0-error", "portable-atomic", + "reqwest", + "rustls", + "rustls-platform-verifier", "ryu", "serde", + "tokio", + "tokio-util", "tracing", ] @@ -4815,6 +4846,8 @@ dependencies = [ "blake3", "bytes", "cfg_aliases", + "clap", + "dashmap", "data-encoding", "derive_more", "getrandom 0.4.3", @@ -4835,17 +4868,28 @@ dependencies = [ "pin-project", "postcard", "rand 0.10.2", + "rcgen 0.14.8", + "reloadable-state", "reqwest", "rustls", + "rustls-cert-file-reader", + "rustls-cert-reloadable-resolver", "rustls-pki-types", "serde", "serde_bytes", + "serde_json", + "sha1 0.11.0", + "simdutf8", "strum", + "time", "tokio", "tokio-rustls", + "tokio-rustls-acme", "tokio-util", "tokio-websockets", + "toml", "tracing", + "tracing-subscriber", "url", "webpki-roots", "ws_stream_wasm", @@ -5562,6 +5606,7 @@ dependencies = [ "rustc-hash 2.1.3", "rustls", "rustls-pki-types", + "rustls-platform-verifier", "slab", "sorted-index-buffer", "thiserror 2.0.20", @@ -6869,6 +6914,7 @@ version = "0.14.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "57f6d249aad744e274e682777a50283a225a32705394ee6d5fcc01efa25e4055" dependencies = [ + "pem", "ring", "rustls-pki-types", "time", @@ -6969,6 +7015,23 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "reloadable-core" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1dc20ac1418988b60072d783c9f68e28a173fb63493c127952f6face3b40c6e0" + +[[package]] +name = "reloadable-state" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3853ef78d45b50f8b989896304a85239539d39b7f866a000e8846b9b72d74ce8" +dependencies = [ + "arc-swap", + "reloadable-core", + "tokio", +] + [[package]] name = "renderdoc-sys" version = "1.1.0" @@ -6998,6 +7061,8 @@ dependencies = [ "rustls", "rustls-pki-types", "rustls-platform-verifier", + "serde", + "serde_json", "sync_wrapper", "tokio", "tokio-rustls", @@ -7143,6 +7208,40 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-cert-file-reader" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8bb47c2a50fdfdaf95b0ac8b12620fc327da1fd4adbb30d0c56d866b005873ff" +dependencies = [ + "rustls-cert-read", + "rustls-pki-types", + "thiserror 2.0.20", + "tokio", +] + +[[package]] +name = "rustls-cert-read" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd46e8c5ae4de3345c4786a83f99ec7aff287209b9e26fa883c473aeb28f19d5" +dependencies = [ + "rustls-pki-types", +] + +[[package]] +name = "rustls-cert-reloadable-resolver" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe1baa8a3a1f05eaa9fc55aed4342867f70e5c170ea3bfed1b38c51a4857c0c8" +dependencies = [ + "futures-util", + "reloadable-state", + "rustls", + "rustls-cert-read", + "thiserror 2.0.20", +] + [[package]] name = "rustls-native-certs" version = "0.8.4" @@ -7372,6 +7471,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "serdect" version = "0.4.3" @@ -7393,6 +7501,17 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", +] + [[package]] name = "sha1_smol" version = "1.0.1" @@ -8117,6 +8236,34 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-rustls-acme" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1af8573b15fdad8d66da116198cd8fd8d87ff62a67c1c6c3df7f62da1170793f" +dependencies = [ + "async-trait", + "base64", + "chrono", + "futures", + "log", + "num-bigint", + "pem", + "proc-macro2", + "rcgen 0.14.8", + "reqwest", + "ring", + "rustls", + "serde", + "serde_json", + "thiserror 2.0.20", + "time", + "tokio", + "tokio-rustls", + "webpki-roots", + "x509-parser", +] + [[package]] name = "tokio-stream" version = "0.1.19" @@ -8183,6 +8330,21 @@ dependencies = [ "tokio-util", ] +[[package]] +name = "toml" +version = "1.1.4+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3aace63f4bbcdfc2c965b059de67119c89c4017a70d633be6c104910f67056f5" +dependencies = [ + "indexmap", + "serde_core", + "serde_spanned", + "toml_datetime", + "toml_parser", + "toml_writer", + "winnow", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -8213,6 +8375,12 @@ dependencies = [ "winnow", ] +[[package]] +name = "toml_writer" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" + [[package]] name = "tower" version = "0.5.3" @@ -8369,7 +8537,7 @@ dependencies = [ "rand 0.9.5", "rustls", "rustls-pki-types", - "sha1", + "sha1 0.10.7", "thiserror 2.0.20", "utf-8", ] diff --git a/Cargo.toml b/Cargo.toml index 7bde3214..dc22da72 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -71,6 +71,7 @@ flume = { version = "0.11.0" } futures = { version = "0.3.30" } gloo-timers = { version = "0.3.0" } iroh = { version = "1.0.3" } +iroh-relay = { version = "1.0.3" } itertools = { version = "0.14.0" } js-sys = { version = "0.3.70" } log = { version = "0.4.26" } diff --git a/crates/aeronet_iroh/Cargo.toml b/crates/aeronet_iroh/Cargo.toml index e50088d3..5eefc722 100644 --- a/crates/aeronet_iroh/Cargo.toml +++ b/crates/aeronet_iroh/Cargo.toml @@ -18,6 +18,12 @@ targets = ["x86_64-unknown-linux-gnu", "wasm32-unknown-unknown"] name = "iroh_peer" path = "examples/iroh_peer.rs" +[features] +default = [] +# Test helpers (`test_utils::run_test_relay`): spins up an in-process loopback +# Iroh relay for integration tests. +test-utils = ["dep:iroh-relay", "iroh-relay/server", "iroh-relay/test-utils"] + [dependencies] aeronet_io = { workspace = true } aeronet_tokio_runtime = { workspace = true } @@ -28,11 +34,14 @@ bytes = { workspace = true } derive_more = { workspace = true, features = ["display", "error"] } futures = { workspace = true } iroh = { workspace = true } +iroh-relay = { workspace = true, optional = true } tracing = { workspace = true } [dev-dependencies] bevy = { workspace = true } clap = { workspace = true } +tokio = { workspace = true } +aeronet_iroh = { path = ".", features = ["test-utils"] } [lints] workspace = true diff --git a/crates/aeronet_iroh/src/lib.rs b/crates/aeronet_iroh/src/lib.rs index 667abf76..7cd1bb04 100644 --- a/crates/aeronet_iroh/src/lib.rs +++ b/crates/aeronet_iroh/src/lib.rs @@ -12,6 +12,8 @@ extern crate alloc; pub mod endpoint; pub mod session; +#[cfg(feature = "test-utils")] +pub mod test_utils; use bevy_app::prelude::*; pub use {aeronet_tokio_runtime::TokioRuntime as IrohRuntime, iroh}; diff --git a/crates/aeronet_iroh/src/session.rs b/crates/aeronet_iroh/src/session.rs index fe4f9d63..f4274662 100644 --- a/crates/aeronet_iroh/src/session.rs +++ b/crates/aeronet_iroh/src/session.rs @@ -237,6 +237,58 @@ pub struct IrohIo { #[derive(Debug, Clone, PartialEq, Eq, Component)] pub struct SelectedPath(pub TransportAddr); +/// Which kind of network path an [`IrohSession`]'s traffic is flowing over. +/// +/// Iroh sessions typically start out on a relayed path and migrate to a +/// direct path once NAT hole punching succeeds. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PathKind { + /// A direct IP path to the peer (hole punching succeeded). + Direct, + /// Traffic is relayed through the given relay server. + Relayed { + /// The relay server currently forwarding this session's traffic. + relay: iroh::RelayUrl, + }, +} + +/// Session telemetry for the currently selected network path. +/// +/// Updated by the IO layer roughly every 100 ms and whenever Iroh selects a +/// new path. Inserted when the session connects. +#[derive(Debug, Clone, Component)] +pub struct PathReport { + /// Which kind of path traffic is currently flowing over. + pub kind: PathKind, + /// RTT estimate of the currently selected path. + pub rtt: Duration, + /// When this report was created (i.e. when the session connected). + pub connected_at: Instant, + /// When a direct path was first selected, if ever. + pub direct_since: Option, +} + +impl PathReport { + fn new(kind: PathKind, rtt: Duration) -> Self { + let direct_since = matches!(kind, PathKind::Direct).then(Instant::now); + Self { + kind, + rtt, + connected_at: Instant::now(), + direct_since, + } + } + + /// Time from session connect until a direct path was first selected, if + /// that has happened yet — the relay-to-direct migration time, the key + /// NAT traversal health metric for a peered session. + #[must_use] + pub fn time_to_direct(&self) -> Option { + self.direct_since + .map(|since| since.saturating_duration_since(self.connected_at)) + } +} + /// Minimum packet MTU that an [`IrohIo`] must support. pub const MIN_MTU: usize = IP_MTU; @@ -413,7 +465,7 @@ pub(crate) fn poll_connecting( Connected { rx_dc_reason }, session, )); - apply_path(&mut entity_commands, next.initial_meta.path); + apply_path(&mut entity_commands, None, next.initial_meta.path); } } @@ -443,10 +495,10 @@ fn try_disconnect( } pub(crate) fn poll( - mut sessions: Query<(Entity, &mut Session, &mut IrohIo)>, + mut sessions: Query<(Entity, &mut Session, &mut IrohIo, Option<&mut PathReport>)>, mut commands: Commands, ) { - 'sessions: for (entity, mut session, mut io) in &mut sessions { + 'sessions: for (entity, mut session, mut io, mut path_report) in &mut sessions { let span = trace_span!("poll", %entity); let _span = span.enter(); @@ -458,7 +510,11 @@ pub(crate) fn poll( }); continue 'sessions; } - apply_path(&mut commands.entity(entity), meta.path); + apply_path( + &mut commands.entity(entity), + path_report.as_deref_mut(), + meta.path, + ); } let mut num_packets = Saturating(0); @@ -479,12 +535,35 @@ pub(crate) fn poll( } } -fn apply_path(entity: &mut EntityCommands, path: Option<(TransportAddr, Duration)>) { +fn apply_path( + entity: &mut EntityCommands, + existing_report: Option<&mut PathReport>, + path: Option<(TransportAddr, Duration)>, +) { let Some((path, rtt)) = path else { - entity.remove::<(SelectedPath, PeerAddr, PacketRtt)>(); + entity.remove::<(SelectedPath, PeerAddr, PacketRtt, PathReport)>(); return; }; + let kind = match &path { + TransportAddr::Relay(relay) => PathKind::Relayed { + relay: relay.clone(), + }, + _ => PathKind::Direct, + }; + match existing_report { + Some(report) => { + if matches!(kind, PathKind::Direct) && !matches!(report.kind, PathKind::Direct) { + report.direct_since.get_or_insert_with(Instant::now); + } + report.kind = kind; + report.rtt = rtt; + } + None => { + entity.try_insert(PathReport::new(kind, rtt)); + } + } + let peer_addr = match &path { TransportAddr::Ip(addr) => Some(*addr), _ => None, @@ -735,9 +814,24 @@ async fn meta_loop( ) -> Result { const META_UPDATE_INTERVAL: Duration = Duration::from_millis(100); + // `PathEventStream` is not a fused stream, but `StreamExt::fuse` wraps + // any stream into one, so `futures::select!` can poll it — no tokio + // dependency needed, and this works on WASM too. + let mut path_events = conn.path_events().fuse(); loop { + // Wait until either the periodic tick fires or the selected path + // changes, so relay-to-direct migration is reported promptly instead + // of up to one interval late. futures::select! { () = IrohRuntime::sleep(META_UPDATE_INTERVAL).fuse() => {}, + event = path_events.next() => { + match event { + Some(iroh::endpoint::PathEvent::Selected { .. } | iroh::endpoint::PathEvent::Closed { .. }) => {} + // not a selection change, or the connection closed (the + // recv loop will surface the error); wait for the tick + _ => IrohRuntime::sleep(META_UPDATE_INTERVAL).await, + } + } _ = rx_closed => return Err(SessionError::FrontendClosed), } diff --git a/crates/aeronet_iroh/src/test_utils.rs b/crates/aeronet_iroh/src/test_utils.rs new file mode 100644 index 00000000..f3ef8e21 --- /dev/null +++ b/crates/aeronet_iroh/src/test_utils.rs @@ -0,0 +1,58 @@ +//! Utilities for testing apps using `aeronet_iroh`. +//! +//! Enable the `test-utils` feature to use this module. + +use { + iroh::{RelayMap, RelayUrl}, + iroh_relay::server::Server, +}; + +/// Starts an Iroh relay server bound to loopback, suitable for tests. +/// +/// Returns the [`RelayMap`] to configure endpoints with +/// (`iroh::RelayMode::Custom`), the [`RelayUrl`] to use as the dial-side +/// addressing hint, and the running server. The relay speaks QUIC as well as +/// HTTPS, so hole punching between test endpoints works the same way it does +/// against a production relay. +/// +/// The server shuts down when the returned [`Server`] is dropped. +/// +/// No external network access is required. +/// +/// # Panics +/// +/// Panics if the relay server fails to bind. +pub async fn run_test_relay() -> (RelayMap, RelayUrl, Server) { + use { + core::net::Ipv4Addr, + iroh_relay::{ + RelayConfig, RelayQuicConfig, + server::{ + CertConfig, QuicConfig, RelayConfig as RelayServerConfig, ServerConfig, TlsConfig, + }, + }, + }; + + let (_certs, server_config) = iroh_relay::server::testing::self_signed_tls_certs_and_config(); + let tls = TlsConfig::new( + (Ipv4Addr::LOCALHOST, 0), + CertConfig::Manual { server_config }, + ); + + let mut relay = RelayServerConfig::new((Ipv4Addr::LOCALHOST, 0)); + relay.tls = Some(tls); + + let mut config = ServerConfig::default(); + config.relay = Some(relay); + config.quic = Some(QuicConfig::new((Ipv4Addr::LOCALHOST, 0))); + + let server = Server::spawn(config) + .await + .expect("failed to spawn test relay server"); + let url = server.https_url().expect("TLS was configured"); + let quic = server + .quic_addr() + .map(|addr| RelayQuicConfig::new(addr.port())); + let map = RelayMap::from(RelayConfig::new(url.clone(), quic)); + (map, url, server) +} diff --git a/crates/aeronet_iroh/tests/relay_path_report.rs b/crates/aeronet_iroh/tests/relay_path_report.rs new file mode 100644 index 00000000..80e08ed0 --- /dev/null +++ b/crates/aeronet_iroh/tests/relay_path_report.rs @@ -0,0 +1,170 @@ +//! End-to-end tests over a real (in-process, loopback) Iroh relay: datagram +//! exchange and path telemetry (`PathReport`). No external network required. + +use { + aeronet_io::Session, + aeronet_iroh::{ + IrohPlugin, + endpoint::IrohEndpoint, + session::{PathKind, PathReport, SessionRequest, SessionResponse}, + test_utils::run_test_relay, + }, + bevy::{ecs::system::EntityCommand, prelude::*}, + bytes::Bytes, + core::time::Duration, + iroh::{RelayMap, endpoint::presets}, + std::{thread, time::Instant}, +}; + +const TIMEOUT: Duration = Duration::from_secs(15); +const ALPN: &[u8] = b"aeronet-iroh/tests-relay/0"; + +fn test_app() -> App { + let mut app = App::new(); + // Run the IO layer's backend tasks on the test's own tokio runtime, + // instead of letting `TokioRuntime`'s `FromWorld` create a runtime that + // would be dropped inside an async context when the `World` drops. + app.insert_resource(aeronet_iroh::IrohRuntime::from( + tokio::runtime::Handle::current(), + )); + app.add_plugins((MinimalPlugins, IrohPlugin)).add_observer( + |mut request: On| { + request.respond(SessionResponse::Accepted); + }, + ); + app +} + +fn open_endpoint(app: &mut App, relay_map: &RelayMap) -> Entity { + let entity = app.world_mut().spawn_empty().id(); + let builder = iroh::Endpoint::builder(presets::Minimal) + .alpns(vec![ALPN.to_vec()]) + .relay_mode(iroh::RelayMode::Custom(relay_map.clone())); + IrohEndpoint::open(builder).apply(app.world_mut().entity_mut(entity)); + entity +} + +fn wait_until(app: &mut App, mut condition: impl FnMut(&mut World) -> bool) { + let deadline = Instant::now() + TIMEOUT; + loop { + app.update(); + if condition(app.world_mut()) { + return; + } + assert!(Instant::now() < deadline, "timed out waiting for condition"); + thread::sleep(Duration::from_millis(5)); + } +} + +/// Two endpoints on a custom loopback relay connect, exchange datagrams, and +/// both populate `PathReport` with a sane path kind and RTT. +#[tokio::test(flavor = "multi_thread")] +async fn relay_connect_datagrams_and_path_report() { + let (relay_map, relay_url, _server) = run_test_relay().await; + + let mut app = test_app(); + let endpoint_a = open_endpoint(&mut app, &relay_map); + let endpoint_b = open_endpoint(&mut app, &relay_map); + wait_until(&mut app, |world| { + world.get::(endpoint_a).is_some() + && world.get::(endpoint_b).is_some() + }); + + // B dials A; the relay URL is the addressing hint and rendezvous. + let target = app + .world() + .get::(endpoint_a) + .unwrap() + .addr() + .with_relay_url(relay_url.clone()); + let connect = app + .world() + .get::(endpoint_b) + .unwrap() + .connect(target, ALPN); + let outgoing = app.world_mut().spawn_empty().id(); + connect.apply(app.world_mut().entity_mut(outgoing)); + + wait_until(&mut app, |world| world.get::(outgoing).is_some()); + let incoming = app + .world_mut() + .query::<(Entity, &Session)>() + .iter(app.world()) + .find_map(|(e, _)| (e != outgoing).then_some(e)) + .expect("accepted session should exist"); + + // datagram in each direction + app.world_mut() + .get_mut::(outgoing) + .unwrap() + .send + .push(Bytes::from_static(b"ping")); + wait_until(&mut app, |world| { + world.get::(incoming).is_some_and(|s| { + s.recv + .iter() + .any(|p| p.payload == Bytes::from_static(b"ping")) + }) + }); + app.world_mut() + .get_mut::(incoming) + .unwrap() + .send + .push(Bytes::from_static(b"pong")); + wait_until(&mut app, |world| { + world.get::(outgoing).is_some_and(|s| { + s.recv + .iter() + .any(|p| p.payload == Bytes::from_static(b"pong")) + }) + }); + + // PathReport populates on both sessions with a non-degenerate RTT. + wait_until(&mut app, |world| { + [outgoing, incoming].iter().all(|e| { + world + .get::(*e) + .is_some_and(|r| r.rtt > Duration::ZERO) + }) + }); + + for entity in [outgoing, incoming] { + let report = app.world().get::(entity).unwrap(); + match &report.kind { + PathKind::Relayed { relay } => { + assert_eq!(*relay, relay_url, "relayed through unexpected relay"); + } + PathKind::Direct => { + assert!( + report.time_to_direct().is_some(), + "direct path without time_to_direct" + ); + } + } + } + + // On loopback with a local relay, hole punching should eventually select a + // direct path on at least one side; tolerate relay-only in environments + // that block even loopback UDP punching. + let deadline = Instant::now() + TIMEOUT; + loop { + app.update(); + let any_direct = [outgoing, incoming].iter().any(|e| { + app.world() + .get::(*e) + .is_some_and(|r| matches!(r.kind, PathKind::Direct)) + }); + if any_direct { + // whichever side migrated must have a time-to-direct + let ttd = [outgoing, incoming] + .iter() + .find_map(|e| app.world().get::(*e)?.time_to_direct()); + assert!(ttd.is_some()); + break; + } + if Instant::now() >= deadline { + break; + } + thread::sleep(Duration::from_millis(5)); + } +}