summaryrefslogtreecommitdiffstats
path: root/src/transport/libp2p.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/libp2p.rs')
-rw-r--r--src/transport/libp2p.rs55
1 files changed, 32 insertions, 23 deletions
diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index ee5bb56..12d3da5 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -68,10 +68,13 @@ use crate::transport::stats::{
Connections, DialFailures, HolePunches, Libp2pStats, Reachability, RelaySlots, TransportStats,
};
use crate::transport::{
- Connection, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe,
- is_probe, wire,
+ Addressed, Connection, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport,
+ hand_on_probe, is_probe, wire,
};
+/// The scheme of this transport's addresses (`transport::paths`).
+pub(crate) const SCHEME: &str = "libp2p";
+
const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1");
/// Sends taking this long are logged with where the time went (at debug).
const SLOW_SEND: Duration = Duration::from_millis(5);
@@ -915,11 +918,11 @@ fn pick_relays(
for (relay, address) in wanted {
match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) {
Ok(listener) => {
- info!(relay = %relay, address = %address, "Picked a member as relay");
+ info!(relay = %Addressed(SCHEME, &relay), address = %address, "Picked a member as relay");
picked.insert(listener, relay);
}
Err(error) => {
- debug!(relay = %relay, error = %error, "Couldn't use a member as relay");
+ debug!(relay = %Addressed(SCHEME, &relay), error = %error, "Couldn't use a member as relay");
}
}
}
@@ -1062,7 +1065,7 @@ async fn run(
}
}
_ => {
- warn!(relay = %relay, "Couldn't reach a relay; trying again later");
+ warn!(relay = %Addressed(SCHEME, &relay), "Couldn't reach a relay; trying again later");
tokio::time::sleep(RELAY_RETRY).await;
let _ = again.send(Command::Reserve { relay, configured });
}
@@ -1075,11 +1078,11 @@ async fn run(
};
match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) {
Ok(listener) => {
- debug!(relay = %relay, "Reserving a slot on a relay");
+ debug!(relay = %Addressed(SCHEME, &relay), "Reserving a slot on a relay");
reservations.insert(listener, (configured, relay_peer));
}
Err(error) => {
- warn!(relay = %relay, error = %error, "Couldn't use a relay; trying again later");
+ warn!(relay = %Addressed(SCHEME, &relay), error = %error, "Couldn't use a relay; trying again later");
reserve_later(configured);
}
}
@@ -1113,7 +1116,7 @@ async fn run(
dedupe_due.insert(peer_id, tokio::time::Instant::now() + DUPLICATE_GRACE);
}
debug!(
- peer = %peer_id,
+ peer = %Addressed(SCHEME, &peer_id),
connection = %connection_id,
dialled = endpoint.is_dialer(),
address = %endpoint.get_remote_address(),
@@ -1138,7 +1141,7 @@ async fn run(
Some(error) => error.to_string(),
};
debug!(
- peer = %peer_id,
+ peer = %Addressed(SCHEME, &peer_id),
connection = %connection_id,
cause = ?cause,
"Connection closed"
@@ -1191,7 +1194,7 @@ async fn run(
book.counts.slots_not_granted += 1;
}
}
- warn!(relay = %relay, reason = ?reason, "Lost the slot on a relay; reserving again later");
+ warn!(relay = %Addressed(SCHEME, &relay), reason = ?reason, "Lost the slot on a relay; reserving again later");
reserve_later(relay);
}
// A relay picked among the members: the next check picks again, another
@@ -1209,10 +1212,10 @@ async fn run(
lost
};
if lost {
- debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying");
+ debug!(relay = %Addressed(SCHEME, &relay), reason = ?reason, "Lost the slot on a member relaying");
} else {
info!(
- relay = %relay,
+ relay = %Addressed(SCHEME, &relay),
pause = ?RELAY_PAUSE,
"A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while"
);
@@ -1234,7 +1237,7 @@ async fn run(
}
}
if !renewal {
- info!(relay = %relay_peer_id, "Reachable through a relay");
+ info!(relay = %Addressed(SCHEME, &relay_peer_id), "Reachable through a relay");
}
}
SwarmEvent::Behaviour(BehaviourEvent::Autonat(
@@ -1260,11 +1263,11 @@ async fn run(
SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
Ok(_) => {
lock(&book).counts.hole_punches.succeeded += 1;
- info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay");
+ info!(peer = %Addressed(SCHEME, &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");
+ debug!(peer = %Addressed(SCHEME, &event.remote_peer_id), error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on");
}
},
SwarmEvent::Behaviour(BehaviourEvent::Identify(
@@ -1274,10 +1277,10 @@ async fn run(
let listen = published(&info.listen_addrs);
// Every new connection identifies again: logged when it says something new.
if book.knows(peer_id, &listen) {
- trace!(peer = %peer_id, connection = %connection_id, "Peer identified again");
+ trace!(peer = %Addressed(SCHEME, &peer_id), connection = %connection_id, "Peer identified again");
} else {
debug!(
- peer = %peer_id,
+ peer = %Addressed(SCHEME, &peer_id),
connection = %connection_id,
listen = ?info.listen_addrs,
observed = %info.observed_addr,
@@ -1315,7 +1318,7 @@ async fn run(
if peer == relaying.local {
continue;
}
- debug!(peer = %peer, address = %address, "Peer on the LAN");
+ debug!(peer = %Addressed(SCHEME, &peer), address = %address, "Peer on the LAN");
let continuum = {
let mut book = lock(&book);
book.remember(peer, address.clone());
@@ -1553,7 +1556,7 @@ impl Transport for Libp2pTransport {
// Where a send's time goes: the swarm task (connecting), the stream's protocol
// negotiation (opening), or the connection (writing).
debug!(
- peer = %peer,
+ peer = %Addressed(SCHEME, &peer),
kind = ?message.kind,
bytes = message.body.len(),
connect_ms = connected.as_secs_f64() * 1000.0,
@@ -1586,17 +1589,23 @@ impl Transport for Libp2pTransport {
match read_message(stream).await {
Ok(body) => match wire::decode(&body) {
Ok(message) if is_probe(&message) => {
- hand_on_probe(&probes, peer.to_string(), message, body.len());
+ hand_on_probe(
+ SCHEME,
+ &probes,
+ peer.to_string(),
+ message,
+ body.len(),
+ );
}
Ok(message) => {
let _ = sender.send((peer, message, body.len())).await;
}
Err(error) => {
- debug!(peer = %peer, error = %error, "Dropping a message that doesn't decode");
+ debug!(peer = %Addressed(SCHEME, &peer), error = %error, "Dropping a message that doesn't decode");
}
},
Err(error) => {
- debug!(peer = %peer, error = %format!("{error:#}"), "Dropping an unreadable stream");
+ debug!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Dropping an unreadable stream");
}
}
drop(permit);
@@ -1605,7 +1614,7 @@ impl Transport for Libp2pTransport {
});
while let Some((peer, message, payload_len)) = received.recv().await {
if let Err(error) = handler(peer.to_string(), message, payload_len).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process inbound message");
+ warn!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Failed to process inbound message");
}
}
Ok(())
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 18:59:18 +0000