decode sync messages lazily

Sync messages are now being decoded only where they are applied: after
membership has been checked, in the sync lane's blocking task.

Debug now shows only the size instead of the whole message.

3 files changed, 40 insertions, 9 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 0fbff05..ca54fae 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -159,7 +159,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     swarm_id,
                     epoch,
                     epoch_seq,
-                    message,
+                    message: transport::SyncPayload::encode(message),
                     ready,
                 };
                 let encoded = transport
@@ -336,6 +336,8 @@ pub(crate) async fn handle_sync(
         // Some((caught_up, heads changed)) when applied, None when the message belongs
         // to another epoch.
         tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
+            // Before the lock: decoding a large message takes a while.
+            let message = message.decode()?;
             let mut replica = doc.blocking_write();
 
             // Authoritative check, under the same lock as the apply: the document may
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index a040993..0d761e0 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -326,13 +326,37 @@ pub struct SyncEnvelope {
     /// ignores an epoch older than the newest it has seen in the same session.
     pub epoch_seq: u32,
     #[serde(with = "crate::transport::wire::codec::sync_message")]
-    pub message: Message,
+    pub message: SyncPayload,
     /// 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 ready: bool,
 }
 
+/// 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.
+#[derive(Clone)]
+pub struct SyncPayload(pub bytes::Bytes);
+
+impl SyncPayload {
+    pub fn encode(message: Message) -> Self {
+        Self(message.encode().into())
+    }
+
+    pub fn decode(&self) -> Result<Message> {
+        Message::decode(&self.0)
+            .map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}"))
+    }
+}
+
+/// Sizes instead of contents: logged messages stay short.
+impl std::fmt::Debug for SyncPayload {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "SyncPayload({} bytes)", self.0.len())
+    }
+}
+
 /// A message serialized for the wire. Cloning is cheap (shared buffer).
 #[derive(Clone, Debug)]
 pub struct EncodedMessage {
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 3c80b11..a0e0f20 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -233,18 +233,23 @@ pub(crate) mod codec {
         }
     }
 
-    /// An Automerge sync message, in Automerge's own encoding.
+    /// An Automerge sync message, in Automerge's own encoding (left encoded:
+    /// `SyncPayload`).
     pub(crate) mod sync_message {
         use super::*;
-        use automerge::sync::Message;
+        use crate::transport::SyncPayload;
 
-        pub(crate) fn serialize<S: Serializer>(value: &Message, s: S) -> Result<S::Ok, S::Error> {
-            Bytes::new(&value.clone().encode()).serialize(s)
+        pub(crate) fn serialize<S: Serializer>(
+            value: &SyncPayload,
+            s: S,
+        ) -> Result<S::Ok, S::Error> {
+            Bytes::new(&value.0).serialize(s)
         }
 
-        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Message, D::Error> {
-            Message::decode(&ByteBuf::deserialize(d)?)
-                .map_err(|error| D::Error::custom(format!("Automerge decode failed: {error:?}")))
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+            d: D,
+        ) -> Result<SyncPayload, D::Error> {
+            Ok(SyncPayload(ByteBuf::deserialize(d)?.into_vec().into()))
         }
     }