diff --git a/Cargo.lock b/Cargo.lock index 73022c62..ad6bd428 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -69,14 +69,15 @@ checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" [[package]] name = "alloy" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50ab0cd8afe573d1f7dc2353698a51b1f93aec362c8211e28cfd3948c6adba39" +checksum = "754c72f6ae867220976be4693d870f8f9866437f37fdeb5640aa24991e5ff97f" dependencies = [ "alloy-consensus", "alloy-contract", "alloy-core", "alloy-eips", + "alloy-ens", "alloy-genesis", "alloy-network", "alloy-provider", @@ -103,9 +104,9 @@ dependencies = [ [[package]] name = "alloy-consensus" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f16daaf7e1f95f62c6c3bf8a3fc3d78b08ae9777810c0bb5e94966c7cd57ef0" +checksum = "99bdf6a3e99089d97e137009956928b41d645035b220cb303eae88182882e9ca" dependencies = [ "alloy-eips", "alloy-primitives", @@ -130,9 +131,9 @@ dependencies = [ [[package]] name = "alloy-consensus-any" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "118998d9015332ab1b4720ae1f1e3009491966a0349938a1f43ff45a8a4c6299" +checksum = "18d98872f189ea87063912de45e7e0d6ae44700a8836f8bfa79b3265f0c38bfe" dependencies = [ "alloy-consensus", "alloy-eips", @@ -144,9 +145,9 @@ dependencies = [ [[package]] name = "alloy-contract" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ac9e0c34dc6bce643b182049cdfcca1b8ce7d9c260cbdd561f511873b7e26cd" +checksum = "f69e65d1b6adc7e46c0e93c2cf3cb4e8646e66a2ced4ba38d542b724acd54774" dependencies = [ "alloy-consensus", "alloy-dyn-abi", @@ -162,7 +163,9 @@ dependencies = [ "futures-util", "serde_json", "thiserror 2.0.18", + "tokio", "tracing", + "wasmtimer", ] [[package]] @@ -234,9 +237,9 @@ dependencies = [ [[package]] name = "alloy-eip7928" -version = "0.3.7" +version = "0.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b827a6d7784fe3eb3489d40699407a4cdcce74271421a01bdffe60cf573bb16" +checksum = "b3b12337f74cbfa451cb04dac173974814a6ff463079e1793aa09600ba8813ab" dependencies = [ "alloy-primitives", "alloy-rlp", @@ -248,9 +251,9 @@ dependencies = [ [[package]] name = "alloy-eips" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6ef28c9fdad22d4eec52d894f5f2673a0895f1e5ef196734568e68c0f6caca8" +checksum = "74bffa6f6440f1ee3c8cd7bc9dc2b01fd837641ef9bd566bd9f8922e75ff8e6c" dependencies = [ "alloy-eip2124", "alloy-eip2930", @@ -269,11 +272,25 @@ dependencies = [ "sha2", ] +[[package]] +name = "alloy-ens" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa8a7af8c504c7896dca006666d3d0c69212af69a0b947209c0605fe35b8e66d" +dependencies = [ + "alloy-contract", + "alloy-primitives", + "alloy-provider", + "alloy-sol-types", + "async-trait", + "thiserror 2.0.18", +] + [[package]] name = "alloy-genesis" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbf9480307b09d22876efb67d30cadd9013134c21f3a17ec9f93fd7536d38024" +checksum = "f48592ff562892af6a2ebb6e214ed1d2795a6c1dc601c601ef0863583c799003" dependencies = [ "alloy-eips", "alloy-primitives", @@ -298,9 +315,9 @@ dependencies = [ [[package]] name = "alloy-json-rpc" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "422d110f1c40f1f8d0e5562b0b649c35f345fccb7093d9f02729943dcd1eef71" +checksum = "31de46f2b1948ac9e378212fd79cb963745cbb29cfe08a13c5b675d7e0934a24" dependencies = [ "alloy-primitives", "alloy-sol-types", @@ -313,9 +330,9 @@ dependencies = [ [[package]] name = "alloy-network" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7197a66d94c4de1591cdc16a9bcea5f8cccd0da81b865b49aef97b1b4016e0fa" +checksum = "fa3ea75a876a4cd302c15dad628cc55f1f0b2db7fd4865f57f4e24cea2629298" dependencies = [ "alloy-consensus", "alloy-consensus-any", @@ -339,9 +356,9 @@ dependencies = [ [[package]] name = "alloy-network-primitives" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb82711d59a43fdfd79727c99f270b974c784ec4eb5728a0d0d22f26716c87ef" +checksum = "f93f09f2bee068dcd80bfdf057109d64e37b318f21d3d64851139ca741366dd8" dependencies = [ "alloy-consensus", "alloy-eips", @@ -380,9 +397,9 @@ dependencies = [ [[package]] name = "alloy-provider" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf6b18b929ef1d078b834c3631e9c925177f3b23ddc6fa08a722d13047205876" +checksum = "4e9070296bd2902aac4d9defd85608116eceb319526746257ab97096da2c5603" dependencies = [ "alloy-chains", "alloy-consensus", @@ -441,9 +458,9 @@ dependencies = [ [[package]] name = "alloy-rpc-client" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94fcc9604042ca80bd37aa5e232ea1cd851f337e31e2babbbb345bc0b1c30de3" +checksum = "87da7170c18c43a5def51f5f166d0260776b721e184c349fae1cd5466cb67362" dependencies = [ "alloy-json-rpc", "alloy-primitives", @@ -464,9 +481,9 @@ dependencies = [ [[package]] name = "alloy-rpc-types" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4faad925d3a669ffc15f43b3deec7fbdf2adeb28a4d6f9cf4bc661698c0f8f4b" +checksum = "88813dc63f261e25d158f07063a71e926f36b3ee8076c1e6d8396348452c5f69" dependencies = [ "alloy-primitives", "alloy-rpc-types-eth", @@ -476,20 +493,24 @@ dependencies = [ [[package]] name = "alloy-rpc-types-any" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3823026d1ed239a40f12364fac50726c8daf1b6ab8077a97212c5123910429ed" +checksum = "6574846e24e090607d78325e90256c9a5ce4010cf0e663b23d06e01fd74356f0" dependencies = [ "alloy-consensus-any", + "alloy-network-primitives", + "alloy-primitives", "alloy-rpc-types-eth", "alloy-serde", + "serde", + "serde_json", ] [[package]] name = "alloy-rpc-types-eth" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59c095f92c4e1ff4981d89e9aa02d5f98c762a1980ab66bec49c44be11349da2" +checksum = "acc71ae98333ab278782d59423bd8317e7658954c356e1f4ce559fcea4f4da85" dependencies = [ "alloy-consensus", "alloy-consensus-any", @@ -508,9 +529,9 @@ dependencies = [ [[package]] name = "alloy-serde" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11ece63b89294b8614ab3f483560c08d016930f842bf36da56bf0b764a15c11e" +checksum = "6f11544d72bc87eab3cf4addc61d1f735bf88f3769482eca3deefc0c2558d2d0" dependencies = [ "alloy-primitives", "serde", @@ -519,9 +540,9 @@ dependencies = [ [[package]] name = "alloy-signer" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43f447aefab0f1c0649f71edc33f590992d4e122bc35fb9cdbbf67d4421ace85" +checksum = "e76dfc1a6c8337e59624b9f6f02245566b7d24e8dab0eb842d57d4fd8777adc3" dependencies = [ "alloy-primitives", "async-trait", @@ -534,9 +555,9 @@ dependencies = [ [[package]] name = "alloy-signer-local" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f721f4bf2e4812e5505aaf5de16ef3065a8e26b9139ac885862d00b5a55a659a" +checksum = "1a9f88d471c719ce2bdf949191f8a1c08bffd0c030606c6a8da0d414454e8bfe" dependencies = [ "alloy-consensus", "alloy-network", @@ -623,9 +644,9 @@ dependencies = [ [[package]] name = "alloy-transport" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8098f965442a9feb620965ba4b4be5e2b320f4ec5a3fff6bfa9e1ff7ef42bed1" +checksum = "f9399b22a1d2a82c8c30fd527541ba61e574795d0b95f3b902d3473d11489be5" dependencies = [ "alloy-json-rpc", "auto_impl", @@ -646,9 +667,9 @@ dependencies = [ [[package]] name = "alloy-transport-http" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8597d36d546e1dab822345ad563243ec3920e199322cb554ce56c8ef1a1e2e7" +checksum = "59b55d7ccf45bec99db692c1266413498d2c9776c390cf148aa590422e31b2d3" dependencies = [ "alloy-json-rpc", "alloy-transport", @@ -678,9 +699,9 @@ dependencies = [ [[package]] name = "alloy-tx-macros" -version = "1.8.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d69722eddcdf1ce096c3ab66cf8116999363f734eb36fe94a148f4f71c85da84" +checksum = "de9792a54c0c8cc009883436ff1a60eb091095aced1c167a764bedbbfa0eee79" dependencies = [ "darling 0.23.0", "proc-macro2", @@ -1540,7 +1561,7 @@ version = "3.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dee98b0db6a962de883bf5d20362dee4d7ca0d12fe39a7c6c73c844e1cd7c1f" dependencies = [ - "darling 0.20.11", + "darling 0.23.0", "ident_case", "prettyplease", "proc-macro2", @@ -2568,7 +2589,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3091,23 +3112,14 @@ dependencies = [ "foldhash 0.1.5", ] -[[package]] -name = "hashbrown" -version = "0.16.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" -dependencies = [ - "allocator-api2", - "equivalent", - "foldhash 0.2.0", -] - [[package]] name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" dependencies = [ + "allocator-api2", + "equivalent", "foldhash 0.2.0", "serde", "serde_core", @@ -3425,7 +3437,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.57.0", + "windows-core 0.62.2", ] [[package]] @@ -4327,11 +4339,11 @@ dependencies = [ [[package]] name = "lru" -version = "0.16.4" +version = "0.18.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39" +checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" dependencies = [ - "hashbrown 0.16.1", + "hashbrown 0.17.1", ] [[package]] @@ -4610,7 +4622,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5181,7 +5193,6 @@ dependencies = [ name = "pluto-cli" version = "1.7.1" dependencies = [ - "backon", "bytes", "chrono", "clap", @@ -6099,7 +6110,7 @@ dependencies = [ "once_cell", "socket2 0.6.4", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6562,7 +6573,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6621,7 +6632,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -7345,7 +7356,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -8319,7 +8330,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 56dce23d..8835e3b6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,7 +39,7 @@ publish = false [workspace.dependencies] async-trait = "0.1.89" -alloy = { version = "1.3", features = ["essentials"] } +alloy = { version = "2.4", features = ["essentials"] } built = { version = "0.8.0", features = ["git2", "chrono", "cargo-lock"] } blst = "0.3" bytes = "1" 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..67edee63 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -346,6 +346,18 @@ 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_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,216 +381,310 @@ 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 backon::{BackoffBuilder, Retryable}; - use std::{str::FromStr, time}; - use tokio::net; + use std::{ + net::{Ipv4Addr, SocketAddr}, + path::Path, + str::FromStr, + sync::LazyLock, + time::{Duration, Instant}, + }; + use tokio::{net, task::JoinHandle}; use tokio_util::sync::CancellationToken; #[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. + 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(); - - assert!( - !addresses.is_empty(), - "Expected at least one multiaddr in response" - ); + let relay = test_relay_server().await.unwrap(); - for addr in addresses { - libp2p::Multiaddr::from_str(&addr).unwrap_or_else(|err| { - panic!("Failed to parse multiaddr '{}': {}", addr, err); - }); - } - }, - ) - .await - .unwrap(); + let response = http_get(&relay.url("/")).await.unwrap(); + let body = response.text().await.unwrap(); + let addresses: Vec = serde_json::from_str(&body).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(); - - assert_eq!(enr.ip(), Some(std::net::Ipv4Addr::new(127, 0, 0, 1))); - }, - ) - .await - .unwrap(); + let relay = test_relay_server().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(); - - assert_eq!(enr.ip(), Some(std::net::Ipv4Addr::new(222, 222, 222, 222))); - }, - ) - .await - .unwrap(); + let relay = + test_relay_server_with(|args| args.p2p.external_ip = Some("222.222.222.222".into())) + .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(time::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(time::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(); + + // 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] 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); - }, - async move |_cfg| { - let response = retry_get(&monitoring_url).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(); + + 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")); } - /// 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. - async fn with_relay_server( - config_fn: FArgs, - test_fn: FTest, - ) -> Result<(), crate::error::CliError> - where - FArgs: FnOnce(&mut super::RelayArgs), - FTest: FnOnce(pluto_relay_server::config::Config) -> Fut, - Fut: std::future::Future, - { - let dir = tempfile::tempdir().unwrap(); + #[tokio::test] + async fn taken_http_port_fails_the_relay() { + let (_taken, addr) = squat_tcp_addr().await; - let tcp_addr = net::TcpListener::bind("127.0.0.1:0") + let err = test_relay_server_with(|args| args.relay.http_address = addr) .await - .unwrap() - .local_addr() - .unwrap() - .to_string(); + .expect_err("relay must not start while its http port is taken"); + + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToBindHttpListener { .. } + ) + ), + "got: {err}" + ); + } + + #[tokio::test] + async fn taken_monitoring_port_fails_the_relay() { + let (_taken, addr) = squat_tcp_addr().await; - let udp_addr = net::UdpSocket::bind("127.0.0.1:0") + let err = test_relay_server_with(|args| args.debug_monitoring.monitor_addr = Some(addr)) .await - .unwrap() - .local_addr() - .unwrap() - .to_string(); + .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}" + ); + } + + #[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"); + + assert!( + matches!( + err, + super::CliError::RelayP2PError( + pluto_relay_server::RelayP2PError::FailedToParseMonitoringAddr(..) + ) + ), + "got: {err}" + ); + } + + #[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; - let http_addr = net::TcpListener::bind("127.0.0.1:0") + relay.stop().await.unwrap(); + + // The relay task has joined, and with it its listener must be gone. + net::TcpListener::bind(http_addr) .await - .unwrap() - .local_addr() - .unwrap() - .to_string(); + .unwrap_or_else(|err| panic!("relay did not release {http_addr}: {err}")); + } + + #[tokio::test] + async fn cancelling_before_startup_stops_the_relay_without_serving() { + let dir = tempfile::tempdir().unwrap(); + let config: pluto_relay_server::config::Config = relay_args(dir.path()).try_into().unwrap(); + let key = super::load_or_create_key(&config).unwrap(); + + let ct = CancellationToken::new(); + ct.cancel(); + + // An already-cancelled token must stop the relay cleanly before it + // binds anything, not run it until some later cancellation check. + let result = tokio::time::timeout( + SHUTDOWN_TIMEOUT, + pluto_relay_server::p2p::run_relay_p2p_node(&config, key, ct), + ) + .await + .expect("relay must return promptly when cancelled at startup"); + assert!(result.is_ok(), "expected clean shutdown, got: {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(); + 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"); + } + + /// Budget for the relay to stop once a test is done with it. + const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); + + /// 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. + const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); - let mut args = super::RelayArgs { + /// 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: http_addr, + http_address: ANY_ADDR.into(), auto_p2p_key: true, p2p_relay_log_level: "info".into(), max_res_per_peer: 0, @@ -593,8 +699,8 @@ mod tests { relays: vec![], external_ip: None, external_host: None, - tcp_addrs: vec![tcp_addr], - udp_addrs: vec![udp_addr], + tcp_addrs: vec![ANY_ADDR.into()], + udp_addrs: vec![ANY_ADDR.into()], disable_reuseport: false, }, log: super::RelayLogFlags { @@ -607,37 +713,141 @@ mod tests { loki_addresses: vec![], loki_service: "".into(), }, - }; - config_fn(&mut args); + } + } + + /// A relay that serves for as long as this value is alive. + /// + /// 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, + } + + /// 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 + } + + /// Starts a relay in a fresh data dir, letting `configure` adjust the + /// [`super::RelayArgs`] it is built from. + /// + /// 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); + + let config: pluto_relay_server::config::Config = args.try_into()?; + let key = super::load_or_create_key(&config)?; - let cfg: pluto_relay_server::config::Config = args.clone().try_into().unwrap(); let ct = CancellationToken::new(); + let bound = pluto_relay_server::p2p::bind_relay(&config, key).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 serve_ct = ct.child_token(); + let handle = tokio::spawn(async move { bound.serve(serve_ct).await.map_err(Into::into) }); + + Ok(TestRelay { + http_addr, + monitoring_addr, + p2p_addrs, + ct, + handle, + _dir: dir, + }) + } - let relay = tokio::spawn(super::run(cfg.clone(), ct.child_token())); + impl TestRelay { + /// URL for `path` on the relay's HTTP server. + fn url(&self, path: &str) -> String { + format!("http://{}{path}", self.http_addr) + } - test_fn(cfg.clone()).await; + /// 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") + } - ct.cancel(); - relay.await.unwrap() + /// 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:?}"), + } + } + } + + 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(); + } } - /// Make an HTTP GET request to the relay server with retries and backoff. - 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 + /// Parses an ENR response body. + fn parse_enr(body: &str) -> pluto_eth2util::enr::Record { + pluto_eth2util::enr::Record::try_from(body).unwrap() + } + + /// Single-shot GET, failing on a non-2xx status. + /// + /// 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 { + CLIENT + .get(url) + .timeout(REQUEST_TIMEOUT) + .send() + .await + .and_then(|response| response.error_for_status()) } - 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 + /// 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 298056a2..4e472222 100644 --- a/crates/p2p/src/utils.rs +++ b/crates/p2p/src/utils.rs @@ -26,23 +26,18 @@ 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. +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 +45,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`]. +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 +71,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); @@ -203,3 +223,67 @@ pub fn filter_direct_quic_addrs(addrs: impl Iterator) -> Vec bool { !is_relay_addr(addr) } + +#[cfg(test)] +mod tests { + use super::*; + + /// 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), + ..Default::default() + } + } + + fn as_strings(addrs: &[Multiaddr]) -> Vec { + addrs.iter().map(ToString::to_string).collect() + } + + #[test] + fn external_multiaddrs_keep_the_bound_ports() { + let cfg = config(Some("1.2.3.4"), Some("relay.example.com")); + // 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_multiaddrs(&cfg, &listen_addrs).unwrap()), + vec![ + "/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 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 8cd118f3..c331ee5e 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,42 @@ 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. A `String` because + /// [`crate::config::Config::http_addr`] is never parsed — the listener + /// binds the configured string directly. + addr: String, + /// Underlying bind error. + #[source] + source: std::io::Error, + }, /// Failed to serve HTTP. #[error("Failed to serve HTTP: {0}")] - FailedToServeHTTP(std::io::Error), + FailedToServeHTTP(#[source] std::io::Error), + + /// A libp2p listener closed before reporting the address it bound. + #[error("libp2p listener closed during startup: {source}")] + ListenerClosedDuringStartup { + /// Why the listener closed. + #[source] + 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. + #[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/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 a3ad389f..d90a77e6 100644 --- a/crates/relay-server/src/p2p.rs +++ b/crates/relay-server/src/p2p.rs @@ -1,42 +1,235 @@ //! Relay P2P node implementation. -use std::{net::SocketAddr, sync::Arc, time::Duration}; +use std::{collections::HashSet, future::Future, net::SocketAddr, sync::Arc}; 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}; 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::{enr_server, 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. +/// +/// Cancellation is honored during startup too: it drops the bind — and with it +/// everything the bind held — and reports a clean shutdown. #[instrument(skip(config, key, ct))] pub async fn run_relay_p2p_node( config: &Config, key: SecretKey, ct: CancellationToken, -) -> Result> { +) -> Result<()> { + let bound = tokio::select! { + biased; + () = ct.cancelled() => { + info!("Relay server shutdown signal received during startup"); + return Ok(()); + } + bound = bind_relay(config, key) => bound?, + }; + + bound.serve(ct).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. +/// +/// 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>, +} + +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 `ct` is cancelled or one of the + /// servers fails. + /// + /// The servers run as `select!` arms rather than spawned tasks, so leaving + /// the loop — for either reason — drops them, and with them their + /// listeners, before this returns: a failed relay never leaves a listener + /// bound behind it. `ct` is cancelled on the way out so tasks scoped to it + /// die with the relay. + pub async fn serve(self, 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" + ); + + let http_addr = self.http_addr(); + let Self { + mut node, + listen_addrs, + state, + enr_listener, + monitoring_server, + } = self; + + let enr_server_fut = serve_if_bound( + enr_listener.map(|listener| enr_server(listener, state, ct.child_token())), + ); + tokio::pin!(enr_server_fut); + + // 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"); + } + + let monitoring_fut = serve_if_bound(monitoring_server.map(serve_monitoring_server)); + tokio::pin!(monitoring_fut); + + let result = loop { + tokio::select! { + biased; + _ = ct.cancelled() => { + info!("Relay server shutdown signal received, shutting down"); + break Ok(()); + }, + result = &mut enr_server_fut => break result, + result = &mut monitoring_fut => break result, + event = node.select_next_some() => { + apply_addr_update(&listen_addrs, handle_swarm_event(&event)).await; + } + } + }; + + ct.cancel(); + result + } +} + +/// Runs a server future when its listener is configured, and stays pending +/// otherwise, so the `select!` arm of an absent server never fires. +async fn serve_if_bound(server: Option>>) -> Result<()> { + match server { + Some(server) => server.await, + None => std::future::pending().await, + } +} + +/// 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 — 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))] +pub async fn bind_relay(config: &Config, key: SecretKey) -> Result { + // 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 + .parse::() + .map_err(|_| RelayP2PError::FailedToParseMonitoringAddr(monitoring_addr))?; + + Some(bind_monitoring_server(bind_addr).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,122 +246,111 @@ 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)?); + let external_addrs = external_multiaddrs(&config.p2p_config, &bound_addrs)?; - 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, + }) +} - // 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 - }; +/// 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(); - 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); - return Err(error); - } - }, - event = node.select_next_some() => { - let address_update = handle_swarm_event(&event, config.filter_private_addrs); - - // 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 => {} - } + while !pending.is_empty() { + let event = node.select_next_some().await; + let addr_update = handle_swarm_event(&event); + + 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); + reason.map_err(|source| RelayP2PError::ListenerClosedDuringStartup { source })?; } + _ => {} } + + apply_addr_update(listen_addrs, addr_update).await; } - ct.cancel(); + Ok(()) +} - match tokio::time::timeout(Duration::from_secs(2), enr_server_handle).await { - Ok(Ok(())) => { - info!("ENR server shutdown complete"); +/// 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); } - Ok(Err(e)) => { - warn!("ENR server shutdown error: {}", e); - } - 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"); - } - } - } - - Ok(node) } /// Result of a swarm event that may require updating the listener address list. @@ -188,21 +370,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 4a5c47a8..dcbdd6a8 100644 --- a/crates/relay-server/src/web.rs +++ b/crates/relay-server/src/web.rs @@ -15,19 +15,16 @@ use axum::{ use k256::SecretKey; use libp2p::{Multiaddr, PeerId, multiaddr}; use pluto_eth2util::enr::{EnrEntry, Record}; -use tokio::{ - net::TcpListener, - sync::{RwLock, mpsc}, -}; +use tokio::{net::TcpListener, sync::RwLock}; 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, + 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 +35,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 +57,7 @@ impl AppState { peer_id: PeerId, addrs: Arc>>, external_addrs: Vec, + filter_private_addrs: bool, ) -> Self { Self { p2p_config, @@ -63,20 +65,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. + /// + /// `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 + /// 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()) { + 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()); } @@ -96,95 +109,88 @@ 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 `ct` is +/// cancelled — a clean `Ok(())` — or the server fails. +/// +/// 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; - }; - +) -> Result<()> { info!("Starting ENR server"); - let state = AppState::new( - config.p2p_config.clone(), - secret_key, - peer_id, - addrs, - external_addrs, - ); - 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); - - 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; - }; + 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)) + }); 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(", "), + peer_name = peer_name(&state.peer_id), + tcp_addrs = state.p2p_config.tcp_addrs.join(", "), + udp_addrs = state.p2p_config.udp_addrs.join(", "), ); + let router = Router::new() + .route("/", get(multiaddr_handler)) + .route("/enr", get(enr_handler)) + .with_state(state); + let ct_clone = ct.child_token(); - if let Err(e) = axum::serve(listener, router) + let result = axum::serve(listener, router) .with_graceful_shutdown(async move { ct_clone.cancelled().await; info!("ENR server shutdown complete"); }) .await - { - warn!("HTTP server error: {}", e); - let _ = server_errors - .send(RelayP2PError::FailedToServeHTTP(e)) - .await; - } + .map_err(RelayP2PError::FailedToServeHTTP); ct.cancel(); if let Some(resolver_handle) = resolver_handle { let _ = resolver_handle.await; } + + result } -/// Starts the Prometheus monitoring server on the given address. -#[instrument(skip(ct))] -pub async fn monitoring_server(bind_addr: SocketAddr, ct: CancellationToken) { - info!("Starting monitoring server on {bind_addr}"); +/// Binds the Prometheus monitoring listener on the given address. +/// +/// 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] +pub(crate) async fn bind_monitoring_server( + bind_addr: SocketAddr, +) -> 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. +/// +/// Returns only on failure: the server has no shutdown signal of its own and +/// is stopped by dropping this future, which closes its listener. +#[instrument(skip_all, fields(addr = %server.local_addr()))] +pub(crate) async fn serve_monitoring_server(server: MetricsServer<'static>) -> Result<()> { + info!("Starting monitoring server"); + + server + .start() + .await + .map_err(RelayP2PError::FailedToServeMonitoring) } /// Error response for HTTP handlers. @@ -405,6 +411,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(); @@ -419,6 +435,7 @@ mod tests { peer_id, Arc::new(RwLock::new(listeners)), external_addrs, + filter_private_addrs, ); (Arc::new(state), peer_id) } @@ -490,6 +507,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..3944611a 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,33 +14,15 @@ 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, - sync::{RwLock, mpsc}, -}; +use tokio::{net::TcpListener, sync::RwLock}; 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,69 +33,53 @@ 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")); +) -> (String, CancellationToken, ServerHandle) { + 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(listener, state, ct_inner)); - (base_url, ct, handle) + (format!("http://{http_addr}"), ct, handle) } -async fn shutdown(ct: CancellationToken, handle: tokio::task::JoinHandle<()>) { +/// Task the `enr_server` runs on, resolving with the server's exit status. +type ServerHandle = tokio::task::JoinHandle>; + +async fn shutdown(ct: CancellationToken, handle: ServerHandle) { ct.cancel(); // The server may take a moment to drain; bound the wait so a hung test // fails loudly instead of hanging CI.