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 ++++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 186 insertions(+), 14 deletions(-) (limited to 'src/swarm/blob.rs') 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() { -- cgit v1.3.1