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/identity.rs | 14 +++++++++++ src/transport/libp2p.rs | 55 +++++++++++++++++++++++++------------------ src/transport/mod.rs | 24 +++++++++++++++++-- src/transport/mycelium/mod.rs | 24 +++++++++++-------- src/transport/paths.rs | 5 ++++ src/transport/tor.rs | 21 ++++++++++------- src/transport/traffic.rs | 4 ++++ 7 files changed, 103 insertions(+), 44 deletions(-) (limited to 'src/transport') diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 502c6b3..432348e 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -474,6 +474,20 @@ impl Transport for IdentityTransport { self.shared.paths.local_addresses() } + fn identities_of(&self, peer: &str) -> Vec { + if peer == self.shared.node_id { + return self.shared.paths.own_addresses(); + } + let book = self.shared.lock(); + let mut addresses = book.proven.get(peer).cloned().unwrap_or_default(); + for hint in book.hints.get(peer).into_iter().flatten() { + if !addresses.contains(hint) { + addresses.push(hint.clone()); + } + } + addresses + } + /// Where `peer` proved itself, else where it is said to be, as others can use it. fn addresses_of(&self, peer: &str) -> Vec { let addresses = { 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(()) diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 839476b..7cd976b 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -570,6 +570,17 @@ pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin; +/// A transport identity (or location) as an address, `:`, the way paths and +/// the identity layer show it: for a transport's own log lines, so that a peer reads the +/// same in every one. +pub(crate) struct Addressed(pub(crate) &'static str, pub(crate) T); + +impl std::fmt::Display for Addressed { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}:{}", self.0, self.1) + } +} + /// Whether `message` is a PING or PONG. The paths layer times them, so a transport's /// consumer hands them on as soon as they are read (`hand_on_probe`), not after the /// messages read before them are handled (which can take a while: sync). @@ -577,8 +588,10 @@ pub(crate) fn is_probe(message: &ParsedMessage) -> bool { matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) } -/// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own. +/// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own; `scheme` is the +/// transport's, for the log. pub(crate) fn hand_on_probe( + scheme: &'static str, handler: &std::sync::Arc, sender: String, message: ParsedMessage, @@ -587,7 +600,7 @@ pub(crate) fn hand_on_probe( let handler = std::sync::Arc::clone(handler); tokio::spawn(async move { if let Err(error) = handler(sender.clone(), message, payload_len).await { - tracing::warn!(peer = %sender, error = %format!("{error:#}"), "Failed to process inbound message"); + tracing::warn!(peer = %Addressed(scheme, &sender), error = %format!("{error:#}"), "Failed to process inbound message"); } }); } @@ -628,6 +641,13 @@ pub(crate) trait Transport: Send + Sync { vec![target.to_owned()] } + /// For a transport under the identity layer: the addresses (`:`, as paths keep them) `peer` proved itself at or is said to be at; this + /// node's own, for its own node ID. By default its identity. + fn identities_of(&self, peer: &str) -> Vec { + vec![peer.to_owned()] + } + /// Where `peer` can be reached, as far as this node knows: passed on to others (in /// a HELLO_REPLY). By default its identity, as for `local_addresses`. fn addresses_of(&self, peer: &str) -> Vec { diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 2bb399c..5e8dc05 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -9,13 +9,17 @@ use url::Url; use crate::transport::wire; use crate::transport::{ - EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe, is_probe, + Addressed, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe, + is_probe, }; pub(crate) mod daemon; pub(crate) use daemon::{MyceliumSetup, MyceliumSpawnConfig}; +/// The scheme of this transport's addresses (`transport::paths`). +pub(crate) const SCHEME: &str = "mycelium"; + const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; /// How long Mycelium tries to deliver a message before dropping it. const DELIVERY_LIMIT: Duration = Duration::from_secs(300); @@ -251,22 +255,22 @@ impl Transport for MyceliumTransport { .map_err(anyhow::Error::from) .and_then(|bytes| Ok(serde_json::from_slice::(&bytes)?)) else { - tracing::debug!(to = %message.peer, "Sent message (couldn't decode response from Mycelium REST API)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), "Sent message (couldn't decode response from Mycelium REST API)"); return Ok(()); }; let Ok(outbound) = serde_json::from_slice::(&message.body) else { - tracing::debug!(to = %message.peer, body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); return Ok(()); }; let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else { - tracing::debug!(to = %message.peer, message = ?outbound, "Sent message (couldn't decode outbound message)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?outbound, "Sent message (couldn't decode outbound message)"); return Ok(()); }; let Some(id) = body.get("id").and_then(|id| id.as_str()) else { - tracing::debug!(to = %message.peer, message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); return Ok(()); }; - tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?parsed, id = %id, "Sent message"); } Ok(()) } @@ -297,7 +301,7 @@ impl Transport for MyceliumTransport { match parse_message(&inbound_message) { Ok((sender, parsed)) => { tracing::debug!( - from = %inbound_message.src_pk, + from = %Addressed(SCHEME, &inbound_message.src_pk), message = ?parsed, id = %inbound_message.id, //payload_bytes = raw_bytes.len(), @@ -305,7 +309,7 @@ impl Transport for MyceliumTransport { ); let payload_len = base64_decoded_len(&inbound_message.payload); if is_probe(&parsed) { - hand_on_probe(&handler, sender, parsed, payload_len); + hand_on_probe(SCHEME, &handler, sender, parsed, payload_len); } else if let Err(error) = handler(sender, parsed, payload_len).await { tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message"); } @@ -317,7 +321,7 @@ impl Transport for MyceliumTransport { } if other_versions.insert(inbound_message.src_pk.clone()) { tracing::warn!( - from = %inbound_message.src_pk, + from = %Addressed(SCHEME, &inbound_message.src_pk), version, ours = wire::VERSION, "Peer speaks another wire version (another build); ignoring its messages" @@ -332,7 +336,7 @@ impl Transport for MyceliumTransport { .and_then(|payload| payload.get(1).copied()); tracing::error!( error = %transport_error, - from = %inbound_message.src_pk, + from = %Addressed(SCHEME, &inbound_message.src_pk), kind = ?kind, payload_bytes = base64_decoded_len(&inbound_message.payload), topic = %inbound_message.topic, diff --git a/src/transport/paths.rs b/src/transport/paths.rs index e966839..2fe534a 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -480,6 +480,11 @@ impl Paths { .map(Link::own_address) } + /// This node's own addresses, one per transport: `:`. + pub(crate) fn own_addresses(&self) -> Vec { + self.links.iter().map(Link::own_address).collect() + } + /// The longest of the transports': until then, something may still be on its way. pub(crate) fn delivery_limit(&self) -> Duration { self.links diff --git a/src/transport/tor.rs b/src/transport/tor.rs index b49cc67..39e0f31 100644 --- a/src/transport/tor.rs +++ b/src/transport/tor.rs @@ -53,13 +53,16 @@ use crate::identity::NodeKey; use crate::locks::MutexExt; use crate::transport::challenge::{self, Role}; use crate::transport::{ - EncodedMessage, HandlerFn, Hex, MessageHandler, ParsedMessage, Transport, hand_on_probe, - is_probe, wire, + Addressed, EncodedMessage, HandlerFn, Hex, MessageHandler, ParsedMessage, Transport, + hand_on_probe, is_probe, wire, }; /// A message read, with the sender's onion service and its payload length. type Inbound = (String, ParsedMessage, usize); +/// The scheme of this transport's addresses (`transport::paths`). +pub(crate) const SCHEME: &str = "tor"; + /// The onion service's nickname in arti's state (one per node). const NICKNAME: &str = "continuum"; /// The largest message read. Sync messages above 16 chunks travel as blobs, and a chunk @@ -432,10 +435,10 @@ impl Shared { self.register(location, stream); } Ok(false) => { - debug!(claimed = %location, key = %Hex(&key), "Dropping a connection to our onion service: another node is at the onion service it claims"); + debug!(claimed = %Addressed(SCHEME, &location), key = %Hex(&key), "Dropping a connection to our onion service: another node is at the onion service it claims"); } Err(error) => { - debug!(claimed = %location, error = %format!("{error:#}"), "Dropping a connection to our onion service: couldn't check the onion service it claims"); + debug!(claimed = %Addressed(SCHEME, &location), error = %format!("{error:#}"), "Dropping a connection to our onion service: couldn't check the onion service it claims"); } } } @@ -450,11 +453,11 @@ impl Shared { writer: tokio::sync::Mutex::new(writer), }); lock(&self.connections).insert(location.clone(), Arc::clone(&connection)); - debug!(peer = %location, connection = id, "Connection opened"); + debug!(peer = %Addressed(SCHEME, &location), connection = id, "Connection opened"); let shared = Arc::clone(self); tokio::spawn(async move { let error = shared.read(&location, reader).await; - debug!(peer = %location, connection = id, error = %format!("{error:#}"), "Connection closed"); + debug!(peer = %Addressed(SCHEME, &location), connection = id, error = %format!("{error:#}"), "Connection closed"); shared.forget(&location, id); }); connection @@ -485,7 +488,7 @@ impl Shared { let message = match wire::decode(&body) { Ok(message) => message, Err(error) => { - debug!(peer = %location, error = %error, "Dropping a message that doesn't decode"); + debug!(peer = %Addressed(SCHEME, &location), error = %error, "Dropping a message that doesn't decode"); continue; } }; @@ -610,13 +613,13 @@ impl Transport for TorTransport { let handler = Arc::clone(&handler); tokio::spawn(async move { while let Some((peer, message, payload_len)) = probes.recv().await { - hand_on_probe(&handler, peer, message, payload_len); + hand_on_probe(SCHEME, &handler, peer, message, payload_len); } }); } while let Some((peer, message, payload_len)) = incoming.recv().await { if let Err(error) = handler(peer.clone(), 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(()) diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 64595cb..32f7257 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -160,6 +160,10 @@ impl Transport for CountingTransport { self.inner.addresses_of(peer) } + fn identities_of(&self, peer: &str) -> Vec { + self.inner.identities_of(peer) + } + fn learn_addresses(&self, peer: &str, addresses: &[String]) { self.inner.learn_addresses(peer, addresses) } -- cgit v1.3.1