diff options
Diffstat (limited to 'src/swarm/sync.rs')
| -rw-r--r-- | src/swarm/sync.rs | 8 |
1 files changed, 8 insertions, 0 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 5be0754..700764f 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -139,6 +139,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let failed = peer_state_mutex.lock().await.failed_whole(Instant::now()); let delta_threshold = state.blobs.delta_threshold_for(rate, failed); + let stats = Arc::clone(&state.stats); // The session the message comes from, for its send's outcome. let (session, generated) = tokio::task::spawn_blocking( move || -> (u64, Result<Option<transport::EncodedMessage>>) { @@ -180,6 +181,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let delivered = level && matches!(body, Some(transport::SyncBody::Message(_))); if delivered { peer_state.sync.assume_delivered(replica.heads()); + if let Some(transport::SyncBody::Message(payload)) = &body { + stats.delivered_sent(payload.0.len()); + } } ( peer_state.sync.id(), @@ -399,6 +403,7 @@ pub(crate) async fn handle_sync( type Applied = (bool, bool, bool, (Vec<String>, bool)); tokio::task::spawn_blocking(move || -> Result<Option<Applied>> { // Before the lock: decoding a large message takes a while. + let payload_len = message.0.len(); let message = crate::document::sync::decode(&message)?; let carried = message.carries_changes(); let mut replica = doc.blocking_write(); @@ -435,6 +440,9 @@ pub(crate) async fn handle_sync( } stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; + if carried && !changed { + stats.nothing_new(payload_len); + } // Whose they are, to pass them on along their writers' trees, however they // came (along a tree, with an answer, pulled). let authors = if changed { |