From dc013e3cd887da2989176853837cea578c84da57 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 4 Nov 2026 23:38:17 +0100 Subject: one delta at a time --- src/swarm/sync.rs | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) (limited to 'src/swarm/sync.rs') diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 94153ff..24b867b 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -575,14 +575,20 @@ pub(crate) async fn handle_sync( /// at a time: the offer is repeated while it runs. fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: transport::DeltaSpec) { if !state.blobs.start_delta(peer) { - trace!(peer = %peer, "Delta offered again while it downloads"); + debug!(peer = %peer, "Delta offered while another downloads; after it"); return; } let state = state.clone(); let peer = peer.to_owned(); tokio::spawn(async move { let applied = download_and_apply_delta(&state, &peer, epoch, spec).await; - state.blobs.end_delta(&peer); + // Those that offered meanwhile send what is still missing: our next message + // says what we have now. + for other in state.blobs.end_delta(&peer) { + if other != peer { + state.request_sync(&other); + } + } match applied { Ok(changed) => { state.notify_store_changed(); @@ -615,7 +621,14 @@ async fn download_and_apply_delta( .filter(|other| other != peer), ); info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob"); + let started = Instant::now(); let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?; + info!( + peer = %peer, + bytes = bytes.len(), + took_ms = started.elapsed().as_millis() as u64, + "Fetched changes offered as a blob" + ); let doc = Arc::clone(&state.doc); let peer_state = state.get_peer_state(peer).await; -- cgit v1.3.1