fix(p2p): populate relay_connections, gate ping metrics to cluster peers

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<B: NetworkBehaviour> Node<B> {
             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<B: NetworkBehaviour> FusedStream for Node<B> {
     }
 }
 
+/// 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<Duration, ping::Failure>,
+) {
+    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();

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!(

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<i64> {
+    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();

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::<Multiaddr>()
+            .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)
+}