sync messages are bytes outside the session

SyncPayload is now an opaque byte container in transport.
SyncSession::generate returns the encoded payload, state::decode_sync(&payload)
decodes it off the document lock (as before), and receive takes the decoded
message, which is a type only the session can open.  transport/mod.rs no longer
imports Automerge.

4 files changed, 31 insertions, 31 deletions

diff --git a/src/state/mod.rs b/src/state/mod.rs
index cf64dc3..673bd56 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -26,7 +26,7 @@ mod timing;
 
 pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing};
 pub(crate) use read::{ReadSource, StaleView};
-pub(crate) use session::SyncSession;
+pub(crate) use session::{SyncSession, decode as decode_sync};
 pub(crate) use timing::Timing;
 
 use std::collections::HashMap;
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));
     }
 }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 18c017c..772932d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -145,8 +145,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                         peer_state
                             .sync
                             .generate(&mut replica.doc, Instant::now(), bytes_before);
-                    let body = message.map(|message| {
-                        let payload = transport::SyncPayload::encode(message);
+                    let body = message.map(|payload| {
                         if payload.0.len() <= delta_threshold {
                             return transport::SyncBody::Message(payload);
                         }
@@ -356,7 +355,7 @@ pub(crate) async fn handle_sync(
         // 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 message = crate::state::decode_sync(&message)?;
             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 c10e0d8..cef31ff 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -1,6 +1,5 @@
 use anyhow::Result;
 use async_trait::async_trait;
-use automerge::sync::Message;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::boxed::Box;
@@ -681,23 +680,13 @@ impl DeltaSpec {
     }
 }
 
-/// 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.
+/// A sync message as the document's engine encodes it (`state::session`): bytes to the
+/// transports. Decoded only where it is applied (`state::decode_sync`), 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(crate) struct SyncPayload(pub(crate) bytes::Bytes);
 
-impl SyncPayload {
-    pub(crate) fn encode(message: Message) -> Self {
-        Self(message.encode().into())
-    }
-
-    pub(crate) 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 {