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/blob.rs | 200 +++++++++++++++++++++++++++++++++++++++++++++++++---- src/swarm/stats.rs | 10 +++ src/swarm/sync.rs | 121 ++++++++++++++++++++++++++++++-- 3 files changed, 312 insertions(+), 19 deletions(-) (limited to 'src/swarm') diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 372d964..679e1c6 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -4,7 +4,9 @@ //! five-minute deadline), and a blob every node holds identically can come from many //! of them. A blob is an Automerge change by its hash (e.g. an epoch's snapshot change, //! which every node on that epoch holds byte for byte), or content by its BLAKE3 hash -//! (e.g. the new document of a `replace` or `import` remake). +//! (e.g. the new document of a `replace` or `import` remake), or the changes between two +//! sets of heads (a delta: what a peer lacks, when one sync message would be too large; +//! `swarm::sync`). //! //! * Serving: a request without chunks gets the manifest (size, chunk size, a BLAKE3 //! hash per chunk); one with chunks gets each chunk as a message of its own. A node @@ -26,14 +28,14 @@ use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; use anyhow::{Result, anyhow, bail}; -use automerge::ChangeHash; +use automerge::{AutoCommit, ChangeHash}; use tokio::sync::{Semaphore, mpsc}; use tokio::time::Instant; use tracing::{debug, info, warn}; use crate::locks::MutexExt; use crate::state::AppState; -use crate::transport::{BlobBody, BlobId, BlobMessage, ParsedMessage}; +use crate::transport::{BlobBody, BlobId, BlobMessage, DeltaSpec, ParsedMessage}; /// Chunk sends in progress at once, over all downloads served. const MAX_SERVING: usize = 8; @@ -76,8 +78,16 @@ pub(crate) struct Blobs { cache: Mutex)>>, /// Answers for downloads in progress, by blob. fetches: Mutex>>, + /// Peers whose offered delta is being downloaded (`swarm::sync`). + deltas: Mutex>, } +/// A sync message larger than this many chunks is sent as a delta blob instead +/// (`swarm::sync`): below it, a blob costs more round trips than it saves; above it, a +/// lost message costs too much to send again whole, and a slow link may not deliver it +/// within Mycelium's deadline. +const DELTA_THRESHOLD_CHUNKS: usize = 16; + struct Prepared { bytes: Arc>, hashes: Vec<[u8; 32]>, @@ -92,9 +102,24 @@ impl Blobs { dropped: Mutex::new((0, None)), cache: Mutex::new(None), fetches: Mutex::new(HashMap::new()), + deltas: Mutex::new(HashSet::new()), } } + /// The largest sync message sent as it is (`DELTA_THRESHOLD_CHUNKS`). + pub(crate) fn delta_threshold(&self) -> usize { + DELTA_THRESHOLD_CHUNKS * self.chunk_size as usize + } + + /// Starts downloading a delta offered by `peer`; false if one from it is running. + pub(crate) fn start_delta(&self, peer: &str) -> bool { + self.deltas.lock_or_recover().insert(peer.to_owned()) + } + + pub(crate) fn end_delta(&self, peer: &str) { + self.deltas.lock_or_recover().remove(peer); + } + fn fetches( &self, ) -> MutexGuard<'_, HashMap>> { @@ -143,7 +168,7 @@ fn chunk_count(size: u64, chunk_size: u32) -> u64 { /// A blob message from `peer`. pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) { match message.body { - BlobBody::Request { chunks } => { + BlobBody::Request { chunks, delta } => { let Ok(permit) = Arc::clone(&state.blobs.answering).try_acquire_owned() else { state.stats.blob_request_dropped(); debug!(peer = %peer, blob = %message.id, "Answering too many blob requests; dropping one"); @@ -160,7 +185,7 @@ pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) { let peer = peer.to_owned(); // Off the inbound loop: answering may send many large messages. tokio::spawn(async move { - serve(&state, &peer, message.id, chunks).await; + serve(&state, &peer, message.id, chunks, delta).await; drop(permit); }); } @@ -178,11 +203,21 @@ pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) { } } -async fn prepare(state: &AppState, id: BlobId) -> Option> { +async fn prepare(state: &AppState, id: BlobId, delta: Option) -> Option> { if let Some(prepared) = state.blobs.cached(id) { return Some(prepared); } let bytes = match id { + BlobId::Delta(delta_id) => { + // The spec must be the one the ID names. + let spec = delta.filter(|spec| spec.id() == delta_id)?; + let doc = Arc::clone(&state.doc); + tokio::task::spawn_blocking(move || delta_bytes(&mut doc.blocking_write().doc, &spec)) + .await + .ok() + .flatten() + .map(Arc::new)? + } BlobId::Change(hash) => { let doc = Arc::clone(&state.doc); tokio::task::spawn_blocking(move || { @@ -225,8 +260,14 @@ async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) { } } -async fn serve(state: &AppState, peer: &str, id: BlobId, chunks: Vec) { - let Some(prepared) = prepare(state, id).await else { +async fn serve( + state: &AppState, + peer: &str, + id: BlobId, + chunks: Vec, + delta: Option, +) { + let Some(prepared) = prepare(state, id, delta).await else { return reply(state, peer, id, BlobBody::Missing).await; }; let chunk_size = state.blobs.chunk_size; @@ -266,13 +307,22 @@ impl Drop for Registration<'_> { } } -async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec) { +async fn request( + state: &AppState, + peer: &str, + id: BlobId, + chunks: Vec, + delta: Option<&DeltaSpec>, +) { let (epoch, epoch_seq) = state.epoch_stamp(); let message = ParsedMessage::Blob(BlobMessage { epoch, epoch_seq, id, - body: BlobBody::Request { chunks }, + body: BlobBody::Request { + chunks, + delta: delta.cloned(), + }, }); if let Err(error) = state.transport.send_message(peer, &message).await { debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob request failed"); @@ -301,6 +351,7 @@ pub(crate) async fn fetch( id: BlobId, peers: Vec, expected_size: Option, + delta: Option<&DeltaSpec>, ) -> Result> { let (sender, mut answers) = mpsc::unbounded_channel(); if state.blobs.fetches().insert(id, sender).is_some() { @@ -331,7 +382,7 @@ pub(crate) async fn fetch( let mut manifest = None; let mut missing: HashSet = HashSet::new(); for peer in &candidates { - request(state, peer, id, Vec::new()).await; + request(state, peer, id, Vec::new(), delta).await; let deadline = Instant::now() + MANIFEST_TIMEOUT; loop { let answer = tokio::time::timeout_at(deadline, answers.recv()).await; @@ -425,7 +476,7 @@ pub(crate) async fn fetch( for index in &batch { source.in_flight.insert(*index, now); } - request(state, &source.peer, id, batch).await; + request(state, &source.peer, id, batch, delta).await; } match tokio::time::timeout(Duration::from_millis(500), answers.recv()).await { @@ -520,7 +571,7 @@ pub(crate) async fn fetch_content( peers: Vec, ) -> Result> { let id = BlobId::Content(hash); - let bytes = fetch(state, id, peers, Some(size)).await?; + let bytes = fetch(state, id, peers, Some(size), None).await?; if *blake3::hash(&bytes).as_bytes() != hash { bail!("downloaded {id} doesn't match its hash"); } @@ -532,7 +583,7 @@ pub(crate) async fn fetch_change( hash: ChangeHash, peers: Vec, ) -> Result { - let bytes = fetch(state, BlobId::Change(hash), peers, None).await?; + let bytes = fetch(state, BlobId::Change(hash), peers, None, None).await?; let change = automerge::Change::from_bytes(bytes) .map_err(|error| anyhow!("downloaded change {hash} doesn't load: {error}"))?; if change.hash() != hash { @@ -541,9 +592,130 @@ pub(crate) async fn fetch_change( Ok(change) } +/// Downloads the changes of `spec` from `peers` (any that hold its `to`). The caller +/// loads them and checks that it then has `to`. +pub(crate) async fn fetch_delta( + state: &AppState, + spec: &DeltaSpec, + peers: Vec, +) -> Result> { + fetch(state, BlobId::Delta(spec.id()), peers, None, Some(spec)).await +} + +/// The bytes of delta `spec` from `doc`: the changes in the history of `spec.to` but +/// not of `spec.from`, each as Automerge encodes it, sorted by hash and concatenated +/// (`AutoCommit::load_incremental` reads that). The same on every node holding `to`, +/// whatever else it has. None if `doc` lacks part of `to`. +pub(crate) fn delta_bytes(doc: &mut AutoCommit, spec: &DeltaSpec) -> Option> { + if spec + .to + .iter() + .any(|head| doc.get_change_by_hash(head).is_none()) + { + return None; + } + // Heads of `from` we don't have can't hide anything we have. + let from: Vec = spec + .from + .iter() + .copied() + .filter(|head| doc.get_change_by_hash(head).is_some()) + .collect(); + let since: HashMap, Vec)> = doc + .get_changes(&from) + .iter() + .map(|change| { + ( + change.hash(), + (change.deps().to_vec(), change.raw_bytes().to_vec()), + ) + }) + .collect(); + // Of those, the ones `to` depends on: what we have beyond `to` stays out. + let mut included: HashSet = HashSet::new(); + let mut pending: Vec = spec.to.clone(); + while let Some(hash) = pending.pop() { + if let Some((deps, _)) = since.get(&hash) + && included.insert(hash) + { + pending.extend(deps.iter().copied()); + } + } + let mut hashes: Vec = included.into_iter().collect(); + hashes.sort_unstable(); + let mut bytes = Vec::new(); + for hash in hashes { + bytes.extend_from_slice(&since[&hash].1); + } + Some(bytes) +} + #[cfg(test)] mod tests { use super::*; + use automerge::{ActorId, ROOT, transaction::Transactable}; + + fn random_write(rng: &mut crate::testrand::Rng, doc: &mut AutoCommit) { + let key = format!("k{}", rng.below(6)); + doc.put(ROOT, key, rng.below(100) as i64).unwrap(); + doc.commit(); + } + + fn sorted(mut heads: Vec) -> Vec { + heads.sort_unstable(); + heads + } + + /// Over random histories: two nodes that reached the same heads through different + /// merge orders, one with more on top, produce the same bytes for a delta to those + /// heads; a document at `from` that loads them is at `to`. + #[test] + fn delta_bytes_are_the_same_everywhere_and_complete() { + crate::testrand::for_seeds(200, |rng| { + let actor = |i: u8| ActorId::from(&[i; 16][..]); + let mut base = AutoCommit::new().with_actor(actor(0)); + for _ in 0..rng.between(0, 5) { + random_write(rng, &mut base); + } + let from = base.get_heads(); + let mut writers: Vec = (1..=rng.between(1, 3)) + .map(|i| base.fork().with_actor(actor(i as u8))) + .collect(); + for _ in 0..rng.between(1, 15) { + let writer = rng.below(writers.len()); + random_write(rng, &mut writers[writer]); + } + let mut one = base.fork().with_actor(actor(10)); + let mut two = base.fork().with_actor(actor(11)); + for writer in writers.iter_mut() { + one.merge(writer).unwrap(); + } + for writer in writers.iter_mut().rev() { + two.merge(writer).unwrap(); + } + let to = one.get_heads(); + assert_eq!(sorted(to.clone()), sorted(two.get_heads())); + // Beyond `to`: must stay out of the delta. + random_write(rng, &mut two); + + let spec = DeltaSpec::new(from.clone(), to.clone()); + let bytes = delta_bytes(&mut one, &spec).unwrap(); + assert_eq!(delta_bytes(&mut two, &spec).unwrap(), bytes); + + let mut behind = base.fork(); + behind.load_incremental(&bytes).unwrap(); + assert_eq!(sorted(behind.get_heads()), sorted(to)); + }); + } + + #[test] + fn no_delta_without_its_heads() { + let mut doc = AutoCommit::new(); + doc.put(ROOT, "k", 1_i64).unwrap(); + doc.commit(); + let spec = DeltaSpec::new(Vec::new(), vec![ChangeHash([7; 32])]); + assert!(delta_bytes(&mut doc, &spec).is_none()); + } #[test] fn chunking() { diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index 8172233..308c2a5 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -31,6 +31,7 @@ struct Inner { stale_hello_replies: u64, relay: Relay, blob_requests_dropped: u64, + largest_sync_received: u64, saved_size: Option, measuring: bool, /// The document version (`AppState::change_tx`) and epoch the saved size is of. @@ -122,6 +123,8 @@ pub(crate) struct Snapshot { pub(crate) stale_hello_replies: u64, pub(crate) relay: Relay, pub(crate) blob_requests_dropped: u64, + /// The largest sync message received (its Automerge part, in bytes). + pub(crate) largest_sync_received: u64, pub(crate) propagation: Propagation, } @@ -181,6 +184,12 @@ impl NodeStats { self.lock().relay.pulls_queued += 1; } + /// A sync message of `bytes` arrived. + pub(crate) fn sync_received(&self, bytes: usize) { + let mut inner = self.lock(); + inner.largest_sync_received = inner.largest_sync_received.max(bytes as u64); + } + /// A blob request went unanswered: too many were being answered. pub(crate) fn blob_request_dropped(&self) { self.lock().blob_requests_dropped += 1; @@ -237,6 +246,7 @@ impl NodeStats { stale_hello_replies: inner.stale_hello_replies, relay: inner.relay, blob_requests_dropped: inner.blob_requests_dropped, + largest_sync_received: inner.largest_sync_received, propagation: inner.propagation, } } 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