summaryrefslogtreecommitdiffstats
path: root/src/transport/p2p.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/p2p.rs')
-rw-r--r--src/transport/p2p.rs212
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));
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 08:04:41 +0000