1 files changed, 65 insertions, 7 deletions

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.