From aa9a6d965c94da5ffb49ffef72a96a49bc18cfbc Mon Sep 17 00:00:00 2001 From: Quang Le Date: Mon, 24 Aug 2026 17:14:02 +0700 Subject: [PATCH 1/2] test(p2p): cover peer, utils, quic_upgrade, force_direct, proto and bootnode --- crates/p2p/Cargo.toml | 2 +- crates/p2p/src/bootnode.rs | 86 +++++++++++++ crates/p2p/src/force_direct.rs | 224 +++++++++++++++++++++++++++++++++ crates/p2p/src/peer.rs | 143 +++++++++++++++++++-- crates/p2p/src/proto.rs | 74 +++++++++++ crates/p2p/src/quic_upgrade.rs | 94 ++++++++++++++ crates/p2p/src/utils.rs | 176 ++++++++++++++++++++++++++ 7 files changed, 788 insertions(+), 11 deletions(-) diff --git a/crates/p2p/Cargo.toml b/crates/p2p/Cargo.toml index 1765c4a1..545394f1 100644 --- a/crates/p2p/Cargo.toml +++ b/crates/p2p/Cargo.toml @@ -37,7 +37,7 @@ pluto-testutil.workspace = true vise-exporter.workspace = true anyhow.workspace = true clap.workspace = true -pluto-cluster.workspace = true +pluto-cluster = { workspace = true, features = ["test-cluster"] } hex.workspace = true libp2p.workspace = true k256.workspace = true diff --git a/crates/p2p/src/bootnode.rs b/crates/p2p/src/bootnode.rs index ea2d2ec3..fc2b4439 100644 --- a/crates/p2p/src/bootnode.rs +++ b/crates/p2p/src/bootnode.rs @@ -404,8 +404,11 @@ fn addr_info_from_p2p_addr(addr: &Multiaddr) -> std::result::Result) -> String { + Record::new(key, entries).expect("build enr").to_string() + } + + #[test] + fn multi_addr_from_enr_str_maps_ports_to_transports() { + let key = k256::SecretKey::random(&mut OsRng); + let peer_id = peer_id_from_key(key.public_key()).expect("peer id from key"); + let ip = EnrEntry::Ipv4(Ipv4Addr::new(1, 2, 3, 4)); + + // The UDP port is advertised as QUIC, and comes first when both ports + // are set so a dialer prefers it over TCP. + let cases = [ + (vec![ip, EnrEntry::Tcp(3610)], vec!["/ip4/1.2.3.4/tcp/3610"]), + ( + vec![ip, EnrEntry::Udp(3630)], + vec!["/ip4/1.2.3.4/udp/3630/quic-v1"], + ), + ( + vec![ip, EnrEntry::Tcp(3610), EnrEntry::Udp(3630)], + vec!["/ip4/1.2.3.4/udp/3630/quic-v1", "/ip4/1.2.3.4/tcp/3610"], + ), + ]; + + for (entries, want) in cases { + let addrs = + multi_addr_from_enr_str(&enr_str(&key, entries)).expect("enr should resolve"); + let want: Vec = want + .iter() + .map(|addr| format!("{addr}/p2p/{peer_id}").parse().expect("multiaddr")) + .collect(); + + assert_eq!(addrs, want); + } + } + + #[test] + fn multi_addr_from_enr_str_rejects_an_enr_without_an_ip() { + let key = k256::SecretKey::random(&mut OsRng); + let enr = enr_str(&key, vec![EnrEntry::Tcp(3610)]); + + let err = multi_addr_from_enr_str(&enr).expect_err("an ip is required"); + + assert!( + matches!(err, BootnodeError::EnrNoIp), + "unexpected error: {err}" + ); + } + + #[test] + fn multi_addr_from_enr_str_rejects_an_enr_without_a_port() { + let key = k256::SecretKey::random(&mut OsRng); + let enr = enr_str(&key, vec![EnrEntry::Ipv4(Ipv4Addr::new(1, 2, 3, 4))]); + + let err = multi_addr_from_enr_str(&enr).expect_err("a port is required"); + + assert!( + matches!(err, BootnodeError::EnrNoPort), + "unexpected error: {err}" + ); + } + + #[test] + fn multi_addr_from_enr_str_rejects_garbage() { + for garbage in [ + "", + "not-an-enr", + // Right prefix, unparsable body. + "enr:not-base64-@@@", + // Valid base64, but not an RLP-encoded record. + "enr:AAAAAAAA", + ] { + let err = multi_addr_from_enr_str(garbage) + .expect_err("garbage must not resolve to an address"); + + assert!( + matches!(err, BootnodeError::ParseEnr(_)), + "unexpected error for {garbage:?}: {err}" + ); + } + } } diff --git a/crates/p2p/src/force_direct.rs b/crates/p2p/src/force_direct.rs index b8c9f49c..b308e297 100644 --- a/crates/p2p/src/force_direct.rs +++ b/crates/p2p/src/force_direct.rs @@ -278,3 +278,227 @@ impl NetworkBehaviour for ForceDirectBehaviour { Poll::Pending } } + +#[cfg(test)] +mod tests { + use libp2p::swarm::ConnectionId; + + use super::*; + use crate::p2p_context::Peer; + + const RELAY_ID: &str = "16Uiu2HAkzdQ5Y9SYT91K1ue5SxXwgmajXntfScGnLYeip5hHyWmT"; + + fn addr(s: &str) -> Multiaddr { + s.parse().unwrap() + } + + fn relayed(transport: &str) -> Multiaddr { + addr(&format!("{transport}/p2p/{RELAY_ID}/p2p-circuit")) + } + + fn conn(id: PeerId, n: usize, remote_addr: Multiaddr) -> Peer { + Peer { + id, + connection_id: ConnectionId::new_unchecked(n), + remote_addr, + } + } + + fn behaviour(local: PeerId, peers: impl IntoIterator) -> ForceDirectBehaviour { + let known: Vec = peers.into_iter().chain(std::iter::once(local)).collect(); + + ForceDirectBehaviour::new(P2PContext::new(known), local) + } + + /// Seeds `conns` and, when `addresses` is `Some`, the identify-reported + /// addresses for `peer`. + fn seed_store( + behaviour: &ForceDirectBehaviour, + peer: PeerId, + conns: Vec, + addresses: Option>, + ) { + // Scoped so the write lock is released before the logic under test + // takes its read lock. + let mut store = behaviour.p2p_context.peer_store_write_lock(); + for conn in conns { + store.add_peer(conn); + } + if let Some(addresses) = addresses { + store.set_peer_addresses(peer, addresses); + } + } + + /// The `Debug` rendering of the queued dial to `peer`. `DialOpts` keeps its + /// address list `pub(crate)`, so this is the only way to see which + /// addresses were selected. + fn dial_debug(behaviour: &ForceDirectBehaviour, peer: &PeerId) -> String { + behaviour + .pending_events + .iter() + .find_map(|event| match event { + ToSwarm::Dial { opts } if opts.get_peer_id().as_ref() == Some(peer) => { + Some(format!("{opts:?}")) + } + _ => None, + }) + .expect("a dial to the peer should be queued") + } + + /// The peers the queued events dial. + fn dialled(behaviour: &ForceDirectBehaviour) -> Vec { + behaviour + .pending_events + .iter() + .filter_map(|event| match event { + ToSwarm::Dial { opts } => opts.get_peer_id(), + _ => None, + }) + .collect() + } + + #[tokio::test] + async fn forces_direct_when_every_connection_is_relayed() { + let local = PeerId::random(); + let peer = PeerId::random(); + let behaviour = &mut behaviour(local, [peer]); + seed_store( + behaviour, + peer, + vec![conn(peer, 1, relayed("/ip4/1.2.3.4/tcp/3610"))], + // Of the two known addresses only the direct one is dialable. + Some(vec![ + relayed("/ip4/1.2.3.4/tcp/3610"), + addr("/ip4/5.6.7.8/tcp/3610"), + ]), + ); + + behaviour.force_direct_connections(); + + assert_eq!(dialled(behaviour), vec![peer]); + assert!(behaviour.pending_forcings.contains(&peer)); + + // The relayed address is filtered out of the dial: forcing a direct + // connection through the relay would be a no-op. + let dial = dial_debug(behaviour, &peer); + assert!(dial.contains("/ip4/5.6.7.8/tcp/3610"), "{dial}"); + assert!(!dial.contains("p2p-circuit"), "{dial}"); + } + + #[tokio::test] + async fn skips_the_local_peer() { + let local = PeerId::random(); + let behaviour = &mut behaviour(local, []); + seed_store( + behaviour, + local, + vec![conn(local, 1, relayed("/ip4/1.2.3.4/tcp/3610"))], + Some(vec![addr("/ip4/5.6.7.8/tcp/3610")]), + ); + + behaviour.force_direct_connections(); + + assert!(dialled(behaviour).is_empty()); + assert!(behaviour.pending_forcings.is_empty()); + } + + #[tokio::test] + async fn skips_a_peer_already_being_forced() { + let local = PeerId::random(); + let peer = PeerId::random(); + let behaviour = &mut behaviour(local, [peer]); + seed_store( + behaviour, + peer, + vec![conn(peer, 1, relayed("/ip4/1.2.3.4/tcp/3610"))], + Some(vec![addr("/ip4/5.6.7.8/tcp/3610")]), + ); + + behaviour.pending_forcings.insert(peer); + + behaviour.force_direct_connections(); + + assert!(dialled(behaviour).is_empty()); + } + + #[tokio::test] + async fn skips_a_peer_without_connections() { + let local = PeerId::random(); + let peer = PeerId::random(); + let behaviour = &mut behaviour(local, [peer]); + // Addresses are known, but there is no relayed connection to replace. + seed_store( + behaviour, + peer, + vec![], + Some(vec![addr("/ip4/5.6.7.8/tcp/3610")]), + ); + + behaviour.force_direct_connections(); + + assert!(dialled(behaviour).is_empty()); + assert!(behaviour.pending_forcings.is_empty()); + } + + #[tokio::test] + async fn skips_a_peer_that_already_has_one_direct_connection() { + let local = PeerId::random(); + let peer = PeerId::random(); + let behaviour = &mut behaviour(local, [peer]); + // The all-relay guard: one direct connection is enough to leave it alone. + seed_store( + behaviour, + peer, + vec![ + conn(peer, 1, relayed("/ip4/1.2.3.4/tcp/3610")), + conn(peer, 2, addr("/ip4/5.6.7.8/tcp/3610")), + ], + Some(vec![addr("/ip4/5.6.7.8/tcp/3610")]), + ); + + behaviour.force_direct_connections(); + + assert!(dialled(behaviour).is_empty()); + assert!(behaviour.pending_forcings.is_empty()); + } + + #[tokio::test] + async fn skips_a_peer_without_known_addresses() { + let local = PeerId::random(); + let peer = PeerId::random(); + let behaviour = &mut behaviour(local, [peer]); + // Identify has not reported an address yet, so there is nothing to dial. + seed_store( + behaviour, + peer, + vec![conn(peer, 1, relayed("/ip4/1.2.3.4/tcp/3610"))], + None, + ); + + behaviour.force_direct_connections(); + + assert!(dialled(behaviour).is_empty()); + assert!(behaviour.pending_forcings.is_empty()); + } + + #[tokio::test] + async fn skips_a_peer_whose_known_addresses_are_all_relayed() { + let local = PeerId::random(); + let peer = PeerId::random(); + let behaviour = &mut behaviour(local, [peer]); + seed_store( + behaviour, + peer, + vec![conn(peer, 1, relayed("/ip4/1.2.3.4/tcp/3610"))], + Some(vec![ + relayed("/ip4/1.2.3.4/tcp/3610"), + relayed("/ip4/1.2.3.4/udp/3610/quic-v1"), + ]), + ); + + behaviour.force_direct_connections(); + + assert!(dialled(behaviour).is_empty()); + assert!(behaviour.pending_forcings.is_empty()); + } +} diff --git a/crates/p2p/src/peer.rs b/crates/p2p/src/peer.rs index 097d651f..6e3c7bff 100644 --- a/crates/p2p/src/peer.rs +++ b/crates/p2p/src/peer.rs @@ -241,8 +241,20 @@ pub fn addr_infos_from_p2p_addrs(addrs: &[Multiaddr]) -> Result> { #[cfg(test)] mod tests { - use super::*; + use std::time::Duration; + + use futures::StreamExt; + use libp2p::{relay, swarm::SwarmEvent}; + use pluto_cluster::test_cluster; use pluto_testutil::random::generate_insecure_k1_key; + use tokio::time::timeout; + + use super::*; + use crate::{ + config::P2PConfig, + p2p::{Node, NodeType}, + p2p_context::P2PContext, + }; #[test] fn new_peer() { @@ -258,22 +270,133 @@ mod tests { ); } - #[test] - #[ignore] - fn new_tcp_host() { - todo!("add this test after implementing p2p.NewNode function"); + /// The cluster's operator ENRs as peers of *this* crate: `Lock::peers` + /// returns the `Peer` of the `pluto-p2p` copy the `pluto-cluster` + /// dev-dependency links, which is a distinct type from the one under test. + fn cluster_peers(lock: &pluto_cluster::lock::Lock) -> Vec { + lock.operators + .iter() + .enumerate() + .map(|(index, operator)| { + let record = Record::try_from(operator.enr.as_str()).unwrap(); + Peer::from_enr(&record, u64::try_from(index).unwrap()).unwrap() + }) + .collect() + } + + fn node(node_type: NodeType, cfg: P2PConfig) -> Node { + Node::new( + cfg, + generate_insecure_k1_key(7), + node_type, + false, + P2PContext::default(), + |builder, _keypair, relay_client| builder.with_inner(relay_client), + ) + .expect("node should build") + } + + /// The address libp2p reports once a client node's single configured + /// listener is bound. + async fn bound_listen_addr(node_type: NodeType, cfg: P2PConfig) -> Multiaddr { + let mut node = node(node_type, cfg); + + assert_eq!(node.node_type(), node_type); + assert_eq!( + node.listener_ids().len(), + 1, + "the one configured address must be registered as a listener" + ); + + timeout(Duration::from_secs(10), async { + loop { + if let SwarmEvent::NewListenAddr { address, .. } = node.select_next_some().await { + return address; + } + } + }) + .await + .expect("listener should report the address it bound") + } + + #[tokio::test] + async fn new_tcp_host() { + let cfg = P2PConfig { + tcp_addrs: vec!["127.0.0.1:0".to_owned()], + ..Default::default() + }; + + let addr = bound_listen_addr(NodeType::TCP, cfg).await; + + assert!(crate::utils::is_tcp_addr(&addr), "bound {addr} is not TCP"); + assert!(!crate::utils::is_quic_addr(&addr)); + // Port 0 was configured, so the kernel picked the listening port. + assert!(crate::utils::tcp_port(&addr).is_some_and(|port| port != 0)); + } + + #[tokio::test] + async fn new_quic_host() { + // A QUIC node listens on its UDP addresses; an empty `tcp_addrs` keeps + // this to the single QUIC listener under test. + let cfg = P2PConfig { + udp_addrs: vec!["127.0.0.1:0".to_owned()], + ..Default::default() + }; + + let addr = bound_listen_addr(NodeType::QUIC, cfg).await; + + assert!( + crate::utils::is_quic_addr(&addr), + "bound {addr} is not QUIC" + ); + assert!(!crate::utils::is_tcp_addr(&addr)); + assert!(crate::utils::udp_port(&addr).is_some_and(|port| port != 0)); + } + + #[tokio::test] + async fn new_host_without_listen_addrs() { + // Charon's TestNewTCPHost/TestNewQUICHost build from an empty + // `p2p.Config`: an outgoing-only node must construct with nothing to + // listen on. + for node_type in [NodeType::TCP, NodeType::QUIC] { + let node = node(node_type, P2PConfig::default()); + + assert_eq!(node.node_type(), node_type); + assert!(node.listener_ids().is_empty()); + } } #[test] - #[ignore] - fn verify_p2p_key() { - todo!("add this test after implementing cluster.NewForT function"); + fn verify_p2p_key_accepts_cluster_keys() { + let (lock, keys, _) = test_cluster::new_for_test(1, 3, 4, 1); + let peers = cluster_peers(&lock); + assert_eq!(peers.len(), keys.len()); + + // Every operator's p2p key matches the public key in its own ENR. + for key in &keys { + verify_p2p_key(&peers, key).expect("cluster p2p key should verify"); + } + + // A key that belongs to no operator matches no ENR. + let outsider = generate_insecure_k1_key(99); + assert!(matches!( + verify_p2p_key(&peers, &outsider), + Err(PeerError::UnknownPublicKey) + )); } #[test] - #[ignore] fn peer_id_key() { - todo!("add this test after implementing peer_id_key function"); + let (lock, keys, _) = test_cluster::new_for_test(1, 3, 4, 1); + let peers = cluster_peers(&lock); + assert_eq!(peers.len(), keys.len()); + + for (peer, key) in peers.iter().zip(keys.iter()) { + let public_key = peer_id_to_public_key(&peer.id).unwrap(); + assert_eq!(public_key, key.public_key()); + + assert_eq!(peer_id_from_key(public_key).unwrap(), peer.id); + } } #[test] diff --git a/crates/p2p/src/proto.rs b/crates/p2p/src/proto.rs index 04f8581a..77dc6583 100644 --- a/crates/p2p/src/proto.rs +++ b/crates/p2p/src/proto.rs @@ -156,6 +156,80 @@ mod tests { use super::*; + #[tokio::test] + async fn length_delimited_round_trip() { + // 127/128 straddle the one-to-two byte varint prefix boundary. + let payloads: [Vec; 5] = [ + vec![], + vec![1_u8], + vec![7_u8; 127], + vec![9_u8; 128], + vec![3_u8; 300], + ]; + + for payload in payloads { + let mut cursor = Cursor::new(Vec::new()); + + write_length_delimited(&mut cursor, &payload) + .await + .expect("write should succeed"); + cursor.set_position(0); + + let decoded = read_length_delimited(&mut cursor, MAX_MESSAGE_SIZE) + .await + .expect("read should succeed"); + + assert_eq!(decoded, payload); + } + } + + #[tokio::test] + async fn length_delimited_frames_are_read_back_in_order() { + let mut cursor = Cursor::new(Vec::new()); + + // Two frames in one stream: the length prefix separates them. + write_length_delimited(&mut cursor, b"first") + .await + .expect("write should succeed"); + write_length_delimited(&mut cursor, b"second") + .await + .expect("write should succeed"); + cursor.set_position(0); + + for want in [b"first".as_slice(), b"second".as_slice()] { + let decoded = read_length_delimited(&mut cursor, MAX_MESSAGE_SIZE) + .await + .expect("read should succeed"); + assert_eq!(decoded, want); + } + } + + #[tokio::test] + async fn oversized_length_delimited_message_fails() { + let mut cursor = Cursor::new(Vec::new()); + write_length_delimited(&mut cursor, &[1, 2, 3, 4]) + .await + .expect("write should succeed"); + cursor.set_position(0); + + // The framing is valid; only the caller's limit rejects it. + let error = read_length_delimited(&mut cursor, 3) + .await + .expect_err("payloads above the limit must fail"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + } + + #[tokio::test] + async fn truncated_length_delimited_payload_fails() { + // The varint announces 4 bytes but only 2 follow. + let mut cursor = Cursor::new(vec![4, 1, 2]); + + let error = read_length_delimited(&mut cursor, MAX_MESSAGE_SIZE) + .await + .expect_err("a truncated payload must fail"); + assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof); + } + #[tokio::test] async fn fixed_size_round_trip() { let payload = vec![1, 2, 3, 4]; diff --git a/crates/p2p/src/quic_upgrade.rs b/crates/p2p/src/quic_upgrade.rs index 0bcae00a..1a8fd496 100644 --- a/crates/p2p/src/quic_upgrade.rs +++ b/crates/p2p/src/quic_upgrade.rs @@ -429,3 +429,97 @@ impl NetworkBehaviour for QuicUpgradeBehaviour { Poll::Pending } } + +#[cfg(test)] +mod tests { + use super::*; + + fn behaviour() -> QuicUpgradeBehaviour { + QuicUpgradeBehaviour::new(P2PContext::default(), PeerId::random(), true) + } + + /// The reason carried by the queued `UpgradeFailed` event for `peer`. + fn failure_reason(behaviour: &QuicUpgradeBehaviour, peer: &PeerId) -> Option { + behaviour + .pending_events + .iter() + .find_map(|event| match event { + ToSwarm::GenerateEvent(QuicUpgradeEvent::UpgradeFailed { + peer: failed, + reason, + }) if failed == peer => Some(reason.clone()), + _ => None, + }) + } + + #[test] + fn backoff_doubles_then_pins_at_the_cap() { + let mut backoff = QuicUpgradeBackoff::new(); + + assert_eq!(QuicUpgradeBackoff::INITIAL, 1); + assert_eq!(QuicUpgradeBackoff::MAX, 512); + assert_eq!(backoff.backoff_duration, 1); + assert_eq!(backoff.tickers_remaining, 1); + + // Each failure doubles up to the cap and re-arms the countdown to the + // full new duration. + for want in [2, 4, 8, 16, 32, 64, 128, 256, 512, 512, 512] { + backoff.record_failure(); + + assert_eq!(backoff.backoff_duration, want); + assert_eq!(backoff.tickers_remaining, want); + } + } + + #[tokio::test] + async fn should_skip_counts_the_backoff_down_then_retries() { + let mut behaviour = behaviour(); + let peer = PeerId::random(); + + // One failure arms a two-ticker backoff, ... + behaviour.record_failure(peer, "dial failed"); + assert_eq!( + failure_reason(&behaviour, &peer).as_deref(), + Some("dial failed") + ); + + // ... so the next two ticks are skipped, ... + assert!(behaviour.should_skip(&peer)); + assert!(behaviour.should_skip(&peer)); + + // ... and every tick after that retries. + assert!(!behaviour.should_skip(&peer)); + assert!(!behaviour.should_skip(&peer)); + } + + #[tokio::test] + async fn backoff_is_tracked_per_peer() { + let mut behaviour = behaviour(); + let backing_off = PeerId::random(); + let other = PeerId::random(); + + behaviour.record_failure(backing_off, "dial failed"); + + assert!(behaviour.should_skip(&backing_off)); + assert!( + !behaviour.should_skip(&other), + "one peer's backoff must not delay another" + ); + } + + #[tokio::test] + async fn clear_backoff_restores_immediate_retries() { + let mut behaviour = behaviour(); + let peer = PeerId::random(); + + behaviour.record_failure(peer, "dial failed"); + behaviour.record_failure(peer, "dial failed"); + assert!(behaviour.backoffs.contains_key(&peer)); + + // A successful upgrade drops the state entirely. + behaviour.clear_backoff(&peer); + + assert!(!behaviour.backoffs.contains_key(&peer)); + assert!(!behaviour.should_skip(&peer)); + } +} diff --git a/crates/p2p/src/utils.rs b/crates/p2p/src/utils.rs index 4e472222..2783a98b 100644 --- a/crates/p2p/src/utils.rs +++ b/crates/p2p/src/utils.rs @@ -241,6 +241,182 @@ mod tests { addrs.iter().map(ToString::to_string).collect() } + const RELAY_ID: &str = "16Uiu2HAkzdQ5Y9SYT91K1ue5SxXwgmajXntfScGnLYeip5hHyWmT"; + + fn addr(s: &str) -> Multiaddr { + s.parse().unwrap() + } + + fn relayed(transport: &str) -> Multiaddr { + addr(&format!("{transport}/p2p/{RELAY_ID}/p2p-circuit")) + } + + fn conn(remote_addr: Multiaddr) -> crate::p2p_context::Peer { + crate::p2p_context::Peer { + id: libp2p::PeerId::random(), + connection_id: libp2p::swarm::ConnectionId::new_unchecked(1), + remote_addr, + } + } + + #[test] + fn is_relay_addr_needs_a_circuit_component() { + assert!(is_relay_addr(&relayed("/ip4/1.2.3.4/tcp/3610"))); + assert!(is_relay_addr(&relayed("/ip4/1.2.3.4/udp/3610/quic-v1"))); + + // A plain address to the relay itself is not a relayed address. + assert!(!is_relay_addr(&addr(&format!( + "/ip4/1.2.3.4/tcp/3610/p2p/{RELAY_ID}" + )))); + assert!(!is_relay_addr(&addr("/ip4/1.2.3.4/tcp/3610"))); + + assert!(is_direct_addr(&addr("/ip4/1.2.3.4/tcp/3610"))); + assert!(!is_direct_addr(&relayed("/ip4/1.2.3.4/tcp/3610"))); + } + + #[test] + fn is_quic_addr_accepts_both_quic_versions() { + assert!(is_quic_addr(&addr("/ip4/1.2.3.4/udp/3610/quic-v1"))); + assert!(is_quic_addr(&addr("/ip4/1.2.3.4/udp/3610/quic"))); + // Relaying is the separate axis: relayed QUIC is still QUIC. + assert!(is_quic_addr(&relayed("/ip4/1.2.3.4/udp/3610/quic-v1"))); + + assert!(!is_quic_addr(&addr("/ip4/1.2.3.4/tcp/3610"))); + // UDP alone is not QUIC. + assert!(!is_quic_addr(&addr("/ip4/1.2.3.4/udp/3610"))); + } + + #[test] + fn is_tcp_addr_needs_a_tcp_component() { + assert!(is_tcp_addr(&addr("/ip4/1.2.3.4/tcp/3610"))); + assert!(is_tcp_addr(&addr("/dns/relay.example.com/tcp/3610"))); + assert!(is_tcp_addr(&relayed("/ip4/1.2.3.4/tcp/3610"))); + + assert!(!is_tcp_addr(&addr("/ip4/1.2.3.4/udp/3610/quic-v1"))); + assert!(!is_tcp_addr(&addr("/ip4/1.2.3.4"))); + } + + #[test] + fn addr_type_and_protocol_classify_the_two_axes() { + assert_eq!( + addr_type(&addr("/ip4/1.2.3.4/tcp/3610")), + ConnectionType::Direct + ); + assert_eq!( + addr_type(&relayed("/ip4/1.2.3.4/tcp/3610")), + ConnectionType::Relay + ); + + assert_eq!( + addr_protocol(&addr("/ip4/1.2.3.4/udp/3610/quic-v1")), + Protocol::Quic + ); + assert_eq!(addr_protocol(&addr("/ip4/1.2.3.4/tcp/3610")), Protocol::Tcp); + assert_eq!(addr_protocol(&addr("/ip4/1.2.3.4")), Protocol::Unknown); + } + + #[test] + fn filter_direct_quic_addrs_keeps_only_unrelayed_quic() { + let quic = addr("/ip4/1.2.3.4/udp/3610/quic-v1"); + let candidates = vec![ + quic.clone(), + addr("/ip4/1.2.3.4/tcp/3610"), + relayed("/ip4/1.2.3.4/udp/3610/quic-v1"), + relayed("/ip4/1.2.3.4/tcp/3610"), + ]; + + assert_eq!(filter_direct_quic_addrs(candidates.into_iter()), vec![quic]); + assert!(filter_direct_quic_addrs(std::iter::empty()).is_empty()); + } + + #[test] + fn quic_is_enabled_only_while_listening_on_quic() { + let tcp = addr("/ip4/1.2.3.4/tcp/3610"); + let quic = addr("/ip4/1.2.3.4/udp/3610/quic-v1"); + + assert!(is_quic_enabled([&tcp, &quic].into_iter())); + assert!(!is_quic_enabled([&tcp].into_iter())); + assert!(!is_quic_enabled(std::iter::empty())); + } + + #[test] + fn direct_conn_checks_ignore_relayed_connections() { + let quic = conn(addr("/ip4/1.2.3.4/udp/3610/quic-v1")); + let tcp = conn(addr("/ip4/1.2.3.4/tcp/3610")); + let relayed_quic = conn(relayed("/ip4/1.2.3.4/udp/3610/quic-v1")); + let relayed_tcp = conn(relayed("/ip4/1.2.3.4/tcp/3610")); + + assert!(has_direct_quic_conn(&[&quic])); + assert!(!has_direct_quic_conn(&[&tcp, &relayed_quic])); + assert!(!has_direct_quic_conn(&[])); + + assert!(has_direct_tcp_conn(&[&tcp])); + assert!(!has_direct_tcp_conn(&[&quic, &relayed_tcp])); + assert!(!has_direct_tcp_conn(&[])); + } + + /// Charon's `TestFilterAdvertisedAddrs` table (`p2p/p2p_internal_test.go`). + /// + /// DEVIATION: charon preserves source order and deduplicates globally + /// across both groups. This implementation sorts each group and + /// deduplicates only within it, so the two duplicate cases advertise an + /// address twice and in a different order. `want` is what this + /// implementation returns today; the inline `charon:` notes record what + /// upstream returns for the cases that differ. + #[test] + fn advertised_addresses_match_the_charon_table() { + const PRIV1: &str = "/ip4/192.168.1.1/tcp/80"; + const PRIV2: &str = "/ip4/127.0.0.1/udp/123"; + const PUB1: &str = "/ip4/1.1.1.1/tcp/80"; + + let cases = [ + ("empty", vec![], vec![], false, vec![]), + ( + "drop one private", + vec![], + vec![PUB1, PRIV1], + true, + vec![PUB1], + ), + ( + "keep one private", + vec![], + vec![PUB1, PRIV1], + false, + vec![PUB1, PRIV1], + ), + ( + // charon: [PRIV1, PUB1]. An external address is never dropped + // for being private, but here it also survives deduplication. + "duplicate public", + vec![PRIV1, PUB1], + vec![PUB1, PRIV1], + true, + vec![PUB1, PRIV1, PUB1], + ), + ( + // charon: [PRIV2, PRIV1] + "duplicate private", + vec![PRIV2, PRIV1], + vec![PRIV1, PRIV2], + false, + vec![PRIV2, PRIV1, PRIV2, PRIV1], + ), + ("drop all private", vec![], vec![PRIV1, PRIV2], true, vec![]), + ]; + + for (name, external, internal, exclude_private, want) in cases { + let got = filter_advertised_addresses( + ExternalAddresses(external.iter().map(|a| addr(a)).collect()), + InternalAddresses(internal.iter().map(|a| addr(a)).collect()), + exclude_private, + ) + .unwrap(); + + assert_eq!(as_strings(&got), want, "case {name:?}"); + } + } + #[test] fn external_multiaddrs_keep_the_bound_ports() { let cfg = config(Some("1.2.3.4"), Some("relay.example.com")); From 82e2037c2ad70ef030c9de8f9387d2c123de76b1 Mon Sep 17 00:00:00 2001 From: Quang Le Date: Wed, 26 Aug 2026 16:38:50 +0700 Subject: [PATCH 2/2] fix: address comments --- Cargo.lock | 1 + crates/p2p/Cargo.toml | 1 + crates/p2p/src/force_direct.rs | 48 +++++-------- crates/p2p/src/p2p.rs | 2 +- crates/p2p/src/peer.rs | 4 +- crates/p2p/src/utils.rs | 121 ++++++++++++++------------------- 6 files changed, 73 insertions(+), 104 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a1efe2a4..5e93e4de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5547,6 +5547,7 @@ dependencies = [ "reqwest 0.13.4", "serde_json", "tempfile", + "test-case", "thiserror 2.0.20", "tokio", "tokio-stream", diff --git a/crates/p2p/Cargo.toml b/crates/p2p/Cargo.toml index 545394f1..633de655 100644 --- a/crates/p2p/Cargo.toml +++ b/crates/p2p/Cargo.toml @@ -44,6 +44,7 @@ k256.workspace = true tokio = { workspace = true, features = ["test-util"] } futures.workspace = true wiremock.workspace = true +test-case.workspace = true [lints] workspace = true diff --git a/crates/p2p/src/force_direct.rs b/crates/p2p/src/force_direct.rs index b308e297..5df6c555 100644 --- a/crates/p2p/src/force_direct.rs +++ b/crates/p2p/src/force_direct.rs @@ -1,7 +1,7 @@ //! Force direct connection behaviour. use std::{ - collections::{HashSet, VecDeque}, + collections::{HashMap, VecDeque}, convert::Infallible, task::{Context, Poll}, }; @@ -34,8 +34,9 @@ pub struct ForceDirectBehaviour { /// Pending events to emit. pending_events: VecDeque>, - /// Pending forcings to emit. - pending_forcings: HashSet, + /// Peers with a force-direct dial in flight, and the direct addresses it + /// was given. + pending_forcings: HashMap>, /// Interval timer for running force direct logic periodically. ticker: Interval, @@ -64,6 +65,8 @@ pub enum ForceDirectEvent { ForceDirectFailure { /// The peer to force direct connection to. peer: PeerId, + /// The direct addresses the failed dial was given. + addresses: Vec, /// The reason for the failure. reason: String, }, @@ -80,7 +83,7 @@ impl ForceDirectBehaviour { local_peer_id, pending_events: VecDeque::new(), ticker, - pending_forcings: HashSet::new(), + pending_forcings: HashMap::new(), } } @@ -100,7 +103,7 @@ impl ForceDirectBehaviour { continue; } - if self.pending_forcings.contains(peer) { + if self.pending_forcings.contains_key(peer) { continue; } @@ -170,7 +173,8 @@ impl ForceDirectBehaviour { direct_addresses.len() ); - self.pending_forcings.insert(*peer); + self.pending_forcings + .insert(*peer, direct_addresses.clone()); self.pending_events.push_back(ToSwarm::Dial { opts: DialOpts::peer_id(*peer) @@ -187,7 +191,7 @@ impl ForceDirectBehaviour { libp2p::core::ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr, }; - if self.pending_forcings.contains(&event.peer_id) && utils::is_direct_addr(addr) { + if self.pending_forcings.contains_key(&event.peer_id) && utils::is_direct_addr(addr) { self.pending_forcings.remove(&event.peer_id); self.pending_events.push_back(ToSwarm::GenerateEvent( ForceDirectEvent::ForceDirectSuccess { @@ -202,10 +206,11 @@ impl ForceDirectBehaviour { return; }; - if self.pending_forcings.remove(&peer_id) { + if let Some(addresses) = self.pending_forcings.remove(&peer_id) { self.pending_events.push_back(ToSwarm::GenerateEvent( ForceDirectEvent::ForceDirectFailure { peer: peer_id, + addresses, reason: "dial failed".to_string(), }, )); @@ -329,22 +334,6 @@ mod tests { } } - /// The `Debug` rendering of the queued dial to `peer`. `DialOpts` keeps its - /// address list `pub(crate)`, so this is the only way to see which - /// addresses were selected. - fn dial_debug(behaviour: &ForceDirectBehaviour, peer: &PeerId) -> String { - behaviour - .pending_events - .iter() - .find_map(|event| match event { - ToSwarm::Dial { opts } if opts.get_peer_id().as_ref() == Some(peer) => { - Some(format!("{opts:?}")) - } - _ => None, - }) - .expect("a dial to the peer should be queued") - } - /// The peers the queued events dial. fn dialled(behaviour: &ForceDirectBehaviour) -> Vec { behaviour @@ -376,13 +365,12 @@ mod tests { behaviour.force_direct_connections(); assert_eq!(dialled(behaviour), vec![peer]); - assert!(behaviour.pending_forcings.contains(&peer)); - // The relayed address is filtered out of the dial: forcing a direct // connection through the relay would be a no-op. - let dial = dial_debug(behaviour, &peer); - assert!(dial.contains("/ip4/5.6.7.8/tcp/3610"), "{dial}"); - assert!(!dial.contains("p2p-circuit"), "{dial}"); + assert_eq!( + behaviour.pending_forcings.get(&peer), + Some(&vec![addr("/ip4/5.6.7.8/tcp/3610")]) + ); } #[tokio::test] @@ -414,7 +402,7 @@ mod tests { Some(vec![addr("/ip4/5.6.7.8/tcp/3610")]), ); - behaviour.pending_forcings.insert(peer); + behaviour.pending_forcings.insert(peer, vec![]); behaviour.force_direct_connections(); diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index ceb4b580..ec1a9af0 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -394,7 +394,7 @@ impl Node { utils::ExternalAddresses(external_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); diff --git a/crates/p2p/src/peer.rs b/crates/p2p/src/peer.rs index 6e3c7bff..c5c2c20d 100644 --- a/crates/p2p/src/peer.rs +++ b/crates/p2p/src/peer.rs @@ -247,7 +247,7 @@ mod tests { use libp2p::{relay, swarm::SwarmEvent}; use pluto_cluster::test_cluster; use pluto_testutil::random::generate_insecure_k1_key; - use tokio::time::timeout; + use tokio::time; use super::*; use crate::{ @@ -308,7 +308,7 @@ mod tests { "the one configured address must be registered as a listener" ); - timeout(Duration::from_secs(10), async { + time::timeout(Duration::from_secs(10), async { loop { if let SwarmEvent::NewListenAddr { address, .. } = node.select_next_some().await { return address; diff --git a/crates/p2p/src/utils.rs b/crates/p2p/src/utils.rs index 2783a98b..143f19ba 100644 --- a/crates/p2p/src/utils.rs +++ b/crates/p2p/src/utils.rs @@ -9,6 +9,7 @@ //! These utilities are primarily used internally by the [`crate::p2p`] module. use std::{ + collections::HashSet, net::{IpAddr, SocketAddr}, time::Duration, }; @@ -118,29 +119,41 @@ pub(crate) struct ExternalAddresses(pub Vec); pub(crate) struct InternalAddresses(pub Vec); -/// Filters the advertised addresses to exclude private addresses if the -/// `exclude_internal_private` flag is set. +/// Returns the unique external and internal addresses to advertise, in source +/// order, optionally excluding private internal addresses. +/// +/// External addresses are never dropped for being private: they were configured +/// explicitly. Deduplication spans both groups, so an address listed as both +/// external and internal is advertised once. +/// /// Since the type of external and internal addresses is the same, we use type /// wrappers to avoid confusion. pub(crate) fn filter_advertised_addresses( external_addrs: ExternalAddresses, internal_addrs: InternalAddresses, exclude_internal_private: bool, -) -> crate::p2p::Result> { - let mut external_addrs = external_addrs.0; - let mut internal_addrs = internal_addrs.0; +) -> Vec { + let mut seen = HashSet::new(); + let mut advertised = Vec::new(); - external_addrs.sort(); - internal_addrs.sort(); + let mut add = |addrs: Vec, exclude_private: bool| { + for addr in addrs { + if !seen.insert(addr.clone()) { + continue; + } - external_addrs.dedup(); - internal_addrs.dedup(); + if exclude_private && addr.is_private() { + continue; + } - if exclude_internal_private { - internal_addrs.retain(|addr| !addr.is_private()); - } + advertised.push(addr); + } + }; + + add(external_addrs.0, false); + add(internal_addrs.0, exclude_internal_private); - Ok(external_addrs.into_iter().chain(internal_addrs).collect()) + advertised } /// Returns the default swarm configuration. @@ -226,6 +239,8 @@ pub fn is_direct_addr(addr: &Multiaddr) -> bool { #[cfg(test)] mod tests { + use test_case::test_case; + use super::*; /// Config with the external overrides under test. @@ -355,66 +370,30 @@ mod tests { assert!(!has_direct_tcp_conn(&[])); } - /// Charon's `TestFilterAdvertisedAddrs` table (`p2p/p2p_internal_test.go`). - /// - /// DEVIATION: charon preserves source order and deduplicates globally - /// across both groups. This implementation sorts each group and - /// deduplicates only within it, so the two duplicate cases advertise an - /// address twice and in a different order. `want` is what this - /// implementation returns today; the inline `charon:` notes record what - /// upstream returns for the cases that differ. - #[test] - fn advertised_addresses_match_the_charon_table() { - const PRIV1: &str = "/ip4/192.168.1.1/tcp/80"; - const PRIV2: &str = "/ip4/127.0.0.1/udp/123"; - const PUB1: &str = "/ip4/1.1.1.1/tcp/80"; - - let cases = [ - ("empty", vec![], vec![], false, vec![]), - ( - "drop one private", - vec![], - vec![PUB1, PRIV1], - true, - vec![PUB1], - ), - ( - "keep one private", - vec![], - vec![PUB1, PRIV1], - false, - vec![PUB1, PRIV1], - ), - ( - // charon: [PRIV1, PUB1]. An external address is never dropped - // for being private, but here it also survives deduplication. - "duplicate public", - vec![PRIV1, PUB1], - vec![PUB1, PRIV1], - true, - vec![PUB1, PRIV1, PUB1], - ), - ( - // charon: [PRIV2, PRIV1] - "duplicate private", - vec![PRIV2, PRIV1], - vec![PRIV1, PRIV2], - false, - vec![PRIV2, PRIV1, PRIV2, PRIV1], - ), - ("drop all private", vec![], vec![PRIV1, PRIV2], true, vec![]), - ]; + const PRIV1: &str = "/ip4/192.168.1.1/tcp/80"; + const PRIV2: &str = "/ip4/127.0.0.1/udp/123"; + const PUB1: &str = "/ip4/1.1.1.1/tcp/80"; - for (name, external, internal, exclude_private, want) in cases { - let got = filter_advertised_addresses( - ExternalAddresses(external.iter().map(|a| addr(a)).collect()), - InternalAddresses(internal.iter().map(|a| addr(a)).collect()), - exclude_private, - ) - .unwrap(); + /// Charon's `TestFilterAdvertisedAddrs` table (`p2p/p2p_internal_test.go`). + #[test_case(&[], &[], false, &[] ; "empty")] + #[test_case(&[], &[PUB1, PRIV1], true, &[PUB1] ; "drop one private")] + #[test_case(&[], &[PUB1, PRIV1], false, &[PUB1, PRIV1] ; "keep one private")] + #[test_case(&[PRIV1, PUB1], &[PUB1, PRIV1], true, &[PRIV1, PUB1] ; "duplicate public")] + #[test_case(&[PRIV2, PRIV1], &[PRIV1, PRIV2], false, &[PRIV2, PRIV1] ; "duplicate private")] + #[test_case(&[], &[PRIV1, PRIV2], true, &[] ; "drop all private")] + fn filters_advertised_addresses( + external: &[&str], + internal: &[&str], + exclude_private: bool, + want: &[&str], + ) { + let got = filter_advertised_addresses( + ExternalAddresses(external.iter().map(|a| addr(a)).collect()), + InternalAddresses(internal.iter().map(|a| addr(a)).collect()), + exclude_private, + ); - assert_eq!(as_strings(&got), want, "case {name:?}"); - } + assert_eq!(as_strings(&got), want); } #[test]