diff options
| author | tv <tv@krebsco.de> | 2026-11-03 06:23:23 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-11-03 06:23:23 +0100 |
| commit | 4559c4427fcdebe1b816ce8c4e651cf11571b7d5 (patch) | |
| tree | beefb9ad0b91145f4a4dbaec67fb09501f4bb525 /src/swarm | |
| parent | b16977b18fdd1e5193f5c015eadcb61a63f4fae1 (diff) | |
tree changes had already
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/sync.rs | 18 |
1 files changed, 14 insertions, 4 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 187dacc..dd5444b 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -376,11 +376,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, carried changes)) 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 carried = message.carries_changes(); let mut replica = doc.blocking_write(); // Authoritative check, under the same lock as the apply: the document may @@ -430,13 +431,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, + carried, ))) }) .await .context("Sync task panicked")?? }; - let Some((caught_up, changed)) = applied else { + let Some((caught_up, changed, carried)) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; @@ -459,6 +461,14 @@ pub(crate) async fn handle_sync( // Caught up: our other neighbours may lack what the source had. if changed || just_caught_up { crate::swarm::relay::announce(state, peer, writer.filter(|_| changed)).await; + } else if carried + && let Some(writer) = writer + && state.initial_sync_done.load(Ordering::Acquire) + { + // A writer's changes along its tree that we had already: they came another way + // first, not saying whose (with an answer to one of our messages, say), so we + // passed nothing on then. Our children in its tree may lack them still. + crate::swarm::tree::forward(state, writer, peer).await; } state |
