From 2b4359f23d20b14bae1896512f678d02b7f3c34c Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 03:25:00 +0100 Subject: introduce eager links --- src/swarm/peer/keepalive.rs | 11 ++++-- src/swarm/relay.rs | 92 ++++++++++++++++++++++++++++++++++++++------- src/swarm/sync.rs | 18 ++++++--- 3 files changed, 100 insertions(+), 21 deletions(-) (limited to 'src/swarm') diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index c426f99..37fcba8 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -176,9 +176,12 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta stamp.epoch_seq, ) }; - let syncs_with_you = match state.get_peer_state(peer).await { - Some(peer_state) => peer_state.lock().await.syncs_on(epoch), - None => false, + let (syncs_with_you, eager) = match state.get_peer_state(peer).await { + Some(peer_state) => { + let peer_state = peer_state.lock().await; + (peer_state.syncs_on(epoch), peer_state.eager_from) + } + None => (false, false), }; transport::Status { heads, @@ -188,6 +191,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta syncs_with_you, leaving: state.is_closing(), announcing: false, + eager, } } @@ -224,6 +228,7 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport peer_state.reports_ready = Some(remote.ready); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); peer_state.last_heads = Some((remote.epoch, remote.heads.clone())); + peer_state.eager_to = remote.eager; drop(peer_state); if left { info!(peer = %peer, "Peer said farewell; dormant until it is heard from again"); diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 26719cd..fcca9c7 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -19,6 +19,16 @@ //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). //! +//! Links start *lazy*: changes are announced over them and pulled. A pull that brought +//! changes we lacked makes the link *eager* in that direction (Plumtree's GRAFT): our +//! STATUS asks the neighbour for its changes as soon as it applies them (`eager`), and +//! it sends them right away as sync messages, without announcing or waiting +//! (`forward_eagerly`). Changes we had already over an eager link make it lazy again +//! (Plumtree's PRUNE): another path brought them first. So eager links form where +//! changes actually flow, over links that deliver, and the announcements on the others +//! are what catches whatever they miss. A link starts lazy again whenever its sync does +//! (`PeerState::reset_sync`: a new neighbour, dormancy, a new run of the peer). +//! //! In a full mesh too: the member list makes every member the writer's neighbour, but //! that says nothing about whether the writer reaches them. One behind NAT may reach a //! single one of them, one on another transport (on Tor alone) none, and no node can @@ -55,6 +65,7 @@ pub(crate) async fn announce(state: &AppState, from: &str) { if !state.initial_sync_done.load(Ordering::Acquire) { return; } + forward_eagerly(state, from).await; let schedule = { let mut pending = state.announce_pending.lock_or_recover(); match pending.as_mut() { @@ -87,6 +98,19 @@ pub(crate) async fn announce(state: &AppState, from: &str) { }); } +/// What `from` sent us, at once to the neighbours that asked for our changes as we apply +/// them (eager links): a sync message each, with what it lacks (the sync worker's). +async fn forward_eagerly(state: &AppState, from: &str) { + for (peer, peer_state) in state.peers_snapshot().await { + if peer == from || !topology::is_neighbour(state, &peer) { + continue; + } + if peer_state.lock().await.eager_to { + state.request_sync(&peer); + } + } +} + /// One announcement round. The peer whose changes we announce has them already, so a /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. @@ -106,7 +130,8 @@ async fn announce_now(state: &AppState, senders: &HashSet) { /// The neighbours to tell of the changes `senders` sent us. The peer whose changes we /// announce has them already, so with a single sender it is skipped; with several, each /// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an -/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have. +/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have; and +/// one we forward to eagerly, which was sent the changes already. async fn unreached(state: &AppState, senders: &HashSet) -> Vec { let skip = match senders.len() { 1 => senders.iter().next().cloned(), @@ -121,13 +146,15 @@ async fn unreached(state: &AppState, senders: &HashSet) -> Vec { if skip.as_deref() == Some(peer.as_str()) || !topology::is_neighbour(state, &peer) { continue; } + let peer_state = peer_state.lock().await; + // Sent what we have as we applied it (`forward_eagerly`). + let eager = peer_state.eager_to; let has_ours = peer_state - .lock() - .await .last_heads .as_ref() .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1); - if !has_ours { + drop(peer_state); + if !eager && !has_ours { unreached.push(peer); } } @@ -211,21 +238,60 @@ pub(crate) async fn pull_if_behind( true } -/// A sync message from `peer` was applied: a pull from it is answered, and the next -/// waiting neighbour's turn comes. -pub(crate) fn pulled(state: &AppState, peer: &str) { - { +/// A sync message from `peer` was applied (`changed`: it brought changes we lacked): a +/// pull from it is answered, and the next waiting neighbour's turn comes. A pull that +/// brought changes makes the link eager (`graft`). +pub(crate) async fn pulled(state: &AppState, peer: &str, changed: bool) { + let next = { let mut pulling = state.pulling.lock_or_recover(); if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) { return; } pulling.current = None; - if pulling.waiting.is_empty() { - return; - } + !pulling.waiting.is_empty() + }; + if changed { + graft(state, peer).await; } - let state = state.clone(); - tokio::spawn(async move { pull_next(&state).await }); + if next { + let state = state.clone(); + tokio::spawn(async move { pull_next(&state).await }); + } +} + +/// `peer` brought changes we lacked when we pulled: from now on we want them as it +/// applies them, and our STATUS says so (Plumtree's GRAFT). +async fn graft(state: &AppState, peer: &str) { + if !set_eager_from(state, peer, true).await { + return; + } + debug!(peer = %peer, "Link eager: we want its changes as it applies them"); + if let Err(error) = announce_heads(state, peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is eager failed"); + } +} + +/// Changes we had already came from `peer` over an eager link: another path brings them +/// first, so announcements will do; our STATUS says so (Plumtree's PRUNE). +pub(crate) async fn prune(state: &AppState, peer: &str) { + if !set_eager_from(state, peer, false).await { + return; + } + debug!(peer = %peer, "Link lazy: its changes reach us first another way"); + if let Err(error) = announce_heads(state, peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed"); + } +} + +/// Sets whether we want `peer`'s changes eagerly; whether that changed it. +async fn set_eager_from(state: &AppState, peer: &str, eager: bool) -> bool { + let Some(peer_state) = state.get_peer_state(peer).await else { + return false; + }; + let mut peer_state = peer_state.lock().await; + let changed = peer_state.eager_from != eager; + peer_state.eager_from = eager; + changed } /// Pulls from the first waiting neighbour that still has heads we lack. 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> { + // Some((caught_up, heads changed, a duplicate)) when applied, None when the + // message belongs to another epoch. + tokio::task::spawn_blocking(move || -> Result> { // 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; -- cgit v1.3.1