From 069debaf62d9ac5fed2e10aab35c31b2d7beeea6 Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 00:02:03 +0100 Subject: delta propagation --- src/swarm/stats.rs | 2 +- src/swarm/sync.rs | 18 +++++++++++++++++- 2 files changed, 18 insertions(+), 2 deletions(-) (limited to 'src') 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 { 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 = + 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")? -- cgit v1.3.1