1 files changed, 24 insertions, 12 deletions

diff --git a/src/state/session.rs b/src/state/session.rs
index 98b4dc3..bf5c83f 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -17,6 +17,18 @@ use tracing::debug;
 use crate::backoff::Backoff;
 use crate::document::Document;
 use crate::swarm::throughput::SyncBytes;
+use crate::transport::SyncPayload;
+
+/// A sync message from a peer, decoded (`decode`), to apply (`SyncSession::receive`).
+pub(crate) struct Received(sync::Message);
+
+/// `payload` decoded. Before taking the document's lock: decoding a large message takes
+/// a while.
+pub(crate) fn decode(payload: &SyncPayload) -> anyhow::Result<Received> {
+    sync::Message::decode(&payload.0)
+        .map(Received)
+        .map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}"))
+}
 
 /// A sync message being handed to our transport (for Mycelium: the HTTP request is in
 /// progress).
@@ -106,7 +118,7 @@ impl SyncSession {
         self.state.their_heads.is_some()
     }
 
-    /// The next sync message for the peer from `doc`, if there is anything to say. A
+    /// The next sync message for the peer from `doc`, encoded, if there is anything to say. A
     /// message is marked as being handed over (`sending`), and an answer to it is
     /// expected, unless the peer's heads are ours already: it has everything, and
     /// Automerge answers only when it has something to say, so waiting for an answer
@@ -117,7 +129,7 @@ impl SyncSession {
         doc: &mut Document,
         now: Instant,
         bytes: SyncBytes,
-    ) -> Option<sync::Message> {
+    ) -> Option<SyncPayload> {
         let sent_before = self.state.sent_hashes.clone();
         self.before_generate = (
             self.state.last_sent_heads.clone(),
@@ -144,7 +156,7 @@ impl SyncSession {
                 regenerate: false,
             });
         }
-        message
+        message.map(|message| SyncPayload(message.encode().into()))
     }
 
     /// Applies a sync message from the peer to `doc`. Doesn't settle the awaited
@@ -152,10 +164,10 @@ impl SyncSession {
     pub(crate) fn receive(
         &mut self,
         doc: &mut Document,
-        message: sync::Message,
+        message: Received,
     ) -> Result<(), AutomergeError> {
         self.delta_offered = false;
-        doc.sync().receive_sync_message(&mut self.state, message)
+        doc.sync().receive_sync_message(&mut self.state, message.0)
     }
 
     /// The message just generated is too large to send in one piece: take back what it
@@ -474,12 +486,12 @@ mod tests {
         // a opens, b answers with its (empty) heads, then a sends the changes.
         let opening = a.generate(&mut ours, now, bytes).unwrap();
         a.send_succeeded(a.id());
-        b.receive(&mut theirs, opening).unwrap();
+        b.receive(&mut theirs, decode(&opening).unwrap()).unwrap();
         let answer = b.generate(&mut theirs, now, bytes).unwrap();
         b.send_succeeded(b.id());
-        a.receive(&mut ours, answer).unwrap();
+        a.receive(&mut ours, decode(&answer).unwrap()).unwrap();
         let carrying = a.generate(&mut ours, now, bytes).unwrap();
-        let carried = carrying.clone().encode().len();
+        let carried = carrying.0.len();
         assert!(carried > 4_000, "the changes took {carried} bytes");
 
         let from = a.withdraw_for_delta();
@@ -494,7 +506,7 @@ mod tests {
         );
         assert!(!a.delta_offered);
         let again = a.generate(&mut ours, now, bytes).unwrap();
-        let carried_again = again.encode().len();
+        let carried_again = again.0.len();
         assert!(carried_again > 4_000, "sent again in {carried_again} bytes");
     }
 
@@ -520,7 +532,7 @@ mod tests {
         // b (joining) opens; a's answer would carry everything: withdrawn for a delta.
         let opening = b.generate(&mut theirs, now, bytes).unwrap();
         b.send_succeeded(b.id());
-        a.receive(&mut ours, opening).unwrap();
+        a.receive(&mut ours, decode(&opening).unwrap()).unwrap();
         assert!(a.generate(&mut ours, now, bytes).is_some());
         a.withdraw_for_delta();
         a.send_succeeded(a.id());
@@ -533,12 +545,12 @@ mod tests {
         );
         let after_delta = b.generate(&mut theirs, now, bytes).unwrap();
         b.send_succeeded(b.id());
-        a.receive(&mut ours, after_delta).unwrap();
+        a.receive(&mut ours, decode(&after_delta).unwrap()).unwrap();
 
         let answer = a
             .generate(&mut ours, now, bytes)
             .expect("a answers with its heads");
-        b.receive(&mut theirs, answer).unwrap();
+        b.receive(&mut theirs, decode(&answer).unwrap()).unwrap();
         assert!(b.have_all_theirs(&mut theirs));
     }
 }