diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/swarm/sync.rs | 4 | ||||
| -rw-r--r-- | src/transport/mod.rs | 26 | ||||
| -rw-r--r-- | src/transport/wire.rs | 19 |
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())) } } |
