diff options
Diffstat (limited to 'src/swarm/sync.rs')
| -rw-r--r-- | src/swarm/sync.rs | 94 |
1 files changed, 78 insertions, 16 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 8190e5c..d28df33 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -138,6 +138,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { .map(|estimate| estimate.bytes_per_sec); let failed = peer_state_mutex.lock().await.failed_whole(Instant::now()); let delta_threshold = state.blobs.delta_threshold_for(rate, failed); + let push_changes = state.push_changes; let stats = Arc::clone(&state.stats); let clock = Arc::clone(&state.clock); @@ -167,7 +168,20 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { delta_threshold, Instant::now(), ); - let body = if let Some(from) = again { + // Along a tree, taken as delivered: the changes as they are, if pushing + // (`AppState::push_changes`). + let pushed = (push_changes && level && writer.is_some() && again.is_none()) + .then(|| peer_state.sync.push(&replica.doc, delta_threshold)) + .flatten(); + let body = if let Some((heads, changes)) = pushed { + Some(transport::SyncBody::Changes { + heads, + changes: changes + .into_iter() + .map(serde_bytes::ByteBuf::from) + .collect(), + }) + } else if let Some(from) = again { let spec = transport::DeltaSpec::new(from, replica.heads().to_vec()); debug!( from = spec.from.len(), @@ -219,11 +233,24 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { Some(transport::SyncBody::Delta(spec)) }) }; - let delivered = level && matches!(body, Some(transport::SyncBody::Message(_))); + let delivered = level + && matches!( + body, + Some( + transport::SyncBody::Message(_) + | transport::SyncBody::Changes { .. } + ) + ); if delivered { peer_state.sync.assume_delivered(replica.heads()); - if let Some(transport::SyncBody::Message(payload)) = &body { - stats.delivered_sent(payload.0.len()); + match &body { + Some(transport::SyncBody::Message(payload)) => { + stats.delivered_sent(payload.0.len()); + } + Some(transport::SyncBody::Changes { changes, .. }) => { + stats.delivered_sent(changes.iter().map(|c| c.len()).sum()); + } + _ => {} } } ( @@ -259,6 +286,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let inner = match &body { transport::SyncBody::Message(payload) | transport::SyncBody::Deflated(payload) => Some(payload.0.len()), + transport::SyncBody::Changes { changes, .. } => { + Some(changes.iter().map(|change| change.len()).sum()) + } transport::SyncBody::Delta(_) => None, }; let ids = rmp_serde::encode::to_vec(&transport::EpochTag::of(epoch)) @@ -434,17 +464,33 @@ pub(crate) async fn handle_sync( let from_ready = envelope.ready; let writer = envelope.writer; let delivered = envelope.delivered; - let (payload, deflated) = match envelope.body { - transport::SyncBody::Message(payload) => (payload, false), - transport::SyncBody::Deflated(payload) => (payload, true), + // Changes pushed along a tree (`SyncBody::Changes`): their heads and bytes, made + // into the Automerge message that would have brought them once the document's lock + // is held (`Received::pushed`). + let (payload, deflated, pushed) = match envelope.body { + transport::SyncBody::Message(payload) => (payload, false, None), + transport::SyncBody::Deflated(payload) => (payload, true, None), + transport::SyncBody::Changes { heads, changes } => { + let changes: Vec<Vec<u8>> = changes + .into_iter() + .map(serde_bytes::ByteBuf::into_vec) + .collect(); + let len = changes.iter().map(Vec::len).sum::<usize>(); + ( + transport::SyncPayload(bytes::Bytes::new()), + false, + Some((heads, changes, len)), + ) + } transport::SyncBody::Delta(spec) => { fetch_delta(state, peer, epoch, spec); return Ok(()); } }; - if payload.0.len() > crate::swarm::blob::MAX_SYNC_MESSAGE { + let wire_len = pushed.as_ref().map_or(payload.0.len(), |(_, _, len)| *len); + if wire_len > crate::swarm::blob::MAX_SYNC_MESSAGE { // Not decoded: a node of this version sends a delta instead. - warn!(peer = %peer, bytes = payload.0.len(), "Dropping a sync message larger than any sender should send"); + warn!(peer = %peer, bytes = wire_len, "Dropping a sync message larger than any sender should send"); return Ok(()); } @@ -485,14 +531,14 @@ pub(crate) async fn handle_sync( } else { message }; - let payload_len = message.0.len(); + let (payload_len, decoded) = match &pushed { + Some((_, _, len)) => (*len, None), + None => ( + message.0.len(), + Some(crate::document::sync::decode(&message)?), + ), + }; stats.sync_received(payload_len); - let message = crate::document::sync::decode(&message)?; - let carried = message.carries_changes(); - let sizes = if carried { message.sizes() } else { (0, 0) }; - if carried { - stats.carrying(sizes); - } let mut replica = doc.blocking_write(); // Authoritative check, under the same lock as the apply: the document may @@ -500,6 +546,22 @@ pub(crate) async fn handle_sync( if replica.epoch != epoch { return Ok(None); } + let message = match decoded { + Some(message) => message, + None => { + let (heads, changes, _) = pushed.expect("pushed or decoded"); + crate::document::sync::Received::pushed( + heads, + changes, + replica.heads().to_vec(), + ) + } + }; + let carried = message.carries_changes(); + let sizes = if carried { message.sizes() } else { (0, 0) }; + if carried { + stats.carrying(sizes); + } let mut peer_state = peer_state_mutex.blocking_lock(); peer_state.reports_ready = Some(from_ready); |