delta propagation

2 files changed, 18 insertions, 2 deletions

diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index de63962..c322305 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -144,7 +144,7 @@ pub(crate) struct Relay {
     pub(crate) nothing_new_change_bytes: u64,
     pub(crate) nothing_new_bloom_bytes: u64,
     /// Changes from elsewhere that came not along a writer's tree (pulled, with an
-    /// answer, sent again), and their delays summed (as `Propagation`).
+    /// answer, sent again, fetched as a delta), and their delays summed (as `Propagation`).
     pub(crate) off_tree: u64,
     pub(crate) off_tree_ms: u64,
 }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 24b867b..dcb29a7 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -633,6 +633,10 @@ async fn download_and_apply_delta(
     let doc = Arc::clone(&state.doc);
     let peer_state = state.get_peer_state(peer).await;
     let stats = Arc::clone(&state.stats);
+    // As for sync messages (`receive`): not while catching up.
+    let measure_propagation = state.initial_sync_done.load(Ordering::Acquire);
+    let clock = Arc::clone(&state.clock);
+    let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
     tokio::task::spawn_blocking(move || -> Result<bool> {
         let mut replica = doc.blocking_write();
         if replica.epoch != epoch {
@@ -646,11 +650,23 @@ async fn download_and_apply_delta(
             bail!("{missing} of the offered heads are still missing after loading");
         }
         stats.heads_seen(replica.heads().len());
+        let changed = replica.heads() != heads_before.as_slice();
+        if changed && measure_propagation {
+            // As in `receive`; a delta goes along no writer's tree.
+            let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms;
+            let delays: Vec<i64> =
+                crate::document::changes::commit_times(&replica.doc, &heads_before)
+                    .into_iter()
+                    .map(|time| now_ms - time)
+                    .collect();
+            stats.off_tree(&delays);
+            stats.propagated(delays);
+        }
         if let Some(peer_state) = peer_state {
             // Sync from it made progress: no heal for a while (`keepalive`).
             peer_state.blocking_lock().last_sync_rx = Some(Instant::now());
         }
-        Ok(replica.heads() != heads_before.as_slice())
+        Ok(changed)
     })
     .await
     .context("Delta apply task panicked")?