1 files changed, 18 insertions, 1 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 09e10f4..3284015 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -14,6 +14,7 @@ use tracing::{debug, info, trace, warn};
 use crate::state::{AppState, Liveness, Sending, Standing};
 use crate::swarm::peer::{peer_swarm, send_hello_once};
 use crate::swarm::source;
+use crate::swarm::throughput::SyncBytes;
 use crate::transport::{self, ParsedMessage};
 
 #[derive(Debug)]
@@ -127,6 +128,8 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
         let transport = Arc::clone(&state.transport);
         let peer_owned = peer.to_owned();
         let ready = state.is_ready().await;
+        // Before this message is counted.
+        let bytes_before = SyncBytes::of(&state.traffic, peer);
 
         let generated =
             tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> {
@@ -157,7 +160,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     if in_step {
                         peer_state.sync.reply_settled();
                     } else if message.is_some() {
-                        peer_state.sync.expect_reply(sent.clone(), Instant::now());
+                        peer_state
+                            .sync
+                            .expect_reply(sent.clone(), Instant::now(), bytes_before);
                     }
                     if message.is_some() {
                         // Marked under the same lock as the generation, so no second
@@ -255,6 +260,8 @@ pub(crate) async fn handle_sync(
     envelope: transport::SyncEnvelope,
 ) -> Result<()> {
     debug!(peer = %peer, "Processing sync envelope");
+    // This message is counted already (`transport::traffic`).
+    let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer));
 
     let peer_state_mutex = state.get_or_create_peer_state(peer).await;
 
@@ -306,6 +313,10 @@ pub(crate) async fn handle_sync(
         let doc = Arc::clone(&state.doc);
         let peer_state_mutex = Arc::clone(&peer_state_mutex);
         let stats = Arc::clone(&state.stats);
+        let throughput = Arc::clone(&state.throughput);
+        let rtt = state
+            .lock_rtt()
+            .effective(peer, Instant::now(), state.timing.rtt_fresh_for());
         let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
         // While catching up, everything arrives late; that says nothing about propagation.
         let measure_propagation = !check_caught_up;
@@ -334,6 +345,12 @@ pub(crate) async fn handle_sync(
                 .receive_sync_message(&mut peer_state.sync.state, message)
                 .context("Automerge synchronization failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
+            if let Some(wait) = &peer_state.sync.reply {
+                throughput
+                    .lock()
+                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .exchange(&peer, wait.started, arrived, rtt);
+            }
             peer_state.sync.reply_settled();
 
             resend_if_needed(&peer, &mut peer_state.sync.state);