From 5a32e746cf7aec3e4ca76616b6b66f023de53988 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 22:08:32 +0100 Subject: add delta blobs --- src/swarm/sync.rs | 121 +++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 116 insertions(+), 5 deletions(-) (limited to 'src/swarm/sync.rs') diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index ca54fae..70ce28e 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -124,6 +124,11 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { // The retry timer enqueues the peer once it's due. return Ok(()); } + if peer_state.sync.delta_offered { + // Its answer, or the reply timeout, brings us back here. + trace!("A delta is offered; waiting for the peer"); + return Ok(()); + } } let (session, encoded) = { @@ -134,13 +139,14 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let ready = state.is_ready().await; // Before this message is counted. let bytes_before = SyncBytes::of(&state.traffic, peer); + let delta_threshold = state.blobs.delta_threshold(); // The session the message comes from, for its send's outcome. let (session, generated) = tokio::task::spawn_blocking( move || -> (u64, Result>) { // Label the message with the epoch of the document it was generated // from, read under the same lock. - let (session, epoch, epoch_seq, message) = { + let (session, epoch, epoch_seq, body) = { let mut replica = doc.blocking_write(); let mut peer_state = psm.blocking_lock(); let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); @@ -150,16 +156,33 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { peer_state .sync .generate(&mut replica.doc, Instant::now(), bytes_before); - (peer_state.sync.id(), epoch, epoch_seq, message) + let body = message.map(|message| { + let payload = transport::SyncPayload::encode(message); + if payload.0.len() <= delta_threshold { + return transport::SyncBody::Message(payload); + } + // Too large for one message: all or nothing, and on a slow link + // within Mycelium's deadline. The peer fetches it in chunks. + let from = peer_state.sync.withdraw_for_delta(); + let spec = transport::DeltaSpec::new(from, replica.heads().to_vec()); + info!( + message_bytes = payload.0.len(), + from = spec.from.len(), + to = spec.to.len(), + "Sync message too large; offering the changes as a blob" + ); + transport::SyncBody::Delta(spec) + }); + (peer_state.sync.id(), epoch, epoch_seq, body) }; - let Some(message) = message else { + let Some(body) = body else { return (session, Ok(None)); }; let envelope = transport::SyncEnvelope { swarm_id, epoch, epoch_seq, - message: transport::SyncPayload::encode(message), + body, ready, }; let encoded = transport @@ -316,6 +339,14 @@ pub(crate) async fn handle_sync( let epoch_label = envelope.epoch; let from_ready = envelope.ready; + let payload = match envelope.body { + transport::SyncBody::Message(payload) => payload, + transport::SyncBody::Delta(spec) => { + fetch_delta(state, peer, envelope.epoch, spec); + return Ok(()); + } + }; + state.stats.sync_received(payload.0.len()); let applied = { let doc = Arc::clone(&state.doc); @@ -329,7 +360,7 @@ pub(crate) async fn handle_sync( // While catching up, everything arrives late; that says nothing about propagation. let measure_propagation = !check_caught_up; let epoch = envelope.epoch; - let message = envelope.message; + let message = payload; let peer = peer.to_owned(); let clock = Arc::clone(&state.clock); @@ -427,6 +458,86 @@ pub(crate) async fn handle_sync( Ok(()) } +/// The peer offered the changes we lack as a blob (`SyncBody::Delta`): fetch them, from +/// it and any other peer that has them, apply them, and sync on. One download per peer +/// 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"); + 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); + match applied { + Ok(changed) => { + state.notify_store_changed(); + if changed { + crate::swarm::relay::announce(&state, &peer).await; + } + } + Err(error) => { + warn!(peer = %peer, error = %format!("{error:#}"), "Delta download failed; syncing on"); + } + } + // Our next message says what we have now; the peer answers with the rest. + let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(peer)); + }); +} + +/// Returns whether our heads changed. +async fn download_and_apply_delta( + state: &AppState, + peer: &str, + epoch: transport::Epoch, + spec: transport::DeltaSpec, +) -> Result { + let mut peers = vec![peer.to_owned()]; + peers.extend( + state + .active_peers_on_local_epoch() + .await + .into_iter() + .filter(|other| other != peer), + ); + info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob"); + let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?; + + let doc = Arc::clone(&state.doc); + let peer_state = state.get_peer_state(peer).await; + let stats = Arc::clone(&state.stats); + tokio::task::spawn_blocking(move || -> Result { + let mut replica = doc.blocking_write(); + if replica.epoch != epoch { + bail!("the epoch changed during the download"); + } + let heads_before = replica.heads().to_vec(); + replica + .doc + .load_incremental(&bytes) + .context("Loading the downloaded changes")?; + replica.refresh_heads(); + let missing = spec + .to + .iter() + .filter(|head| replica.doc.get_change_by_hash(head).is_none()) + .count(); + if missing > 0 { + bail!("{missing} of the offered heads are still missing after loading"); + } + stats.heads_seen(replica.heads().len()); + 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()) + }) + .await + .context("Delta apply task panicked")? +} + pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver) { let mut dirty_peers = HashSet::::new(); let mut document_dirty = false; -- cgit v1.3.1