aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/sync.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/sync.rs')
-rw-r--r--src/swarm/sync.rs18
1 files changed, 13 insertions, 5 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 3bfbf7d..37c3b0a 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -372,11 +372,12 @@ pub(crate) async fn handle_sync(
let clock = Arc::clone(&state.clock);
let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
- // Some((caught_up, heads changed)) when applied, None when the message belongs
- // to another epoch.
- tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
+ // Some((caught_up, heads changed, a duplicate)) when applied, None when the
+ // message belongs to another epoch.
+ tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool, bool)>> {
// Before the lock: decoding a large message takes a while.
let message = crate::document::sync::decode(&message)?;
+ let carries_changes = message.carries_changes();
let mut replica = doc.blocking_write();
// Authoritative check, under the same lock as the apply: the document may
@@ -406,6 +407,9 @@ pub(crate) async fn handle_sync(
let heads_after = replica.heads().to_vec();
stats.heads_seen(heads_after.len());
let changed = heads_after != heads_before;
+ // Changes we had already, over a link we asked to be eager: the link isn't
+ // needed (`relay::prune`).
+ let duplicate = carries_changes && !changed && peer_state.eager_from;
if changed && measure_propagation {
// Now by the peers' clocks (their median): this node's own clock skew
// doesn't count; the writer's still does.
@@ -426,13 +430,14 @@ pub(crate) async fn handle_sync(
// before it registered itself).
check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc),
changed,
+ duplicate,
)))
})
.await
.context("Sync task panicked")??
};
- let Some((caught_up, changed)) = applied else {
+ let Some((caught_up, changed, duplicate)) = applied else {
debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
return Ok(());
};
@@ -451,7 +456,10 @@ pub(crate) async fn handle_sync(
state.notify_store_changed();
- crate::swarm::relay::pulled(state, peer);
+ crate::swarm::relay::pulled(state, peer, changed).await;
+ if duplicate {
+ crate::swarm::relay::prune(state, peer).await;
+ }
// Caught up: our other neighbours may lack what the source had.
if changed || just_caught_up {
crate::swarm::relay::announce(state, peer).await;