deflate blob chunks

1 files changed, 32 insertions, 5 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 394e589..8190e5c 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -238,11 +238,27 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                 let Some(body) = body else {
                     return (session, Ok(None));
                 };
+                // A large message deflated, if it shrinks (`compress`).
+                let body = match body {
+                    transport::SyncBody::Message(payload) => {
+                        match crate::compress::deflate(&payload.0) {
+                            Some(deflated) => {
+                                stats.deflated(payload.0.len(), deflated.len());
+                                transport::SyncBody::Deflated(transport::SyncPayload(
+                                    deflated.into(),
+                                ))
+                            }
+                            None => transport::SyncBody::Message(payload),
+                        }
+                    }
+                    other => other,
+                };
                 // What the envelope adds to an Automerge message (`swarm::stats`), and of
                 // that, the epoch's tag (msgpack: 12 bytes, up to 20 for generations past
                 // 127).
                 let inner = match &body {
-                    transport::SyncBody::Message(payload) => Some(payload.0.len()),
+                    transport::SyncBody::Message(payload)
+                    | transport::SyncBody::Deflated(payload) => Some(payload.0.len()),
                     transport::SyncBody::Delta(_) => None,
                 };
                 let ids = rmp_serde::encode::to_vec(&transport::EpochTag::of(epoch))
@@ -418,8 +434,9 @@ pub(crate) async fn handle_sync(
     let from_ready = envelope.ready;
     let writer = envelope.writer;
     let delivered = envelope.delivered;
-    let payload = match envelope.body {
-        transport::SyncBody::Message(payload) => payload,
+    let (payload, deflated) = match envelope.body {
+        transport::SyncBody::Message(payload) => (payload, false),
+        transport::SyncBody::Deflated(payload) => (payload, true),
         transport::SyncBody::Delta(spec) => {
             fetch_delta(state, peer, epoch, spec);
             return Ok(());
@@ -430,7 +447,6 @@ pub(crate) async fn handle_sync(
         warn!(peer = %peer, bytes = payload.0.len(), "Dropping a sync message larger than any sender should send");
         return Ok(());
     }
-    state.stats.sync_received(payload.0.len());
 
     let applied = {
         let doc = Arc::clone(&state.doc);
@@ -458,8 +474,19 @@ pub(crate) async fn handle_sync(
             (Vec<(String, Vec<crate::document::ChangeHash>)>, bool),
         );
         tokio::task::spawn_blocking(move || -> Result<Option<Applied>> {
-            // Before the lock: decoding a large message takes a while.
+            // Before the lock: inflating and decoding a large message takes a while. No
+            // larger than any sender should send, deflated or not.
+            let message = if deflated {
+                transport::SyncPayload(
+                    crate::compress::inflate(&message.0, crate::swarm::blob::MAX_SYNC_MESSAGE)
+                        .context("Inflating a sync message")?
+                        .into(),
+                )
+            } else {
+                message
+            };
             let payload_len = message.0.len();
+            stats.sync_received(payload_len);
             let message = crate::document::sync::decode(&message)?;
             let carried = message.carries_changes();
             let sizes = if carried { message.sizes() } else { (0, 0) };