summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/config/mod.rs9
-rw-r--r--src/logging.rs173
-rw-r--r--src/main.rs11
-rw-r--r--src/simtest/identity.rs3
-rw-r--r--src/state/peer.rs5
-rw-r--r--src/swarm/state_file.rs12
-rw-r--r--src/swarm/topology.rs14
-rw-r--r--src/transport/identity.rs14
-rw-r--r--src/transport/libp2p.rs55
-rw-r--r--src/transport/mod.rs24
-rw-r--r--src/transport/mycelium/mod.rs24
-rw-r--r--src/transport/paths.rs5
-rw-r--r--src/transport/tor.rs21
-rw-r--r--src/transport/traffic.rs4
14 files changed, 323 insertions, 51 deletions
diff --git a/src/config/mod.rs b/src/config/mod.rs
index d4a7c9c..9c92b7e 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -110,6 +110,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
"HTTP API users (from htpasswd) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake. Such writes act on the whole swarm and are logged. Epoch metadata and the swarm ID are never writable.",
),
option(
+ "log-raw-ids",
+ Kind::Flag,
+ "BOOL",
+ Some("false"),
+ "Log node IDs and transport identities as they are. Without it, a member's read `<scheme>:<alias>` wherever they appear (`continuum:<alias>`, `libp2p:<alias>`), as the member list names it.",
+ ),
+ option(
"debug-api",
Kind::Flag,
"BOOL",
@@ -376,6 +383,8 @@ pub(crate) struct Settings {
#[serde(default)]
pub(crate) debug_api: bool,
#[serde(default)]
+ pub(crate) log_raw_ids: bool,
+ #[serde(default)]
pub(crate) seeds: Vec<String>,
#[serde(default)]
pub(crate) genesis: bool,
diff --git a/src/logging.rs b/src/logging.rs
index 4496125..093d49e 100644
--- a/src/logging.rs
+++ b/src/logging.rs
@@ -3,16 +3,161 @@
//!
//! The filter starts from `RUST_LOG`, or `info` without it. It applies to the fmt
//! logger only, so the console layer keeps getting the runtime events it needs.
+//!
+//! Each line is written with members named by their alias (`Renaming`): wherever a
+//! member's node ID or one of its addresses appears, in a field, a message or a dump of
+//! a value, it reads `<scheme>:<alias>` (`continuum:<alias>` for the node ID,
+//! `libp2p:<alias>` for its libp2p peer ID, with its scheme or bare). The names and
+//! aliases are as `set_aliases` last gave them (`swarm::topology`, whenever it looks at
+//! the member list); at start, as the state file remembers them, so they show from the
+//! first line (stale, perhaps: a member's alias that changed meanwhile is logged as such
+//! once the member list says so). `--log-raw-ids` leaves IDs as they are.
-use std::sync::OnceLock;
+use std::borrow::Cow;
+use std::collections::{BTreeMap, HashMap};
+use std::io::Write;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{OnceLock, RwLock};
use anyhow::{Result, anyhow};
+use tracing_subscriber::fmt::MakeWriter;
use tracing_subscriber::{EnvFilter, Registry, prelude::*, reload};
+use crate::locks::RwLockExt;
+
type FilterHandle = reload::Handle<EnvFilter, Registry>;
static FILTER: OnceLock<FilterHandle> = OnceLock::new();
+/// Members' aliases, by the names that stand for them (`set_aliases`), and the
+/// replacements made of them, longest first.
+static ALIASES: RwLock<Aliases> = RwLock::new(Aliases {
+ given: None,
+ renames: Vec::new(),
+});
+/// `--log-raw-ids`.
+static RAW_IDS: AtomicBool = AtomicBool::new(false);
+
+struct Aliases {
+ given: Option<HashMap<String, String>>,
+ renames: Vec<(String, String)>,
+}
+
+/// Leave IDs in the log as they are (`--log-raw-ids`).
+pub(crate) fn set_raw_ids(raw: bool) {
+ RAW_IDS.store(raw, Ordering::Relaxed);
+}
+
+/// The aliases to name members by from now on, by the names that stand for them: node
+/// IDs and addresses (`<scheme>:<transport identity>`).
+pub(crate) fn set_aliases(aliases: HashMap<String, String>) {
+ let previous = {
+ let mut current = ALIASES.write_or_recover();
+ if current.given.as_ref() == Some(&aliases) {
+ return;
+ }
+ current.renames = renames(&aliases);
+ current.given.replace(aliases.clone())
+ };
+ // A member's alias that changed, after the lock: the line itself is renamed.
+ let Some(previous) = previous else {
+ return;
+ };
+ for (name, before) in &previous {
+ if !crate::identity::is_node_id(name) {
+ continue;
+ }
+ match aliases.get(name) {
+ Some(after) if after != before => {
+ tracing::info!(peer = %name, previous = %before, "A member's alias changed");
+ }
+ None => tracing::info!(peer = %name, previous = %before, "A member's alias is gone"),
+ _ => {}
+ }
+ }
+}
+
+/// What each name becomes: `<scheme>:<alias>`, for the name with its scheme and bare.
+/// Longest first, so that an address is renamed whole before its bare identity could be.
+fn renames(aliases: &HashMap<String, String>) -> Vec<(String, String)> {
+ let mut renames: BTreeMap<String, String> = BTreeMap::new();
+ for (name, alias) in aliases {
+ let Some((scheme, identity)) = name.split_once(':') else {
+ continue;
+ };
+ let renamed = format!("{scheme}:{alias}");
+ renames.insert(name.clone(), renamed.clone());
+ // A node ID's hex alone is no name of it.
+ if !crate::identity::is_node_id(name) && !identity.is_empty() {
+ renames.entry(identity.to_owned()).or_insert(renamed);
+ }
+ }
+ let mut renames: Vec<(String, String)> = renames.into_iter().collect();
+ renames.sort_by(|a, b| b.0.len().cmp(&a.0.len()).then_with(|| a.0.cmp(&b.0)));
+ renames
+}
+
+/// The aliases `set_aliases` was last given, sorted (for the state file).
+pub(crate) fn aliases() -> BTreeMap<String, String> {
+ ALIASES
+ .read_or_recover()
+ .given
+ .iter()
+ .flatten()
+ .map(|(name, alias)| (name.clone(), alias.clone()))
+ .collect()
+}
+
+/// `line` with members named by their alias (module docs).
+fn rename(line: &str) -> Cow<'_, str> {
+ if RAW_IDS.load(Ordering::Relaxed) {
+ return Cow::Borrowed(line);
+ }
+ rename_with(line, &ALIASES.read_or_recover().renames)
+}
+
+fn rename_with<'a>(line: &'a str, renames: &[(String, String)]) -> Cow<'a, str> {
+ let mut renamed = Cow::Borrowed(line);
+ for (name, alias) in renames {
+ if renamed.contains(name.as_str()) {
+ renamed = Cow::Owned(renamed.replace(name.as_str(), alias));
+ }
+ }
+ renamed
+}
+
+/// Writes the log to stdout, a line (one event) at a time, renamed (`rename`).
+struct Renaming;
+
+impl<'a> MakeWriter<'a> for Renaming {
+ type Writer = RenamingLine;
+
+ fn make_writer(&'a self) -> Self::Writer {
+ RenamingLine(Vec::new())
+ }
+}
+
+/// One event's text, written renamed when the logger is done with it.
+struct RenamingLine(Vec<u8>);
+
+impl Write for RenamingLine {
+ fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+ self.0.extend_from_slice(bytes);
+ Ok(bytes.len())
+ }
+
+ fn flush(&mut self) -> std::io::Result<()> {
+ Ok(())
+ }
+}
+
+impl Drop for RenamingLine {
+ fn drop(&mut self) {
+ let text = String::from_utf8_lossy(&self.0);
+ let _ = std::io::stdout().write_all(rename(&text).as_bytes());
+ }
+}
+
/// Installs the global subscriber. Call once, first thing in main.
pub(crate) fn init() {
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
@@ -21,6 +166,7 @@ pub(crate) fn init() {
let registry = tracing_subscriber::registry().with(
tracing_subscriber::fmt::layer()
.compact()
+ .with_writer(Renaming)
.with_filter(filter),
);
@@ -56,3 +202,28 @@ fn handle() -> Result<&'static FilterHandle> {
.get()
.ok_or_else(|| anyhow!("Logging is not initialised"))
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn members_are_named_by_their_alias() {
+ let node = crate::identity::node_id_of(&[1; 32]);
+ let renames = renames(&HashMap::from([
+ (node.clone(), "kitchen".to_owned()),
+ ("libp2p:12D3KooWexample".to_owned(), "kitchen".to_owned()),
+ ("tor:abc.onion:9662".to_owned(), "kitchen".to_owned()),
+ ]));
+ let line = format!(
+ "peer={node} to=libp2p:12D3KooWexample event=PeerId(\"12D3KooWexample\") at=abc.onion:9662 other=12D3KooWother"
+ );
+ assert_eq!(
+ rename_with(&line, &renames),
+ "peer=continuum:kitchen to=libp2p:kitchen event=PeerId(\"libp2p:kitchen\") at=tor:kitchen other=12D3KooWother"
+ );
+ // A node ID's hex alone stays.
+ let hex = node.split_once(':').unwrap().1;
+ assert_eq!(rename_with(hex, &renames), hex);
+ }
+}
diff --git a/src/main.rs b/src/main.rs
index f174c4e..e6a27f4 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -114,7 +114,9 @@ async fn main() -> Result<()> {
}
None => (None, None),
};
+ logging::set_raw_ids(args.log_raw_ids);
if let Some(remembered) = &remembered {
+ logging::set_aliases(remembered.aliases.clone().into_iter().collect());
info!(
swarm_id = %remembered.swarm_id,
seeds = remembered.seeds.len(),
@@ -176,7 +178,7 @@ async fn main() -> Result<()> {
.take()
.context("Mycelium named twice in transports")?;
(
- "mycelium",
+ transport::mycelium::SCHEME,
Arc::new(mycelium::MyceliumTransport::connect(setup).await?),
)
}
@@ -212,7 +214,7 @@ async fn main() -> Result<()> {
let key =
transport::libp2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?;
(
- "libp2p",
+ transport::libp2p::SCHEME,
Arc::new(transport::libp2p::Libp2pTransport::start(
transport::libp2p::Libp2pConfig {
key,
@@ -241,7 +243,10 @@ async fn main() -> Result<()> {
}) => transport?,
_ = stopped(stop.clone()) => return Ok(()),
};
- ("tor", Arc::new(transport) as Arc<dyn Transport>)
+ (
+ transport::tor::SCHEME,
+ Arc::new(transport) as Arc<dyn Transport>,
+ )
}
other => anyhow::bail!("transports: no {other} in this build"),
};
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index 3d235b0..789b011 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -54,6 +54,9 @@ async fn nodes_are_their_node_ids() {
// Its only path, measured by keepalives once there were any.
assert_eq!(destination.paths.len(), 1);
assert_eq!(destination.paths[0].address, "sim:b");
+ // The names that stand for b, through every layer (the logs' aliases use them).
+ assert_eq!(a.transport.identities_of(b.local_id()), ["sim:b"]);
+ assert_eq!(a.transport.identities_of(a.local_id()), ["sim:a"]);
let stats = a.transport.stats();
assert_eq!(stats.identity.unwrap().proven["sim"], 2);
assert!(stats.libp2p.is_none());
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 9f2da0f..923c6d2 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -444,7 +444,10 @@ mod tests {
let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
let mut heard = PeerState::new(true, &timing);
assert_eq!(heard.liveness, Liveness::Unknown);
- assert!(!heard.mark_seen(heard.added_at), "no re-handshake: none yet");
+ assert!(
+ !heard.mark_seen(heard.added_at),
+ "no re-handshake: none yet"
+ );
assert_eq!(heard.liveness, Liveness::Active);
let mut silent = PeerState::new(true, &timing);
assert!(silent.check_dormancy(&timing, silent.added_at + timing.dormant_after));
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index 3693d89..a65fa5b 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -1,5 +1,6 @@
//! The state file (`--state-file`): what a node remembers across restarts, so it
-//! rejoins its swarm on its own: the swarm ID and some members to reach it through.
+//! rejoins its swarm on its own: the swarm ID and some members to reach it through; and
+//! the members' aliases, for the logs from the start (`logging`).
//! Not the document: a restarted node still starts empty and syncs.
//!
//! Written rarely: once the node is ready, then when the remembered members change
@@ -35,6 +36,9 @@ pub(crate) struct Remembered {
/// them (`swarm::members::addresses`).
#[serde(default)]
pub(crate) seeds: Vec<String>,
+ /// Members' aliases, by the names that stand for them (`logging::aliases`).
+ #[serde(default)]
+ pub(crate) aliases: std::collections::BTreeMap<String, String>,
}
pub(crate) struct StateFile {
@@ -136,7 +140,11 @@ pub(crate) async fn update(state: &AppState, now: bool) {
state.clock.now_millis().as_u64(),
)
};
- let remembered = Remembered { swarm_id, seeds };
+ let remembered = Remembered {
+ swarm_id,
+ seeds,
+ aliases: crate::logging::aliases(),
+ };
{
let inner = file.lock();
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index c9d1266..2ba07b7 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -137,6 +137,19 @@ pub(crate) fn affects_neighbours(path: &[String]) -> bool {
path.is_empty() || path.starts_with(&members::members_dir())
}
+/// Hands the members' `aliases` to the logger, under every name that stands for each:
+/// its node ID, and its addresses (`logging`, which renames them bare too).
+fn set_aliases(state: &AppState, aliases: std::collections::HashMap<String, String>) {
+ let mut names = std::collections::HashMap::new();
+ for (member, alias) in aliases {
+ for address in state.transport.identities_of(&member) {
+ names.insert(address, alias.clone());
+ }
+ names.insert(member, alias);
+ }
+ crate::logging::set_aliases(names);
+}
+
/// Recomputes the neighbours once ready, if the member list, the reachable members or
/// `k` changed.
pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
@@ -146,6 +159,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
let (current, stable, transports, alive) = {
let replica = state.doc.read().await;
members::learn_addresses(state, &replica.doc);
+ set_aliases(state, members::aliases(&replica.doc));
let current = members::members(&replica.doc);
// Those that said they are alive within the electorate window.
let alive: BTreeSet<String> = members::electorate(
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)
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 18:06:23 +0000