add delta blobs

11 files changed, 590 insertions, 31 deletions

diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index f81ed4c..ae28c91 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -1264,12 +1264,14 @@ components:
               type: integer
         sync:
           type: object
-          required: [divergedPeers]
+          required: [divergedPeers, largestReceivedBytes]
           properties:
             lastFromNeighbourSecs:
               type: [integer, 'null']
             divergedPeers:
               type: integer
+            largestReceivedBytes:
+              type: integer
         relay:
           type: object
           required: [announceRounds, pulls, pullsQueued, blobRequestsDropped]
diff --git a/src/api/stats.rs b/src/api/stats.rs
index a34d4ed..fabbffd 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -114,6 +114,9 @@ struct SyncStats {
     last_from_neighbour_secs: Option<u64>,
     /// Peers whose heads differ from ours while sync is idle (`peers.md`, healing).
     diverged_peers: usize,
+    /// The largest sync message received since start (its Automerge part). Larger
+    /// transfers come as deltas (blobs).
+    largest_received_bytes: u64,
 }
 
 /// Passing changes on (`swarm::relay`) and serving blobs, since start.
@@ -256,6 +259,7 @@ async fn collect(state: &AppState) -> StatsResponse {
         sync: SyncStats {
             last_from_neighbour_secs: last_from_neighbour.map(|at| at.elapsed().as_secs()),
             diverged_peers,
+            largest_received_bytes: node.largest_sync_received,
         },
         relay: RelayStats {
             announce_rounds: node.relay.announce_rounds,
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 30bb966..f9eb3bf 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -323,6 +323,12 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
         );
     }
     out.single(
+        "largest_sync_received_bytes",
+        "gauge",
+        "The largest sync message received since start; larger transfers come as deltas.",
+        stats.sync.largest_received_bytes as f64,
+    );
+    out.single(
         "diverged_peers",
         "gauge",
         "Peers whose heads differ from ours while sync is idle.",
diff --git a/src/simtest/deltas.rs b/src/simtest/deltas.rs
new file mode 100644
index 0000000..ce74607
--- /dev/null
+++ b/src/simtest/deltas.rs
@@ -0,0 +1,98 @@
+//! Large transfers as deltas (`docs/peers.md`): a node joining a swarm with much
+//! history, and one returning after much was written, get the bulk as a blob, and no
+//! sync message they receive is larger than the threshold.
+//!
+//! The nodes use 1 KiB chunks, so the threshold is 16 KiB, and a few hundred KiB of
+//! writes are well above it.
+
+use std::time::Duration;
+
+use serde_json::json;
+
+use super::{Sim, read, write};
+use crate::node::NodeConfig;
+use crate::state::AppState;
+use crate::swarm::bootstrap::Start;
+
+const SOON: Duration = Duration::from_secs(10);
+/// Catching up a few hundred KiB in 1 KiB chunks.
+const CATCH_UP: Duration = Duration::from_secs(60);
+const CHUNK: u32 = 1024;
+/// 16 chunks (`swarm::blob`).
+const THRESHOLD: u64 = 16 * CHUNK as u64;
+
+fn small_chunks(config: &mut NodeConfig) {
+    config.blob_chunk_bytes = CHUNK;
+}
+
+/// About 1 KB that doesn't compress (Automerge compresses large changes).
+fn noise(seed: u64) -> String {
+    let mut rng = crate::testrand::Rng::new(seed);
+    (0..63)
+        .map(|_| format!("{:016x}", rng.next_u64()))
+        .collect()
+}
+
+/// `count` writes of about 1 KB each, under `prefix`.
+async fn write_much(node: &AppState, prefix: &str, count: usize) {
+    for i in 0..count {
+        write(node, &format!("{prefix}/{i}"), json!(noise(i as u64))).await;
+    }
+}
+
+fn largest_sync_received(node: &AppState) -> u64 {
+    node.stats.snapshot().largest_sync_received
+}
+
+fn blob_bytes_received(node: &AppState) -> u64 {
+    node.traffic.total().received.blob.bytes
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn joining_node_gets_the_history_as_a_delta() {
+    let sim = Sim::new(41);
+    let a = sim.start_with("a", &[], Start::Genesis, small_chunks);
+    sim.wait_ready(&[&a], SOON).await;
+    write_much(&a, "history", 200).await;
+
+    let b = sim.start_with("b", &["a"], Start::Join, small_chunks);
+    sim.wait_ready(&[&b], CATCH_UP).await;
+    sim.wait_converged(&[&a, &b], CATCH_UP).await;
+    assert_eq!(read(&b, "history/199").await, Some(json!(noise(199))));
+
+    assert!(
+        blob_bytes_received(&b) > 100 * 1_000,
+        "{} blob bytes received",
+        blob_bytes_received(&b)
+    );
+    assert!(
+        largest_sync_received(&b) <= THRESHOLD,
+        "a sync message of {} bytes",
+        largest_sync_received(&b)
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn returning_node_gets_what_it_missed_as_a_delta() {
+    let sim = Sim::new(42);
+    let a = sim.start_with("a", &[], Start::Genesis, small_chunks);
+    let b = sim.start_with("b", &["a"], Start::Join, small_chunks);
+    let (a, b) = (&a, &b);
+    sim.wait_ready(&[a, b], SOON).await;
+    write(a, "early", json!(1)).await;
+    sim.wait_converged(&[a, b], SOON).await;
+
+    sim.net.take_down("b");
+    write_much(a, "while/away", 200).await;
+    let blobs_before = blob_bytes_received(b);
+    sim.net.bring_up("b");
+
+    sim.wait_converged(&[a, b], CATCH_UP).await;
+    assert_eq!(read(b, "while/away/199").await, Some(json!(noise(199))));
+    assert!(blob_bytes_received(b) - blobs_before > 100 * 1_000);
+    assert!(
+        largest_sync_received(b) <= THRESHOLD,
+        "a sync message of {} bytes",
+        largest_sync_received(b)
+    );
+}
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index f7af61b..5e9cc2f 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -9,6 +9,7 @@
 //! `RUST_LOG=continuum=debug cargo test simtest -- --nocapture` shows the nodes' logs
 //! (all nodes interleaved; messages name peers by ID).
 
+mod deltas;
 mod remakes;
 mod soak;
 mod timing;
diff --git a/src/state/session.rs b/src/state/session.rs
index 618d65b..6efe538 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -52,6 +52,11 @@ pub(crate) struct SyncSession {
     /// The wait after the next refusal.
     pub(crate) send_backoff: Backoff,
 
+    /// We offered the peer a delta instead of a message too large to send
+    /// (`withdraw_for_delta`): no more messages until it answers, or the answer is
+    /// overdue (then the offer goes out again).
+    pub(crate) delta_offered: bool,
+
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
     /// How long to wait for it; doubles while answers stay out.
@@ -80,6 +85,7 @@ impl SyncSession {
             sending: None,
             retry_at: None,
             send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
+            delta_offered: false,
             reply: None,
             reply_backoff,
         }
@@ -134,9 +140,30 @@ impl SyncSession {
         doc: &mut AutoCommit,
         message: sync::Message,
     ) -> Result<(), AutomergeError> {
+        self.delta_offered = false;
         doc.sync().receive_sync_message(&mut self.state, message)
     }
 
+    /// The message just generated is too large to send in one piece: take back what it
+    /// carried, as if it had never been generated, and offer a delta instead. The
+    /// message being handed over becomes the offer (it carries nothing); the answer is
+    /// still awaited, and if it doesn't come, the offer is made again. Returns the
+    /// peer's heads as far as we know them: where the delta starts.
+    pub(crate) fn withdraw_for_delta(&mut self) -> Vec<automerge::ChangeHash> {
+        if let Some(sending) = &mut self.sending {
+            let sent = std::mem::take(&mut sending.sent);
+            for hash in &sent {
+                self.state.sent_hashes.remove(hash);
+            }
+            if let Some(wait) = &mut self.reply {
+                wait.sent.retain(|hash| !sent.contains(hash));
+            }
+        }
+        self.state.in_flight = false;
+        self.delta_offered = true;
+        self.state.their_heads.clone().unwrap_or_default()
+    }
+
     /// Whether `doc` holds everything the peer had when it last told us its heads: they
     /// are known, not empty (empty heads are "contained" in every document, so they
     /// prove nothing), and all in `doc`.
@@ -282,6 +309,7 @@ impl SyncSession {
             self.state.sent_hashes.remove(hash);
         }
         self.state.in_flight = false;
+        self.delta_offered = false;
         self.reply_backoff.grow();
         Some(wait.sent.len())
     }
@@ -323,4 +351,51 @@ mod tests {
         assert!(!new.send_succeeded(new.id()));
         assert!(new.sending.is_none());
     }
+
+    /// A message too large to send is taken back: the peer, after the offer and an
+    /// overdue answer, gets the same changes in the next message.
+    #[test]
+    fn withdrawing_for_a_delta_keeps_the_changes_for_later() {
+        use automerge::{ROOT, transaction::Transactable};
+
+        // 16 KiB of hex, about 8 KiB of information: Automerge compresses what it sends
+        // (a peer with nothing gets the document saved), so this stays above 4 KB.
+        let mut rng = crate::testrand::Rng::new(1);
+        let big: String = (0..1024)
+            .map(|_| format!("{:016x}", rng.next_u64()))
+            .collect();
+        let mut ours = AutoCommit::new();
+        ours.put(ROOT, "big", big).unwrap();
+        ours.commit();
+        let mut theirs = AutoCommit::new();
+        let (mut a, mut b) = (session(), session());
+        let now = Instant::now();
+        let bytes = SyncBytes::default();
+
+        // a opens, b answers with its (empty) heads, then a sends the changes.
+        let opening = a.generate(&mut ours, now, bytes).unwrap();
+        a.send_succeeded(a.id());
+        b.receive(&mut theirs, opening).unwrap();
+        let answer = b.generate(&mut theirs, now, bytes).unwrap();
+        b.send_succeeded(b.id());
+        a.receive(&mut ours, answer).unwrap();
+        let carrying = a.generate(&mut ours, now, bytes).unwrap();
+        let carried = carrying.clone().encode().len();
+        assert!(carried > 4_000, "the changes took {carried} bytes");
+
+        let from = a.withdraw_for_delta();
+        assert!(from.is_empty());
+        assert!(a.delta_offered);
+        assert!(!a.send_succeeded(a.id()));
+
+        // No answer to the offer: it lapses, and the changes go out again.
+        assert!(
+            a.take_overdue_reply(now + Duration::from_secs(3600))
+                .is_some()
+        );
+        assert!(!a.delta_offered);
+        let again = a.generate(&mut ours, now, bytes).unwrap();
+        let carried_again = again.encode().len();
+        assert!(carried_again > 4_000, "sent again in {carried_again} bytes");
+    }
 }
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() {
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<SavedSize>,
     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<Option<transport::EncodedMessage>>) {
                 // 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<bool> {
+    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<bool> {
+        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<SyncEvent>) {
     let mut dirty_peers = HashSet::<String>::new();
     let mut document_dirty = false;
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index c109b0c..3bc88e4 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -147,6 +147,9 @@ pub(crate) enum BlobId {
     Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash),
     /// Bytes by their BLAKE3 hash: e.g. the content of a remake.
     Content(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]),
+    /// The changes of a `DeltaSpec`, by its ID (`DeltaSpec::id`). The requests carry
+    /// the spec, so any node holding its `to` can serve it.
+    Delta(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]),
 }
 
 impl std::fmt::Display for BlobId {
@@ -157,6 +160,10 @@ impl std::fmt::Display for BlobId {
                 write!(f, "content:")?;
                 hash.iter().try_for_each(|byte| write!(f, "{byte:02x}"))
             }
+            BlobId::Delta(id) => {
+                write!(f, "delta:")?;
+                id.iter().try_for_each(|byte| write!(f, "{byte:02x}"))
+            }
         }
     }
 }
@@ -176,8 +183,12 @@ pub(crate) struct BlobMessage {
 
 #[derive(Clone, Serialize, Deserialize)]
 pub(crate) enum BlobBody {
-    /// Asks for these chunks; for the manifest if empty.
-    Request { chunks: Vec<u32> },
+    /// Asks for these chunks; for the manifest if empty. `delta`: the spec of a
+    /// `BlobId::Delta`, which the ID alone doesn't say.
+    Request {
+        chunks: Vec<u32>,
+        delta: Option<DeltaSpec>,
+    },
     /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter.
     Manifest {
         size: u64,
@@ -198,9 +209,11 @@ pub(crate) enum BlobBody {
 impl std::fmt::Debug for BlobBody {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            BlobBody::Request { chunks } => {
-                f.debug_struct("Request").field("chunks", chunks).finish()
-            }
+            BlobBody::Request { chunks, delta } => f
+                .debug_struct("Request")
+                .field("chunks", chunks)
+                .field("delta", &delta.is_some())
+                .finish(),
             BlobBody::Manifest {
                 size,
                 chunk_size,
@@ -325,14 +338,59 @@ pub(crate) struct SyncEnvelope {
     /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver
     /// ignores an epoch older than the newest it has seen in the same session.
     pub(crate) epoch_seq: u32,
-    #[serde(with = "crate::transport::wire::codec::sync_message")]
-    pub(crate) message: SyncPayload,
+    pub(crate) body: SyncBody,
     /// The sender was ready when it generated the message: its document is the swarm's,
     /// not one it is still filling (joining, rejoining). Only such a peer can tell a
     /// catching-up node that it has everything; an empty document says nothing else.
     pub(crate) ready: bool,
 }
 
+/// What a sync envelope carries.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) enum SyncBody {
+    /// An Automerge sync message.
+    Message(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload),
+    /// Instead of a message too large to send in one piece: fetch these changes as a
+    /// blob (`BlobId::Delta`), then sync on.
+    Delta(DeltaSpec),
+}
+
+/// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of
+/// a document at `to`. Every node holding `to` produces the same bytes for it
+/// (`swarm::blob::delta_bytes`).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct DeltaSpec {
+    /// Sorted.
+    #[serde(with = "crate::transport::wire::codec::change_hashes")]
+    pub(crate) from: Vec<ChangeHash>,
+    /// Sorted.
+    #[serde(with = "crate::transport::wire::codec::change_hashes")]
+    pub(crate) to: Vec<ChangeHash>,
+}
+
+impl DeltaSpec {
+    pub(crate) fn new(mut from: Vec<ChangeHash>, mut to: Vec<ChangeHash>) -> Self {
+        from.sort_unstable();
+        from.dedup();
+        to.sort_unstable();
+        to.dedup();
+        Self { from, to }
+    }
+
+    /// Names the spec as a blob: SHA-256 over both lists.
+    pub(crate) fn id(&self) -> [u8; 32] {
+        let mut hasher = Sha256::new();
+        hasher.update(b"continuum/delta/1");
+        for heads in [&self.from, &self.to] {
+            hasher.update((heads.len() as u32).to_be_bytes());
+            for head in heads {
+                hasher.update(head.0);
+            }
+        }
+        hasher.finalize().into()
+    }
+}
+
 /// An Automerge sync message in Automerge's own encoding. Decoded only where it is
 /// applied (`decode`), after the sender proved to be a member and off the consumer: a
 /// large message from anyone else costs nothing but its bytes.
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 67d2fb3..009380f 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 2;
+pub(crate) const VERSION: u8 = 3;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -233,6 +233,28 @@ pub(crate) mod codec {
         }
     }
 
+    /// Change hashes, each a 32-byte binary.
+    pub(crate) mod change_hashes {
+        use super::*;
+        use automerge::ChangeHash;
+
+        pub(crate) fn serialize<S: Serializer>(
+            value: &[ChangeHash],
+            s: S,
+        ) -> Result<S::Ok, S::Error> {
+            s.collect_seq(value.iter().map(|hash| Bytes::new(&hash.0)))
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+            d: D,
+        ) -> Result<Vec<ChangeHash>, D::Error> {
+            Vec::<ByteBuf>::deserialize(d)?
+                .iter()
+                .map(|hash| Ok(ChangeHash(fixed(hash, "a change hash")?)))
+                .collect()
+        }
+    }
+
     /// An Automerge sync message, in Automerge's own encoding (left encoded:
     /// `SyncPayload`).
     pub(crate) mod sync_message {
@@ -416,7 +438,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&hello).unwrap()),
-            format!("020097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+            format!("030097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
         );
 
         let reply = ParsedMessage::HelloReply(HelloReply {
@@ -433,7 +455,7 @@ mod tests {
         assert_eq!(
             hex(&encode(&reply).unwrap()),
             format!(
-                "020199a5737761726d91a170{epoch_hex}01c400c2{}05{}",
+                "030199a5737761726d91a170{epoch_hex}01c400c2{}05{}",
                 run("04"),
                 run("06")
             )
@@ -448,7 +470,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&ping).unwrap()),
-            format!("0203959290c400{epoch_hex}01c3c2")
+            format!("0303959290c400{epoch_hex}01c3c2")
         );
     }
 }