diff options
Diffstat (limited to 'src/swarm/sync.rs')
| -rw-r--r-- | src/swarm/sync.rs | 18 |
1 files changed, 17 insertions, 1 deletions
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")? |