measure 2-hop neighbourhood

7 files changed, 136 insertions, 8 deletions

diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 8259ed5..322170b 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -10,6 +10,7 @@ pub(crate) mod links;
 pub(crate) mod members;
 pub(crate) mod pair;
 pub(crate) mod peer;
+pub(crate) mod reach;
 pub(crate) mod relay;
 pub(crate) mod rtt;
 pub(crate) mod skew;
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 37fcba8..c04b1ad 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -34,11 +34,14 @@ pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
 
 /// Sends `peer` a STATUS announcing our new heads (`swarm::relay`): no keepalive, so it
 /// leaves the keepalives' schedule as it is, and it is answered only if the peer's heads
-/// differ (`Status::announcing`).
-pub(crate) async fn announce_heads(state: &AppState, peer: &str) -> Result<()> {
+/// differ (`Status::announcing`). Whom we reach only `with_reaches` (`swarm::reach`):
+/// announcements of heads go often.
+pub(crate) async fn announce_heads(state: &AppState, peer: &str, with_reaches: bool) -> Result<()> {
+    let local = local_status(state, peer).await;
     let status = transport::Status {
         announcing: true,
-        ..local_status(state, peer).await
+        reaches: local.reaches.filter(|_| with_reaches),
+        ..local
     };
     state
         .transport
@@ -192,6 +195,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
         leaving: state.is_closing(),
         announcing: false,
         eager,
+        reaches: Some(crate::swarm::reach::published(state)),
     }
 }
 
@@ -229,6 +233,9 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport
         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()));
+        }
         drop(peer_state);
         if left {
             info!(peer = %peer, "Peer said farewell; dormant until it is heard from again");
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index a803922..45e8d8f 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -164,6 +164,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             forget_strangers(&state).await;
             crate::swarm::members::ensure_registered(&state).await;
             crate::swarm::stability::tick(&state).await;
+            crate::swarm::reach::check(&state).await;
             crate::swarm::skew::check(&state).await;
             crate::swarm::stats::measure_saved_size(&state);
             crate::swarm::source::warn_if_stalled(&state);
diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs
new file mode 100644
index 0000000..17b189d
--- /dev/null
+++ b/src/swarm/reach.rs
@@ -0,0 +1,119 @@
+//! Whom each neighbour reaches: a node's 2-hop neighbourhood, as measured, not as the
+//! member list says.
+//!
+//! A node knows its own links, and from them whom it reaches: the members it has an
+//! active session with (handshake done, neither dormant nor unreachable). 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
+//! STATUS_REPLY, and soon (the peer worker looks every 5 s) to every neighbour when it
+//! changed,
+//! but not in announcements of heads, which go often. So every node knows whom its
+//! neighbours reach, as of their latest STATUS: for as long as that is fresh
+//! (`held_for`), then not, until the next. Nothing goes into the document, and nobody
+//! knows more than two hops out.
+//!
+//! What it is for: passing changes on around links that don't work, and to exactly the
+//! members the sender doesn't reach; a member a neighbour reaches is one worth trying,
+//! and a way to it. Not for choosing multipoint relays (as OLSR does): over the member
+//! list's graph, which is close to random, neighbourhoods hardly overlap, and nearly
+//! every neighbour would be one (13.7 of 13.7 at 250 members and more), saving next to
+//! nothing but in a full mesh.
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::debug;
+
+use crate::locks::MutexExt;
+use crate::state::{AppState, HelloState, Standing};
+use crate::swarm::peer::announce_heads;
+use crate::swarm::{members, topology};
+
+/// Whom we reach, as last told our neighbours.
+#[derive(Default)]
+pub(crate) struct Reach {
+    published: Option<Vec<u64>>,
+}
+
+/// The pair keys of the members we reach now, sorted: those with an active session.
+async fn measure(state: &AppState) -> Vec<u64> {
+    let mut reached = Vec::new();
+    for (peer, peer_state) in state.peers_snapshot().await {
+        let peer_state = peer_state.lock().await;
+        if peer_state.hello_state == HelloState::Established
+            && !peer_state.liveness.is_dormant()
+            && !peer_state.is_unreachable()
+            && matches!(peer_state.standing, Standing::Member { .. })
+        {
+            reached.push(topology::pair_key(&peer));
+        }
+    }
+    reached.sort_unstable();
+    reached
+}
+
+/// 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()
+}
+
+/// With the peer worker's periodic pass (every 5 s): whom we reach now; if it changed,
+/// our neighbours are told at once (a STATUS each, unanswered if the heads are the
+/// same). A change that is undone within a pass isn't told.
+pub(crate) async fn check(state: &AppState) {
+    let now = measure(state).await;
+    {
+        let mut reach = state.reach.lock_or_recover();
+        if reach.published.as_ref() == Some(&now) {
+            return;
+        }
+        reach.published = Some(now.clone());
+    }
+    debug!(
+        reached = now.len(),
+        "Whom we reach changed; telling our neighbours"
+    );
+    for (peer, peer_state) in state.peers_snapshot().await {
+        let established = peer_state.lock().await.hello_state == HelloState::Established;
+        if !established || !topology::is_neighbour(state, &peer) {
+            continue;
+        }
+        if let Err(error) = tell(state, &peer).await {
+            debug!(peer = %peer, error = %format!("{error:#}"), "Telling whom we reach failed");
+        }
+    }
+}
+
+/// A STATUS to `peer` with whom we reach: as an announcement (unanswered if the heads
+/// are the same), but with `reaches`.
+async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> {
+    announce_heads(state, peer, true).await
+}
+
+/// How long what a neighbour said it reaches holds: past two of its regular STATUSes,
+/// each of which says it again.
+pub(crate) fn held_for(state: &AppState) -> Duration {
+    state.timing.status_refresh * 2 + state.timing.keepalive
+}
+
+/// The members `peer` reaches, as it last said, while that holds (`held_for`); by
+/// pair key.
+pub(crate) fn of(state: &AppState, said: Option<&(Instant, Vec<u64>)>) -> Option<Vec<u64>> {
+    said.filter(|(at, _)| at.elapsed() < held_for(state))
+        .map(|(_, reaches)| reaches.clone())
+}
+
+/// The members of our member list by pair key, to name what neighbours say they reach.
+pub(crate) async fn names(state: &AppState) -> HashMap<u64, String> {
+    let replica = state.doc.read().await;
+    members::members(&replica.doc)
+        .into_iter()
+        .map(|member| (topology::pair_key(&member), member))
+        .collect()
+}
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index e11bd69..69d73d3 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -122,7 +122,7 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
     }
     state.stats.announce_round();
     for peer in neighbours {
-        if let Err(error) = announce_heads(state, &peer).await {
+        if let Err(error) = announce_heads(state, &peer, false).await {
             debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
         }
     }
@@ -268,7 +268,7 @@ async fn graft(state: &AppState, peer: &str) {
     }
     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).await {
+    if let Err(error) = announce_heads(state, peer, false).await {
         debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is eager failed");
     }
 }
@@ -281,7 +281,7 @@ pub(crate) async fn prune(state: &AppState, peer: &str) {
     }
     state.stats.link_pruned();
     debug!(peer = %peer, "Link lazy: its changes reach us first another way");
-    if let Err(error) = announce_heads(state, peer).await {
+    if let Err(error) = announce_heads(state, peer, false).await {
         debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed");
     }
 }
diff --git a/src/swarm/topology/graph.rs b/src/swarm/topology/graph.rs
index fb33170..5401506 100644
--- a/src/swarm/topology/graph.rs
+++ b/src/swarm/topology/graph.rs
@@ -203,7 +203,7 @@ pub(crate) fn pair_score(a: &str, b: &str) -> u64 {
 }
 
 /// What a member contributes to its pair scores: computed once per member.
-pub(super) fn pair_key(member: &str) -> u64 {
+pub(crate) fn pair_key(member: &str) -> u64 {
     let digest = hash(PAIR_LABEL, &[member]);
     u64::from_be_bytes(digest[..8].try_into().expect("8 bytes"))
 }
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index e5dacdd..1c1a6c4 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -35,8 +35,8 @@ use crate::swarm::members;
 
 mod graph;
 
-pub(crate) use graph::pair_score;
 use graph::*;
+pub(crate) use graph::{pair_key, pair_score};
 
 /// This node's current neighbours, and the member list they were computed from.
 #[derive(Debug, Default)]