diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index 4cf88307..895812e2 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -89,6 +89,7 @@ use std::{ pin::Pin, task::{Context, Poll}, + time::Duration, }; use futures::{Stream, StreamExt, stream::FusedStream}; @@ -608,17 +609,7 @@ impl Node { SwarmEvent::Behaviour(PlutoBehaviourEvent::Ping(ping::Event { peer, result, .. })) => { - let peer_label = peer_name(peer); - match result { - Ok(duration) => { - P2P_METRICS.ping_latency_secs[&peer_label].observe(duration.as_secs_f64()); - P2P_METRICS.ping_success[&peer_label].set(1); - } - Err(_) => { - P2P_METRICS.ping_error_total[&peer_label].inc(); - P2P_METRICS.ping_success[&peer_label].set(0); - } - } + record_ping_metrics(&self.p2p_context, peer, result); } // AutoNAT reachability status @@ -697,6 +688,45 @@ impl FusedStream for Node { } } +/// Records the outcome of a ping, gated to known cluster peers. +/// +/// Charon pings an explicit allowlist of cluster operator peers +/// (`manifest.ClusterPeerIDs` wired into `p2p.NewPingService`, +/// `charon/app/app.go:390`, `charon/p2p/ping.go:35-65`), so relays never appear +/// in `p2p_ping_*`. Pluto instead uses libp2p's ping behaviour, which pings +/// every connected peer — relay servers included, since they are ordinary +/// transport peers — so the allowlist is applied here, at emission time. +/// Without it a relay shows up as a phantom peer on the dashboard's peer +/// panels (`max(p2p_ping_success) by (peer)`) and inflates the +/// `insufficient_connected_peers` health check, which counts non-zero +/// `p2p_ping_success` labels. +/// +/// A relay server node tracks no cluster peers at all +/// ([`P2PContext::default`], `pluto_relay_server::p2p`), so it deliberately +/// publishes no `p2p_ping_*` series: Charon's relay likewise never starts a +/// ping service, and this keeps the label set of a public relay bounded. +fn record_ping_metrics( + ctx: &P2PContext, + peer: &PeerId, + result: &std::result::Result, +) { + if !ctx.is_known_peer(peer) { + return; + } + + let peer_label = peer_name(peer); + match result { + Ok(duration) => { + P2P_METRICS.ping_latency_secs[&peer_label].observe(duration.as_secs_f64()); + P2P_METRICS.ping_success[&peer_label].set(1); + } + Err(_) => { + P2P_METRICS.ping_error_total[&peer_label].inc(); + P2P_METRICS.ping_success[&peer_label].set(0); + } + } +} + /// Stores identify-reported listen addresses for a peer, gated to known cluster /// peers only. Addresses from unknown peers are dropped (and not cloned), since /// the only consumers of `peer_addresses` look up known peers exclusively — so @@ -712,6 +742,8 @@ fn store_identify_addrs(ctx: &P2PContext, peer_id: &PeerId, addrs: &[Multiaddr]) #[cfg(test)] mod tests { + use vise::{Counter, Gauge}; + use super::*; fn random_peer_id() -> PeerId { @@ -752,6 +784,57 @@ mod tests { assert!(ctx.peer_store_lock().peer_addresses(&unknown).is_none()); } + #[test] + fn ping_metrics_recorded_for_known_peer() { + let known = random_peer_id(); + let ctx = P2PContext::new([known]); + let label = peer_name(&known); + + record_ping_metrics(&ctx, &known, &Ok(Duration::from_millis(20))); + + assert_eq!( + P2P_METRICS.ping_success.get(&label).map(Gauge::get), + Some(1) + ); + assert!( + P2P_METRICS.ping_latency_secs.contains(&label), + "latency must be observed for a known cluster peer" + ); + + record_ping_metrics(&ctx, &known, &Err(ping::Failure::Timeout)); + + assert_eq!( + P2P_METRICS.ping_success.get(&label).map(Gauge::get), + Some(0) + ); + assert_eq!( + P2P_METRICS.ping_error_total.get(&label).map(Counter::get), + Some(1) + ); + } + + #[test] + fn ping_metrics_skipped_for_relay_or_unknown_peer() { + let known = random_peer_id(); + // A relay server is an ordinary transport peer that libp2p pings, but + // it is not a cluster peer, so it must not reach `p2p_ping_*`. + let relay = random_peer_id(); + let ctx = P2PContext::new([known]); + let label = peer_name(&relay); + + record_ping_metrics(&ctx, &relay, &Ok(Duration::from_millis(20))); + record_ping_metrics(&ctx, &relay, &Err(ping::Failure::Timeout)); + + // Read with `get` (not indexing) so the assertions don't create the + // very series they are checking for. + assert!(!P2P_METRICS.ping_success.contains(&label)); + assert!(!P2P_METRICS.ping_latency_secs.contains(&label)); + assert!( + !P2P_METRICS.ping_error_total.contains(&label), + "the error path must be gated too, not just success/latency" + ); + } + #[test] fn identify_addrs_capped_for_known_peer() { let known = random_peer_id(); diff --git a/crates/p2p/src/relay/manager.rs b/crates/p2p/src/relay/manager.rs index b7a2ad04..5930576d 100644 --- a/crates/p2p/src/relay/manager.rs +++ b/crates/p2p/src/relay/manager.rs @@ -26,6 +26,8 @@ use super::{ event::{RelayDialError, RelayDialType, RelayManagerEvent}, }; use crate::{ + metrics::P2P_METRICS, + name::peer_name, p2p_context::P2PContext, peer::{MutablePeer, Peer}, }; @@ -195,7 +197,8 @@ impl RelayManager { self.set_relay_state(relay.id, RelayConnectionState::Dialing); } - /// Updates the connection state for a relay, logging the transition and + /// Updates the connection state for a relay, logging the transition, + /// reporting reservation availability on `p2p_relay_connections`, and /// maintaining the `established_at` watchdog timestamp. fn set_relay_state(&mut self, relay_id: PeerId, next: RelayConnectionState) { let prev = self.connection_states.insert(relay_id, next); @@ -207,6 +210,7 @@ impl RelayManager { "Relay connection state transition" ); } + Self::report_relay_connection(relay_id, matches!(next, RelayConnectionState::Reserved)); match next { // Entering or refreshing the no-reservation-yet state: start (or // restart, on demote from Reserved) the stuck-Established timer. @@ -223,6 +227,19 @@ impl RelayManager { } } + /// Reports a relay's reservation availability on `p2p_relay_connections`. + /// + /// Charon parity: the gauge tracks whether a relay *reservation* is + /// currently held, not how many transport connections exist — it is set to + /// `0` before each reserve attempt and to `1` once the circuit is reserved + /// (`charon/p2p/relay.go:40-44,60-101`). A transport count would not fit + /// the metric anyway: it carries only a `peer` label while a relay can hold + /// both a tcp and a quic connection, so per-transport writes would be + /// last-writer-wins. + fn report_relay_connection(relay_id: PeerId, reserved: bool) { + P2P_METRICS.relay_connections[&peer_name(&relay_id)].set(i64::from(reserved)); + } + /// Polls every active dial state once, queuing a `ToSwarm::Dial` event for /// any whose backoff has elapsed. Wakers for the remaining (pending) ones /// are registered via the underlying `Sleep` futures. @@ -714,6 +731,9 @@ impl RelayManager { "Relay closed but addresses no longer tracked; cannot redial" ); self.connection_states.remove(&relay_id); + // The only path that drops a relay's state without going through + // `set_relay_state`, so clear the reservation gauge here. + Self::report_relay_connection(relay_id, false); return; }; tracing::debug!( diff --git a/crates/p2p/src/relay/manager/tests.rs b/crates/p2p/src/relay/manager/tests.rs index 3d9cfb33..9c50e4fe 100644 --- a/crates/p2p/src/relay/manager/tests.rs +++ b/crates/p2p/src/relay/manager/tests.rs @@ -863,6 +863,84 @@ async fn poll_fires_swept_peer_dial_within_the_same_watchdog_pass() { ); } +// ---- relay_connections metric -------------------------------------- + +/// Current `p2p_relay_connections` value for a relay, or `None` if the relay +/// has no series yet. Reads with `get` (not indexing) so the helper doesn't +/// create the series it reports on. +fn relay_connections(relay_id: PeerId) -> Option { + P2P_METRICS + .relay_connections + .get(&peer_name(&relay_id)) + .map(vise::Gauge::get) +} + +#[tokio::test] +async fn relay_connections_tracks_reservation_lifecycle() { + let mut mgr = manager(); + let relay_id = PeerId::random(); + let circuit = addr(&format!( + "/ip4/10.0.0.1/tcp/9000/p2p/{relay_id}/p2p-circuit" + )); + + assert_eq!( + relay_connections(relay_id), + None, + "no series before the relay is known" + ); + + // Dialing: reported, but no reservation yet. + mgr.queue_relay_update(relay_peer(relay_id, vec![addr("/ip4/10.0.0.1/tcp/9000")])); + assert_eq!(relay_connections(relay_id), Some(0)); + + // Transport connected, reservation still unconfirmed. + mgr.on_connection_established(relay_id); + assert_eq!( + relay_connections(relay_id), + Some(0), + "a transport connection alone is not a reservation" + ); + + // Reservation confirmed. + mgr.on_new_listen_addr(&circuit); + assert_eq!(relay_connections(relay_id), Some(1)); + + // Reservation lost while the transport connection stays up: libp2p's relay + // client owns refreshes, so this happens without any ConnectionClosed. + mgr.on_expired_listen_addr(&circuit); + assert_eq!( + mgr.connection_states.get(&relay_id), + Some(&RelayConnectionState::Established), + "precondition: demoted without losing the transport connection" + ); + assert_eq!(relay_connections(relay_id), Some(0)); + + // Transport connection drops → redial campaign, still no reservation. + mgr.on_connection_closed(relay_id); + assert_eq!(relay_connections(relay_id), Some(0)); + + // Reconnect and re-reserve. + mgr.on_connection_established(relay_id); + mgr.on_new_listen_addr(&circuit); + assert_eq!(relay_connections(relay_id), Some(1)); +} + +#[tokio::test] +async fn relay_connections_cleared_when_relay_state_is_dropped() { + // `redial_relay` gives up (and drops the connection state) when the relay's + // addresses are no longer tracked — the one path that doesn't go through + // `set_relay_state`, so the gauge must be cleared explicitly. + let mut mgr = manager(); + let relay_id = PeerId::random(); + mgr.set_relay_state(relay_id, RelayConnectionState::Reserved); + assert_eq!(relay_connections(relay_id), Some(1)); + + mgr.on_connection_closed(relay_id); + + assert!(!mgr.connection_states.contains_key(&relay_id)); + assert_eq!(relay_connections(relay_id), Some(0)); +} + #[tokio::test] async fn sweep_is_noop_without_reserved_relays() { let target = PeerId::random(); diff --git a/crates/p2p/tests/common/mod.rs b/crates/p2p/tests/common/mod.rs new file mode 100644 index 00000000..0ee19ae5 --- /dev/null +++ b/crates/p2p/tests/common/mod.rs @@ -0,0 +1,80 @@ +//! Shared fixtures for the `pluto-p2p` integration tests. + +use std::time::Duration; + +use futures::StreamExt as _; +use k256::SecretKey; +use libp2p::{Multiaddr, PeerId, relay, swarm::SwarmEvent}; +use pluto_p2p::{ + config::P2PConfig, + p2p::{Node, NodeType}, + p2p_context::P2PContext, + peer::peer_id_from_key, +}; +use tokio::{task::JoinHandle, time::timeout}; + +/// How long any single step of an integration test may take before it is +/// treated as hung. +pub const TEST_TIMEOUT: Duration = Duration::from_secs(30); + +/// Starts an in-process relay server node ([`Node::new_server`] plus +/// [`relay::Behaviour`], the production wiring) on loopback TCP and drives it +/// in the background. +/// +/// Returns the relay's peer id, the concrete address it ended up listening on, +/// and the handle of the task polling its swarm — abort the handle to stop the +/// relay. +pub async fn spawn_relay_server(key: SecretKey) -> (PeerId, Multiaddr, JoinHandle<()>) { + let peer_id = peer_id_from_key(key.public_key()).expect("relay peer id"); + + let mut node = Node::new_server( + P2PConfig::default(), + key, + NodeType::TCP, + false, + // Relay servers don't track cluster peers - they serve all connections. + P2PContext::default(), + None, + |builder, keypair| { + // Start from the rust-libp2p defaults so the per-peer and per-IP + // reservation / circuit rate limiters survive: an exhaustive struct + // literal drops them, which is the bug fixed in + // `pluto_relay_server::config::create_relay_config`. + builder.with_inner(relay::Behaviour::new( + keypair.public().to_peer_id(), + relay::Config::default(), + )) + }, + ) + .expect("build relay server node"); + + node.listen_on( + "/ip4/127.0.0.1/tcp/0" + .parse::() + .expect("parse relay listen multiaddr"), + ) + .expect("relay listen_on"); + + let addr = timeout(TEST_TIMEOUT, async { + loop { + if let SwarmEvent::NewListenAddr { address, .. } = node.select_next_some().await { + return address; + } + } + }) + .await + .expect("timed out waiting for the relay listen address"); + + // The relay must advertise a reachable address, otherwise reservations are + // rejected client-side with `NoAddressesInReservation`. + node.add_external_address(addr.clone()); + + // Keep the relay driven so it can service reservations and circuits. + let handle = tokio::spawn(async move { + loop { + node.select_next_some().await; + } + }); + + (peer_id, addr, handle) +} diff --git a/crates/p2p/tests/relay_circuit.rs b/crates/p2p/tests/relay_circuit.rs index 396b396c..6eb08517 100644 --- a/crates/p2p/tests/relay_circuit.rs +++ b/crates/p2p/tests/relay_circuit.rs @@ -14,10 +14,10 @@ //! and relay client paths over real sockets — the relay reservation and circuit //! hop, not just a direct dial. -use std::time::Duration; +mod common; use futures::StreamExt as _; -use libp2p::{Multiaddr, PeerId, multiaddr::Protocol, relay, swarm::SwarmEvent}; +use libp2p::{PeerId, multiaddr::Protocol, relay, swarm::SwarmEvent}; use pluto_p2p::{ config::P2PConfig, p2p::{Node, NodeType}, @@ -28,71 +28,18 @@ use pluto_p2p::{ use pluto_testutil::random::generate_insecure_k1_key; use tokio::time::timeout; -const TEST_TIMEOUT: Duration = Duration::from_secs(30); +use common::{TEST_TIMEOUT, spawn_relay_server}; #[tokio::test] async fn two_nodes_connect_through_relay_circuit() { - let relay_key = generate_insecure_k1_key(1); let listener_key = generate_insecure_k1_key(2); let dialer_key = generate_insecure_k1_key(3); - let relay_peer = peer_id_from_key(relay_key.public_key()).expect("relay peer id"); let listener_peer = peer_id_from_key(listener_key.public_key()).expect("listener peer id"); let dialer_peer = peer_id_from_key(dialer_key.public_key()).expect("dialer peer id"); - // --- Relay server node. --- - let relay_config = relay::Config { - max_reservations: 16, - max_reservations_per_peer: 4, - reservation_duration: Duration::from_secs(3600), - reservation_rate_limiters: vec![], - max_circuits: 16, - max_circuits_per_peer: 4, - max_circuit_duration: Duration::from_secs(120), - max_circuit_bytes: 32 * 1024 * 1024, - circuit_src_rate_limiters: vec![], - }; - let mut relay_node = Node::new_server( - P2PConfig::default(), - relay_key, - NodeType::TCP, - false, - P2PContext::default(), - None, - move |builder, keypair| { - let behaviour = relay::Behaviour::new(keypair.public().to_peer_id(), relay_config); - builder.with_inner(behaviour) - }, - ) - .expect("build relay server node"); - - let relay_listen = "/ip4/127.0.0.1/tcp/0" - .parse::() - .expect("parse relay listen multiaddr"); - relay_node.listen_on(relay_listen).expect("relay listen_on"); - - // Wait for the relay's concrete TCP address, then keep the relay driven in - // the background so it can service reservations and circuits. - let relay_addr = timeout(TEST_TIMEOUT, async { - loop { - let event = relay_node.select_next_some().await; - if let SwarmEvent::NewListenAddr { address, .. } = event { - return address; - } - } - }) - .await - .expect("timed out waiting for the relay listen address"); - - // The relay must advertise a reachable address, otherwise reservations are - // rejected client-side with `NoAddressesInReservation`. - relay_node.add_external_address(relay_addr.clone()); - - let relay_handle = tokio::spawn(async move { - loop { - relay_node.select_next_some().await; - } - }); + let (relay_peer, relay_addr, relay_handle) = + spawn_relay_server(generate_insecure_k1_key(1)).await; // Full relay address including its peer id, plus the circuit suffix. let relay_with_id = relay_addr.with(Protocol::P2p(relay_peer)); diff --git a/crates/p2p/tests/relay_metrics.rs b/crates/p2p/tests/relay_metrics.rs new file mode 100644 index 00000000..7a9ef412 --- /dev/null +++ b/crates/p2p/tests/relay_metrics.rs @@ -0,0 +1,128 @@ +//! End-to-end check of relay connectivity metrics on the production path: a +//! real [`Node`] whose [`RelayManager`] reserves a circuit on an in-process +//! relay server over loopback TCP. +//! +//! The `relay::manager` unit tests drive the behaviour's `FromSwarm` handlers +//! directly; this test exercises the whole swarm plumbing the live cluster +//! runs, which is where both metric bugs showed up on the Charon dashboard: +//! +//! 1. `p2p_relay_connections` had no writer at all, so "Connected Relays" +//! (`sum(p2p_relay_connections) by (peer) > 0`) was blank; +//! 2. the relay server — an ordinary transport peer, so libp2p's ping behaviour +//! pings it — leaked into the per-peer `p2p_ping_*` series and appeared as a +//! phantom cluster peer on `max(p2p_ping_success) by (peer)`. + +mod common; + +use futures::StreamExt as _; +use libp2p::{ + PeerId, ping, relay, + swarm::{NetworkBehaviour, SwarmEvent}, +}; +use pluto_p2p::{ + behaviours::pluto::PlutoBehaviourEvent, + config::P2PConfig, + metrics::P2P_METRICS, + name::peer_name, + p2p::{Node, NodeType}, + p2p_context::P2PContext, + peer::{AddrInfo, MutablePeer, Peer}, + relay::{RelayManager, RelayManagerEvent}, +}; +use pluto_testutil::random::generate_insecure_k1_key; +use tokio::time::timeout; +use vise::Gauge; + +use common::{TEST_TIMEOUT, spawn_relay_server}; + +/// Client behaviour mirroring the app wiring: the relay client transport plus +/// the [`RelayManager`] that keeps the reservation alive. Ping is not listed +/// here because it lives in the outer `PlutoBehaviour`, as it does in the app. +#[derive(NetworkBehaviour)] +struct ClientBehaviour { + relay: relay::client::Behaviour, + relay_manager: RelayManager, +} + +#[tokio::test] +async fn relay_reservation_sets_relay_connections_and_emits_no_ping_metrics() { + let (relay_peer, relay_addr, relay_handle) = + spawn_relay_server(generate_insecure_k1_key(11)).await; + let relay_label = peer_name(&relay_peer); + + let relay_mutable = MutablePeer::new(Peer::new_relay_peer(&AddrInfo { + id: relay_peer, + addrs: vec![relay_addr], + })); + + // The relay is deliberately absent from the known-peer set: the app builds + // `P2PContext` from cluster peers only, handing relays to the conn gater + // and the relay manager instead (`app/src/node/behaviour.rs`). + let mut client: Node = Node::new( + P2PConfig::default(), + generate_insecure_k1_key(12), + NodeType::TCP, + false, + P2PContext::new(Vec::::new()), + move |builder, _keypair, relay_client| { + let p2p_context = builder.p2p_context(); + builder.with_inner(ClientBehaviour { + relay: relay_client, + relay_manager: RelayManager::new(vec![relay_mutable], p2p_context), + }) + }, + ) + .expect("build relay client node"); + + // Drive the client until it holds a reservation *and* has pinged the relay. + // The manager dials the relay and listens on its circuit address on its + // own, and libp2p pings a fresh connection immediately, so both usually + // land in the same handful of events. Metrics are recorded in + // `Node::handle_event` before an event is yielded, so by the time these + // are observed the gauge write and the ping gate have already run. + let mut reserved = false; + let mut pinged = false; + timeout(TEST_TIMEOUT, async { + while !(reserved && pinged) { + match client.select_next_some().await { + SwarmEvent::Behaviour(PlutoBehaviourEvent::Inner( + ClientBehaviourEvent::RelayManager(RelayManagerEvent::RelayReserved(peer)), + )) if peer == relay_peer => reserved = true, + SwarmEvent::Behaviour(PlutoBehaviourEvent::Ping(ping::Event { peer, .. })) + if peer == relay_peer => + { + pinged = true; + } + _ => {} + } + } + }) + .await + .expect("timed out waiting for the relay reservation and a ping of the relay"); + + assert_eq!( + P2P_METRICS + .relay_connections + .get(&relay_label) + .map(Gauge::get), + Some(1), + "a held reservation must show up as p2p_relay_connections{{peer}}=1" + ); + + // Read with `contains` rather than indexing so these assertions don't + // create the very series they are checking for. + assert!( + !P2P_METRICS.ping_success.contains(&relay_label), + "the relay must not appear in p2p_ping_success" + ); + assert!( + !P2P_METRICS.ping_latency_secs.contains(&relay_label), + "the relay must not appear in p2p_ping_latency_secs" + ); + assert!( + !P2P_METRICS.ping_error_total.contains(&relay_label), + "the relay must not appear in p2p_ping_error_total" + ); + + relay_handle.abort(); +}