summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/main.rs2
-rw-r--r--src/swarm/sync.rs16
-rw-r--r--src/transport/libp2p.rs42
-rw-r--r--src/transport/mycelium/daemon.rs17
-rw-r--r--src/transport/mycelium/mod.rs18
-rw-r--r--src/transport/tor.rs8
6 files changed, 48 insertions, 55 deletions
diff --git a/src/main.rs b/src/main.rs
index 708e43d..3bdf0ed 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -353,7 +353,7 @@ async fn main() -> Result<()> {
tokio::select! {
exit_status = state.transport.wait_for_exit() => {
- error!("Mycelium runtime process exited. Status: {:?}", exit_status);
+ error!(status = ?exit_status, "A transport stopped");
let _ = shutdown_tx.send(());
let _ = server_fut.as_mut().await;
}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 011f049..df4bb2d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -39,7 +39,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
return Ok(());
}
let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
- trace!("Peer was forgotten; skipping synchronization");
+ trace!("Peer was forgotten; skipping sync");
return Ok(());
};
@@ -50,7 +50,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
if matches!(peer_state.liveness, Liveness::Dormant { .. }) {
// Reactivation re-handshakes, and PeerValidated then restarts sync.
- trace!("Skipping synchronization with dormant peer");
+ trace!("Skipping sync with dormant peer");
return Ok(());
}
@@ -59,7 +59,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
.is_some_and(|epoch| epoch != state.local_epoch())
{
// Unrelated document histories must never be synced.
- trace!("Skipping synchronization with peer on another epoch");
+ trace!("Skipping sync with peer on another epoch");
return Ok(());
}
@@ -222,7 +222,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
tokio::spawn(async move {
if let Err(error) = send_generated(&state, &peer, &peer_state_mutex, session, encoded).await
{
- warn!(peer = %peer, error = %format!("{error:#}"), "Synchronization processing failed");
+ warn!(peer = %peer, error = %format!("{error:#}"), "Sync processing failed");
}
});
Ok(())
@@ -393,7 +393,7 @@ pub(crate) async fn handle_sync(
peer_state
.sync
.receive(&mut replica.doc, message)
- .context("Automerge synchronization failed")?;
+ .context("Automerge sync failed")?;
peer_state.last_sync_rx = Some(Instant::now());
if let Some(wait) = &peer_state.sync.reply {
throughput
@@ -551,7 +551,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
let mut dirty_peers = HashSet::<String>::new();
let mut document_dirty = false;
- info!("Starting synchronization worker");
+ info!("Starting the sync worker");
loop {
if dirty_peers.is_empty() && !document_dirty {
@@ -567,7 +567,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
}
},
None => {
- warn!("Synchronization event channel closed");
+ warn!("Sync event channel closed");
return;
}
}
@@ -617,7 +617,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
warn!(
peer = %peer,
error = %format!("{error:#}"),
- "Synchronization processing failed"
+ "Sync processing failed"
);
}
}
diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index 30fdb23..8655f23 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -512,7 +512,7 @@ impl Libp2pTransport {
},
stopping,
));
- info!(peer_id = %peer_id, "libp2p transport started");
+ info!(peer_id = %peer_id, "Transport started");
Ok(Self {
local_id: peer_id.to_string(),
peer_id,
@@ -802,11 +802,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 libp2p relay");
+ info!(relay = %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 libp2p relay");
+ debug!(relay = %relay, error = %error, "Couldn't use a member as relay");
}
}
}
@@ -861,7 +861,7 @@ async fn run(
_ = prune.tick() => {
let forgotten = lock(&book).prune(tokio::time::Instant::now(), |peer| swarm.is_connected(peer));
if forgotten > 0 {
- debug!(forgotten, "Forgot libp2p peers nobody mentions any more");
+ debug!(forgotten, "Forgot peers nobody mentions any more");
}
}
_ = dht_bootstrap.tick(), if relaying.public => {
@@ -924,7 +924,7 @@ async fn run(
}
}
_ => {
- warn!(relay = %relay, "Couldn't reach a libp2p relay; trying again later");
+ warn!(relay = %relay, "Couldn't reach a relay; trying again later");
tokio::time::sleep(RELAY_RETRY).await;
let _ = again.send(Command::Reserve { relay, configured });
}
@@ -937,11 +937,11 @@ async fn run(
};
match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) {
Ok(listener) => {
- debug!(relay = %relay, "Reserving a slot on a libp2p relay");
+ debug!(relay = %relay, "Reserving a slot on a relay");
reservations.insert(listener, (configured, relay_peer));
}
Err(error) => {
- warn!(relay = %relay, error = %error, "Couldn't use a libp2p relay; trying again later");
+ warn!(relay = %relay, error = %error, "Couldn't use a relay; trying again later");
reserve_later(configured);
}
}
@@ -972,7 +972,7 @@ async fn run(
dialled = endpoint.is_dialer(),
address = %endpoint.get_remote_address(),
relayed,
- "libp2p connection opened"
+ "Connection opened"
);
}
for done in connecting.remove(&peer_id).into_iter().flatten() {
@@ -989,7 +989,7 @@ async fn run(
peer = %peer_id,
connection = %connection_id,
cause = ?cause,
- "libp2p connection closed"
+ "Connection closed"
);
}
SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => {
@@ -1009,7 +1009,7 @@ async fn run(
}
}
SwarmEvent::NewListenAddr { address, .. } => {
- info!(address = %address, "libp2p listening");
+ info!(address = %address, "Listening");
// A relay tells those reserving where it is: its external addresses.
if serving && !is_local(&address) && !is_relayed(&address) {
swarm.add_external_address(address.clone());
@@ -1020,7 +1020,7 @@ async fn run(
}
}
SwarmEvent::ExpiredListenAddr { address, .. } => {
- info!(address = %address, "libp2p no longer listening");
+ info!(address = %address, "No longer listening");
if serving {
swarm.remove_external_address(&address);
}
@@ -1039,7 +1039,7 @@ async fn run(
book.counts.slots_not_granted += 1;
}
}
- warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later");
+ warn!(relay = %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
@@ -1068,7 +1068,7 @@ async fn run(
}
}
SwarmEvent::ListenerError { error, .. } => {
- warn!(error = %error, "libp2p listener failed");
+ warn!(error = %error, "Listener failed");
}
SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. },
@@ -1082,13 +1082,13 @@ async fn run(
}
}
if !renewal {
- info!(relay = %relay_peer_id, "Reachable through a libp2p relay");
+ info!(relay = %relay_peer_id, "Reachable through a relay");
}
}
SwarmEvent::Behaviour(BehaviourEvent::Autonat(
::libp2p::autonat::Event::StatusChanged { old, new },
)) => {
- info!(old = ?old, new = ?new, "libp2p reachability (AutoNAT)");
+ info!(old = ?old, new = ?new, "Reachability (AutoNAT)");
{
use ::libp2p::autonat::NatStatus;
let mut book = lock(&book);
@@ -1103,7 +1103,7 @@ async fn run(
nat = new;
}
SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => {
- debug!(event = ?event, "libp2p relay");
+ debug!(event = ?event, "Relay event");
}
SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
Ok(_) => {
@@ -1122,7 +1122,7 @@ 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, "libp2p peer identified again");
+ trace!(peer = %peer_id, connection = %connection_id, "Peer identified again");
} else {
debug!(
peer = %peer_id,
@@ -1130,7 +1130,7 @@ async fn run(
listen = ?info.listen_addrs,
observed = %info.observed_addr,
agent = %info.agent_version,
- "libp2p peer identified"
+ "Peer identified"
);
}
// Last first: `remember` puts each in front.
@@ -1163,7 +1163,7 @@ async fn run(
if peer == relaying.local {
continue;
}
- debug!(peer = %peer, address = %address, "libp2p peer on the LAN");
+ debug!(peer = %peer, address = %address, "Peer on the LAN");
let continuum = {
let mut book = lock(&book);
book.remember(peer, address.clone());
@@ -1404,7 +1404,7 @@ impl Transport for Libp2pTransport {
let _ = sender.send((peer, body)).await;
}
Err(error) => {
- debug!(peer = %peer, error = %format!("{error:#}"), "Dropping an unreadable libp2p stream");
+ debug!(peer = %peer, error = %format!("{error:#}"), "Dropping an unreadable stream");
}
}
drop(permit);
@@ -1419,7 +1419,7 @@ impl Transport for Libp2pTransport {
}
}
Err(error) => {
- debug!(peer = %peer, error = %error, "Dropping a libp2p message that doesn't decode");
+ debug!(peer = %peer, error = %error, "Dropping a message that doesn't decode");
}
}
}
diff --git a/src/transport/mycelium/daemon.rs b/src/transport/mycelium/daemon.rs
index f986d52..58ebe0c 100644
--- a/src/transport/mycelium/daemon.rs
+++ b/src/transport/mycelium/daemon.rs
@@ -66,19 +66,14 @@ impl MyceliumSetup {
pub(crate) async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> {
// TODO warn for all ignored options if rest_url_opt is set
if let Some(url) = config.rest_url_opt {
- tracing::info!(
- "Configured to interface with an external standalone Mycelium instance at: {}",
- url
- );
+ tracing::info!(url = %url, "Using a daemon running elsewhere");
return Ok(Self {
runtime: MyceliumRuntime::External,
rest_url: url,
});
}
- tracing::info!(
- "Configured to initialize an embedded user-space Mycelium background worker"
- );
+ tracing::info!("Starting the supervised daemon");
let key = match config.key_file_opt {
Some(key_file) => KeySource::Path(key_file),
@@ -101,7 +96,7 @@ impl MyceliumSetup {
let supervised_node =
SupervisedMycelium::spawn(key, private_net_config, config.tcp_port, &config.peers)
.await
- .context("Failed to securely bootstrap embedded Mycelium background worker")?;
+ .context("Starting the supervised Mycelium daemon")?;
let resolved_url = Url::parse(&format!(
"http://127.0.0.1:{}",
@@ -204,7 +199,7 @@ impl SupervisedMycelium {
}
// Own process group: a Ctrl-C in the terminal goes to continuum only,
// which then stops Mycelium itself, instead of both exiting at once (which
- // shows up as "Mycelium runtime process exited").
+ // shows up as "A transport stopped").
if libc::setpgid(0, 0) != 0 {
return Err(std::io::Error::last_os_error());
}
@@ -252,9 +247,7 @@ impl SupervisedMycelium {
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
- let mut child = cmd
- .spawn()
- .context("Failed to execute mycelium process daemon")?;
+ let mut child = cmd.spawn().context("Running `mycelium`")?;
// Its output goes where ours does.
if let Some(mut stdout) = child.stdout.take() {
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 43d14a8..61aab85 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -144,11 +144,11 @@ impl MyceliumTransport {
Ok(response) if response.status() == StatusCode::NO_CONTENT => break,
Ok(response) if response.status().is_success() => discarded += 1,
Ok(response) => {
- tracing::warn!(status = %response.status(), "Couldn't drain the Mycelium inbox");
+ tracing::warn!(status = %response.status(), "Couldn't drain the inbox");
break;
}
Err(error) => {
- tracing::warn!(error = %error, "Couldn't drain the Mycelium inbox");
+ tracing::warn!(error = %error, "Couldn't drain the inbox");
break;
}
}
@@ -274,7 +274,7 @@ impl Transport for MyceliumTransport {
let mut other_versions = std::collections::HashSet::new();
tracing::info!(
endpoint = self.messages_endpoint.as_str(),
- "Starting Mycelium message consumer"
+ "Starting the consumer"
);
// Several polls at once, so one message's HTTP round trip overlaps the next one's
@@ -351,7 +351,7 @@ async fn poll_inbox(
tx: tokio::sync::mpsc::Sender<InboundMessage>,
) {
loop {
- tracing::trace!("Polling Mycelium inbox");
+ tracing::trace!("Polling the inbox");
let response = match client
.get(endpoint.clone())
@@ -362,7 +362,7 @@ async fn poll_inbox(
{
Ok(resp) => resp,
Err(error) => {
- tracing::warn!(error = %error, "Mycelium inbox poll failed");
+ tracing::warn!(error = %error, "Polling the inbox failed");
tokio::time::sleep(Duration::from_millis(250)).await;
continue;
}
@@ -371,13 +371,13 @@ async fn poll_inbox(
let status = response.status();
if status == StatusCode::NO_CONTENT {
- tracing::trace!("Mycelium inbox is empty");
+ tracing::trace!("The inbox is empty");
tokio::time::sleep(EMPTY_QUEUE_DELAY).await;
continue;
}
if !status.is_success() {
- tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status");
+ tracing::warn!(status = %status, "Polling the inbox failed");
tokio::time::sleep(Duration::from_millis(500)).await;
continue;
}
@@ -392,7 +392,7 @@ async fn poll_inbox(
continue;
}
Err(error) => {
- tracing::warn!(error = %error, "Failed to read raw Mycelium response stream");
+ tracing::warn!(error = %error, "Couldn't read a message from the inbox");
tokio::time::sleep(Duration::from_millis(100)).await;
continue;
}
@@ -404,7 +404,7 @@ async fn poll_inbox(
tracing::error!(
error = %error,
raw_payload = %String::from_utf8_lossy(&raw_bytes),
- "Failed to decode inbound Mycelium message structure"
+ "Dropping a message that doesn't decode"
);
tokio::time::sleep(Duration::from_millis(500)).await;
continue;
diff --git a/src/transport/tor.rs b/src/transport/tor.rs
index d4a5604..643e628 100644
--- a/src/transport/tor.rs
+++ b/src/transport/tor.rs
@@ -203,7 +203,7 @@ impl TorTransport {
let _ = stopping.send(true);
});
}
- info!(address = %onion, "Tor transport started (the onion service may take a minute to be reachable)");
+ info!(address = %onion, "Transport started (the onion service may take a minute to be reachable)");
Ok(Self {
shared,
client,
@@ -335,11 +335,11 @@ impl Shared {
writer: tokio::sync::Mutex::new(writer),
});
lock(&self.connections).insert(key.clone(), connection);
- debug!(peer = %key, connection = id, "Tor connection up");
+ debug!(peer = %key, connection = id, "Connection opened");
let shared = Arc::clone(self);
tokio::spawn(async move {
let error = shared.read(&key, reader).await;
- debug!(peer = %key, connection = id, error = %format!("{error:#}"), "Tor connection closed");
+ debug!(peer = %key, connection = id, error = %format!("{error:#}"), "Connection closed");
shared.forget(&key, id);
});
}
@@ -505,7 +505,7 @@ impl Transport for TorTransport {
}
}
Err(error) => {
- debug!(peer = %peer, error = %error, "Dropping a Tor message that doesn't decode");
+ debug!(peer = %peer, error = %error, "Dropping a message that doesn't decode");
}
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 07:42:54 +0000