introduce eager links

3 files changed, 100 insertions, 21 deletions

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<String>) {
 /// 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<String>) -> Vec<String> {
     let skip = match senders.len() {
         1 => senders.iter().next().cloned(),
@@ -121,13 +146,15 @@ async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> {
         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<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;