add transport stats

1 files changed, 194 insertions, 18 deletions

diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs
index ebda922..c03a0c4 100644
--- a/src/transport/p2p.rs
+++ b/src/transport/p2p.rs
@@ -58,7 +58,12 @@ use tokio::sync::{Semaphore, mpsc, oneshot, watch};
 use tracing::{debug, info, warn};
 
 use crate::locks::MutexExt;
-use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport, wire};
+use crate::transport::stats::{
+    Connections, DialFailures, HolePunches, Libp2pStats, Reachability, RelaySlots, TransportStats,
+};
+use crate::transport::{
+    Connection, EncodedMessage, MessageHandler, ParsedMessage, Transport, wire,
+};
 
 const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1");
 /// The largest message read: anyone who can connect can send anything, so this bounds
@@ -193,6 +198,15 @@ struct Book {
     public_relays: HashMap<PeerId, Multiaddr>,
     /// Relays picked that never granted a slot, and when (`RELAY_PAUSE`).
     failed_relays: HashMap<PeerId, tokio::time::Instant>,
+    /// Relays that granted us a slot we hold.
+    slots: std::collections::HashSet<PeerId>,
+    /// Open connections: to whom, and whether through a relay.
+    connections: HashMap<ConnectionId, (PeerId, bool)>,
+    /// What AutoNAT says, and the address it confirmed while public.
+    reachability: Reachability,
+    public_address: Option<Multiaddr>,
+    /// Since start (`stats`).
+    counts: Counts,
     /// When each peer was last mentioned (`STALE_AFTER`).
     seen: HashMap<PeerId, tokio::time::Instant>,
     /// Peers found on the LAN (mDNS) not known yet to speak continuum.
@@ -204,7 +218,72 @@ struct Book {
     discovered: Vec<Multiaddr>,
 }
 
+/// What `Libp2pStats` counts since start.
+#[derive(Default)]
+struct Counts {
+    slots_granted: u64,
+    slots_lost: u64,
+    slots_not_granted: u64,
+    hole_punches: HolePunches,
+    dial_failures: DialFailures,
+}
+
+/// Counts a dial that failed with `error` (not one that wasn't needed).
+fn count_dial_failure(failures: &mut DialFailures, error: &::libp2p::swarm::DialError) {
+    use ::libp2p::swarm::DialError;
+    match error {
+        DialError::DialPeerConditionFalse(_) => {}
+        DialError::Transport(_) => failures.unreachable += 1,
+        DialError::WrongPeerId { .. } => failures.wrong_peer_id += 1,
+        DialError::NoAddresses => failures.no_addresses += 1,
+        _ => failures.other += 1,
+    }
+}
+
 impl Book {
+    /// How we are connected to `peer`: directly if any connection is direct.
+    fn connection(&self, peer: &PeerId) -> Option<Connection> {
+        let mut found = None;
+        for (connected, relayed) in self.connections.values() {
+            if connected == peer {
+                if !relayed {
+                    return Some(Connection::Direct);
+                }
+                found = Some(Connection::Relayed);
+            }
+        }
+        found
+    }
+
+    fn stats(&self) -> Libp2pStats {
+        let mut connections = Connections::default();
+        for (_, relayed) in self.connections.values() {
+            if *relayed {
+                connections.relayed += 1;
+            } else {
+                connections.direct += 1;
+            }
+        }
+        Libp2pStats {
+            reachability: self.reachability,
+            public_address: self.public_address.as_ref().map(Multiaddr::to_string),
+            connections,
+            relay_slots: RelaySlots {
+                held: self.slots.len(),
+                paused: self
+                    .failed_relays
+                    .values()
+                    .filter(|at| at.elapsed() < RELAY_PAUSE)
+                    .count(),
+                granted: self.counts.slots_granted,
+                lost: self.counts.slots_lost,
+                not_granted: self.counts.slots_not_granted,
+            },
+            hole_punches: self.counts.hole_punches,
+            dial_failures: self.counts.dial_failures,
+        }
+    }
+
     /// `peer` was mentioned just now.
     fn touch(&mut self, peer: PeerId) {
         self.seen.insert(peer, tokio::time::Instant::now());
@@ -675,8 +754,10 @@ fn pick_relays(
 ) {
     if nat.is_public() && !picked.is_empty() {
         info!("Reachable directly (AutoNAT); letting the relays picked go");
-        for listener in std::mem::take(picked).into_keys() {
+        let mut book = lock(book);
+        for (listener, relay) in std::mem::take(picked) {
             swarm.remove_listener(listener);
+            book.slots.remove(&relay);
         }
         return;
     }
@@ -735,15 +816,14 @@ async fn run(
 ) {
     let serving = relaying.serving;
     let mut nat = ::libp2p::autonat::NatStatus::Unknown;
-    // Relays picked among the members, by listener, and those that granted a slot.
+    // Relays picked among the members, by listener.
     let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new();
-    let mut granted: std::collections::HashSet<PeerId> = std::collections::HashSet::new();
     let mut check = tokio::time::interval(RELAY_CHECK);
     check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
     let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP);
     dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-    // Relays we listen through, by listener, as configured.
-    let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> =
+    // Relays we listen through, by listener: as configured, and their peer.
+    let mut reservations: HashMap<::libp2p::core::transport::ListenerId, (Multiaddr, PeerId)> =
         HashMap::new();
     let reserve_later = |configured: Multiaddr| {
         let again = again.clone();
@@ -805,6 +885,7 @@ async fn run(
                         opts = opts.extend_addresses_through_behaviour();
                     }
                     if let Err(error) = swarm.dial(opts.build()) {
+                        count_dial_failure(&mut lock(&book).counts.dial_failures, &error);
                         for done in connecting.remove(&peer).into_iter().flatten() {
                             let _ = done.send(Err(error.to_string()));
                         }
@@ -837,10 +918,13 @@ async fn run(
                     });
                 }
                 Some(Command::Reserve { relay, configured }) => {
+                    let Some(relay_peer) = split_peer(&relay) else {
+                        continue; // The arm above takes those.
+                    };
                     match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) {
                         Ok(listener) => {
                             debug!(relay = %relay, "Reserving a slot on a libp2p relay");
-                            reservations.insert(listener, configured);
+                            reservations.insert(listener, (configured, relay_peer));
                         }
                         Err(error) => {
                             warn!(relay = %relay, error = %error, "Couldn't use a libp2p relay; trying again later");
@@ -862,8 +946,13 @@ async fn run(
                 }
             },
             event = swarm.select_next_some() => match event {
-                SwarmEvent::ConnectionEstablished { peer_id, connection_id, .. } => {
-                    lock(&book).touch(peer_id);
+                SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, .. } => {
+                    {
+                        let mut book = lock(&book);
+                        book.touch(peer_id);
+                        let relayed = endpoint.is_relayed();
+                        book.connections.insert(connection_id, (peer_id, relayed));
+                    }
                     for done in connecting.remove(&peer_id).into_iter().flatten() {
                         let _ = done.send(Ok(()));
                     }
@@ -872,7 +961,11 @@ async fn run(
                         let _ = done.send(Ok(peer_id));
                     }
                 }
+                SwarmEvent::ConnectionClosed { connection_id, .. } => {
+                    lock(&book).connections.remove(&connection_id);
+                }
                 SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => {
+                    count_dial_failure(&mut lock(&book).counts.dial_failures, &error);
                     if let Some((_, done)) = identifying.remove(&connection_id) {
                         let _ = done.send(Err(error.to_string()));
                     }
@@ -909,14 +1002,33 @@ async fn run(
                     lock(&book)
                         .listening
                         .retain(|known| !addresses.contains(known));
-                    if let Some(relay) = reservations.remove(&listener_id) {
+                    if let Some((relay, relay_peer)) = reservations.remove(&listener_id) {
+                        {
+                            let mut book = lock(&book);
+                            if book.slots.remove(&relay_peer) {
+                                book.counts.slots_lost += 1;
+                            } else {
+                                book.counts.slots_not_granted += 1;
+                            }
+                        }
                         warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later");
                         reserve_later(relay);
                     }
                     // A relay picked among the members: the next check picks again, another
                     // one if this one never granted the slot.
                     if let Some(relay) = picked.remove(&listener_id) {
-                        if granted.remove(&relay) {
+                        let lost = {
+                            let mut book = lock(&book);
+                            let lost = book.slots.remove(&relay);
+                            if lost {
+                                book.counts.slots_lost += 1;
+                            } else {
+                                book.counts.slots_not_granted += 1;
+                                book.failed_relays.insert(relay, tokio::time::Instant::now());
+                            }
+                            lost
+                        };
+                        if lost {
                             debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying");
                         } else {
                             info!(
@@ -924,9 +1036,6 @@ async fn run(
                                 pause = ?RELAY_PAUSE,
                                 "A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while"
                             );
-                            lock(&book)
-                                .failed_relays
-                                .insert(relay, tokio::time::Instant::now());
                         }
                     }
                 }
@@ -936,8 +1045,14 @@ async fn run(
                 SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
                     ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. },
                 )) => {
-                    granted.insert(relay_peer_id);
-                    lock(&book).failed_relays.remove(&relay_peer_id);
+                    {
+                        let mut book = lock(&book);
+                        book.slots.insert(relay_peer_id);
+                        book.failed_relays.remove(&relay_peer_id);
+                        if !renewal {
+                            book.counts.slots_granted += 1;
+                        }
+                    }
                     if !renewal {
                         info!(relay = %relay_peer_id, "Reachable through a libp2p relay");
                     }
@@ -946,14 +1061,31 @@ async fn run(
                     ::libp2p::autonat::Event::StatusChanged { old, new },
                 )) => {
                     info!(old = ?old, new = ?new, "libp2p reachability (AutoNAT)");
+                    {
+                        use ::libp2p::autonat::NatStatus;
+                        let mut book = lock(&book);
+                        (book.reachability, book.public_address) = match &new {
+                            NatStatus::Unknown => (Reachability::Unknown, None),
+                            NatStatus::Private => (Reachability::Private, None),
+                            NatStatus::Public(address) => {
+                                (Reachability::Public, Some(address.clone()))
+                            }
+                        };
+                    }
                     nat = new;
                 }
                 SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => {
                     debug!(event = ?event, "libp2p relay");
                 }
                 SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
-                    Ok(_) => info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay"),
-                    Err(error) => debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on"),
+                    Ok(_) => {
+                        lock(&book).counts.hole_punches.succeeded += 1;
+                        info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay");
+                    }
+                    Err(error) => {
+                        lock(&book).counts.hole_punches.failed += 1;
+                        debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on");
+                    }
                 },
                 SwarmEvent::Behaviour(BehaviourEvent::Identify(
                     ::libp2p::identify::Event::Received { peer_id, info, .. },
@@ -1124,6 +1256,18 @@ impl Transport for Libp2pTransport {
         Some(peer.to_string())
     }
 
+    fn connection(&self, target: &str) -> Option<Connection> {
+        let peer = PeerId::from_str(target).ok()?;
+        lock(&self.book).connection(&peer)
+    }
+
+    fn stats(&self) -> TransportStats {
+        TransportStats {
+            libp2p: Some(lock(&self.book).stats()),
+            ..TransportStats::default()
+        }
+    }
+
     /// Nodes found on the LAN (mDNS) that speak continuum.
     fn discovered_seeds(&self) -> Vec<String> {
         std::mem::take(&mut lock(&self.book).discovered)
@@ -1352,6 +1496,12 @@ mod tests {
         a.send_message(b.local_id(), &ping()).await.unwrap();
         let (from, _) = at_b.recv().await.unwrap();
         assert_eq!(from, a.local_id());
+
+        assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
+        assert_eq!(b.connection(a.local_id()), Some(Connection::Direct));
+        let stats = a.stats().libp2p.unwrap();
+        assert!(stats.connections.direct >= 1);
+        assert_eq!(stats.connections.relayed, 0);
     }
 
     #[tokio::test]
@@ -1474,11 +1624,16 @@ mod tests {
             "{relayed}"
         );
 
+        assert!(eventually(|| b.stats().libp2p.unwrap().relay_slots.held == 1).await);
+        let slots = b.stats().libp2p.unwrap().relay_slots;
+        assert_eq!((slots.granted, slots.not_granted), (1, 0));
+
         let (a, _) = memory_node();
         assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
         a.send_message(b.local_id(), &ping()).await.unwrap();
         let (from, _) = at_b.recv().await.unwrap();
         assert_eq!(from, a.local_id());
+        assert!(a.connection(b.local_id()).is_some());
     }
 
     fn peer() -> PeerId {
@@ -1603,5 +1758,26 @@ mod tests {
             .to_string();
         assert!(b.send_message(&stranger, &ping()).await.is_err());
         assert!(b.resolve_seed(&stranger).await.is_err());
+        assert_eq!(b.connection(&stranger), None);
+    }
+
+    #[test]
+    fn a_direct_connection_wins_over_a_relayed_one() {
+        let mut book = Book::default();
+        let (relayed_only, both) = (peer(), peer());
+        let mut id = 0;
+        let mut connect = |book: &mut Book, peer: PeerId, relayed: bool| {
+            id += 1;
+            book.connections
+                .insert(ConnectionId::new_unchecked(id), (peer, relayed));
+        };
+        connect(&mut book, relayed_only, true);
+        connect(&mut book, both, true);
+        connect(&mut book, both, false);
+        assert_eq!(book.connection(&relayed_only), Some(Connection::Relayed));
+        assert_eq!(book.connection(&both), Some(Connection::Direct));
+        assert_eq!(book.connection(&peer()), None);
+        let connections = book.stats().connections;
+        assert_eq!((connections.direct, connections.relayed), (1, 2));
     }
 }