deflate blob chunks

8 files changed, 120 insertions, 7 deletions

diff --git a/src/compress.rs b/src/compress.rs
new file mode 100644
index 0000000..77e634d
--- /dev/null
+++ b/src/compress.rs
@@ -0,0 +1,60 @@
+//! Deflate for what goes over the wire: large sync messages (`swarm::sync`). Automerge
+//! sends changes as they are stored, uncompressed; text and JSON in them shrink to about
+//! half.
+
+use std::io::{Read, Write};
+
+use anyhow::{Result, bail};
+use flate2::Compression;
+use flate2::read::DeflateDecoder;
+use flate2::write::DeflateEncoder;
+
+/// Below this, deflating saves too little to be worth it.
+pub(crate) const WORTH_FROM: usize = 1024;
+
+/// `bytes` deflated, if that makes them at least a tenth smaller.
+pub(crate) fn deflate(bytes: &[u8]) -> Option<Vec<u8>> {
+    if bytes.len() < WORTH_FROM {
+        return None;
+    }
+    let mut encoder = DeflateEncoder::new(Vec::with_capacity(bytes.len() / 2), Compression::fast());
+    encoder.write_all(bytes).ok()?;
+    let deflated = encoder.finish().ok()?;
+    (deflated.len() < bytes.len() - bytes.len() / 10).then_some(deflated)
+}
+
+/// `bytes` inflated, if they come to at most `limit` bytes: a small message can't make
+/// us hold a large one.
+pub(crate) fn inflate(bytes: &[u8], limit: usize) -> Result<Vec<u8>> {
+    let mut inflated = Vec::with_capacity((bytes.len() * 3).min(limit));
+    DeflateDecoder::new(bytes)
+        .take(limit as u64 + 1)
+        .read_to_end(&mut inflated)?;
+    if inflated.len() > limit {
+        bail!("inflates to more than {limit} bytes");
+    }
+    Ok(inflated)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// What deflates inflates back; little, or what doesn't shrink, isn't deflated; more
+    /// than the limit is refused.
+    #[test]
+    fn deflates_and_inflates_within_limits() {
+        let text = "a key, a value; ".repeat(1000);
+        let deflated = deflate(text.as_bytes()).expect("text shrinks");
+        assert!(deflated.len() < text.len() / 4);
+        assert_eq!(inflate(&deflated, text.len()).unwrap(), text.as_bytes());
+        assert!(
+            inflate(&deflated, text.len() - 1).is_err(),
+            "over the limit"
+        );
+        assert!(deflate(b"short").is_none());
+        let mut rng = crate::testrand::Rng::new(1);
+        let noise: Vec<u8> = (0..4096).map(|_| rng.next_u64() as u8).collect();
+        assert!(deflate(&noise).is_none(), "noise doesn't shrink");
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index c3d7458..128e5a0 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -14,6 +14,7 @@ use tracing::{debug, error, info, warn};
 mod api;
 mod backoff;
 mod clock;
+mod compress;
 mod config;
 mod document;
 mod files;
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 5b713f3..ae22fd1 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -335,6 +335,9 @@ impl Totals {
             relay.framed += stats.relay.framed;
             relay.framing_bytes += stats.relay.framing_bytes;
             relay.framing_id_bytes += stats.relay.framing_id_bytes;
+            relay.deflated += stats.relay.deflated;
+            relay.deflated_from += stats.relay.deflated_from;
+            relay.deflated_to += stats.relay.deflated_to;
             relay.nothing_new += stats.relay.nothing_new;
             relay.nothing_new_bytes += stats.relay.nothing_new_bytes;
             relay.nothing_new_along_trees += stats.relay.nothing_new_along_trees;
@@ -729,6 +732,14 @@ fn report(
         each(r1.framing_bytes - r0.framing_bytes, framed),
         each(r1.framing_id_bytes - r0.framing_id_bytes, framed),
     );
+    let deflated = r1.deflated - r0.deflated;
+    let _ = writeln!(
+        out,
+        "    deflated: {:.1} sync messages per write, {:.0} bytes each to {:.0}",
+        deflated as f64 / writes,
+        each(r1.deflated_from - r0.deflated_from, deflated),
+        each(r1.deflated_to - r0.deflated_to, deflated),
+    );
     let (lately, later) = (
         r1.asked_known_lately - r0.asked_known_lately,
         r1.asked_known_later - r0.asked_known_later,
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index ed91db0..6332901 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -133,6 +133,10 @@ pub(crate) struct Relay {
     pub(crate) framed: u64,
     pub(crate) framing_bytes: u64,
     pub(crate) framing_id_bytes: u64,
+    /// Sync messages deflated (`compress`), their bytes before and after.
+    pub(crate) deflated: u64,
+    pub(crate) deflated_from: u64,
+    pub(crate) deflated_to: u64,
     /// Sync messages received that carried changes, all of which we had already, and
     /// their payload's bytes: changes that came twice. Of those, how many came along a
     /// writer's tree (`SyncEnvelope::writer`), and how many were taken as delivered by
@@ -254,6 +258,14 @@ impl NodeStats {
         self.lock().relay.tree_forwards += 1;
     }
 
+    /// A sync message of `from` bytes went out deflated to `to`.
+    pub(crate) fn deflated(&self, from: usize, to: usize) {
+        let mut inner = self.lock();
+        inner.relay.deflated += 1;
+        inner.relay.deflated_from += from as u64;
+        inner.relay.deflated_to += to as u64;
+    }
+
     /// A sync message went out whose envelope added `bytes` to its Automerge message,
     /// `ids` of them the epoch.
     pub(crate) fn sync_framing(&self, bytes: usize, ids: usize) {
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) };
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 32be29b..1e9e5a9 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -729,6 +729,8 @@ impl std::fmt::Display for EpochTag {
 pub(crate) enum SyncBody {
     /// An Automerge sync message.
     Message(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload),
+    /// An Automerge sync message, deflated (`compress`): a large one, which shrank.
+    Deflated(#[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),
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index b7907fb..df95bce 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -198,7 +198,7 @@ fn related(peer: &str, message: &ParsedMessage) -> Option<String> {
             SyncBody::Delta(spec) => {
                 format!("blob {}", crate::transport::BlobId::Delta(spec.id()))
             }
-            SyncBody::Message(_) => return None,
+            SyncBody::Message(_) | SyncBody::Deflated(_) => return None,
         },
         _ => return None,
     })
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index d1ac14d..a722065 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 = 30;
+pub(crate) const VERSION: u8 = 31;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {