cleanup logging

5 files changed, 69 insertions, 56 deletions

diff --git a/src/bootstrap.rs b/src/bootstrap.rs
index df47ff6..470ca67 100644
--- a/src/bootstrap.rs
+++ b/src/bootstrap.rs
@@ -4,7 +4,7 @@ use std::sync::atomic::Ordering;
 use std::time::Duration;
 
 use anyhow::{bail, Result};
-use tracing::{debug, error, info, instrument, warn};
+use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
 use crate::peer::{remember_peer, send_hello_once};
@@ -13,7 +13,6 @@ use crate::state::{AppState, PeerMembership};
 const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300);
 const BOOTSTRAP_RETRY: Duration = Duration::from_millis(500);
 
-#[instrument(skip(state, seeds))]
 pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> {
     if seeds.is_empty() {
         info!("Initializing genesis node (no seed peers configured)");
diff --git a/src/main.rs b/src/main.rs
index f5cec55..44f9dd4 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -15,7 +15,7 @@ use anyhow::{Context, Result};
 use automerge::AutoCommit;
 use clap::Parser;
 use tokio::sync::{broadcast, mpsc, watch, RwLock};
-use tracing::{error, info};
+use tracing::{debug, error, info, trace};
 use url::Url;
 
 mod api;
@@ -114,12 +114,15 @@ async fn main() -> Result<()> {
 
     info!(
         bind_addr = args.bind_addr.to_string(),
-        mycelium_rest_url = mycelium_rest_url.to_string(),
-        "Starting Automerge replication service"
+        "Starting replication service"
+    );
+    debug!(
+        url = mycelium_rest_url.to_string(),
+        "Using Mycelium REST API"
     );
 
     let seeds: Vec<String> = args.seeds.unwrap_or_default();
-    info!(seed_count = seeds.len(), "Loaded bootstrap configuration");
+    debug!(seed_count = seeds.len(), "Loaded bootstrap configuration");
 
     let (sync_tx, sync_rx) = mpsc::unbounded_channel();
     let (change_tx, _) = watch::channel(0_u64);
@@ -133,12 +136,12 @@ async fn main() -> Result<()> {
         None => Epoch::GENESIS,
     };
     let (epoch_tx, _) = watch::channel(initial_epoch);
-    info!(epoch = %initial_epoch, "Local epoch");
+    debug!(epoch = %initial_epoch, "Local epoch");
 
     let firewall = Arc::new(Firewall::default());
 
     let mycelium_transport = Arc::new(mycelium::MyceliumTransport::connect(mycelium_setup).await?);
-    info!(local_id = mycelium_transport.local_id(), "Transport ready");
+    debug!(local_id = mycelium_transport.local_id(), "Transport ready");
 
     let state = AppState {
         doc: Arc::new(RwLock::new(Replica::new(initial_epoch, AutoCommit::new()))),
@@ -160,7 +163,7 @@ async fn main() -> Result<()> {
         compaction: Arc::new(compaction::Compaction::default()),
     };
 
-    info!("Application state initialized");
+    trace!("Application state initialized");
 
     let consumer_state = state.clone();
     tokio::spawn(async move {
@@ -173,25 +176,25 @@ async fn main() -> Result<()> {
             }))
             .await;
     });
-    info!("Mycelium consumer task spawned");
+    trace!("Mycelium consumer task spawned");
 
     let sync_state = state.clone();
     tokio::spawn(async move {
         sync::run_sync_worker(sync_state, sync_rx).await;
     });
-    info!("Sync worker task spawned");
+    trace!("Sync worker task spawned");
 
     let peer_worker_state = state.clone();
     tokio::spawn(async move {
         peer::run_peer_worker(peer_worker_state).await;
     });
-    info!("Peer worker task spawned");
+    trace!("Peer worker task spawned");
 
     // Switches to committed epochs and catches up with newer ones. Starts before
     // bootstrapping: a node joining a compacted cluster can only finish its initial sync
     // after fetching the current epoch from a peer.
     tokio::spawn(compaction::run_worker(state.clone()));
-    info!("Compaction worker task spawned");
+    trace!("Compaction worker task spawned");
 
     bootstrap::bootstrap(&state, &seeds)
         .await
diff --git a/src/mycelium.rs b/src/mycelium.rs
index e623659..d2deebe 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -42,14 +42,16 @@ struct InboundMessage {
     payload: String,
 }
 
-#[derive(Debug, Serialize)]
+// Deserialize only for debugging
+#[derive(Debug, Serialize, Deserialize)]
 struct OutboundMessage {
     dst: Destination,
     topic: String,
     payload: String,
 }
 
-#[derive(Debug, Serialize)]
+// Deserialize only for debugging
+#[derive(Debug, Serialize, Deserialize)]
 struct Destination {
     pk: String,
 }
@@ -554,7 +556,7 @@ impl Transport for MyceliumTransport {
     }
 
     async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
-        tracing::debug!(payload_bytes = message.body.len(), "Sending message");
+        tracing::trace!(payload_bytes = message.body.len(), "Sending message");
         let response = self
             .client
             .post(self.messages_endpoint.clone())
@@ -576,16 +578,28 @@ impl Transport for MyceliumTransport {
         // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing
         // a message that never arrived.
         if tracing::enabled!(tracing::Level::DEBUG) {
-            if let Ok(body) = response
+            let Ok(body) = response
                 .bytes()
                 .await
                 .map_err(anyhow::Error::from)
                 .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?))
-            {
-                if let Some(id) = body.get("id").and_then(|id| id.as_str()) {
-                    tracing::debug!(peer = %message.peer, message_id = id, "Mycelium accepted message");
-                }
-            }
+            else {
+                tracing::debug!(to = %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)");
+                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)");
+                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)");
+                return Ok(());
+            };
+            tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message");
         }
         Ok(())
     }
@@ -651,14 +665,15 @@ impl Transport for MyceliumTransport {
                 }
             };
 
-            tracing::debug!(
-                id = %inbound_message.id,
-                payload_bytes = raw_bytes.len(),
-                "Received inbound Mycelium message"
-            );
-
             match parse_message(&inbound_message) {
                 Ok((sender, parsed)) => {
+                    tracing::debug!(
+                        from = %inbound_message.src_pk,
+                        message = ?parsed,
+                        id = %inbound_message.id,
+                        //payload_bytes = raw_bytes.len(),
+                        "Received message"
+                    );
                     if let Err(error) = handler(sender, parsed).await {
                         tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
                     }
@@ -694,7 +709,8 @@ fn encode_payload(payload: &Vec<u8>) -> String {
     BASE64.encode(payload)
 }
 
-fn decode_payload(topic: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
+fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
+    let topic: &str = &decode_topic(topic_encoded)?;
     let bytes = BASE64.decode(payload)?;
     match topic {
         TOPIC_HELLO => {
@@ -722,8 +738,7 @@ fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), Tr
         return Err(TransportError::NoSender);
     }
 
-    let topic = decode_topic(&message.topic)?;
-    let parsed = decode_payload(&topic, &message.payload)?;
+    let parsed = decode_payload(&message.topic, &message.payload)?;
 
     Ok((sender, parsed))
 }
diff --git a/src/peer.rs b/src/peer.rs
index 54ba0a4..d27cda8 100644
--- a/src/peer.rs
+++ b/src/peer.rs
@@ -148,7 +148,6 @@ async fn heal_if_diverged(
 
 // --- peer discovery and HELLO ------------------------------------------------
 
-#[instrument(skip(state, peer))]
 pub(crate) async fn remember_peer(state: &AppState, peer: &str) {
     if peer.is_empty() {
         warn!("Ignoring empty peer public key");
@@ -165,18 +164,18 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) {
     if !peer_states.contains_key(peer) {
         peer_states.entry(peer.to_owned()).or_default();
         info!(
+            peer = %peer,
             peer_count = peer_states.len(),
-            "Discovered and registered peer"
+            "Discovered peer"
         );
     } else {
-        trace!("Peer already registered");
+        trace!(peer = %peer, "Peer already registered");
     }
 }
 
-#[instrument(skip(state), fields(peer = %peer))]
 pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> {
     let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
-        trace!("Peer was forgotten; not sending HELLO");
+        trace!(peer = %peer, "Peer was forgotten; not sending HELLO");
         return Ok(());
     };
 
@@ -188,7 +187,7 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
             };
             true
         } else {
-            trace!(hello_state = ?peer_state.hello_state, "not sending");
+            trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending");
             false
         }
     };
@@ -197,11 +196,10 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
         return Ok(());
     }
 
-    info!("Sending HELLO");
+    debug!(to = %peer, "Sending HELLO");
     transmit_hello(state, peer, HELLO_RETRY_INITIAL).await
 }
 
-#[instrument(skip(state), fields(peer = %peer))]
 async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
     let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
         return Ok(());
@@ -222,6 +220,7 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
     };
 
     debug!(
+        peer = %peer,
         next_retry_ms = delay.as_millis(),
         "No HELLO_REPLY yet; re-sending HELLO"
     );
@@ -297,7 +296,6 @@ async fn flush_owed_hello_replies(state: &AppState) {
     }
 }
 
-#[instrument(skip(state), fields(peer = %peer))]
 async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
     let cluster_id = state
         .cluster_id()
@@ -331,7 +329,7 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
 
     let ready = state.is_ready().await;
 
-    debug!(ready, "Prepared HELLO_REPLY");
+    debug!(to = %peer, ready, "Sending HELLO_REPLY");
 
     state
         .transport
@@ -339,12 +337,12 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
         .await
         .context("Sending HELLO_REPLY")?;
 
-    debug!("HELLO_REPLY sent");
+    trace!(to = %peer, "HELLO_REPLY sent");
     Ok(())
 }
 
-#[instrument(skip(state), fields(from_peer = %from_peer, advertised_peers = ?advertised_peers))]
 async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[String]) {
+    // TODO here we could add a trace t oshwo advertised_peers
     for discovered_peer in advertised_peers {
         if discovered_peer == from_peer {
             continue;
@@ -432,7 +430,6 @@ pub(crate) async fn run_peer_worker(state: AppState) {
 
 // --- peer membership ---------------------------------------------------------
 
-#[instrument(skip(state), fields(peer = %peer, received_cluster = %cluster_id))]
 async fn establish_peer(
     state: &AppState,
     peer: &str,
@@ -482,12 +479,12 @@ async fn establish_peer(
     if peer_state.hello_state != HelloState::Established {
         peer_state.hello_state = HelloState::Established;
         peer_state.reset_sync();
-        info!("Handshake complete; sync state reset");
+        info!(peer = %peer, "Handshake complete; sync state reset");
     }
 
     let pending = std::mem::take(&mut peer_state.pending_inbound_sync);
 
-    info!(cluster_id = %cluster_id, pending_messages = pending.len(), "Peer membership established");
+    info!(peer = %peer, cluster_id = %cluster_id, pending_messages = pending.len(), "Peer membership established");
     Ok(pending)
 }
 
@@ -563,7 +560,6 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
 
 // --- inbound dispatch --------------------------------------------------------
 
-#[instrument(skip(state, message))]
 pub(crate) async fn handle_message(
     state: &AppState,
     peer: String,
@@ -590,7 +586,7 @@ pub(crate) async fn handle_message(
 
     let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await;
 
-    debug!(message = ?message, "Received message");
+    //debug!(from = %peer, message = ?message, "Received message");
 
     match message {
         transport::ParsedMessage::Hello(_) => {
@@ -626,6 +622,7 @@ pub(crate) async fn handle_message(
 
             if !pending.is_empty() {
                 debug!(
+                    to = %peer,
                     pending_messages = pending.len(),
                     "Replaying pending sync messages"
                 );
diff --git a/src/sync.rs b/src/sync.rs
index 7def081..9db4b40 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -10,7 +10,7 @@ use anyhow::{anyhow, bail, Context, Result};
 use automerge::sync::SyncDoc;
 use tokio::sync::mpsc;
 use tokio::time::Instant;
-use tracing::{debug, info, instrument, trace, warn};
+use tracing::{debug, info, trace, warn};
 
 use crate::peer::{peer_cluster, send_hello_once};
 use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound};
@@ -34,7 +34,6 @@ pub(crate) struct ChangeSet {
     pub(crate) paths: Vec<Vec<String>>,
 }
 
-#[instrument(skip(state), fields(peer = %peer, retry_delivery))]
 async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> {
     let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
         trace!("Peer was forgotten; skipping synchronization");
@@ -155,7 +154,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                 std::mem::take(&mut peer_state.needs_generate)
             };
 
-            debug!(regenerate, "Outbound sync message accepted by Mycelium");
+            trace!(regenerate, "Outbound sync message accepted by Mycelium");
 
             if regenerate {
                 state
@@ -198,13 +197,12 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
     }
 }
 
-#[instrument(skip(state, envelope), fields(peer = %peer, cluster_id = %envelope.cluster_id))]
 pub(crate) async fn handle_sync(
     state: &AppState,
     peer: &str,
     envelope: transport::SyncEnvelope,
 ) -> Result<()> {
-    debug!("Processing sync envelope");
+    debug!(peer = %peer, "Processing sync envelope");
 
     let peer_state_mutex = state.get_or_create_peer_state(peer).await;
 
@@ -221,11 +219,12 @@ pub(crate) async fn handle_sync(
                 if !duplicate {
                     peer_state.pending_inbound_sync.push(envelope.clone());
                     info!(
+                        peer = %peer,
                         pending_count = peer_state.pending_inbound_sync.len(),
                         "Queued sync envelope pending peer membership"
                     );
                 } else {
-                    trace!("Ignoring duplicate pending sync envelope");
+                    trace!(peer = %peer, "Ignoring duplicate pending sync envelope");
                 }
                 true
             }
@@ -316,7 +315,7 @@ pub(crate) async fn handle_sync(
     };
 
     let Some(caught_up) = applied else {
-        debug!(epoch = %epoch_label, "Dropped sync message from another epoch");
+        debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
 
@@ -399,7 +398,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
     let mut retry_peers = HashSet::<String>::new();
     let mut document_dirty = false;
 
-    info!("Starting event-driven synchronization worker");
+    info!("Starting synchronization worker");
 
     loop {
         if dirty_peers.is_empty() && retry_peers.is_empty() && !document_dirty {