add delta blobs

1 files changed, 186 insertions, 14 deletions

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<Option<(BlobId, Arc<Prepared>)>>,
     /// Answers for downloads in progress, by blob.
     fetches: Mutex<HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>>,
+    /// Peers whose offered delta is being downloaded (`swarm::sync`).
+    deltas: Mutex<HashSet<String>>,
 }
 
+/// 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<Vec<u8>>,
     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<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>> {
@@ -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<Arc<Prepared>> {
+async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Option<Arc<Prepared>> {
     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<u32>) {
-    let Some(prepared) = prepare(state, id).await else {
+async fn serve(
+    state: &AppState,
+    peer: &str,
+    id: BlobId,
+    chunks: Vec<u32>,
+    delta: Option<DeltaSpec>,
+) {
+    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<u32>) {
+async fn request(
+    state: &AppState,
+    peer: &str,
+    id: BlobId,
+    chunks: Vec<u32>,
+    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<String>,
     expected_size: Option<u64>,
+    delta: Option<&DeltaSpec>,
 ) -> Result<Vec<u8>> {
     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<String> = 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<String>,
 ) -> Result<Vec<u8>> {
     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<String>,
 ) -> Result<automerge::Change> {
-    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<String>,
+) -> Result<Vec<u8>> {
+    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<Vec<u8>> {
+    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<ChangeHash> = spec
+        .from
+        .iter()
+        .copied()
+        .filter(|head| doc.get_change_by_hash(head).is_some())
+        .collect();
+    let since: HashMap<ChangeHash, (Vec<ChangeHash>, Vec<u8>)> = 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<ChangeHash> = HashSet::new();
+    let mut pending: Vec<ChangeHash> = 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<ChangeHash> = 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<ChangeHash>) -> Vec<ChangeHash> {
+        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<AutoCommit> = (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() {