From 2c0416d53a17400e3c796a6a1e4e196aa5318c56 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 14:03:53 +0100 Subject: add support for renaming IDs in the logger --- src/transport/libp2p.rs | 55 ++++++++++++++++++++++++++++--------------------- 1 file changed, 32 insertions(+), 23 deletions(-) (limited to 'src/transport/libp2p.rs') 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(()) -- cgit v1.3.1