1 files changed, 46 insertions, 15 deletions

diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs
index 8c22ef6..e6cee2f 100644
--- a/src/swarm/reach.rs
+++ b/src/swarm/reach.rs
@@ -7,7 +7,8 @@
 //! `Timing::unreached_after`, asked again halfway, with nothing else heard from it; a
 //! member gone shows within seconds, not after `dormant_after`). It tells its
 //! neighbours so in its STATUS (`Status::reaches`, each member by its pair key,
-//! `topology::pair_key`, 8 bytes instead of a node ID): in every keepalive STATUS and
+//! `topology::pair_key`, 8 bytes instead of a node ID, with its round trip to each as a
+//! bucket: `Status::round_trips`, `rtt::bucket`): in every keepalive STATUS and
 //! STATUS_REPLY, and soon to every neighbour when it changed (the peer worker looks
 //! every 5 s, and at once when a member we reach stops answering), but not in
 //! announcements of heads, which go often. So every node knows whom its neighbours
@@ -32,17 +33,20 @@ use tracing::debug;
 use crate::locks::MutexExt;
 use crate::state::{AppState, HelloState, Standing};
 use crate::swarm::peer::announce_heads;
-use crate::swarm::{members, topology};
+use crate::swarm::{members, rtt, topology};
 
-/// Whom we reach, as last told our neighbours.
+/// Whom we reach, as last told our neighbours, and our round trip to each.
 #[derive(Default)]
 pub(crate) struct Reach {
     published: Option<Vec<u64>>,
+    /// One per member in `published`, in its order (`rtt::bucket`).
+    round_trips: Vec<u8>,
 }
 
 /// The pair keys of the members we reach now, sorted: those with an active session
-/// that answer.
-async fn measure(state: &AppState) -> Vec<u64> {
+/// that answer; and our round trip to each, as a bucket, held near a bound to the one
+/// told before (`before`, by pair key: `rtt::bucket`).
+async fn measure(state: &AppState, before: &HashMap<u64, u8>) -> (Vec<u64>, Vec<u8>) {
     let now = Instant::now();
     let mut reached = Vec::new();
     for (peer, peer_state) in state.peers_snapshot().await {
@@ -53,21 +57,37 @@ async fn measure(state: &AppState) -> Vec<u64> {
             && !peer_state.answer_overdue(now)
             && matches!(peer_state.standing, Standing::Member { .. })
         {
-            reached.push(topology::pair_key(&peer));
+            reached.push((topology::pair_key(&peer), peer));
         }
     }
     reached.sort_unstable();
-    reached
+    let rtts = state.lock_rtt();
+    let round_trips = reached
+        .iter()
+        .map(|(key, peer)| {
+            let rtt = rtts.effective(peer, now, state.timing.rtt_fresh_for());
+            rtt::bucket(rtt, before.get(key).copied())
+        })
+        .collect();
+    (
+        reached.into_iter().map(|(key, _)| key).collect(),
+        round_trips,
+    )
 }
 
 /// Whom we reach, as we tell it (`Status::reaches`); empty until first measured.
 pub(crate) fn published(state: &AppState) -> Vec<u64> {
-    state
-        .reach
-        .lock_or_recover()
-        .published
-        .clone()
-        .unwrap_or_default()
+    published_with_round_trips(state).0
+}
+
+/// Whom we reach and our round trip to each, as we tell it (`Status::reaches`,
+/// `Status::round_trips`); empty until first measured.
+pub(crate) fn published_with_round_trips(state: &AppState) -> (Vec<u64>, Vec<u8>) {
+    let reach = state.reach.lock_or_recover();
+    match &reach.published {
+        Some(keys) => (keys.clone(), reach.round_trips.clone()),
+        None => (Vec::new(), Vec::new()),
+    }
 }
 
 /// With the peer worker's periodic pass (every 5 s), and at once when a member we said
@@ -75,13 +95,24 @@ pub(crate) fn published(state: &AppState) -> Vec<u64> {
 /// our neighbours are told at once (a STATUS each, unanswered unless they lack some of
 /// our heads). A change that is undone within a pass isn't told.
 pub(crate) async fn check(state: &AppState) {
-    let now = measure(state).await;
+    let before: HashMap<u64, u8> = {
+        let reach = state.reach.lock_or_recover();
+        reach
+            .published
+            .iter()
+            .flatten()
+            .copied()
+            .zip(reach.round_trips.iter().copied())
+            .collect()
+    };
+    let (now, round_trips) = measure(state, &before).await;
     {
         let mut reach = state.reach.lock_or_recover();
-        if reach.published.as_ref() == Some(&now) {
+        if reach.published.as_ref() == Some(&now) && reach.round_trips == round_trips {
             return;
         }
         reach.published = Some(now.clone());
+        reach.round_trips = round_trips;
     }
     debug!(
         reached = now.len(),