add support for renaming IDs in the logger

7 files changed, 103 insertions, 44 deletions

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<String> {
+        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<String> {
         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<Box<dyn
     + 'static;
 pub(crate) type MessageHandler = Box<HandlerFn>;
 
+/// A transport identity (or location) as an address, `<scheme>:<it>`, 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<T>(pub(crate) &'static str, pub(crate) T);
+
+impl<T: std::fmt::Display> std::fmt::Display for Addressed<T> {
+    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<HandlerFn>,
     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 (`<scheme>:<transport
+    /// identity>`, 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<String> {
+        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<String> {
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::<serde_json::Value>(&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::<OutboundMessage>(&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: `<scheme>:<its identity there>`.
+    pub(crate) fn own_addresses(&self) -> Vec<String> {
+        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<String> {
+        self.inner.identities_of(peer)
+    }
+
     fn learn_addresses(&self, peer: &str, addresses: &[String]) {
         self.inner.learn_addresses(peer, addresses)
     }