diff options
| -rw-r--r-- | src/state/mod.rs | 2 | ||||
| -rw-r--r-- | src/state/session.rs | 36 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 5 | ||||
| -rw-r--r-- | src/transport/mod.rs | 19 |
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 { |