deflate blob chunks

5 files changed, 45 insertions, 8 deletions

diff --git a/docs/peers.md b/docs/peers.md
index a6b69b4..e098b01 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -172,8 +172,9 @@ sent: it would be all or nothing, sent again whole after a lost copy, and on a s
 may not arrive within the transport's delivery limit (Mycelium's five minutes). The sender takes it back and offers a
 *delta* instead: the changes in the history of its heads but not of the peer's (as far
 as it knows them; of those, the ones it holds, and for any it lacks, the heads they
-share). The peer fetches them as a blob, in chunks, from the sender and any other peer
-holding both sets of heads (every such peer produces the same bytes), loads them,
+share). The peer fetches them as a blob, in chunks (each deflated where that makes it
+at least a tenth smaller), from the sender and any other peer holding both sets of
+heads (every such peer produces the same bytes), loads them,
 and syncs on; the rest goes in ordinary messages. Until the peer answers, the sender
 sends it nothing; if the answer is overdue, it offers again (the same offer, without
 generating the message again, while neither side's heads moved). This is how a join without a
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 11bf3dc..0bd2d37 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -9,7 +9,8 @@
 //! 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
+//!   hash per chunk); one with chunks gets each chunk as a message of its own, deflated
+//!   where that shrinks it (the hash is of the chunk, so any source's matches). A node
 //!   that doesn't hold the blob answers `Missing`. The bytes of the blob served last
 //!   are kept, so serving a download doesn't extract them for every chunk. Serving runs
 //!   in tasks of its own, at most [`MAX_ANSWERING`] requests and [`MAX_SERVING`] chunk
@@ -335,8 +336,19 @@ async fn serve(
         let Ok(_permit) = state.blobs.serving.acquire().await else {
             return;
         };
-        let data = bytes::Bytes::copy_from_slice(&prepared.bytes[start..end]);
-        reply(state, peer, id, BlobBody::Chunk { index, data }).await;
+        let chunk = &prepared.bytes[start..end];
+        // Deflated where it shrinks (`compress`): changes are stored uncompressed.
+        let body = match crate::compress::deflate(chunk) {
+            Some(deflated) => BlobBody::DeflatedChunk {
+                index,
+                data: deflated.into(),
+            },
+            None => BlobBody::Chunk {
+                index,
+                data: bytes::Bytes::copy_from_slice(chunk),
+            },
+        };
+        reply(state, peer, id, body).await;
     }
 }
 
@@ -527,7 +539,18 @@ pub(crate) async fn fetch(
             request(state, &source.peer, id, batch, delta).await;
         }
 
-        match tokio::time::timeout(Duration::from_millis(500), answers.recv()).await {
+        let answer = tokio::time::timeout(Duration::from_millis(500), answers.recv()).await;
+        // A deflated chunk inflated, to no more than a chunk; one that doesn't is bad.
+        let answer = match answer {
+            Ok(Some((from, BlobBody::DeflatedChunk { index, data }))) => {
+                let data = crate::compress::inflate(&data, manifest.chunk_size as usize)
+                    .map(bytes::Bytes::from)
+                    .unwrap_or_default();
+                Ok(Some((from, BlobBody::Chunk { index, data })))
+            }
+            other => other,
+        };
+        match answer {
             Ok(Some((from, BlobBody::Chunk { index, data }))) => {
                 let Some(source) = sources.iter_mut().find(|source| source.peer == from) else {
                     continue;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index c75f66b..1542d4f 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -174,7 +174,8 @@ pub(crate) async fn handle_message(
             // for: the request's round trip is no exchange of sync bytes to time).
             if let Some(delay_ms) = crate::swarm::hop::note(state, &peer, blob.sent).await
                 && delay_ms > 0
-                && let transport::BlobBody::Chunk { data, .. } = &blob.body
+                && let transport::BlobBody::Chunk { data, .. }
+                | transport::BlobBody::DeflatedChunk { data, .. } = &blob.body
             {
                 let rtt =
                     state
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 1e9e5a9..eeb9d34 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -322,6 +322,13 @@ pub(crate) enum BlobBody {
     },
     /// The sender doesn't have the blob.
     Missing,
+    /// A chunk, deflated (`compress`): the receiver inflates it to the chunk, which the
+    /// manifest's hash is of, whoever sent it.
+    DeflatedChunk {
+        index: u32,
+        #[serde(with = "crate::transport::wire::codec::shared_bytes")]
+        data: bytes::Bytes,
+    },
 }
 
 /// Sizes instead of contents: logged messages stay short.
@@ -349,6 +356,11 @@ impl std::fmt::Debug for BlobBody {
                 .field("bytes", &data.len())
                 .finish(),
             BlobBody::Missing => f.write_str("Missing"),
+            BlobBody::DeflatedChunk { index, data } => f
+                .debug_struct("DeflatedChunk")
+                .field("index", index)
+                .field("bytes", &data.len())
+                .finish(),
         }
     }
 }
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index a722065..2794e69 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, 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 = 31;
+pub(crate) const VERSION: u8 = 32;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {