replace eager links by trees

6 files changed, 35 insertions, 137 deletions

diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index e0d483d..c430d58 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -1,9 +1,8 @@
 //! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a
 //! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its
 //! neighbours and the peers it holds dormant (`own_links`), each with the transport its
-//! messages go over, whether only through a relay, the round trip, whether it is
-//! dormant, and which way the link is eager (`swarm::relay`): as the member measures
-//! them, which no other node can.
+//! messages go over, whether only through a relay, the round trip and whether it is
+//! dormant: as the member measures them, which no other node can.
 //!
 //! Every node answers, whatever its build: one with a dashboard asks any member, which
 //! may have none. Only members (in the member list) are answered, as the links name
@@ -11,8 +10,8 @@
 //! is written to the document: asked for while someone looks, forgotten after.
 //!
 //! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again
-//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, eager,
-//! or a round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
+//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a
+//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
 //! subscribes (round trips change without telling), until the subscription runs out
 //! (the asker renews it with its next LINKS, so one that is gone stops it by not
 //! renewing) or a LINKS with 0 ends it.
@@ -232,14 +231,7 @@ impl Links {
 pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
     let mut links = Vec::new();
     for (peer, peer_state) in state.peers_snapshot().await {
-        let (dormant, eager_to, eager_from) = {
-            let peer_state = peer_state.lock().await;
-            (
-                peer_state.liveness.is_dormant(),
-                peer_state.eager_to,
-                peer_state.eager_from,
-            )
-        };
+        let dormant = peer_state.lock().await.liveness.is_dormant();
         if !dormant && !topology::is_neighbour(state, &peer) {
             continue;
         }
@@ -264,8 +256,6 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
                 .rtt(&peer)
                 .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
             dormant,
-            eager_to,
-            eager_from,
             clock: state.transport.clock_offset(&peer),
             peer,
         });
@@ -1020,8 +1010,6 @@ fn changed(old: &[Link], new: &[Link]) -> bool {
                 || a.relayed != b.relayed
                 || a.down != b.down
                 || a.dormant != b.dormant
-                || a.eager_to != b.eager_to
-                || a.eager_from != b.eager_from
                 || moved(a.rtt_us, b.rtt_us)
                 || clock_moved(a.clock, b.clock)
                 || a.others.len() != b.others.len()
@@ -1074,8 +1062,6 @@ mod tests {
             down: false,
             rtt_us,
             dormant: false,
-            eager_to: false,
-            eager_from: false,
             others: Vec::new(),
             clock: None,
         }
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index c04b1ad..8362d73 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -179,12 +179,9 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
             stamp.epoch_seq,
         )
     };
-    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),
+    let syncs_with_you = match state.get_peer_state(peer).await {
+        Some(peer_state) => peer_state.lock().await.syncs_on(epoch),
+        None => false,
     };
     transport::Status {
         heads,
@@ -194,7 +191,6 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
         syncs_with_you,
         leaving: state.is_closing(),
         announcing: false,
-        eager,
         reaches: Some(crate::swarm::reach::published(state)),
     }
 }
@@ -232,7 +228,6 @@ 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;
         if let Some(reaches) = &remote.reaches {
             peer_state.reaches = Some((Instant::now(), reaches.clone()));
         }
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 7dbc5b0..f4eaf5a 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -19,15 +19,9 @@
 //! 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).
+//! Changes that say whose they are (the writer's own, and what passes on along its tree)
+//! go at once to the receiver's children in the writer's tree (`swarm::tree`), which
+//! the announcement round then leaves out.
 //!
 //! 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
@@ -74,7 +68,6 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>)
     if !state.initial_sync_done.load(Ordering::Acquire) {
         return;
     }
-    forward_eagerly(state, from).await;
     let sent = match writer {
         Some(writer) => crate::swarm::tree::forward(state, writer, from).await,
         None => Vec::new(),
@@ -106,20 +99,6 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>)
     });
 }
 
-/// 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.stats.eager_forward();
-            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.
@@ -140,8 +119,7 @@ async fn announce_now(state: &AppState, pending: &Pending) {
 /// 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; and
-/// one we forward to eagerly, or `sent` the changes along a writer's tree, which has
-/// them already.
+/// one `sent` the changes along a writer's tree, which has them already.
 async fn unreached(
     state: &AppState,
     senders: &HashSet<String>,
@@ -163,15 +141,13 @@ async fn unreached(
         {
             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);
-        drop(peer_state);
-        if !eager && !has_ours {
+        if !has_ours {
             unreached.push(peer);
         }
     }
@@ -255,62 +231,21 @@ pub(crate) async fn pull_if_behind(
     true
 }
 
-/// 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 = {
+/// 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) {
+    {
         let mut pulling = state.pulling.lock_or_recover();
         if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
             return;
         }
         pulling.current = None;
-        !pulling.waiting.is_empty()
-    };
-    if changed {
-        graft(state, peer).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;
-    }
-    state.stats.link_grafted();
-    debug!(peer = %peer, "Link eager: we want its changes as it applies them");
-    if let Err(error) = announce_heads(state, peer, false).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;
-    }
-    state.stats.link_pruned();
-    debug!(peer = %peer, "Link lazy: its changes reach us first another way");
-    if let Err(error) = announce_heads(state, peer, false).await {
-        debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed");
+        if pulling.waiting.is_empty() {
+            return;
+        }
     }
-}
-
-/// 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
+    let state = state.clone();
+    tokio::spawn(async move { pull_next(&state).await });
 }
 
 /// Pulls from the first waiting neighbour that still has heads we lack.
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index dd56bb0..8a00629 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -102,12 +102,9 @@ pub(crate) struct Relay {
     pub(crate) pulls: u64,
     /// Announcements of heads we lacked that had to wait for the pull in progress.
     pub(crate) pulls_queued: u64,
-    /// Links made eager (a pull brought changes over them) and lazy again (changes came
-    /// over them that another path brought first).
-    pub(crate) grafts: u64,
-    pub(crate) prunes: u64,
-    /// Changes forwarded over eager links: a sync message asked for per neighbour.
-    pub(crate) eager_forwards: u64,
+    /// Changes passed on along writers' trees (`swarm::tree`): a sync message asked for
+    /// per child.
+    pub(crate) tree_forwards: u64,
 }
 
 /// When the initial sync finished, and what it took.
@@ -190,16 +187,8 @@ impl NodeStats {
         self.lock().relay.pulls_queued += 1;
     }
 
-    pub(crate) fn link_grafted(&self) {
-        self.lock().relay.grafts += 1;
-    }
-
-    pub(crate) fn link_pruned(&self) {
-        self.lock().relay.prunes += 1;
-    }
-
-    pub(crate) fn eager_forward(&self) {
-        self.lock().relay.eager_forwards += 1;
+    pub(crate) fn tree_forward(&self) {
+        self.lock().relay.tree_forwards += 1;
     }
 
     /// A sync message of `bytes` arrived.
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 403e0c1..187dacc 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -376,12 +376,11 @@ 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, a duplicate)) when applied, None when the
-        // message belongs to another epoch.
-        tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool, bool)>> {
+        // Some((caught_up, heads changed)) when applied, None when the message belongs
+        // to another epoch.
+        tokio::task::spawn_blocking(move || -> Result<Option<(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
@@ -411,9 +410,6 @@ 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.
@@ -434,14 +430,13 @@ 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, duplicate)) = applied else {
+    let Some((caught_up, changed)) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
@@ -460,10 +455,7 @@ pub(crate) async fn handle_sync(
 
     state.notify_store_changed();
 
-    crate::swarm::relay::pulled(state, peer, changed).await;
-    if duplicate {
-        crate::swarm::relay::prune(state, peer).await;
-    }
+    crate::swarm::relay::pulled(state, peer);
     // 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;
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index 489bc22..1dff92e 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -146,6 +146,7 @@ pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Vec<St
         if let Some(peer_state) = state.get_peer_state(child).await {
             peer_state.lock().await.tree_writer = Some(writer);
         }
+        state.stats.tree_forward();
         state.request_sync(child);
     }
     if !children.is_empty() {