From 34eed823b91210af782e4cc183267642cda171a0 Mon Sep 17 00:00:00 2001 From: Quang Le Date: Thu, 6 Aug 2026 16:27:02 +0700 Subject: [PATCH 1/3] fix(relay): remove the bind race in the relay test fixture --- Cargo.lock | 1 - crates/cli/Cargo.toml | 1 - crates/cli/src/commands/relay.rs | 518 +++++++++++++++++++++++++++---- crates/p2p/src/utils.rs | 55 ++++ crates/relay-server/src/error.rs | 31 +- crates/relay-server/src/p2p.rs | 46 ++- crates/relay-server/src/web.rs | 70 ++++- 7 files changed, 622 insertions(+), 100 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 73022c62..42078e16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5181,7 +5181,6 @@ dependencies = [ name = "pluto-cli" version = "1.7.1" dependencies = [ - "backon", "bytes", "chrono", "clap", diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index ce98d69d..26a00608 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -50,7 +50,6 @@ sha2.workspace = true [dev-dependencies] tempfile.workspace = true test-case.workspace = true -backon.workspace = true wiremock.workspace = true pluto-cluster = { workspace = true, features = ["test-cluster"] } pluto-testutil.workspace = true diff --git a/crates/cli/src/commands/relay.rs b/crates/cli/src/commands/relay.rs index 4ae7317d..99b10d03 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -377,15 +377,19 @@ async fn serve_relay( #[cfg(test)] mod tests { - use backon::{BackoffBuilder, Retryable}; - use std::{str::FromStr, time}; - use tokio::net; + use std::{ + cell::Cell, + io, + str::FromStr, + time::{Duration, Instant}, + }; + use tokio::{net, task::JoinHandle}; use tokio_util::sync::CancellationToken; #[tokio::test] async fn run_bootnode() { with_relay_server( - |args| { + |args, _| { args.relay.auto_p2p_key = false; pluto_p2p::k1::new_saved_priv_key(&args.data_dir.data_dir).unwrap(); }, @@ -398,7 +402,7 @@ mod tests { #[tokio::test] async fn run_bootnode_auto_p2p() { let first_run = with_relay_server( - |args| { + |args, _| { args.relay.auto_p2p_key = false; }, async |_| { /* Relay server does not start due to missing p2p key */ }, @@ -412,7 +416,7 @@ mod tests { )); let second_run = with_relay_server( - |_| {}, + |_, _| {}, async |_| { /* Relay server starts with auto-generated p2p key */ }, ) .await; @@ -422,7 +426,7 @@ mod tests { #[tokio::test] async fn serve_addr_multiaddrs() { with_relay_server( - |_| {}, + |_, _| {}, async |cfg| { let response = relay_server_get(cfg, "/").await.unwrap(); let body = response.text().await.unwrap(); @@ -447,7 +451,7 @@ mod tests { #[tokio::test] async fn serve_addr_enr() { with_relay_server( - |_| {}, + |_, _| {}, async |cfg| { let response = relay_server_get(cfg, "/enr").await.unwrap(); let body = response.text().await.unwrap(); @@ -463,7 +467,7 @@ mod tests { #[tokio::test] async fn serve_addr_enr_ext_ip() { with_relay_server( - |args| args.p2p.external_ip = Some("222.222.222.222".into()), + |args, _| args.p2p.external_ip = Some("222.222.222.222".into()), async |cfg| { let response = relay_server_get(cfg, "/enr").await.unwrap(); let body = response.text().await.unwrap(); @@ -479,12 +483,12 @@ mod tests { #[tokio::test] async fn serve_addr_enr_ext_host() { with_relay_server( - |args| args.p2p.external_host = Some("www.google.com".into()), + |args, _| args.p2p.external_host = Some("www.google.com".into()), async |cfg| { // Resolution happens asynchronously on a tick, so poll until the // ENR reflects a non-loopback IP (mirrors the Go test using // `assert.Eventually`). - tokio::time::timeout(time::Duration::from_secs(10), async { + tokio::time::timeout(Duration::from_secs(10), async { loop { let response = relay_server_get(cfg.clone(), "/enr").await.unwrap(); let body = response.text().await.unwrap(); @@ -495,7 +499,7 @@ mod tests { break; } - tokio::time::sleep(time::Duration::from_millis(200)).await; + tokio::time::sleep(Duration::from_millis(200)).await; } }) .await @@ -508,20 +512,15 @@ mod tests { #[tokio::test] async fn serve_addr_metrics() { - let monitoring_addr = net::TcpListener::bind("127.0.0.1:0") - .await - .unwrap() - .local_addr() - .unwrap() - .to_string(); - let monitoring_url = format!("http://{monitoring_addr}/metrics"); - with_relay_server( - move |args| { - args.debug_monitoring.monitor_addr = Some(monitoring_addr); + |args, monitoring_addr| { + args.debug_monitoring.monitor_addr = Some(monitoring_addr.into()); }, - async move |_cfg| { - let response = retry_get(&monitoring_url).await.unwrap(); + async |cfg| { + let monitoring_addr = cfg.monitoring_addr.unwrap(); + let response = http_get(&format!("http://{monitoring_addr}/metrics")) + .await + .unwrap(); let body = response.text().await.unwrap(); assert!(body.contains("relay_p2p_connection_total")); @@ -536,49 +535,135 @@ mod tests { .unwrap(); } + /// Number of complete relay startup attempts — each with a fresh data dir + /// and freshly allocated HTTP ports — before the fixture gives up on a bind + /// race and surfaces the error. + const MAX_STARTUP_ATTEMPTS: usize = 5; + + /// Per-attempt budget for the relay's HTTP servers to start serving. Sized + /// for a heavily loaded CI machine: the relay either serves or exits with + /// an error long before this, so exceeding it means it hung. + const SERVING_TIMEOUT: Duration = Duration::from_secs(30); + + /// Budget for the relay to stop once the test function has returned. + const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); + + /// libp2p listen address for the fixture: port 0 lets the kernel assign the + /// port inside libp2p's own `bind`, so no other process can claim it in + /// between. + /// + /// The HTTP listeners can't do that: their addresses are inbound config the + /// relay never reports back (as in Charon, whose relay test passes + /// `HTTPAddr` in), so the fixture allocates them up front and retries the + /// race instead. A p2p race could not be retried anyway — libp2p buries the + /// `AddrInUse` inside `io::Error::other(Transport(..))`, out of reach of + /// `Error::source` and therefore of [`is_addr_in_use`]. + /// + /// The cost is that external multiaddrs derived from these listen ports + /// carry port 0; no test asserts on them (Charon's assert only on the ENR's + /// IP), and the listen-port-to-advertised-port mapping is covered + /// separately by `external_multiaddrs_keep_the_listen_ports` in + /// `pluto_p2p::utils`. + const P2P_LISTEN_ADDR: &str = "127.0.0.1:0"; + + /// Allocates a loopback address by binding port 0, reading back the + /// assigned port and dropping the socket. + /// + /// The relay rebinds that port moments later, so another process can claim + /// it in between; [`with_relay_server`] retries the whole startup with a + /// new address when that happens. + async fn free_tcp_addr() -> String { + squat_tcp_addr().await.1 + } + + /// A relay that has been observed serving every HTTP endpoint it was + /// configured with. + struct ServingRelay { + /// Config the relay was started with. + cfg: pluto_relay_server::config::Config, + /// Cancels the relay. + ct: CancellationToken, + /// Relay task, resolving with the relay's exit status. + handle: JoinHandle>, + /// Data dir of this attempt, kept alive while the relay runs. + dir: tempfile::TempDir, + } + /// Run a function in the context of a running relay server. /// /// The server can be configured before initialization through - /// [`super::RelayArgs`], while the test function receives a function to - /// make HTTP requests to the running relay server. + /// [`super::RelayArgs`]; the closure also receives a loopback address + /// allocated for the attempt, for tests that opt into the monitoring + /// server. It is invoked once per startup attempt, so it must not assume it + /// runs only once. + /// + /// The test function runs only once the relay serves every HTTP endpoint it + /// was configured with, and receives the config the relay was started with. + /// Startup and shutdown errors are returned to the caller instead of + /// showing up as connection failures inside the test function. async fn with_relay_server( - config_fn: FArgs, + mut config_fn: FArgs, test_fn: FTest, ) -> Result<(), crate::error::CliError> where - FArgs: FnOnce(&mut super::RelayArgs), + FArgs: FnMut(&mut super::RelayArgs, &str), FTest: FnOnce(pluto_relay_server::config::Config) -> Fut, Fut: std::future::Future, { - let dir = tempfile::tempdir().unwrap(); + let mut attempts: usize = 0; + + let ServingRelay { + cfg, + ct, + handle, + dir, + } = loop { + attempts = attempts.saturating_add(1); + + match start_relay(&mut config_fn).await { + Ok(relay) => break relay, + // Another process claimed one of the freshly allocated ports + // before the relay could bind it. The relay task is gone for + // good — request retries in the test function could never + // recover — so start over with new ports, boundedly. + Err(err) if is_addr_in_use(&err) && attempts < MAX_STARTUP_ATTEMPTS => { + tracing::debug!("relay lost the race for a port, retrying: {err}"); + } + Err(err) => return Err(err), + } + }; - let tcp_addr = net::TcpListener::bind("127.0.0.1:0") - .await - .unwrap() - .local_addr() - .unwrap() - .to_string(); + test_fn(cfg).await; - let udp_addr = net::UdpSocket::bind("127.0.0.1:0") - .await - .unwrap() - .local_addr() - .unwrap() - .to_string(); + ct.cancel(); + let exit = match tokio::time::timeout(SHUTDOWN_TIMEOUT, handle).await { + Ok(Ok(exit)) => exit, + Ok(Err(err)) => resume_relay_panic(err), + Err(_) => panic!("relay did not shut down within {SHUTDOWN_TIMEOUT:?}"), + }; - let http_addr = net::TcpListener::bind("127.0.0.1:0") - .await - .unwrap() - .local_addr() - .unwrap() - .to_string(); + // The relay has stopped, so nothing reads the data dir anymore. + drop(dir); + + exit + } + + /// Starts one relay with freshly allocated addresses and waits until it + /// either serves or exits. + async fn start_relay( + config_fn: &mut impl FnMut(&mut super::RelayArgs, &str), + ) -> Result { + let dir = tempfile::tempdir().unwrap(); + // Only bound when a test opts into the monitoring server by setting + // `super::RelayDebugMonitoringArgs::monitor_addr` to it. + let monitoring_addr = free_tcp_addr().await; let mut args = super::RelayArgs { data_dir: super::RelayDataDirArgs { data_dir: dir.path().to_path_buf(), }, relay: super::RelayRelayArgs { - http_address: http_addr, + http_address: free_tcp_addr().await, auto_p2p_key: true, p2p_relay_log_level: "info".into(), max_res_per_peer: 0, @@ -593,8 +678,8 @@ mod tests { relays: vec![], external_ip: None, external_host: None, - tcp_addrs: vec![tcp_addr], - udp_addrs: vec![udp_addr], + tcp_addrs: vec![P2P_LISTEN_ADDR.into()], + udp_addrs: vec![P2P_LISTEN_ADDR.into()], disable_reuseport: false, }, log: super::RelayLogFlags { @@ -608,36 +693,335 @@ mod tests { loki_service: "".into(), }, }; - config_fn(&mut args); + config_fn(&mut args, &monitoring_addr); - let cfg: pluto_relay_server::config::Config = args.clone().try_into().unwrap(); + let cfg: pluto_relay_server::config::Config = args.try_into().unwrap(); let ct = CancellationToken::new(); + let mut handle = tokio::spawn(super::run(cfg.clone(), ct.child_token())); + + // Wait for the relay to serve, or to exit trying. A failed listener + // bind takes the relay down permanently, and awaiting the relay only + // after the test function would let a request `unwrap()` mask it as + // `ConnectionRefused`. + let serving = tokio::select! { + joined = &mut handle => { + // The relay is gone; cancel so nothing it spawned outlives it + // and keeps a listener bound into the next attempt. + ct.cancel(); + + return match joined { + Ok(Ok(())) => panic!("relay exited before serving {:?}", cfg.http_addr), + Ok(Err(err)) => Err(err), + Err(err) => resume_relay_panic(err), + }; + }, + serving = wait_until_serving(&cfg) => serving, + }; - let relay = tokio::spawn(super::run(cfg.clone(), ct.child_token())); + if let Err(err) = serving { + // The relay is alive but never served: not a bind race. Take it + // down so the failure isn't followed by a leaked relay. + ct.cancel(); + let _ = tokio::time::timeout(SHUTDOWN_TIMEOUT, handle).await; + panic!("{err}"); + } - test_fn(cfg.clone()).await; + Ok(ServingRelay { + cfg, + ct, + handle, + dir, + }) + } - ct.cancel(); - relay.await.unwrap() + /// Polls every HTTP endpoint the relay is configured to serve until each + /// one answers, returning a description of whichever never came up within + /// [`SERVING_TIMEOUT`]. + /// + /// `/enr` stands in for the ENR server as a whole: it shares a listener + /// with `/` and only succeeds once libp2p has reported its listen + /// addresses, so test functions can request either without racing startup. + async fn wait_until_serving(cfg: &pluto_relay_server::config::Config) -> Result<(), String> { + let client = reqwest::Client::new(); + let started = Instant::now(); + + let urls = [ + cfg.http_addr + .as_ref() + .map(|addr| format!("http://{addr}/enr")), + cfg.monitoring_addr + .as_ref() + .map(|addr| format!("http://{addr}/metrics")), + ] + .into_iter() + .flatten(); + + for url in urls { + while let Err(err) = client + .get(&url) + .timeout(Duration::from_secs(1)) + .send() + .await + .and_then(|response| response.error_for_status()) + { + if started.elapsed() >= SERVING_TIMEOUT { + return Err(format!( + "{url} still not serving {SERVING_TIMEOUT:?} into startup: {err}" + )); + } + + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + Ok(()) + } + + /// Reports whether `err`'s source chain contains an + /// [`io::ErrorKind::AddrInUse`] error, i.e. whether a listener lost the + /// race for a port that had just been allocated. + /// + /// Walks the chain instead of matching error strings so the typed + /// `io::ErrorKind` decides: both relay bind errors + /// ([`pluto_relay_server::RelayP2PError::FailedToBindHttpListener`] and its + /// monitoring counterpart) keep the original `io::Error`. + /// + /// Those two are the only bind races this fixture can hit; the p2p + /// listeners use port 0 instead — see [`P2P_LISTEN_ADDR`]. + fn is_addr_in_use(err: &(dyn std::error::Error + 'static)) -> bool { + let mut next = Some(err); + + while let Some(err) = next { + if let Some(io_err) = err.downcast_ref::() + && io_err.kind() == io::ErrorKind::AddrInUse + { + return true; + } + + next = err.source(); + } + + false + } + + /// Re-raises a panic from the relay task in the test thread, so the + /// original panic message is what the test reports. + fn resume_relay_panic(err: tokio::task::JoinError) -> ! { + if err.is_panic() { + std::panic::resume_unwind(err.into_panic()); + } + + panic!("relay task was cancelled: {err}"); + } + + /// Binds a loopback port and holds it, so anything else binding the + /// returned address fails with [`io::ErrorKind::AddrInUse`] — the failure a + /// concurrently started process causes by claiming a port between + /// [`free_tcp_addr`] and the relay's own bind. + async fn squat_tcp_addr() -> (net::TcpListener, String) { + let listener = net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + (listener, addr) + } + + #[tokio::test] + async fn fixture_recovers_from_transient_bind_race() { + let (squatter, squatted) = squat_tcp_addr().await; + // Released at the start of the second attempt, so exactly one bind + // loses the race — the transient failure seen when tests run + // concurrently. + let mut squatter = Some(squatter); + let attempts = Cell::new(0usize); + + with_relay_server( + |args, _| { + attempts.set(attempts.get().saturating_add(1)); + if attempts.get() > 1 { + squatter.take(); + } + args.relay.http_address = squatted.clone(); + }, + async |cfg| { + let response = relay_server_get(cfg, "/enr").await.unwrap(); + + assert!(response.status().is_success(), "{}", response.status()); + }, + ) + .await + .unwrap(); + + assert_eq!( + attempts.get(), + 2, + "the relay should have started on the second attempt" + ); + } + + /// Runs the fixture with `configure` pointing one of the relay's listeners + /// at a port held for the whole run, so every attempt loses the race for it + /// as if a concurrent process had claimed it. + /// + /// Asserts that the fixture retried boundedly and gave up with an + /// `AddrInUse` error, and returns that error so the caller can check which + /// listener reported it. + async fn exhaust_retries_on_taken_port( + configure: impl Fn(&mut super::RelayArgs, String), + ) -> super::CliError { + let (_squatter, squatted) = squat_tcp_addr().await; + let attempts = Cell::new(0usize); + + let result = with_relay_server( + |args, _| { + attempts.set(attempts.get().saturating_add(1)); + configure(args, squatted.clone()); + }, + async |_| panic!("test function must not run when a listener cannot bind"), + ) + .await; + + let err = result.expect_err("relay must not serve while one of its ports is taken"); + assert!( + is_addr_in_use(&err), + "expected an AddrInUse error, got: {err}" + ); + assert_eq!( + attempts.get(), + MAX_STARTUP_ATTEMPTS, + "bind races must be retried, but boundedly" + ); + + err + } + + #[tokio::test] + async fn fixture_retries_bind_races_boundedly() { + let err = exhaust_retries_on_taken_port(|args, addr| args.relay.http_address = addr).await; + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToBindHttpListener { .. } + ) + ), + "expected the bind error to be surfaced, got: {err}" + ); + + // The monitoring listener behaves the same way. It used to be only + // `warn!`-ed, which left the relay running and the port unserved — the + // most frequent pre-fix failure in the stress runs. + let err = exhaust_retries_on_taken_port(|args, addr| { + args.debug_monitoring.monitor_addr = Some(addr); + }) + .await; + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToBindMonitoringListener { .. } + ) + ), + "expected the monitoring bind error to be surfaced, got: {err}" + ); } - /// Make an HTTP GET request to the relay server with retries and backoff. + #[tokio::test] + async fn fixture_leaves_no_listener_bound_when_startup_fails() { + let http_addr = Cell::new(None); + let attempts = Cell::new(0usize); + + let result = with_relay_server( + |args, _| { + attempts.set(attempts.get().saturating_add(1)); + http_addr.set(Some(args.relay.http_address.clone())); + // Rejected while starting up, after the HTTP address has been + // configured: the relay must fail without leaving its HTTP + // listener bound behind. + args.debug_monitoring.monitor_addr = Some("not-an-address".into()); + }, + async |_| panic!("test function must not run when startup fails"), + ) + .await; + + let err = result.expect_err("an unusable monitoring address must fail the relay"); + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToParseMonitoringAddr(..) + ) + ), + "expected the startup error to be surfaced, got: {err}" + ); + assert_eq!(attempts.get(), 1, "this failure is not retryable"); + + let http_addr = http_addr.take().expect("the fixture configured the relay"); + net::TcpListener::bind(&http_addr) + .await + .unwrap_or_else(|err| panic!("failed relay left {http_addr} bound: {err}")); + } + + #[tokio::test] + async fn fixture_stops_relay_and_releases_http_port() { + let http_addr = Cell::new(None); + + with_relay_server( + |_, _| {}, + async |cfg| { + http_addr.set(cfg.http_addr.clone()); + }, + ) + .await + .unwrap(); + + // The fixture returned, so the relay task has joined — and with it, its + // listener must be gone. + let http_addr = http_addr.take().expect("relay served an http address"); + net::TcpListener::bind(&http_addr) + .await + .unwrap_or_else(|err| panic!("relay did not release {http_addr}: {err}")); + } + + #[test] + fn is_addr_in_use_detects_bind_races_through_error_chain() { + let http_bind = super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToBindHttpListener { + addr: "127.0.0.1:1".into(), + source: io::Error::from(io::ErrorKind::AddrInUse), + }, + ); + assert!(is_addr_in_use(&http_bind)); + + let monitoring_bind = super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToBindMonitoringListener { + addr: "127.0.0.1:1".parse().unwrap(), + source: io::Error::from(io::ErrorKind::AddrInUse), + }, + ); + assert!(is_addr_in_use(&monitoring_bind)); + + let unrelated = + super::CliError::RelayP2PError(pluto_relay_server::RelayP2PError::FailedToServeHTTP( + io::Error::from(io::ErrorKind::ConnectionReset), + )); + assert!(!is_addr_in_use(&unrelated)); + } + + /// Make an HTTP GET request to the relay server. + /// + /// Single-shot on purpose: [`with_relay_server`] runs the test function + /// only once the relay serves, so a failure here is a real failure + /// rather than a startup race that retries would paper over. async fn relay_server_get( cfg: pluto_relay_server::config::Config, path: &str, ) -> Result { let http_address = cfg.http_addr.unwrap(); - retry_get(&format!("http://{}{}", http_address, path)).await + http_get(&format!("http://{http_address}{path}")).await } - async fn retry_get(url: &str) -> Result { - let request = async || reqwest::get(url).await.and_then(|r| r.error_for_status()); - let mut backoff = backon::ExponentialBuilder::default() - .with_min_delay(time::Duration::from_millis(200)) - .with_max_delay(time::Duration::from_secs(2)) - .with_factor(1.0) - .with_max_times(8) - .build(); - request.retry(&mut backoff).await + async fn http_get(url: &str) -> Result { + reqwest::get(url) + .await + .and_then(|response| response.error_for_status()) } } diff --git a/crates/p2p/src/utils.rs b/crates/p2p/src/utils.rs index 298056a2..05564239 100644 --- a/crates/p2p/src/utils.rs +++ b/crates/p2p/src/utils.rs @@ -203,3 +203,58 @@ pub fn filter_direct_quic_addrs(addrs: impl Iterator) -> Vec bool { !is_relay_addr(addr) } + +#[cfg(test)] +mod tests { + use super::*; + + /// Config with the listen addresses and external overrides under test. + fn config(external_ip: Option<&str>, external_host: Option<&str>) -> P2PConfig { + P2PConfig { + external_ip: external_ip.map(String::from), + external_host: external_host.map(String::from), + tcp_addrs: vec!["127.0.0.1:3610".to_string(), "127.0.0.1:3611".to_string()], + udp_addrs: vec!["127.0.0.1:3620".to_string(), "127.0.0.1:3621".to_string()], + ..Default::default() + } + } + + fn as_strings(addrs: &[Multiaddr]) -> Vec { + addrs.iter().map(ToString::to_string).collect() + } + + #[test] + fn external_multiaddrs_keep_the_listen_ports() { + let cfg = config(Some("1.2.3.4"), Some("relay.example.com")); + + // The external address replaces the listen IP but must advertise the + // port the node actually listens on — one address per listen port, IP + // forms first, then hostname forms. + assert_eq!( + as_strings(&external_tcp_multiaddrs(&cfg).unwrap()), + vec![ + "/ip4/1.2.3.4/tcp/3610", + "/ip4/1.2.3.4/tcp/3611", + "/dns/relay.example.com/tcp/3610", + "/dns/relay.example.com/tcp/3611", + ] + ); + assert_eq!( + as_strings(&external_udp_multiaddrs(&cfg).unwrap()), + vec![ + "/ip4/1.2.3.4/udp/3620/quic-v1", + "/ip4/1.2.3.4/udp/3621/quic-v1", + "/dns/relay.example.com/udp/3620/quic-v1", + "/dns/relay.example.com/udp/3621/quic-v1", + ] + ); + } + + #[test] + fn no_external_multiaddrs_without_external_config() { + let cfg = config(None, None); + + assert!(external_tcp_multiaddrs(&cfg).unwrap().is_empty()); + assert!(external_udp_multiaddrs(&cfg).unwrap().is_empty()); + } +} diff --git a/crates/relay-server/src/error.rs b/crates/relay-server/src/error.rs index 8cd118f3..8cf567c6 100644 --- a/crates/relay-server/src/error.rs +++ b/crates/relay-server/src/error.rs @@ -1,3 +1,5 @@ +use std::net::SocketAddr; + use libp2p::multiaddr; use pluto_p2p::p2p::P2PError; @@ -18,12 +20,35 @@ pub enum RelayP2PError { P2PConfigError(#[from] pluto_p2p::config::P2PConfigError), /// Failed to bind HTTP listener. - #[error("Failed to bind HTTP listener: {0}")] - FailedToBindHttpListener(String), + #[error("Failed to bind HTTP listener {addr}: {source}")] + FailedToBindHttpListener { + /// Address the listener could not be bound to. + addr: String, + /// Underlying bind error. Kept typed so callers can tell an + /// [`std::io::ErrorKind::AddrInUse`] race apart from a real + /// misconfiguration. + #[source] + source: std::io::Error, + }, /// Failed to serve HTTP. #[error("Failed to serve HTTP: {0}")] - FailedToServeHTTP(std::io::Error), + FailedToServeHTTP(#[source] std::io::Error), + + /// Failed to bind the monitoring listener. + #[error("Failed to bind monitoring listener {addr}: {source}")] + FailedToBindMonitoringListener { + /// Address the monitoring listener could not be bound to. + addr: SocketAddr, + /// Underlying bind error, kept typed for the same reason as + /// [`RelayP2PError::FailedToBindHttpListener`]. + #[source] + source: std::io::Error, + }, + + /// Failed to serve the monitoring API. + #[error("Failed to serve monitoring API: {0}")] + FailedToServeMonitoring(#[source] std::io::Error), /// Failed to parse multiaddress. #[error("Failed to parse multiaddress: {0}")] diff --git a/crates/relay-server/src/p2p.rs b/crates/relay-server/src/p2p.rs index a3ad389f..6f6badd4 100644 --- a/crates/relay-server/src/p2p.rs +++ b/crates/relay-server/src/p2p.rs @@ -15,7 +15,7 @@ use crate::{ config::{Config, create_relay_config}, error::RelayP2PError, metrics::{PeerWithPeerClusterLabels, RELAY_METRICS}, - web::{enr_server, monitoring_server}, + web::{bind_monitoring_server, enr_server, serve_monitoring_server}, }; use pluto_p2p::{ BandwidthFactory, PeerConnectionMetrics, @@ -77,6 +77,25 @@ pub async fn run_relay_p2p_node( let mut external_addrs = external_tcp_multiaddrs(&config.p2p_config)?; external_addrs.extend(external_udp_multiaddrs(&config.p2p_config)?); + // Bind the monitoring listener before starting the ENR server: an unusable + // address or a lost race for the port then fails the relay while nothing + // has been spawned, instead of returning past a running ENR server and + // leaving its listener bound. It also means a serving ENR endpoint implies + // every listener of this relay is bound. + let monitoring_server = match config.monitoring_addr.clone() { + Some(monitoring_addr) => { + let bind_addr = monitoring_addr + .parse::() + .map_err(|_| RelayP2PError::FailedToParseMonitoringAddr(monitoring_addr))?; + + Some(bind_monitoring_server(bind_addr, ct.child_token()).await?) + } + None => { + info!("Prometheus monitoring not available, since monitoring-address flag is not set"); + None + } + }; + let enr_server_handle = tokio::spawn(enr_server( server_errors.clone(), config.clone(), @@ -93,16 +112,13 @@ pub async fn run_relay_p2p_node( info!("Runtime multiaddrs not available via http, since http-address flag is not set"); } - // Start monitoring server if configured - let monitoring_handle = if let Some(monitoring_addr) = config.monitoring_addr.clone() { - let bind_addr = monitoring_addr - .parse::() - .map_err(|_| RelayP2PError::FailedToParseMonitoringAddr(monitoring_addr))?; - Some(tokio::spawn(monitoring_server(bind_addr, ct.child_token()))) - } else { - info!("Prometheus monitoring not available, since monitoring-address flag is not set"); - None - }; + // Serve the monitoring listener bound above. + let monitoring_handle = monitoring_server + .map(|server| tokio::spawn(serve_monitoring_server(server_errors.clone(), server))); + + // Set when one of the HTTP servers fails; returned once the shutdown below + // has run, so a failed relay never leaves listeners bound behind it. + let mut server_error = None; loop { tokio::select! { @@ -114,7 +130,8 @@ pub async fn run_relay_p2p_node( error = server_errors_receiver.recv() => { if let Some(error) = error { warn!("Server error: {}", error); - return Err(error); + server_error = Some(error); + break; } }, event = node.select_next_some() => { @@ -168,7 +185,10 @@ pub async fn run_relay_p2p_node( } } - Ok(node) + match server_error { + Some(error) => Err(error), + None => Ok(node), + } } /// Result of a swarm event that may require updating the listener address list. diff --git a/crates/relay-server/src/web.rs b/crates/relay-server/src/web.rs index 4a5c47a8..1dfe84f8 100644 --- a/crates/relay-server/src/web.rs +++ b/crates/relay-server/src/web.rs @@ -21,11 +21,11 @@ use tokio::{ }; use tokio_util::sync::CancellationToken; use tracing::{debug, info, instrument, warn}; -use vise_exporter::MetricsExporter; +use vise_exporter::{MetricsExporter, MetricsServer}; use crate::{ config::{Config, EXTERNAL_HOST_RESOLVE_INTERVAL}, - error::RelayP2PError, + error::{RelayP2PError, Result}, }; use pluto_p2p::{config::P2PConfig, name::peer_name}; @@ -114,6 +114,23 @@ pub async fn enr_server( info!("Starting ENR server"); + // Bind before spawning anything else, so a failed bind has nothing to + // clean up. The error keeps its `io::ErrorKind` so callers can tell a lost + // race for the port from an unusable address. + let listener = match TcpListener::bind(&http_addr).await { + Ok(listener) => listener, + Err(err) => { + warn!("Failed to bind HTTP listener to {http_addr}: {err}"); + let _ = server_errors + .send(RelayP2PError::FailedToBindHttpListener { + addr: http_addr, + source: err, + }) + .await; + return; + } + }; + let state = AppState::new( config.p2p_config.clone(), secret_key, @@ -139,14 +156,6 @@ pub async fn enr_server( .route("/enr", get(enr_handler)) .with_state(state_arc); - let Ok(listener) = TcpListener::bind(&http_addr).await else { - warn!("Failed to bind HTTP listener to {}", http_addr); - let _ = server_errors - .send(RelayP2PError::FailedToBindHttpListener(http_addr)) - .await; - return; - }; - info!( "Relay started {peer_name} on {tcp_addrs} and {udp_addrs}", peer_name = peer_name(&peer_id), @@ -175,16 +184,47 @@ pub async fn enr_server( } } -/// Starts the Prometheus monitoring server on the given address. +/// Binds the Prometheus monitoring listener on the given address. +/// +/// Binding is separate from serving so that a bind failure is reported to the +/// caller synchronously — before any other listener is started — and keeps its +/// `io::ErrorKind`, letting callers tell a lost race for the port from a real +/// misconfiguration. #[instrument(skip(ct))] -pub async fn monitoring_server(bind_addr: SocketAddr, ct: CancellationToken) { - info!("Starting monitoring server on {bind_addr}"); +pub(crate) async fn bind_monitoring_server( + bind_addr: SocketAddr, + ct: CancellationToken, +) -> Result> { + info!("Binding monitoring server"); MetricsExporter::default() .with_graceful_shutdown(ct.cancelled_owned()) - .start(bind_addr) + .bind(bind_addr) .await - .unwrap_or_else(|e| warn!("Monitoring server error: {e}")); + .map_err(|source| RelayP2PError::FailedToBindMonitoringListener { + addr: bind_addr, + source, + }) +} + +/// Serves an already bound monitoring listener until shutdown. +/// +/// Serve failures are reported on `server_errors` (mirroring Charon, where the +/// monitoring server's `ListenAndServe` error terminates the relay) so they +/// cannot go unnoticed behind a log line. +#[instrument(skip_all, fields(addr = %server.local_addr()))] +pub(crate) async fn serve_monitoring_server( + server_errors: mpsc::Sender, + server: MetricsServer<'static>, +) { + info!("Starting monitoring server"); + + if let Err(err) = server.start().await { + warn!("Monitoring server error: {err}"); + let _ = server_errors + .send(RelayP2PError::FailedToServeMonitoring(err)) + .await; + } } /// Error response for HTTP handlers. From 480c6fb01577e11a25ca00ee9b956835f0ff36e0 Mon Sep 17 00:00:00 2001 From: Quang Le Date: Fri, 7 Aug 2026 16:34:18 +0700 Subject: [PATCH 2/3] fix(relay): bind and report every listener before serving --- crates/cli/src/commands/relay.rs | 893 +++++++----------- crates/p2p/src/p2p.rs | 61 +- crates/p2p/src/utils.rs | 139 ++- crates/relay-server/src/error.rs | 18 +- crates/relay-server/src/lib.rs | 2 +- crates/relay-server/src/p2p.rs | 440 ++++++--- crates/relay-server/src/web.rs | 148 +-- crates/relay-server/tests/http_integration.rs | 95 +- 8 files changed, 944 insertions(+), 852 deletions(-) diff --git a/crates/cli/src/commands/relay.rs b/crates/cli/src/commands/relay.rs index 99b10d03..3fe9ab36 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -346,6 +346,19 @@ async fn serve_relay( info!("{LICENSE}"); info!(config = ?config); + let key = load_or_create_key(config)?; + + pluto_relay_server::p2p::run_relay_p2p_node(config, key, ct) + .await + .map(|_| ()) + .map_err(Into::into) +} + +/// Loads the relay's p2p key from its data dir, generating and persisting one +/// when it is missing and `--auto-p2pkey` is set. +fn load_or_create_key( + config: &pluto_relay_server::config::Config, +) -> Result { let key = match pluto_p2p::k1::load_priv_key(&config.data_dir) { Ok(key) => Ok(key), Err(pluto_p2p::k1::K1Error::K1UtilError(pluto_k1util::K1UtilError::FailedToReadFile( @@ -369,18 +382,16 @@ async fn serve_relay( e => e, }?; - pluto_relay_server::p2p::run_relay_p2p_node(config, key, ct) - .await - .map(|_| ()) - .map_err(Into::into) + Ok(key) } #[cfg(test)] mod tests { use std::{ - cell::Cell, - io, + net::{Ipv4Addr, SocketAddr}, + path::Path, str::FromStr, + sync::LazyLock, time::{Duration, Instant}, }; use tokio::{net, task::JoinHandle}; @@ -388,282 +399,257 @@ mod tests { #[tokio::test] async fn run_bootnode() { - with_relay_server( - |args, _| { - args.relay.auto_p2p_key = false; - pluto_p2p::k1::new_saved_priv_key(&args.data_dir.data_dir).unwrap(); - }, - async |_| { /* Relay server starts with existing p2p key */ }, - ) + // Relay server starts with the existing p2p key. + let _relay = test_relay_server_with(|args| { + args.relay.auto_p2p_key = false; + pluto_p2p::k1::new_saved_priv_key(&args.data_dir.data_dir).unwrap(); + }) .await .unwrap(); } #[tokio::test] async fn run_bootnode_auto_p2p() { - let first_run = with_relay_server( - |args, _| { - args.relay.auto_p2p_key = false; - }, - async |_| { /* Relay server does not start due to missing p2p key */ }, - ) - .await; + // Relay server does not start due to the missing p2p key. + let missing_key = test_relay_server_with(|args| args.relay.auto_p2p_key = false).await; assert!(matches!( - first_run, + missing_key, Err(super::CliError::RelayP2PError( pluto_relay_server::RelayP2PError::FailedToLoadPrivateKey(..) )) )); - let second_run = with_relay_server( - |_, _| {}, - async |_| { /* Relay server starts with auto-generated p2p key */ }, - ) - .await; - assert!(matches!(second_run, Ok(()))); + // The success path — starting with an auto-generated key — is what + // every other test here does, since `relay_args` sets + // `auto_p2p_key`. + } + + #[tokio::test] + async fn run_exits_when_cancelled() { + let dir = tempfile::tempdir().unwrap(); + let args = relay_args(dir.path()); + + // Covers the CLI entry point that the fixture bypasses: tracing init and + // the Loki drain. A pre-cancelled token is deterministic because the + // shutdown arm of the serve loop is the `biased` first branch — the same + // way charon's relay test starts (`cmd/relay/relay_internal_test.go:40`). + let ct = CancellationToken::new(); + ct.cancel(); + + super::run(args.try_into().unwrap(), ct).await.unwrap(); } #[tokio::test] async fn serve_addr_multiaddrs() { - with_relay_server( - |_, _| {}, - async |cfg| { - let response = relay_server_get(cfg, "/").await.unwrap(); - let body = response.text().await.unwrap(); - let addresses: Vec = serde_json::from_str(&body).unwrap(); + let relay = test_relay_server().await.unwrap(); - assert!( - !addresses.is_empty(), - "Expected at least one multiaddr in response" - ); + let response = http_get(&relay.url("/")).await.unwrap(); + let body = response.text().await.unwrap(); + let addresses: Vec = serde_json::from_str(&body).unwrap(); - for addr in addresses { - libp2p::Multiaddr::from_str(&addr).unwrap_or_else(|err| { - panic!("Failed to parse multiaddr '{}': {}", addr, err); - }); - } - }, - ) - .await - .unwrap(); + assert!( + !addresses.is_empty(), + "Expected at least one multiaddr in response" + ); + + for addr in addresses { + libp2p::Multiaddr::from_str(&addr).unwrap_or_else(|err| { + panic!("Failed to parse multiaddr '{}': {}", addr, err); + }); + } } #[tokio::test] async fn serve_addr_enr() { - with_relay_server( - |_, _| {}, - async |cfg| { - let response = relay_server_get(cfg, "/enr").await.unwrap(); - let body = response.text().await.unwrap(); - let enr = pluto_eth2util::enr::Record::try_from(body.as_str()).unwrap(); + let relay = test_relay_server().await.unwrap(); - assert_eq!(enr.ip(), Some(std::net::Ipv4Addr::new(127, 0, 0, 1))); - }, - ) - .await - .unwrap(); + let response = http_get(&relay.url("/enr")).await.unwrap(); + let enr = parse_enr(&response.text().await.unwrap()); + + assert_eq!(enr.ip(), Some(Ipv4Addr::new(127, 0, 0, 1))); } #[tokio::test] async fn serve_addr_enr_ext_ip() { - with_relay_server( - |args, _| args.p2p.external_ip = Some("222.222.222.222".into()), - async |cfg| { - let response = relay_server_get(cfg, "/enr").await.unwrap(); - let body = response.text().await.unwrap(); - let enr = pluto_eth2util::enr::Record::try_from(body.as_str()).unwrap(); + let relay = + test_relay_server_with(|args| args.p2p.external_ip = Some("222.222.222.222".into())) + .await + .unwrap(); - assert_eq!(enr.ip(), Some(std::net::Ipv4Addr::new(222, 222, 222, 222))); - }, - ) - .await - .unwrap(); + let response = http_get(&relay.url("/enr")).await.unwrap(); + let enr = parse_enr(&response.text().await.unwrap()); + + assert_eq!(enr.ip(), Some(Ipv4Addr::new(222, 222, 222, 222))); + // The external IP is advertised on the ports libp2p bound, not on the + // port 0 that was configured — which would be undialable. + assert_eq!(enr.tcp(), Some(relay.p2p_port(pluto_p2p::utils::tcp_port))); + assert_eq!(enr.udp(), Some(relay.p2p_port(pluto_p2p::utils::udp_port))); } #[tokio::test] async fn serve_addr_enr_ext_host() { - with_relay_server( - |args, _| args.p2p.external_host = Some("www.google.com".into()), - async |cfg| { - // Resolution happens asynchronously on a tick, so poll until the - // ENR reflects a non-loopback IP (mirrors the Go test using - // `assert.Eventually`). - tokio::time::timeout(Duration::from_secs(10), async { - loop { - let response = relay_server_get(cfg.clone(), "/enr").await.unwrap(); - let body = response.text().await.unwrap(); - let enr = pluto_eth2util::enr::Record::try_from(body.as_str()).unwrap(); - let ip = enr.ip().unwrap(); - - if !ip.is_loopback() { - break; - } - - tokio::time::sleep(Duration::from_millis(200)).await; - } - }) + let relay = + test_relay_server_with(|args| args.p2p.external_host = Some("www.google.com".into())) .await - .expect("external host never resolved to non-loopback ip"); - }, - ) - .await - .unwrap(); + .unwrap(); + + // Resolution happens asynchronously on a tick, so wait until the ENR + // reflects a non-loopback IP (mirrors the Go test using + // `assert.Eventually`). + relay + .get_until("/enr", |body| !parse_enr(body).ip().unwrap().is_loopback()) + .await; } #[tokio::test] async fn serve_addr_metrics() { - with_relay_server( - |args, monitoring_addr| { - args.debug_monitoring.monitor_addr = Some(monitoring_addr.into()); - }, - async |cfg| { - let monitoring_addr = cfg.monitoring_addr.unwrap(); - let response = http_get(&format!("http://{monitoring_addr}/metrics")) - .await - .unwrap(); - let body = response.text().await.unwrap(); - - assert!(body.contains("relay_p2p_connection_total")); - assert!(body.contains("relay_p2p_active_connections")); - assert!(body.contains("relay_p2p_ping_latency")); - assert!(body.contains("relay_p2p_network_sent_bytes")); - assert!(body.contains("relay_p2p_network_receive_bytes")); - assert!(body.ends_with("# EOF\n")); - }, - ) + // The monitoring port used to be guessed by the fixture, and losing the + // race for it was the most frequent pre-fix failure. It is now assigned + // by the kernel inside the relay's own bind and read back off it. + let relay = test_relay_server_with(|args| { + args.debug_monitoring.monitor_addr = Some(ANY_ADDR.into()); + }) .await .unwrap(); - } - /// Number of complete relay startup attempts — each with a fresh data dir - /// and freshly allocated HTTP ports — before the fixture gives up on a bind - /// race and surfaces the error. - const MAX_STARTUP_ATTEMPTS: usize = 5; + let monitoring_addr = relay.monitoring_addr.expect("monitoring was configured"); + let response = http_get(&format!("http://{monitoring_addr}/metrics")) + .await + .unwrap(); + let body = response.text().await.unwrap(); + + assert!(body.contains("relay_p2p_connection_total")); + assert!(body.contains("relay_p2p_active_connections")); + assert!(body.contains("relay_p2p_ping_latency")); + assert!(body.contains("relay_p2p_network_sent_bytes")); + assert!(body.contains("relay_p2p_network_receive_bytes")); + assert!(body.ends_with("# EOF\n")); + } - /// Per-attempt budget for the relay's HTTP servers to start serving. Sized - /// for a heavily loaded CI machine: the relay either serves or exits with - /// an error long before this, so exceeding it means it hung. - const SERVING_TIMEOUT: Duration = Duration::from_secs(30); + #[tokio::test] + async fn taken_http_port_fails_the_relay() { + let (_taken, addr) = squat_tcp_addr().await; - /// Budget for the relay to stop once the test function has returned. - const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); + let err = test_relay_server_with(|args| args.relay.http_address = addr) + .await + .expect_err("relay must not start while its http port is taken"); - /// libp2p listen address for the fixture: port 0 lets the kernel assign the - /// port inside libp2p's own `bind`, so no other process can claim it in - /// between. - /// - /// The HTTP listeners can't do that: their addresses are inbound config the - /// relay never reports back (as in Charon, whose relay test passes - /// `HTTPAddr` in), so the fixture allocates them up front and retries the - /// race instead. A p2p race could not be retried anyway — libp2p buries the - /// `AddrInUse` inside `io::Error::other(Transport(..))`, out of reach of - /// `Error::source` and therefore of [`is_addr_in_use`]. - /// - /// The cost is that external multiaddrs derived from these listen ports - /// carry port 0; no test asserts on them (Charon's assert only on the ENR's - /// IP), and the listen-port-to-advertised-port mapping is covered - /// separately by `external_multiaddrs_keep_the_listen_ports` in - /// `pluto_p2p::utils`. - const P2P_LISTEN_ADDR: &str = "127.0.0.1:0"; - - /// Allocates a loopback address by binding port 0, reading back the - /// assigned port and dropping the socket. - /// - /// The relay rebinds that port moments later, so another process can claim - /// it in between; [`with_relay_server`] retries the whole startup with a - /// new address when that happens. - async fn free_tcp_addr() -> String { - squat_tcp_addr().await.1 + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToBindHttpListener { .. } + ) + ), + "got: {err}" + ); } - /// A relay that has been observed serving every HTTP endpoint it was - /// configured with. - struct ServingRelay { - /// Config the relay was started with. - cfg: pluto_relay_server::config::Config, - /// Cancels the relay. - ct: CancellationToken, - /// Relay task, resolving with the relay's exit status. - handle: JoinHandle>, - /// Data dir of this attempt, kept alive while the relay runs. - dir: tempfile::TempDir, + #[tokio::test] + async fn taken_monitoring_port_fails_the_relay() { + let (_taken, addr) = squat_tcp_addr().await; + + let err = test_relay_server_with(|args| args.debug_monitoring.monitor_addr = Some(addr)) + .await + .expect_err("relay must not start while its monitoring port is taken"); + + // A monitoring bind failure used to be only `warn!`-ed, which left the + // relay running with the port unserved. + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToBindMonitoringListener { .. } + ) + ), + "got: {err}" + ); } - /// Run a function in the context of a running relay server. - /// - /// The server can be configured before initialization through - /// [`super::RelayArgs`]; the closure also receives a loopback address - /// allocated for the attempt, for tests that opt into the monitoring - /// server. It is invoked once per startup attempt, so it must not assume it - /// runs only once. - /// - /// The test function runs only once the relay serves every HTTP endpoint it - /// was configured with, and receives the config the relay was started with. - /// Startup and shutdown errors are returned to the caller instead of - /// showing up as connection failures inside the test function. - async fn with_relay_server( - mut config_fn: FArgs, - test_fn: FTest, - ) -> Result<(), crate::error::CliError> - where - FArgs: FnMut(&mut super::RelayArgs, &str), - FTest: FnOnce(pluto_relay_server::config::Config) -> Fut, - Fut: std::future::Future, - { - let mut attempts: usize = 0; - - let ServingRelay { - cfg, - ct, - handle, - dir, - } = loop { - attempts = attempts.saturating_add(1); - - match start_relay(&mut config_fn).await { - Ok(relay) => break relay, - // Another process claimed one of the freshly allocated ports - // before the relay could bind it. The relay task is gone for - // good — request retries in the test function could never - // recover — so start over with new ports, boundedly. - Err(err) if is_addr_in_use(&err) && attempts < MAX_STARTUP_ATTEMPTS => { - tracing::debug!("relay lost the race for a port, retrying: {err}"); - } - Err(err) => return Err(err), - } - }; + #[tokio::test] + async fn unusable_monitoring_addr_fails_the_relay() { + let err = test_relay_server_with(|args| { + args.debug_monitoring.monitor_addr = Some("not-an-address".into()); + }) + .await + .expect_err("an unusable monitoring address must fail the relay"); - test_fn(cfg).await; + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToParseMonitoringAddr(..) + ) + ), + "got: {err}" + ); + } - ct.cancel(); - let exit = match tokio::time::timeout(SHUTDOWN_TIMEOUT, handle).await { - Ok(Ok(exit)) => exit, - Ok(Err(err)) => resume_relay_panic(err), - Err(_) => panic!("relay did not shut down within {SHUTDOWN_TIMEOUT:?}"), - }; + #[tokio::test] + async fn stopping_the_relay_releases_its_http_port() { + let mut relay = test_relay_server().await.unwrap(); + let http_addr = relay.http_addr; - // The relay has stopped, so nothing reads the data dir anymore. - drop(dir); + relay.stop().await.unwrap(); - exit + // The relay task has joined, and with it its listener must be gone. + net::TcpListener::bind(http_addr) + .await + .unwrap_or_else(|err| panic!("relay did not release {http_addr}: {err}")); } - /// Starts one relay with freshly allocated addresses and waits until it - /// either serves or exits. - async fn start_relay( - config_fn: &mut impl FnMut(&mut super::RelayArgs, &str), - ) -> Result { + #[test] + fn advertise_priv_inverts_filter_private_addrs() { + // The flag is the inverse of the config knob it feeds, and every test + // above depends on the inversion holding: with private addresses + // filtered, the loopback listeners never reach `/enr`. let dir = tempfile::tempdir().unwrap(); - // Only bound when a test opts into the monitoring server by setting - // `super::RelayDebugMonitoringArgs::monitor_addr` to it. - let monitoring_addr = free_tcp_addr().await; + let mut args = relay_args(dir.path()); + + let config: pluto_relay_server::config::Config = args.clone().try_into().unwrap(); + assert!(!config.filter_private_addrs, "advertise_priv: true"); + + args.relay.advertise_priv = false; + let config: pluto_relay_server::config::Config = args.try_into().unwrap(); + assert!(config.filter_private_addrs, "advertise_priv: false"); + } - let mut args = super::RelayArgs { + /// Budget for the relay to stop once a test is done with it. + const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); + + /// Budget for an endpoint to reach the state a test waits for. Sized for a + /// heavily loaded CI machine. + const SERVING_TIMEOUT: Duration = Duration::from_secs(30); + + /// Per-request budget, so a server that accepts the connection but never + /// answers fails the test instead of hanging it until the harness gives up. + const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); + + /// Loopback address with an ephemeral port, used for *every* listener in + /// these tests. + /// + /// Port 0 lets the kernel assign the port inside the relay's own `bind` and + /// [`TestRelay`] reads back what was bound, so nothing here names a port + /// another process could take first. That is what makes the bind race these + /// tests used to lose unrepresentable rather than merely unlikely. + const ANY_ADDR: &str = "127.0.0.1:0"; + + /// Shared HTTP client: building one per request re-reads the system CA + /// store and throws away the connection pool. + static CLIENT: LazyLock = LazyLock::new(reqwest::Client::new); + + /// Relay arguments every test starts from: all listeners on [`ANY_ADDR`], + /// quiet logs, no relays to dial. + /// + /// `advertise_priv` is load-bearing: without it, `filter_private_addrs` + /// drops the loopback listen addresses, and `/enr` answers 500 forever. + fn relay_args(data_dir: &Path) -> super::RelayArgs { + super::RelayArgs { data_dir: super::RelayDataDirArgs { - data_dir: dir.path().to_path_buf(), + data_dir: data_dir.to_path_buf(), }, relay: super::RelayRelayArgs { - http_address: free_tcp_addr().await, + http_address: ANY_ADDR.into(), auto_p2p_key: true, p2p_relay_log_level: "info".into(), max_res_per_peer: 0, @@ -678,8 +664,8 @@ mod tests { relays: vec![], external_ip: None, external_host: None, - tcp_addrs: vec![P2P_LISTEN_ADDR.into()], - udp_addrs: vec![P2P_LISTEN_ADDR.into()], + tcp_addrs: vec![ANY_ADDR.into()], + udp_addrs: vec![ANY_ADDR.into()], disable_reuseport: false, }, log: super::RelayLogFlags { @@ -692,336 +678,189 @@ mod tests { loki_addresses: vec![], loki_service: "".into(), }, - }; - config_fn(&mut args, &monitoring_addr); - - let cfg: pluto_relay_server::config::Config = args.try_into().unwrap(); - let ct = CancellationToken::new(); - let mut handle = tokio::spawn(super::run(cfg.clone(), ct.child_token())); - - // Wait for the relay to serve, or to exit trying. A failed listener - // bind takes the relay down permanently, and awaiting the relay only - // after the test function would let a request `unwrap()` mask it as - // `ConnectionRefused`. - let serving = tokio::select! { - joined = &mut handle => { - // The relay is gone; cancel so nothing it spawned outlives it - // and keeps a listener bound into the next attempt. - ct.cancel(); - - return match joined { - Ok(Ok(())) => panic!("relay exited before serving {:?}", cfg.http_addr), - Ok(Err(err)) => Err(err), - Err(err) => resume_relay_panic(err), - }; - }, - serving = wait_until_serving(&cfg) => serving, - }; - - if let Err(err) = serving { - // The relay is alive but never served: not a bind race. Take it - // down so the failure isn't followed by a leaked relay. - ct.cancel(); - let _ = tokio::time::timeout(SHUTDOWN_TIMEOUT, handle).await; - panic!("{err}"); } - - Ok(ServingRelay { - cfg, - ct, - handle, - dir, - }) } - /// Polls every HTTP endpoint the relay is configured to serve until each - /// one answers, returning a description of whichever never came up within - /// [`SERVING_TIMEOUT`]. + /// A relay that serves for as long as this value is alive. /// - /// `/enr` stands in for the ENR server as a whole: it shares a listener - /// with `/` and only succeeds once libp2p has reported its listen - /// addresses, so test functions can request either without racing startup. - async fn wait_until_serving(cfg: &pluto_relay_server::config::Config) -> Result<(), String> { - let client = reqwest::Client::new(); - let started = Instant::now(); - - let urls = [ - cfg.http_addr - .as_ref() - .map(|addr| format!("http://{addr}/enr")), - cfg.monitoring_addr - .as_ref() - .map(|addr| format!("http://{addr}/metrics")), - ] - .into_iter() - .flatten(); - - for url in urls { - while let Err(err) = client - .get(&url) - .timeout(Duration::from_secs(1)) - .send() - .await - .and_then(|response| response.error_for_status()) - { - if started.elapsed() >= SERVING_TIMEOUT { - return Err(format!( - "{url} still not serving {SERVING_TIMEOUT:?} into startup: {err}" - )); - } - - tokio::time::sleep(Duration::from_millis(20)).await; - } - } + /// It is already serving when a test receives it, and stops when the value + /// goes out of scope, so a test body is just requests and assertions. + #[derive(Debug)] + struct TestRelay { + /// Address the ENR/multiaddr HTTP server is bound to — the one the + /// kernel assigned, read back off the bound listener. + http_addr: SocketAddr, + /// Address the monitoring server is bound to, when one was configured. + monitoring_addr: Option, + /// Addresses libp2p bound, with the ports the kernel assigned. + p2p_addrs: Vec, + /// Cancels the relay. + ct: CancellationToken, + /// Relay task, resolving with the relay's exit status. + handle: JoinHandle>, + /// Data dir, kept alive for as long as the relay is. + _dir: tempfile::TempDir, + } - Ok(()) + /// Starts a serving relay with the default test arguments. See + /// [`test_relay_server_with`]. + async fn test_relay_server() -> Result { + test_relay_server_with(|_| {}).await } - /// Reports whether `err`'s source chain contains an - /// [`io::ErrorKind::AddrInUse`] error, i.e. whether a listener lost the - /// race for a port that had just been allocated. + /// Starts a relay in a fresh data dir, letting `configure` adjust the + /// [`super::RelayArgs`] it is built from. /// - /// Walks the chain instead of matching error strings so the typed - /// `io::ErrorKind` decides: both relay bind errors - /// ([`pluto_relay_server::RelayP2PError::FailedToBindHttpListener`] and its - /// monitoring counterpart) keep the original `io::Error`. - /// - /// Those two are the only bind races this fixture can hit; the p2p - /// listeners use port 0 instead — see [`P2P_LISTEN_ADDR`]. - fn is_addr_in_use(err: &(dyn std::error::Error + 'static)) -> bool { - let mut next = Some(err); - - while let Some(err) = next { - if let Some(io_err) = err.downcast_ref::() - && io_err.kind() == io::ErrorKind::AddrInUse - { - return true; - } + /// Returns once every listener is bound *and* libp2p has reported the + /// addresses it got, so requests against the returned addresses are served + /// straight away — no readiness poll, and a failure in the test that + /// follows is a real failure rather than a startup race. Every startup + /// failure — an unloadable key, an unusable address, a port that is + /// genuinely taken — comes back as `Err` here rather than reaching a + /// test as a connection failure further down. + async fn test_relay_server_with( + configure: impl FnOnce(&mut super::RelayArgs), + ) -> Result { + let dir = tempfile::tempdir().unwrap(); + let mut args = relay_args(dir.path()); + configure(&mut args); - next = err.source(); - } + let config: pluto_relay_server::config::Config = args.try_into()?; + let key = super::load_or_create_key(&config)?; - false + let ct = CancellationToken::new(); + let bound = pluto_relay_server::p2p::bind_relay(&config, key, ct.child_token()).await?; + + // Read the addresses off the bound relay before serving consumes it. + let http_addr = bound + .http_addr() + .expect("`relay_args` configures an http address"); + let monitoring_addr = bound.monitoring_addr(); + let p2p_addrs = bound.p2p_addrs().await; + + let handle = + tokio::spawn(async move { bound.serve().await.map(|_| ()).map_err(Into::into) }); + + Ok(TestRelay { + http_addr, + monitoring_addr, + p2p_addrs, + ct, + handle, + _dir: dir, + }) } - /// Re-raises a panic from the relay task in the test thread, so the - /// original panic message is what the test reports. - fn resume_relay_panic(err: tokio::task::JoinError) -> ! { - if err.is_panic() { - std::panic::resume_unwind(err.into_panic()); + impl TestRelay { + /// URL for `path` on the relay's HTTP server. + fn url(&self, path: &str) -> String { + format!("http://{}{path}", self.http_addr) } - panic!("relay task was cancelled: {err}"); - } - - /// Binds a loopback port and holds it, so anything else binding the - /// returned address fails with [`io::ErrorKind::AddrInUse`] — the failure a - /// concurrently started process causes by claiming a port between - /// [`free_tcp_addr`] and the relay's own bind. - async fn squat_tcp_addr() -> (net::TcpListener, String) { - let listener = net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap().to_string(); - (listener, addr) - } - - #[tokio::test] - async fn fixture_recovers_from_transient_bind_race() { - let (squatter, squatted) = squat_tcp_addr().await; - // Released at the start of the second attempt, so exactly one bind - // loses the race — the transient failure seen when tests run - // concurrently. - let mut squatter = Some(squatter); - let attempts = Cell::new(0usize); - - with_relay_server( - |args, _| { - attempts.set(attempts.get().saturating_add(1)); - if attempts.get() > 1 { - squatter.take(); - } - args.relay.http_address = squatted.clone(); - }, - async |cfg| { - let response = relay_server_get(cfg, "/enr").await.unwrap(); - - assert!(response.status().is_success(), "{}", response.status()); - }, - ) - .await - .unwrap(); - - assert_eq!( - attempts.get(), - 2, - "the relay should have started on the second attempt" - ); - } - - /// Runs the fixture with `configure` pointing one of the relay's listeners - /// at a port held for the whole run, so every attempt loses the race for it - /// as if a concurrent process had claimed it. - /// - /// Asserts that the fixture retried boundedly and gave up with an - /// `AddrInUse` error, and returns that error so the caller can check which - /// listener reported it. - async fn exhaust_retries_on_taken_port( - configure: impl Fn(&mut super::RelayArgs, String), - ) -> super::CliError { - let (_squatter, squatted) = squat_tcp_addr().await; - let attempts = Cell::new(0usize); - - let result = with_relay_server( - |args, _| { - attempts.set(attempts.get().saturating_add(1)); - configure(args, squatted.clone()); - }, - async |_| panic!("test function must not run when a listener cannot bind"), - ) - .await; - - let err = result.expect_err("relay must not serve while one of its ports is taken"); - assert!( - is_addr_in_use(&err), - "expected an AddrInUse error, got: {err}" - ); - assert_eq!( - attempts.get(), - MAX_STARTUP_ATTEMPTS, - "bind races must be retried, but boundedly" - ); + /// Port of the relay's libp2p listen address selected by `port_of`, + /// e.g. [`pluto_p2p::utils::tcp_port`]. + fn p2p_port(&self, port_of: impl Fn(&libp2p::Multiaddr) -> Option) -> u16 { + self.p2p_addrs + .iter() + .find_map(port_of) + .expect("`relay_args` configures both transports") + } - err - } + /// Fetches `path` until it answers 2xx *and* `ready` accepts the body, + /// returning that body. + /// + /// This is not race tolerance — the relay is fully bound and serving + /// before a test gets hold of it, so a request can never be refused and + /// a transport error panics instead of being retried. The one thing it + /// waits out is DNS: `--p2p-external-hostname` is resolved on a tick by + /// a background task, so the ENR reflects it only after the first + /// lookup answers. Charon waits the same way (`assert.Eventually`, + /// `cmd/relay/relay_internal_test.go:208`). + async fn get_until(&self, path: &str, ready: impl Fn(&str) -> bool) -> String { + let started = Instant::now(); + + loop { + let response = CLIENT + .get(self.url(path)) + .timeout(REQUEST_TIMEOUT) + .send() + .await + .unwrap_or_else(|err| { + panic!( + "GET {path} failed: {err} (relay exited: {})", + self.handle.is_finished() + ) + }); - #[tokio::test] - async fn fixture_retries_bind_races_boundedly() { - let err = exhaust_retries_on_taken_port(|args, addr| args.relay.http_address = addr).await; - assert!( - matches!( - err, - super::CliError::RelayP2PError( - pluto_relay_server::RelayP2PError::FailedToBindHttpListener { .. } - ) - ), - "expected the bind error to be surfaced, got: {err}" - ); + let status = response.status(); + let body = response.text().await.unwrap(); - // The monitoring listener behaves the same way. It used to be only - // `warn!`-ed, which left the relay running and the port unserved — the - // most frequent pre-fix failure in the stress runs. - let err = exhaust_retries_on_taken_port(|args, addr| { - args.debug_monitoring.monitor_addr = Some(addr); - }) - .await; - assert!( - matches!( - err, - super::CliError::RelayP2PError( - pluto_relay_server::RelayP2PError::FailedToBindMonitoringListener { .. } - ) - ), - "expected the monitoring bind error to be surfaced, got: {err}" - ); - } + if status.is_success() && ready(&body) { + return body; + } - #[tokio::test] - async fn fixture_leaves_no_listener_bound_when_startup_fails() { - let http_addr = Cell::new(None); - let attempts = Cell::new(0usize); - - let result = with_relay_server( - |args, _| { - attempts.set(attempts.get().saturating_add(1)); - http_addr.set(Some(args.relay.http_address.clone())); - // Rejected while starting up, after the HTTP address has been - // configured: the relay must fail without leaving its HTTP - // listener bound behind. - args.debug_monitoring.monitor_addr = Some("not-an-address".into()); - }, - async |_| panic!("test function must not run when startup fails"), - ) - .await; + // The relay is the only thing serving this address, so if it is + // gone nothing will ever satisfy the poll. + assert!( + !self.handle.is_finished(), + "relay exited while waiting for {path} to serve" + ); + assert!( + started.elapsed() < SERVING_TIMEOUT, + "{path} not ready {SERVING_TIMEOUT:?} into startup; last status {status}: {body}" + ); - let err = result.expect_err("an unusable monitoring address must fail the relay"); - assert!( - matches!( - err, - super::CliError::RelayP2PError( - pluto_relay_server::RelayP2PError::FailedToParseMonitoringAddr(..) - ) - ), - "expected the startup error to be surfaced, got: {err}" - ); - assert_eq!(attempts.get(), 1, "this failure is not retryable"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + } - let http_addr = http_addr.take().expect("the fixture configured the relay"); - net::TcpListener::bind(&http_addr) - .await - .unwrap_or_else(|err| panic!("failed relay left {http_addr} bound: {err}")); + /// Cancels the relay, waits for it to stop, and returns its exit + /// status. + /// + /// Only needed by tests that assert on what stopping released; dropping + /// the value is enough everywhere else. + async fn stop(&mut self) -> Result<(), super::CliError> { + self.ct.cancel(); + + match tokio::time::timeout(SHUTDOWN_TIMEOUT, &mut self.handle).await { + Ok(Ok(exit)) => exit, + Ok(Err(err)) => panic!("relay task did not join: {err}"), + Err(_) => panic!("relay did not shut down within {SHUTDOWN_TIMEOUT:?}"), + } + } } - #[tokio::test] - async fn fixture_stops_relay_and_releases_http_port() { - let http_addr = Cell::new(None); - - with_relay_server( - |_, _| {}, - async |cfg| { - http_addr.set(cfg.http_addr.clone()); - }, - ) - .await - .unwrap(); - - // The fixture returned, so the relay task has joined — and with it, its - // listener must be gone. - let http_addr = http_addr.take().expect("relay served an http address"); - net::TcpListener::bind(&http_addr) - .await - .unwrap_or_else(|err| panic!("relay did not release {http_addr}: {err}")); + impl Drop for TestRelay { + /// Best-effort: a `Drop` cannot await the task, which is why + /// [`TestRelay::stop`] exists for tests that need it fully stopped. + fn drop(&mut self) { + self.ct.cancel(); + } } - #[test] - fn is_addr_in_use_detects_bind_races_through_error_chain() { - let http_bind = super::CliError::RelayP2PError( - pluto_relay_server::RelayP2PError::FailedToBindHttpListener { - addr: "127.0.0.1:1".into(), - source: io::Error::from(io::ErrorKind::AddrInUse), - }, - ); - assert!(is_addr_in_use(&http_bind)); - - let monitoring_bind = super::CliError::RelayP2PError( - pluto_relay_server::RelayP2PError::FailedToBindMonitoringListener { - addr: "127.0.0.1:1".parse().unwrap(), - source: io::Error::from(io::ErrorKind::AddrInUse), - }, - ); - assert!(is_addr_in_use(&monitoring_bind)); - - let unrelated = - super::CliError::RelayP2PError(pluto_relay_server::RelayP2PError::FailedToServeHTTP( - io::Error::from(io::ErrorKind::ConnectionReset), - )); - assert!(!is_addr_in_use(&unrelated)); + /// Parses an ENR response body. + fn parse_enr(body: &str) -> pluto_eth2util::enr::Record { + pluto_eth2util::enr::Record::try_from(body).unwrap() } - /// Make an HTTP GET request to the relay server. + /// Single-shot GET, failing on a non-2xx status. /// - /// Single-shot on purpose: [`with_relay_server`] runs the test function - /// only once the relay serves, so a failure here is a real failure - /// rather than a startup race that retries would paper over. - async fn relay_server_get( - cfg: pluto_relay_server::config::Config, - path: &str, - ) -> Result { - let http_address = cfg.http_addr.unwrap(); - http_get(&format!("http://{http_address}{path}")).await - } - + /// Single-shot on purpose: the relay is serving before a test gets hold of + /// it, so a failure here is a real failure rather than a startup race. async fn http_get(url: &str) -> Result { - reqwest::get(url) + CLIENT + .get(url) + .timeout(REQUEST_TIMEOUT) + .send() .await .and_then(|response| response.error_for_status()) } + + /// Binds a loopback port and holds it, so anything else binding the + /// returned address fails with `AddrInUse`. + async fn squat_tcp_addr() -> (net::TcpListener, String) { + let listener = net::TcpListener::bind(ANY_ADDR).await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + (listener, addr) + } } diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index 4cf88307..f7027bc1 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -93,7 +93,9 @@ use std::{ use futures::{Stream, StreamExt, stream::FusedStream}; use libp2p::{ - Multiaddr, PeerId, Swarm, SwarmBuilder, autonat, identify, + Multiaddr, PeerId, Swarm, SwarmBuilder, autonat, + core::transport::ListenerId, + identify, identity::Keypair, noise, ping, relay, swarm::{ListenError, NetworkBehaviour, SwarmEvent}, @@ -229,6 +231,9 @@ pub struct Node { /// Node type. node_type: NodeType, + + /// Listeners registered through [`Node::listen_on`], in registration order. + listener_ids: Vec, } impl Node { @@ -338,7 +343,6 @@ impl Node { fn apply_config(&mut self, cfg: &P2PConfig, filter_private_addrs: bool) -> Result<()> { let mut addrs = cfg.tcp_multiaddrs()?; - let mut external_addrs = utils::external_tcp_multiaddrs(cfg)?; if self.node_type == NodeType::QUIC { let udp_addrs = cfg.udp_multiaddrs()?; @@ -348,10 +352,6 @@ impl Node { } addrs.extend(udp_addrs); - - let external_udp_addrs = utils::external_udp_multiaddrs(cfg)?; - - external_addrs.extend(external_udp_addrs); } if addrs.is_empty() { @@ -362,16 +362,40 @@ impl Node { // Listen on internal addresses only for addr in &addrs { - self.swarm.listen_on(addr.clone())?; + self.listen_on(addr.clone())?; } + self.set_advertised_addrs(cfg, filter_private_addrs, &addrs) + } + + /// Advertises the external IP / hostname from `cfg` on the ports of + /// `listen_addrs`, together with `listen_addrs` themselves. + /// + /// Replaces everything the node advertises, including addresses added + /// through [`Node::add_external_address`]. + /// + /// Callers that listen on port 0 should call this again once libp2p has + /// reported the kernel-assigned ports: the configured addresses advertise + /// port 0, which is not dialable. + pub fn set_advertised_addrs( + &mut self, + cfg: &P2PConfig, + filter_private_addrs: bool, + listen_addrs: &[Multiaddr], + ) -> Result<()> { + let external_addrs = utils::external_multiaddrs(cfg, listen_addrs)?; + // Advertise filtered addresses (external + optionally filtered internal) let advertised_addrs = utils::filter_advertised_addresses( utils::ExternalAddresses(external_addrs), - utils::InternalAddresses(addrs), + utils::InternalAddresses(listen_addrs.to_vec()), filter_private_addrs, )?; + for addr in self.swarm.external_addresses().cloned().collect::>() { + self.swarm.remove_external_address(&addr); + } + for addr in advertised_addrs { self.swarm.add_external_address(addr); } @@ -429,6 +453,7 @@ impl Node { swarm, node_type: NodeType::QUIC, p2p_context, + listener_ids: Vec::new(), }) } @@ -464,6 +489,7 @@ impl Node { swarm, node_type: NodeType::TCP, p2p_context, + listener_ids: Vec::new(), }) } @@ -482,6 +508,7 @@ impl Node { swarm, node_type: NodeType::QUIC, p2p_context, + listener_ids: Vec::new(), }) } @@ -500,6 +527,7 @@ impl Node { swarm, node_type: NodeType::TCP, p2p_context, + listener_ids: Vec::new(), }) } @@ -566,9 +594,20 @@ impl Node { } /// Listens on an address. - pub fn listen_on(&mut self, addr: Multiaddr) -> Result<()> { - self.swarm.listen_on(addr)?; - Ok(()) + /// + /// The listener is bound before this returns, but reports the address it + /// bound — the kernel-assigned one when the port is 0 — later, as a + /// [`SwarmEvent::NewListenAddr`] carrying the returned [`ListenerId`]. + pub fn listen_on(&mut self, addr: Multiaddr) -> Result { + let listener_id = self.swarm.listen_on(addr)?; + self.listener_ids.push(listener_id); + Ok(listener_id) + } + + /// Returns the listeners registered through [`Node::listen_on`], in + /// registration order. + pub fn listener_ids(&self) -> &[ListenerId] { + &self.listener_ids } /// Adds an external address to the peer store. diff --git a/crates/p2p/src/utils.rs b/crates/p2p/src/utils.rs index 05564239..62f1c267 100644 --- a/crates/p2p/src/utils.rs +++ b/crates/p2p/src/utils.rs @@ -26,23 +26,21 @@ use crate::{ manet::Manet, }; -/// Returns the external IP and Hostname fields as multiaddrs using the listen -/// TCP addresses ports. -pub fn external_tcp_multiaddrs(cfg: &P2PConfig) -> crate::p2p::Result> { - let addrs = cfg.parse_tcp_addrs()?; - - let mut ports = vec![]; - - for addr in &addrs { - ports.push(addr.port()); - } - +/// Returns the external IP and Hostname fields as TCP multiaddrs on `ports`. +/// +/// `ports` must be the ports the node actually listens on: a configured port of +/// 0 means the kernel picks one, so the configured value would advertise +/// nothing dialable. +pub fn external_tcp_multiaddrs( + cfg: &P2PConfig, + ports: &[u16], +) -> crate::p2p::Result> { let mut resp = vec![]; if let Some(external_ip) = cfg.external_ip.as_ref() { let ip = external_ip.parse::()?; - for port in &ports { + for port in ports { let maddr = config::multi_addr_from_ip_tcp_port(SocketAddr::new(ip, *port))?; resp.push(maddr); @@ -50,7 +48,7 @@ pub fn external_tcp_multiaddrs(cfg: &P2PConfig) -> crate::p2p::Result crate::p2p::Result crate::p2p::Result> { - let addrs = cfg.parse_udp_addrs()?; - - let mut ports = vec![]; - - for addr in &addrs { - ports.push(addr.port()); - } - +/// Returns the external IP and Hostname fields as QUIC multiaddrs on `ports`. +/// +/// `ports` must be the ports the node actually listens on, as in +/// [`external_tcp_multiaddrs`]. +pub fn external_udp_multiaddrs( + cfg: &P2PConfig, + ports: &[u16], +) -> crate::p2p::Result> { let mut resp = vec![]; if let Some(external_ip) = cfg.external_ip.as_ref() { let ip = external_ip.parse::()?; - for port in &ports { + for port in ports { let maddr = config::multi_addr_from_ip_udp_port(SocketAddr::new(ip, *port))?; resp.push(maddr); @@ -82,7 +77,7 @@ pub fn external_udp_multiaddrs(cfg: &P2PConfig) -> crate::p2p::Result crate::p2p::Result crate::p2p::Result> { + let tcp_ports: Vec = listen_addrs.iter().filter_map(tcp_port).collect(); + let udp_ports: Vec = listen_addrs.iter().filter_map(udp_port).collect(); + + let mut addrs = external_tcp_multiaddrs(cfg, &tcp_ports)?; + addrs.extend(external_udp_multiaddrs(cfg, &udp_ports)?); + + Ok(addrs) +} + +/// Returns the TCP port of a multiaddr. +pub fn tcp_port(addr: &Multiaddr) -> Option { + addr.iter().find_map(|protocol| match protocol { + MaProtocol::Tcp(port) => Some(port), + _ => None, + }) +} + +/// Returns the UDP port of a multiaddr. +pub fn udp_port(addr: &Multiaddr) -> Option { + addr.iter().find_map(|protocol| match protocol { + MaProtocol::Udp(port) => Some(port), + _ => None, + }) +} + pub(crate) struct ExternalAddresses(pub Vec); pub(crate) struct InternalAddresses(pub Vec); @@ -208,13 +234,11 @@ pub fn is_direct_addr(addr: &Multiaddr) -> bool { mod tests { use super::*; - /// Config with the listen addresses and external overrides under test. + /// Config with the external overrides under test. fn config(external_ip: Option<&str>, external_host: Option<&str>) -> P2PConfig { P2PConfig { external_ip: external_ip.map(String::from), external_host: external_host.map(String::from), - tcp_addrs: vec!["127.0.0.1:3610".to_string(), "127.0.0.1:3611".to_string()], - udp_addrs: vec!["127.0.0.1:3620".to_string(), "127.0.0.1:3621".to_string()], ..Default::default() } } @@ -224,37 +248,48 @@ mod tests { } #[test] - fn external_multiaddrs_keep_the_listen_ports() { + fn external_multiaddrs_keep_the_bound_ports() { let cfg = config(Some("1.2.3.4"), Some("relay.example.com")); - - // The external address replaces the listen IP but must advertise the - // port the node actually listens on — one address per listen port, IP - // forms first, then hostname forms. + // What libp2p reports once bound: the kernel-assigned ports of a `:0` + // listen config, which is what must be advertised — not the 0 that was + // asked for. + let listen_addrs = vec![ + "/ip4/127.0.0.1/tcp/40001".parse().unwrap(), + "/ip4/127.0.0.1/tcp/40002".parse().unwrap(), + "/ip4/127.0.0.1/udp/40003/quic-v1".parse().unwrap(), + ]; + + // The external address replaces the listen IP but keeps its port — one + // address per listen port, IP forms first, then hostname forms, TCP + // before QUIC. assert_eq!( - as_strings(&external_tcp_multiaddrs(&cfg).unwrap()), + as_strings(&external_multiaddrs(&cfg, &listen_addrs).unwrap()), vec![ - "/ip4/1.2.3.4/tcp/3610", - "/ip4/1.2.3.4/tcp/3611", - "/dns/relay.example.com/tcp/3610", - "/dns/relay.example.com/tcp/3611", - ] - ); - assert_eq!( - as_strings(&external_udp_multiaddrs(&cfg).unwrap()), - vec![ - "/ip4/1.2.3.4/udp/3620/quic-v1", - "/ip4/1.2.3.4/udp/3621/quic-v1", - "/dns/relay.example.com/udp/3620/quic-v1", - "/dns/relay.example.com/udp/3621/quic-v1", + "/ip4/1.2.3.4/tcp/40001", + "/ip4/1.2.3.4/tcp/40002", + "/dns/relay.example.com/tcp/40001", + "/dns/relay.example.com/tcp/40002", + "/ip4/1.2.3.4/udp/40003/quic-v1", + "/dns/relay.example.com/udp/40003/quic-v1", ] ); } #[test] fn no_external_multiaddrs_without_external_config() { - let cfg = config(None, None); - - assert!(external_tcp_multiaddrs(&cfg).unwrap().is_empty()); - assert!(external_udp_multiaddrs(&cfg).unwrap().is_empty()); + let listen_addrs = vec!["/ip4/127.0.0.1/tcp/40001".parse().unwrap()]; + + // Nothing to advertise without an override, and nothing to advertise on + // when the node listens nowhere. + assert!( + external_multiaddrs(&config(None, None), &listen_addrs) + .unwrap() + .is_empty() + ); + assert!( + external_multiaddrs(&config(Some("1.2.3.4"), None), &[]) + .unwrap() + .is_empty() + ); } } diff --git a/crates/relay-server/src/error.rs b/crates/relay-server/src/error.rs index 8cf567c6..17cef505 100644 --- a/crates/relay-server/src/error.rs +++ b/crates/relay-server/src/error.rs @@ -22,11 +22,11 @@ pub enum RelayP2PError { /// Failed to bind HTTP listener. #[error("Failed to bind HTTP listener {addr}: {source}")] FailedToBindHttpListener { - /// Address the listener could not be bound to. + /// Address the listener could not be bound to. A `String` because + /// [`crate::config::Config::http_addr`] is never parsed — the listener + /// binds the configured string directly. addr: String, - /// Underlying bind error. Kept typed so callers can tell an - /// [`std::io::ErrorKind::AddrInUse`] race apart from a real - /// misconfiguration. + /// Underlying bind error. #[source] source: std::io::Error, }, @@ -35,13 +35,19 @@ pub enum RelayP2PError { #[error("Failed to serve HTTP: {0}")] FailedToServeHTTP(#[source] std::io::Error), + /// A libp2p listener closed before reporting the address it bound. + #[error("libp2p listener closed during startup: {reason}")] + ListenerClosedDuringStartup { + /// Why the listener closed. + reason: String, + }, + /// Failed to bind the monitoring listener. #[error("Failed to bind monitoring listener {addr}: {source}")] FailedToBindMonitoringListener { /// Address the monitoring listener could not be bound to. addr: SocketAddr, - /// Underlying bind error, kept typed for the same reason as - /// [`RelayP2PError::FailedToBindHttpListener`]. + /// Underlying bind error. #[source] source: std::io::Error, }, diff --git a/crates/relay-server/src/lib.rs b/crates/relay-server/src/lib.rs index dd6d0616..eba99304 100644 --- a/crates/relay-server/src/lib.rs +++ b/crates/relay-server/src/lib.rs @@ -23,4 +23,4 @@ pub use error::RelayP2PError; pub(crate) use error::Result; #[doc(hidden)] -pub use web::enr_server; +pub use web::{AppState, enr_server}; diff --git a/crates/relay-server/src/p2p.rs b/crates/relay-server/src/p2p.rs index 6f6badd4..2ef41866 100644 --- a/crates/relay-server/src/p2p.rs +++ b/crates/relay-server/src/p2p.rs @@ -1,42 +1,251 @@ //! Relay P2P node implementation. -use std::{net::SocketAddr, sync::Arc, time::Duration}; +use std::{collections::HashSet, net::SocketAddr, sync::Arc, time::Duration}; use futures::StreamExt; use k256::SecretKey; -use libp2p::{PeerId, relay, swarm::SwarmEvent}; +use libp2p::{Multiaddr, PeerId, core::transport::ListenerId, relay, swarm::SwarmEvent}; use pluto_p2p::{behaviours::pluto::PlutoBehaviourEvent, name::peer_name}; -use tokio::sync::{RwLock, mpsc}; +use tokio::{ + net::TcpListener, + sync::{RwLock, mpsc}, + task::JoinHandle, +}; use tokio_util::sync::CancellationToken; use tracing::{debug, info, instrument, warn}; +use vise_exporter::MetricsServer; use crate::{ Result, config::{Config, create_relay_config}, error::RelayP2PError, metrics::{PeerWithPeerClusterLabels, RELAY_METRICS}, - web::{bind_monitoring_server, enr_server, serve_monitoring_server}, + web::{AppState, bind_monitoring_server, enr_server, serve_monitoring_server}, }; use pluto_p2p::{ BandwidthFactory, PeerConnectionMetrics, - manet::Manet, p2p::{Node, NodeType}, p2p_context::P2PContext, - utils::{external_tcp_multiaddrs, external_udp_multiaddrs}, + utils::external_multiaddrs, }; -/// Runs a relay P2P node. +/// Runs a relay P2P node: binds every listener, then serves until `ct` is +/// cancelled or a server fails. #[instrument(skip(config, key, ct))] pub async fn run_relay_p2p_node( config: &Config, key: SecretKey, ct: CancellationToken, ) -> Result> { + bind_relay(config, key, ct).await?.serve().await +} + +/// A relay whose listeners are all bound, but which is not serving yet. +/// +/// Startup is split in two so that binding — the fallible, all-or-nothing part +/// — completes before anything is served. A caller therefore learns of an +/// unusable or already-taken address while there is still nothing to clean up, +/// and can read back the addresses that were actually bound. +/// +/// "Before anything is served" includes p2p: [`bind_relay`] binds the HTTP +/// listeners before it creates the swarm, so no peer is ever accepted by a +/// relay that then fails to start. +/// +/// Dropping a `BoundRelay` without calling [`BoundRelay::serve`] releases every +/// listener it holds, but does not cancel the [`CancellationToken`] passed to +/// [`bind_relay`] — that token belongs to the caller. +/// +/// Production code should use [`run_relay_p2p_node`]; the halves are only +/// pulled apart by tests that need the ports the kernel assigned. +#[doc(hidden)] +pub struct BoundRelay { + /// The relay swarm, listening on every configured address and having + /// reported each of them. + node: Node, + /// Addresses libp2p is listening on, shared with the HTTP handlers. + listen_addrs: Arc>>, + /// State the HTTP handlers answer from. + state: Arc, + /// Bound ENR/multiaddr HTTP listener, unless no HTTP address is configured. + enr_listener: Option, + /// Bound Prometheus monitoring server, unless monitoring is disabled. + monitoring_server: Option>, + /// Cancels the relay and everything it spawns. + ct: CancellationToken, +} + +impl BoundRelay { + /// Address the ENR/multiaddr HTTP server is bound to, or `None` when no + /// HTTP address is configured. + /// + /// This is the address that was actually bound, which is the configured one + /// only when it named a fixed port. + pub fn http_addr(&self) -> Option { + self.enr_listener + .as_ref() + .and_then(|listener| listener.local_addr().ok()) + } + + /// Address the Prometheus monitoring server is bound to, or `None` when + /// monitoring is disabled. + pub fn monitoring_addr(&self) -> Option { + self.monitoring_server + .as_ref() + .map(|server| server.local_addr()) + } + + /// Addresses libp2p is listening on, as it reported them — so the + /// kernel-assigned ports when the configured ones were 0. + pub async fn p2p_addrs(&self) -> Vec { + self.listen_addrs.read().await.clone() + } + + /// Serves every bound listener until the relay is cancelled or one of its + /// servers fails. + pub async fn serve(self) -> Result> { + let http_addr = self.http_addr(); + let Self { + mut node, + listen_addrs, + state, + enr_listener, + monitoring_server, + ct, + } = self; + + let (server_errors, mut server_errors_receiver) = mpsc::channel(3); + + let enr_server_handle = enr_listener.map(|listener| { + tokio::spawn(enr_server( + server_errors.clone(), + listener, + state, + ct.child_token(), + )) + }); + + // The bound address, not the configured one: they differ whenever the + // configured port was 0. + if let Some(http_addr) = http_addr { + info!("Runtime multiaddrs available via http at {http_addr}"); + } else { + info!("Runtime multiaddrs not available via http, since http-address flag is not set"); + } + + // Serve the monitoring listener bound by `bind_relay`. + let monitoring_handle = monitoring_server + .map(|server| tokio::spawn(serve_monitoring_server(server_errors.clone(), server))); + + // A server failure is returned only once the shutdown below has run, so + // a failed relay never leaves listeners bound behind it. + let server_error = loop { + tokio::select! { + biased; + _ = ct.cancelled() => { + info!("Relay server shutdown signal received, shutting down gracefully"); + break None; + }, + error = server_errors_receiver.recv() => { + if let Some(error) = error { + warn!("Server error: {}", error); + break Some(error); + } + }, + event = node.select_next_some() => { + apply_addr_update(&listen_addrs, handle_swarm_event(&event)).await; + } + } + }; + + ct.cancel(); + + if let Some(handle) = enr_server_handle { + join_or_abort("ENR server", handle).await; + } + + if let Some(handle) = monitoring_handle { + join_or_abort("Monitoring server", handle).await; + } + + match server_error { + Some(error) => Err(error), + None => Ok(node), + } + } +} + +/// Binds every listener the relay is configured with, without serving any of +/// them. +/// +/// Returns once the swarm, the ENR HTTP listener and the monitoring listener +/// are all bound, so any bind failure — an unusable address, or a port another +/// process holds — fails here rather than partway through a running relay. +/// +/// Order matters: the HTTP listeners are bound before the swarm is created, and +/// the swarm is polled only once they are. Polling is what makes the relay +/// service p2p connections, so a bind that failed after it would take down +/// peers the relay had already accepted. +#[doc(hidden)] +#[instrument(skip(config, key, ct))] +pub async fn bind_relay( + config: &Config, + key: SecretKey, + ct: CancellationToken, +) -> Result { + let (git_hash, build_time) = pluto_core::version::git_commit(); + info!( + version = %*pluto_core::version::VERSION, + git_hash = %git_hash, + build_time = %build_time, + "Pluto relay starting" + ); + + // The HTTP listeners are bound before the swarm exists, and the swarm is not + // polled until they are. Polling accepts and services p2p connections — it + // completes handshakes, counts them, and can hand out circuit reservations — + // so binding these afterwards would mean a bind failure tore down peers the + // relay had already taken on. + // + // Binding here rather than inside each server's task is also what lets an + // unusable address or a lost race for a port fail the relay while nothing + // has been spawned. + let monitoring_server = match config.monitoring_addr.clone() { + Some(monitoring_addr) => { + let bind_addr = monitoring_addr + .parse::() + .map_err(|_| RelayP2PError::FailedToParseMonitoringAddr(monitoring_addr))?; + + Some(bind_monitoring_server(bind_addr, ct.child_token()).await?) + } + None => { + info!("Prometheus monitoring not available, since monitoring-address flag is not set"); + None + } + }; + + let enr_listener = match config.http_addr.as_deref() { + Some(http_addr) => { + info!("Binding ENR server on {http_addr}"); + let listener = TcpListener::bind(http_addr).await.map_err(|source| { + RelayP2PError::FailedToBindHttpListener { + addr: http_addr.to_owned(), + source, + } + })?; + Some(listener) + } + None => { + warn!("HTTP address is not set, skipping ENR server"); + None + } + }; + let relay_config = create_relay_config(config); let bandwidth: BandwidthFactory = std::sync::Arc::new(|peer_id| PeerConnectionMetrics { sent: RELAY_METRICS.network_sent_bytes_total[&relay_labels(peer_id)].clone(), received: RELAY_METRICS.network_receive_bytes_total[&relay_labels(peer_id)].clone(), }); + // Binds the configured TCP listeners; `listen_on` below binds the UDP ones. let mut node = Node::new_server( config.p2p_config.clone(), key.clone(), @@ -53,141 +262,136 @@ pub async fn run_relay_p2p_node( }, )?; - let (git_hash, build_time) = pluto_core::version::git_commit(); - info!( - version = %*pluto_core::version::VERSION, - git_hash = %git_hash, - build_time = %build_time, - "Pluto relay starting" - ); - for udp_addr in config.p2p_config.udp_multiaddrs()? { debug!("Listening on UDP address {}", udp_addr); node.listen_on(udp_addr)?; } - let (server_errors, mut server_errors_receiver) = mpsc::channel(3); + // First poll of the swarm, and so the first point at which this relay + // services anything. Every other listener is already bound. + let listen_addrs = Arc::new(RwLock::new(Vec::new())); + wait_for_listen_addrs(&mut node, &listen_addrs).await?; + let bound_addrs = listen_addrs.read().await.clone(); - let listeners = Arc::new(RwLock::new(Vec::new())); + // Advertise the ports libp2p bound rather than the configured ones, which + // carry port 0 whenever the kernel picked the port. + node.set_advertised_addrs( + &config.p2p_config, + config.filter_private_addrs, + &bound_addrs, + )?; // Compute external multiaddrs from external_ip / external_host config so // they're advertised on `/` and folded into ENR responses on `/enr` even // when libp2p only sees private listen addresses (e.g., K8s pods behind // NodePort). - let mut external_addrs = external_tcp_multiaddrs(&config.p2p_config)?; - external_addrs.extend(external_udp_multiaddrs(&config.p2p_config)?); - - // Bind the monitoring listener before starting the ENR server: an unusable - // address or a lost race for the port then fails the relay while nothing - // has been spawned, instead of returning past a running ENR server and - // leaving its listener bound. It also means a serving ENR endpoint implies - // every listener of this relay is bound. - let monitoring_server = match config.monitoring_addr.clone() { - Some(monitoring_addr) => { - let bind_addr = monitoring_addr - .parse::() - .map_err(|_| RelayP2PError::FailedToParseMonitoringAddr(monitoring_addr))?; + let external_addrs = external_multiaddrs(&config.p2p_config, &bound_addrs)?; - Some(bind_monitoring_server(bind_addr, ct.child_token()).await?) - } - None => { - info!("Prometheus monitoring not available, since monitoring-address flag is not set"); - None - } - }; - - let enr_server_handle = tokio::spawn(enr_server( - server_errors.clone(), - config.clone(), - key.clone(), + let state = Arc::new(AppState::new( + config.p2p_config.clone(), + key, *node.local_peer_id(), - listeners.clone(), + listen_addrs.clone(), external_addrs, - ct.child_token(), + config.filter_private_addrs, )); - if let Some(http_addr) = config.http_addr.clone() { - info!("Runtime multiaddrs available via http at {http_addr}"); - } else { - info!("Runtime multiaddrs not available via http, since http-address flag is not set"); - } + Ok(BoundRelay { + node, + listen_addrs, + state, + enr_listener, + monitoring_server, + ct, + }) +} - // Serve the monitoring listener bound above. - let monitoring_handle = monitoring_server - .map(|server| tokio::spawn(serve_monitoring_server(server_errors.clone(), server))); +/// Waits until every listener registered on `node` has reported the address it +/// bound, collecting them into `listen_addrs`. +/// +/// libp2p binds inside `listen_on` but reports the bound address — the +/// kernel-assigned one when the configured port was 0 — only as a swarm event, +/// so this is what turns a bound relay into one that knows its own addresses +/// and can answer `/enr`. +/// +/// This terminates for any listener that bound, whether its address is public +/// or private: private addresses are withheld when the ENR is rendered, not +/// when they are ingested, so a node that only ever listens on RFC 1918 +/// addresses still finishes starting. +async fn wait_for_listen_addrs( + node: &mut Node, + listen_addrs: &Arc>>, +) -> Result<()> { + let mut pending: HashSet = node.listener_ids().iter().copied().collect(); - // Set when one of the HTTP servers fails; returned once the shutdown below - // has run, so a failed relay never leaves listeners bound behind it. - let mut server_error = None; + while !pending.is_empty() { + let event = node.select_next_some().await; - loop { - tokio::select! { - biased; - _ = ct.cancelled() => { - info!("Relay server shutdown signal received, shutting down gracefully"); - break; - }, - error = server_errors_receiver.recv() => { - if let Some(error) = error { - warn!("Server error: {}", error); - server_error = Some(error); - break; - } - }, - event = node.select_next_some() => { - let address_update = handle_swarm_event(&event, config.filter_private_addrs); + match &event { + SwarmEvent::NewListenAddr { listener_id, .. } => { + pending.remove(listener_id); + } + // A listener that closes will never report an address. If it closed + // because of an error the relay cannot start; a clean close just + // means one fewer listener to wait for. + SwarmEvent::ListenerClosed { + listener_id, + reason, + .. + } => { + pending.remove(listener_id); - // Update listener address list - match address_update { - AddrUpdate::Add(address) => { - listeners.write().await.push(address); - } - AddrUpdate::Remove(address) => { - listeners.write().await.retain(|a| *a != address); - } - AddrUpdate::RemoveAll(addresses) => { - listeners - .write() - .await - .retain(|a| !addresses.contains(a)); - } - AddrUpdate::None => {} + if let Err(err) = reason { + return Err(RelayP2PError::ListenerClosedDuringStartup { + reason: err.to_string(), + }); } } + _ => {} } + + apply_addr_update(listen_addrs, handle_swarm_event(&event)).await; } - ct.cancel(); + Ok(()) +} - match tokio::time::timeout(Duration::from_secs(2), enr_server_handle).await { - Ok(Ok(())) => { - info!("ENR server shutdown complete"); - } - Ok(Err(e)) => { - warn!("ENR server shutdown error: {}", e); +/// Applies an [`AddrUpdate`] to the listen addresses shared with the HTTP +/// handlers. +async fn apply_addr_update(listen_addrs: &Arc>>, update: AddrUpdate) { + match update { + AddrUpdate::Add(address) => listen_addrs.write().await.push(address), + AddrUpdate::Remove(address) => { + listen_addrs.write().await.retain(|addr| *addr != address); } - Err(_) => { - warn!("ENR server shutdown timeout"); + AddrUpdate::RemoveAll(addresses) => { + listen_addrs + .write() + .await + .retain(|addr| !addresses.contains(addr)); } + AddrUpdate::None => {} } +} - if let Some(handle) = monitoring_handle { - match tokio::time::timeout(Duration::from_secs(2), handle).await { - Ok(Ok(())) => { - info!("Monitoring server shutdown complete"); - } - Ok(Err(e)) => { - warn!("Monitoring server shutdown error: {}", e); - } - Err(_) => { - warn!("Monitoring server shutdown timeout"); - } - } - } +/// Grace period for a server task to finish shutting down before it is aborted. +const SERVER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); - match server_error { - Some(error) => Err(error), - None => Ok(node), +/// Waits for a server task to shut down, aborting it if it overruns +/// [`SERVER_SHUTDOWN_TIMEOUT`]. +/// +/// The abort is awaited: dropping a `JoinHandle` only *detaches* the task, +/// which would leave it holding its listener after the relay has reported that +/// it stopped. +async fn join_or_abort(name: &str, mut handle: JoinHandle<()>) { + match tokio::time::timeout(SERVER_SHUTDOWN_TIMEOUT, &mut handle).await { + Ok(Ok(())) => info!("{name} shutdown complete"), + Ok(Err(err)) => warn!("{name} shutdown error: {err}"), + Err(_) => { + warn!("{name} shutdown timed out, aborting"); + handle.abort(); + let _ = handle.await; + } } } @@ -208,21 +412,13 @@ enum AddrUpdate { /// Returns an [`AddrUpdate`] describing any change to the listener address /// list that the caller should apply. /// -/// `filter_private_addrs` drops private listen addresses (e.g. loopback, -/// RFC 1918) from the advertised set — parity with Go charon's -/// `filterAdvertisedAddrs(excludeInternalPrivate=true)`. -fn handle_swarm_event( - event: &SwarmEvent>, - filter_private_addrs: bool, -) -> AddrUpdate { +/// Every listen address is tracked, private ones included; whether they are +/// advertised is decided when a response is rendered. +fn handle_swarm_event(event: &SwarmEvent>) -> AddrUpdate { match event { // Track listener address changes SwarmEvent::NewListenAddr { address, .. } => { debug!(%address, "listening on new address"); - if filter_private_addrs && address.is_private() { - debug!(%address, "skipping private listen address"); - return AddrUpdate::None; - } AddrUpdate::Add(address.clone()) } SwarmEvent::ListenerClosed { addresses, .. } => { diff --git a/crates/relay-server/src/web.rs b/crates/relay-server/src/web.rs index 1dfe84f8..85802808 100644 --- a/crates/relay-server/src/web.rs +++ b/crates/relay-server/src/web.rs @@ -24,10 +24,10 @@ use tracing::{debug, info, instrument, warn}; use vise_exporter::{MetricsExporter, MetricsServer}; use crate::{ - config::{Config, EXTERNAL_HOST_RESOLVE_INTERVAL}, + config::EXTERNAL_HOST_RESOLVE_INTERVAL, error::{RelayP2PError, Result}, }; -use pluto_p2p::{config::P2PConfig, name::peer_name}; +use pluto_p2p::{config::P2PConfig, manet::Manet, name::peer_name}; /// Shared application state for HTTP handlers. #[derive(Clone)] @@ -38,12 +38,16 @@ pub struct AppState { secret_key: SecretKey, /// The peer ID of this node. peer_id: PeerId, - /// The libp2p-discovered listen addresses of this node. + /// The libp2p-discovered listen addresses of this node, private ones + /// included: they are withheld at read time, not at ingest. addrs: Arc>>, - /// External multiaddrs derived from `external_ip` / `external_host` config. - /// Fixed at startup. Includes `/ip4//...` and - /// `/dns//...` variants for both TCP and UDP/QUIC. + /// External multiaddrs derived from `external_ip` / `external_host` config + /// and the ports libp2p bound. Fixed at startup. Includes + /// `/ip4//...` and `/dns//...` variants for + /// both TCP and UDP/QUIC. external_addrs: Vec, + /// Whether private listen addresses are withheld from what is advertised. + filter_private_addrs: bool, /// The resolved external host IP (if configured). external_host_ip: Arc>>, } @@ -56,6 +60,7 @@ impl AppState { peer_id: PeerId, addrs: Arc>>, external_addrs: Vec, + filter_private_addrs: bool, ) -> Self { Self { p2p_config, @@ -63,20 +68,31 @@ impl AppState { peer_id, addrs, external_addrs, + filter_private_addrs, external_host_ip: Arc::new(RwLock::new(None)), } } - /// Returns the union of configured external multiaddrs and the live - /// libp2p listen addresses, externals first, deduped while preserving - /// order. Mirrors Go charon's `filterAdvertisedAddrs(externalAddrs, - /// internalAddrs, …)` — listeners are already filtered for private - /// addresses at ingest time when `filter_private_addrs` is set. + /// Returns the union of configured external multiaddrs and the live libp2p + /// listen addresses, externals first, deduped while preserving order. + /// + /// Mirrors Go charon's `filterAdvertisedAddrs(externalAddrs, internalAddrs, + /// excludeInternalPrivate)`: `filter_private_addrs` withholds private + /// listen addresses (loopback, RFC 1918) but never the external ones. + /// + /// Filtering happens here rather than when a listen address is ingested so + /// that the relay always knows every address it bound — startup waits on + /// that, and a node whose only addresses are private must still finish + /// starting. async fn advertised_addrs(&self) -> Vec { let listeners = self.addrs.read().await; + let listeners = listeners + .iter() + .filter(|addr| !(self.filter_private_addrs && addr.is_private())); + let mut seen: HashSet<&Multiaddr> = HashSet::new(); let mut union: Vec = Vec::new(); - for addr in self.external_addrs.iter().chain(listeners.iter()) { + for addr in self.external_addrs.iter().chain(listeners) { if seen.insert(addr) { union.push(addr.clone()); } @@ -96,72 +112,38 @@ impl AppState { } } -/// Starts the ENR HTTP server. -#[instrument(skip(server_errors, config, secret_key, peer_id, addrs, external_addrs, ct))] +/// Serves the ENR HTTP API on an already bound listener until shutdown. +/// +/// The listener is bound by the caller so that a bind failure is reported +/// before any task is spawned, and so the caller learns the address that was +/// actually bound. +#[instrument(skip_all)] pub async fn enr_server( server_errors: mpsc::Sender, - config: Config, - secret_key: SecretKey, - peer_id: PeerId, - addrs: Arc>>, - external_addrs: Vec, + listener: TcpListener, + state: Arc, ct: CancellationToken, ) { - let Some(http_addr) = config.http_addr.clone() else { - warn!("HTTP address is not set, skipping ENR server"); - return; - }; - info!("Starting ENR server"); - // Bind before spawning anything else, so a failed bind has nothing to - // clean up. The error keeps its `io::ErrorKind` so callers can tell a lost - // race for the port from an unusable address. - let listener = match TcpListener::bind(&http_addr).await { - Ok(listener) => listener, - Err(err) => { - warn!("Failed to bind HTTP listener to {http_addr}: {err}"); - let _ = server_errors - .send(RelayP2PError::FailedToBindHttpListener { - addr: http_addr, - source: err, - }) - .await; - return; - } - }; + // Start external host resolver task if configured + let resolver_handle = state.p2p_config.external_host.clone().map(|external_host| { + let state = state.clone(); + let ct = ct.child_token(); + tokio::spawn(resolve_external_host_periodically(state, external_host, ct)) + }); - let state = AppState::new( - config.p2p_config.clone(), - secret_key, - peer_id, - addrs, - external_addrs, + info!( + "Relay started {peer_name} on {tcp_addrs} and {udp_addrs}", + peer_name = peer_name(&state.peer_id), + tcp_addrs = state.p2p_config.tcp_addrs.join(", "), + udp_addrs = state.p2p_config.udp_addrs.join(", "), ); - let state_arc = Arc::new(state); - - // Start external host resolver task if configured - let resolver_handle = if let Some(external_host) = config.p2p_config.external_host { - let state_clone = state_arc.clone(); - let ct_clone = ct.child_token(); - Some(tokio::spawn(async move { - resolve_external_host_periodically(state_clone, external_host, ct_clone).await; - })) - } else { - None - }; let router = Router::new() .route("/", get(multiaddr_handler)) .route("/enr", get(enr_handler)) - .with_state(state_arc); - - info!( - "Relay started {peer_name} on {tcp_addrs} and {udp_addrs}", - peer_name = peer_name(&peer_id), - tcp_addrs = config.p2p_config.tcp_addrs.join(", "), - udp_addrs = config.p2p_config.udp_addrs.join(", "), - ); + .with_state(state); let ct_clone = ct.child_token(); if let Err(e) = axum::serve(listener, router) @@ -186,10 +168,8 @@ pub async fn enr_server( /// Binds the Prometheus monitoring listener on the given address. /// -/// Binding is separate from serving so that a bind failure is reported to the -/// caller synchronously — before any other listener is started — and keeps its -/// `io::ErrorKind`, letting callers tell a lost race for the port from a real -/// misconfiguration. +/// Binding is separate from serving so a bind failure fails the relay before +/// anything is spawned, and so the caller can read back the bound address. #[instrument(skip(ct))] pub(crate) async fn bind_monitoring_server( bind_addr: SocketAddr, @@ -445,6 +425,16 @@ mod tests { external_host: Option<&str>, external_addrs: Vec, listeners: Vec, + ) -> (Arc, PeerId) { + test_state_filtered(external_ip, external_host, external_addrs, listeners, false) + } + + fn test_state_filtered( + external_ip: Option<&str>, + external_host: Option<&str>, + external_addrs: Vec, + listeners: Vec, + filter_private_addrs: bool, ) -> (Arc, PeerId) { let secret_key = SecretKey::random(&mut OsRng); let peer_id = Keypair::generate_secp256k1().public().to_peer_id(); @@ -459,6 +449,7 @@ mod tests { peer_id, Arc::new(RwLock::new(listeners)), external_addrs, + filter_private_addrs, ); (Arc::new(state), peer_id) } @@ -530,6 +521,23 @@ mod tests { assert!(state.advertised_addrs().await.is_empty()); } + #[tokio::test] + async fn advertised_addrs_withholds_private_listen_addrs_when_filtering() { + let externals = vec![ma("/ip4/1.2.3.4/tcp/3610")]; + let listeners = vec![ + ma("/ip4/127.0.0.1/tcp/3610"), + ma("/ip4/10.0.0.1/tcp/3610"), + ma("/ip4/8.8.8.8/tcp/3610"), + ]; + let (state, _) = test_state_filtered(Some("1.2.3.4"), None, externals, listeners, true); + + // Externals are never filtered; private listen addresses are. + assert_eq!( + state.advertised_addrs().await, + vec![ma("/ip4/1.2.3.4/tcp/3610"), ma("/ip4/8.8.8.8/tcp/3610")] + ); + } + // ------ multiaddr_handler ------ #[tokio::test] diff --git a/crates/relay-server/tests/http_integration.rs b/crates/relay-server/tests/http_integration.rs index 1bf6dca1..b61c159d 100644 --- a/crates/relay-server/tests/http_integration.rs +++ b/crates/relay-server/tests/http_integration.rs @@ -1,8 +1,8 @@ //! End-to-end integration tests for the relay HTTP layer. //! //! Spins up the real `enr_server` axum app on an ephemeral port and asserts -//! `/` and `/enr` over a live HTTP socket via `reqwest`. Tests are isolated -//! by binding to `127.0.0.1:0`-equivalent (find free port, then bind), +//! `/` and `/enr` over a live HTTP socket via `reqwest`. Tests are isolated by +//! binding `127.0.0.1:0` and reading the assigned port back off the listener, //! shutting down via `CancellationToken`, and using config-only knobs so no //! libp2p swarm is started. //! @@ -14,11 +14,7 @@ use std::{net::Ipv4Addr, sync::Arc, time::Duration}; use k256::SecretKey; use libp2p::{Multiaddr, identity::Keypair}; use pluto_eth2util::enr::Record; -use pluto_p2p::{ - config::P2PConfig, - utils::{external_tcp_multiaddrs, external_udp_multiaddrs}, -}; -use pluto_relay_server::config::Config; +use pluto_p2p::{config::P2PConfig, utils::external_multiaddrs}; use rand::rngs::OsRng; use tokio::{ net::TcpListener, @@ -26,21 +22,10 @@ use tokio::{ }; use tokio_util::sync::CancellationToken; -/// Ephemeral port helper: bind 127.0.0.1:0, capture the assigned port, then -/// drop the listener so `enr_server` can bind it. Small TOCTOU window, fine -/// for tests. -async fn pick_free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0") - .await - .expect("bind ephemeral"); - let port = l.local_addr().expect("local_addr").port(); - drop(l); - port -} - /// Constructs a `P2PConfig` with sensible listen addrs so the external-addr -/// helpers produce something to advertise. The listen ports are advisory: -/// `enr_server` only binds the HTTP listener, not p2p sockets. +/// helpers produce something to advertise. The listen ports are the ports the +/// externals are advertised on; no p2p socket is bound, `enr_server` only +/// serves the HTTP listener. fn p2p_config(external_ip: Option<&str>, external_host: Option<&str>, port: u16) -> P2PConfig { P2PConfig { tcp_addrs: vec![format!("127.0.0.1:{port}")], @@ -51,66 +36,50 @@ fn p2p_config(external_ip: Option<&str>, external_host: Option<&str>, port: u16) } } -/// Spawn an `enr_server` task bound to a free port and return the base URL -/// plus a cancellation handle. +/// Spawn an `enr_server` task on a listener bound to an ephemeral port, and +/// return the base URL plus a cancellation handle. +/// +/// The listener is bound here and handed over, so the returned URL names a port +/// that is already accepting connections: no free-port guess, and no readiness +/// poll for the bind. async fn spawn_server( p2p_config: P2PConfig, listeners: Vec, ) -> (String, CancellationToken, tokio::task::JoinHandle<()>) { - let http_port = pick_free_port().await; - let http_addr = format!("127.0.0.1:{http_port}"); - - let config = Config::builder() - .http_addr(http_addr.clone()) - .p2p_config(p2p_config.clone()) - .max_res_per_peer(8) - .max_conns(64) - .build(); - - let external_addrs = { - let mut v = external_tcp_multiaddrs(&p2p_config).expect("tcp externals"); - v.extend(external_udp_multiaddrs(&p2p_config).expect("udp externals")); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral"); + let http_addr = listener.local_addr().expect("local_addr"); + + // No swarm runs here, so the configured listen addresses stand in for the + // ones libp2p would report having bound. + let bound_addrs = { + let mut v = p2p_config.tcp_multiaddrs().expect("tcp listen addrs"); + v.extend(p2p_config.udp_multiaddrs().expect("udp listen addrs")); v }; + let external_addrs = external_multiaddrs(&p2p_config, &bound_addrs).expect("externals"); let secret_key = SecretKey::random(&mut OsRng); let peer_id = Keypair::generate_secp256k1().public().to_peer_id(); - let listeners = Arc::new(RwLock::new(listeners)); let ct = CancellationToken::new(); let (errs, _errs_rx) = mpsc::channel(4); - let ct_inner = ct.clone(); - let handle = tokio::spawn(pluto_relay_server::enr_server( - errs, - config, + let state = Arc::new(pluto_relay_server::AppState::new( + p2p_config, secret_key, peer_id, - listeners, + Arc::new(RwLock::new(listeners)), external_addrs, - ct_inner, + false, )); - // Wait until the server is actually accepting connections — the spawn is - // racy with the bind, and `reqwest` would otherwise hit `ConnectionRefused`. - let base_url = format!("http://{http_addr}"); - let start = std::time::Instant::now(); - let timeout = Duration::from_secs(5); - loop { - match reqwest::Client::new() - .get(format!("{base_url}/")) - .timeout(Duration::from_millis(200)) - .send() - .await - { - Ok(_) => break, - Err(_) if start.elapsed() < timeout => { - tokio::time::sleep(Duration::from_millis(20)).await; - } - Err(e) => panic!("server never came up: {e}"), - } - } + let ct_inner = ct.clone(); + let handle = tokio::spawn(pluto_relay_server::enr_server( + errs, listener, state, ct_inner, + )); - (base_url, ct, handle) + (format!("http://{http_addr}"), ct, handle) } async fn shutdown(ct: CancellationToken, handle: tokio::task::JoinHandle<()>) { From 81a9ccb549ba3a73be54a993e84d68f85be3efb0 Mon Sep 17 00:00:00 2001 From: Quang Le Date: Fri, 7 Aug 2026 17:04:10 +0700 Subject: [PATCH 3/3] refactor: simplify code --- crates/cli/src/commands/relay.rs | 86 ++++++++++---------------------- crates/p2p/src/utils.rs | 10 +--- crates/relay-server/src/p2p.rs | 44 ++++++++-------- crates/relay-server/src/web.rs | 14 +++--- 4 files changed, 59 insertions(+), 95 deletions(-) diff --git a/crates/cli/src/commands/relay.rs b/crates/cli/src/commands/relay.rs index 3fe9ab36..4e04e58f 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -431,8 +431,7 @@ mod tests { // Covers the CLI entry point that the fixture bypasses: tracing init and // the Loki drain. A pre-cancelled token is deterministic because the - // shutdown arm of the serve loop is the `biased` first branch — the same - // way charon's relay test starts (`cmd/relay/relay_internal_test.go:40`). + // shutdown arm of the serve loop is the `biased` first branch. let ct = CancellationToken::new(); ct.cancel(); @@ -493,12 +492,29 @@ mod tests { .await .unwrap(); - // Resolution happens asynchronously on a tick, so wait until the ENR - // reflects a non-loopback IP (mirrors the Go test using - // `assert.Eventually`). - relay - .get_until("/enr", |body| !parse_enr(body).ip().unwrap().is_loopback()) - .await; + // The relay is already serving, so this waits on one thing only: the + // hostname is resolved by a background task on a tick, and the ENR + // reflects it once DNS answers. A transport error or a non-2xx panics + // rather than being retried. + let deadline = Instant::now() + DNS_TIMEOUT; + loop { + let body = http_get(&relay.url("/enr")) + .await + .unwrap() + .text() + .await + .unwrap(); + + if !parse_enr(&body).ip().unwrap().is_loopback() { + break; + } + + assert!( + Instant::now() < deadline, + "external host not resolved into the ENR {DNS_TIMEOUT:?} in; last body: {body}" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } } #[tokio::test] @@ -617,9 +633,9 @@ mod tests { /// Budget for the relay to stop once a test is done with it. const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); - /// Budget for an endpoint to reach the state a test waits for. Sized for a - /// heavily loaded CI machine. - const SERVING_TIMEOUT: Duration = Duration::from_secs(30); + /// Budget for a DNS lookup to answer and reach the ENR. Sized for a heavily + /// loaded CI machine. + const DNS_TIMEOUT: Duration = Duration::from_secs(30); /// Per-request budget, so a server that accepts the connection but never /// answers fails the test instead of hanging it until the harness gives up. @@ -766,54 +782,6 @@ mod tests { .expect("`relay_args` configures both transports") } - /// Fetches `path` until it answers 2xx *and* `ready` accepts the body, - /// returning that body. - /// - /// This is not race tolerance — the relay is fully bound and serving - /// before a test gets hold of it, so a request can never be refused and - /// a transport error panics instead of being retried. The one thing it - /// waits out is DNS: `--p2p-external-hostname` is resolved on a tick by - /// a background task, so the ENR reflects it only after the first - /// lookup answers. Charon waits the same way (`assert.Eventually`, - /// `cmd/relay/relay_internal_test.go:208`). - async fn get_until(&self, path: &str, ready: impl Fn(&str) -> bool) -> String { - let started = Instant::now(); - - loop { - let response = CLIENT - .get(self.url(path)) - .timeout(REQUEST_TIMEOUT) - .send() - .await - .unwrap_or_else(|err| { - panic!( - "GET {path} failed: {err} (relay exited: {})", - self.handle.is_finished() - ) - }); - - let status = response.status(); - let body = response.text().await.unwrap(); - - if status.is_success() && ready(&body) { - return body; - } - - // The relay is the only thing serving this address, so if it is - // gone nothing will ever satisfy the poll. - assert!( - !self.handle.is_finished(), - "relay exited while waiting for {path} to serve" - ); - assert!( - started.elapsed() < SERVING_TIMEOUT, - "{path} not ready {SERVING_TIMEOUT:?} into startup; last status {status}: {body}" - ); - - tokio::time::sleep(Duration::from_millis(20)).await; - } - } - /// Cancels the relay, waits for it to stop, and returns its exit /// status. /// diff --git a/crates/p2p/src/utils.rs b/crates/p2p/src/utils.rs index 62f1c267..4e472222 100644 --- a/crates/p2p/src/utils.rs +++ b/crates/p2p/src/utils.rs @@ -31,10 +31,7 @@ use crate::{ /// `ports` must be the ports the node actually listens on: a configured port of /// 0 means the kernel picks one, so the configured value would advertise /// nothing dialable. -pub fn external_tcp_multiaddrs( - cfg: &P2PConfig, - ports: &[u16], -) -> crate::p2p::Result> { +fn external_tcp_multiaddrs(cfg: &P2PConfig, ports: &[u16]) -> crate::p2p::Result> { let mut resp = vec![]; if let Some(external_ip) = cfg.external_ip.as_ref() { @@ -60,10 +57,7 @@ pub fn external_tcp_multiaddrs( /// /// `ports` must be the ports the node actually listens on, as in /// [`external_tcp_multiaddrs`]. -pub fn external_udp_multiaddrs( - cfg: &P2PConfig, - ports: &[u16], -) -> crate::p2p::Result> { +fn external_udp_multiaddrs(cfg: &P2PConfig, ports: &[u16]) -> crate::p2p::Result> { let mut resp = vec![]; if let Some(external_ip) = cfg.external_ip.as_ref() { diff --git a/crates/relay-server/src/p2p.rs b/crates/relay-server/src/p2p.rs index 2ef41866..b3a2282d 100644 --- a/crates/relay-server/src/p2p.rs +++ b/crates/relay-server/src/p2p.rs @@ -36,7 +36,7 @@ pub async fn run_relay_p2p_node( config: &Config, key: SecretKey, ct: CancellationToken, -) -> Result> { +) -> Result<()> { bind_relay(config, key, ct).await?.serve().await } @@ -102,7 +102,7 @@ impl BoundRelay { /// Serves every bound listener until the relay is cancelled or one of its /// servers fails. - pub async fn serve(self) -> Result> { + pub async fn serve(self) -> Result<()> { let http_addr = self.http_addr(); let Self { mut node, @@ -159,17 +159,25 @@ impl BoundRelay { ct.cancel(); - if let Some(handle) = enr_server_handle { - join_or_abort("ENR server", handle).await; - } - - if let Some(handle) = monitoring_handle { - join_or_abort("Monitoring server", handle).await; - } + // Concurrently: the two are unrelated, and each gets its own grace + // period, so joining them in sequence would double the worst-case + // shutdown against an orchestrator's SIGTERM budget. + tokio::join!( + async { + if let Some(handle) = enr_server_handle { + join_or_abort("ENR server", handle).await; + } + }, + async { + if let Some(handle) = monitoring_handle { + join_or_abort("Monitoring server", handle).await; + } + }, + ); match server_error { Some(error) => Err(error), - None => Ok(node), + None => Ok(()), } } } @@ -183,8 +191,9 @@ impl BoundRelay { /// /// Order matters: the HTTP listeners are bound before the swarm is created, and /// the swarm is polled only once they are. Polling is what makes the relay -/// service p2p connections, so a bind that failed after it would take down -/// peers the relay had already accepted. +/// service p2p connections — it completes handshakes and can hand out circuit +/// reservations — so a bind that failed after it would take down peers the +/// relay had already accepted. #[doc(hidden)] #[instrument(skip(config, key, ct))] pub async fn bind_relay( @@ -200,15 +209,8 @@ pub async fn bind_relay( "Pluto relay starting" ); - // The HTTP listeners are bound before the swarm exists, and the swarm is not - // polled until they are. Polling accepts and services p2p connections — it - // completes handshakes, counts them, and can hand out circuit reservations — - // so binding these afterwards would mean a bind failure tore down peers the - // relay had already taken on. - // - // Binding here rather than inside each server's task is also what lets an - // unusable address or a lost race for a port fail the relay while nothing - // has been spawned. + // Bound here rather than inside each server's task, so an unusable address + // or a lost race for a port fails the relay while nothing has been spawned. let monitoring_server = match config.monitoring_addr.clone() { Some(monitoring_addr) => { let bind_addr = monitoring_addr diff --git a/crates/relay-server/src/web.rs b/crates/relay-server/src/web.rs index 85802808..c3d4e53b 100644 --- a/crates/relay-server/src/web.rs +++ b/crates/relay-server/src/web.rs @@ -76,9 +76,9 @@ impl AppState { /// Returns the union of configured external multiaddrs and the live libp2p /// listen addresses, externals first, deduped while preserving order. /// - /// Mirrors Go charon's `filterAdvertisedAddrs(externalAddrs, internalAddrs, - /// excludeInternalPrivate)`: `filter_private_addrs` withholds private - /// listen addresses (loopback, RFC 1918) but never the external ones. + /// `filter_private_addrs` withholds private listen addresses (loopback, + /// RFC 1918) but never the external ones, which are advertised as + /// configured. /// /// Filtering happens here rather than when a listen address is ingested so /// that the relay always knows every address it bound — startup waits on @@ -91,7 +91,7 @@ impl AppState { .filter(|addr| !(self.filter_private_addrs && addr.is_private())); let mut seen: HashSet<&Multiaddr> = HashSet::new(); - let mut union: Vec = Vec::new(); + let mut union: Vec = Vec::with_capacity(self.external_addrs.len()); for addr in self.external_addrs.iter().chain(listeners) { if seen.insert(addr) { union.push(addr.clone()); @@ -189,9 +189,9 @@ pub(crate) async fn bind_monitoring_server( /// Serves an already bound monitoring listener until shutdown. /// -/// Serve failures are reported on `server_errors` (mirroring Charon, where the -/// monitoring server's `ListenAndServe` error terminates the relay) so they -/// cannot go unnoticed behind a log line. +/// Serve failures are reported on `server_errors`, so a monitoring server that +/// dies takes the relay down with it instead of going unnoticed behind a log +/// line and leaving the port unserved. #[instrument(skip_all, fields(addr = %server.local_addr()))] pub(crate) async fn serve_monitoring_server( server_errors: mpsc::Sender,