diff options
Diffstat (limited to 'src/sync.rs')
| -rw-r--r-- | src/sync.rs | 15 |
1 files changed, 7 insertions, 8 deletions
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 { |
