working links recorded

8 files changed, 339 insertions, 18 deletions

diff --git a/docs/peers.md b/docs/peers.md
index f19e699..0b31556 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -358,6 +358,16 @@ pair exchanges changes it sends no keepalives, and so no STATUS, however long th
 writes go on. `GET /peers` shows it
 (`localReaches`, and `reaches` per peer).
 
+Where its links differ from the member list's graph for a while, a member records so in
+its registry entry ([`swarm/working_links.rs`](../src/swarm/working_links.rs), `links`):
+its neighbours by the graph it doesn't reach (`unreached`), once down for four keepalive
+intervals, and its neighbours it reaches that the graph doesn't link it to (`reached`:
+a ring going round a member it doesn't reach, a member reaching out from behind NAT),
+once up for one; each comes out again after the
+other hold. On a stable swarm the entry has none. A member that went away is recorded
+as unreached by its neighbours (they can't tell it from a link that broke), and taken
+out once they reach it again.
+
 A writer's changes pass on along its tree ([`swarm/tree.rs`](../src/swarm/tree.rs)), the same on every node:
 built on the graph the member list gives (as if every member reached every other) and
 on whom each member says it reaches. Members are ordered by their distance from the
diff --git a/src/node.rs b/src/node.rs
index 5bfce31..9322b75 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -144,6 +144,7 @@ impl Node {
             announce_slow: Default::default(),
             slow_rounds: Default::default(),
             reach: Default::default(),
+            working_links: Default::default(),
             trees: Default::default(),
             rtt: Default::default(),
             throughput: Default::default(),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index e127877..d1a32df 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -160,6 +160,8 @@ pub(crate) struct AppState {
     pub(crate) trees: Arc<std::sync::Mutex<crate::swarm::tree::Trees>>,
     /// Whom we reach, as told our neighbours (`swarm::reach`).
     pub(crate) reach: Arc<std::sync::Mutex<crate::swarm::reach::Reach>>,
+    /// Links that differ from our record, and since when (`swarm::working_links`).
+    pub(crate) working_links: Arc<std::sync::Mutex<crate::swarm::working_links::Pending>>,
     /// Our neighbours, once known (`swarm::topology`).
     pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,
     /// Something the neighbours are computed from may have changed: the member list or a
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index a534c04..1cb4062 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -12,11 +12,12 @@
 //! for people reading /peers and metrics (not unique and never an identity: the public
 //! key stays that), where people reach its dashboard (`--dashboard-url`, `dashboard`),
 //! where it can be reached (`addresses`, as its transport gives them:
-//! hints for the others' transports), and an `alive` stamp the node renews now and
+//! hints for the others' transports), an `alive` stamp the node renews now and
 //! then: the members renewed within `--electorate-window` are the electorate of a
-//! remake.
+//! remake, and where its links differ from the member list's graph (`links`,
+//! `swarm::working_links`).
 
-use std::collections::HashMap;
+use std::collections::{BTreeSet, HashMap};
 use std::time::Duration;
 
 use serde_json::{Value as JsonValue, json};
@@ -426,6 +427,77 @@ pub(crate) async fn ensure_registered(state: &AppState) {
     }
 }
 
+fn links_path(pk: &str) -> Vec<String> {
+    let mut path = member_path(pk);
+    path.push("links".to_owned());
+    path
+}
+
+/// The most members a member may record in each of `RecordedLinks`' sets: more than it
+/// could hold sessions with.
+pub(crate) const MAX_RECORDED_LINKS: usize = 256;
+
+/// Where a member's links differ from the member list's graph, as it recorded them
+/// (`links`, `swarm::working_links`): its neighbours by the graph it doesn't reach
+/// (`unreached`), and the members it reaches the graph doesn't link it to (`reached`).
+/// Both empty while every link the graph gives works, and no other does.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub(crate) struct RecordedLinks {
+    pub(crate) unreached: BTreeSet<String>,
+    pub(crate) reached: BTreeSet<String>,
+}
+
+impl RecordedLinks {
+    fn is_empty(&self) -> bool {
+        self.unreached.is_empty() && self.reached.is_empty()
+    }
+}
+
+/// `pk`'s recorded links; none if its entry doesn't say (at most `MAX_RECORDED_LINKS`
+/// in each set).
+pub(crate) fn links(doc: &impl ReadDoc, pk: &str) -> RecordedLinks {
+    let Ok(Some(JsonValue::Object(links))) = store::read_path(doc, &links_path(pk)) else {
+        return RecordedLinks::default();
+    };
+    let set = |name: &str| -> BTreeSet<String> {
+        match links.get(name) {
+            Some(JsonValue::Array(items)) => items
+                .iter()
+                .filter_map(JsonValue::as_str)
+                .take(MAX_RECORDED_LINKS)
+                .map(str::to_owned)
+                .collect(),
+            _ => BTreeSet::new(),
+        }
+    };
+    RecordedLinks {
+        unreached: set("unreached"),
+        reached: set("reached"),
+    }
+}
+
+/// Records `pk`'s `links` (at most `MAX_RECORDED_LINKS` in each set); none removes them.
+pub(crate) fn set_links(
+    doc: &mut impl Transactable,
+    pk: &str,
+    links: &RecordedLinks,
+) -> Result<(), StoreError> {
+    if links.is_empty() {
+        return match store::delete_path(doc, &links_path(pk), false) {
+            Err(StoreError::NotFound) => Ok(()),
+            result => result,
+        };
+    }
+    let list = |set: &BTreeSet<String>| -> Vec<String> {
+        set.iter().take(MAX_RECORDED_LINKS).cloned().collect()
+    };
+    let value = json!({
+        "unreached": list(&links.unreached),
+        "reached": list(&links.reached),
+    });
+    store::write_path(doc, &links_path(pk), &value)
+}
+
 /// Removes `pk`'s entry (an operator forgot it). Absent is fine.
 pub(crate) fn unregister(doc: &mut impl Transactable, pk: &str) -> Result<(), StoreError> {
     match store::delete_path(doc, &member_path(pk), false) {
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index ac62c64..7014438 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -24,3 +24,4 @@ pub(crate) mod sync;
 pub(crate) mod throughput;
 pub(crate) mod topology;
 pub(crate) mod tree;
+pub(crate) mod working_links;
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 9ddf177..e17eb34 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -215,6 +215,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             crate::swarm::members::ensure_registered(&state).await;
             crate::swarm::stability::tick(&state).await;
             crate::swarm::reach::check(&state).await;
+            crate::swarm::working_links::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
index d4949ef..760bcb5 100644
--- a/src/swarm/reach.rs
+++ b/src/swarm/reach.rs
@@ -15,8 +15,9 @@
 //! reach, as of their latest STATUS, for as long as the session lasts (a
 //! neighbour that goes dormant, or starts anew, says it again). Not for a set time:
 //! while a pair exchanges changes it sends no keepalives, and so no STATUS, for as
-//! long as the writes go on. Nothing goes into the document, and nobody knows more than
-//! two hops out.
+//! long as the writes go on. Nothing of it goes into the document (but where a link
+//! stays down for a while: `swarm::working_links`), 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,
@@ -25,7 +26,7 @@
 //! 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::collections::{BTreeSet, HashMap};
 
 use tokio::time::Instant;
 use tracing::debug;
@@ -77,18 +78,11 @@ pub(crate) struct Reach {
 /// 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 {
-        let peer_state = peer_state.lock().await;
-        if peer_state.hello_state == HelloState::Established
-            && !peer_state.liveness.is_dormant()
-            && !peer_state.is_unreachable()
-            && !peer_state.answer_overdue(now)
-            && matches!(peer_state.standing, Standing::Member { .. })
-        {
-            reached.push((topology::pair_key(&peer), peer));
-        }
-    }
+    let mut reached: Vec<(u64, String)> = reached_members(state, now)
+        .await
+        .into_iter()
+        .map(|peer| (topology::pair_key(&peer), peer))
+        .collect();
     reached.sort_unstable();
     let rtts = state.lock_rtt();
     let round_trips = reached
@@ -104,6 +98,24 @@ async fn measure(state: &AppState, before: &HashMap<u64, u8>) -> (Vec<u64>, Vec<
     )
 }
 
+/// The members we reach now: those with an active session (handshake done, neither
+/// dormant nor unreachable) that answer.
+pub(crate) async fn reached_members(state: &AppState, now: Instant) -> BTreeSet<String> {
+    let mut reached = BTreeSet::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()
+            && !peer_state.answer_overdue(now)
+            && matches!(peer_state.standing, Standing::Member { .. })
+        {
+            reached.insert(peer);
+        }
+    }
+    reached
+}
+
 /// Whom we reach, as we tell it (`Status::reaches`); empty until first measured.
 pub(crate) fn published(state: &AppState) -> Vec<u64> {
     published_with_round_trips(state).0
diff --git a/src/swarm/working_links.rs b/src/swarm/working_links.rs
new file mode 100644
index 0000000..621ea6c
--- /dev/null
+++ b/src/swarm/working_links.rs
@@ -0,0 +1,222 @@
+//! Which links work, as every member records its own in its registry entry (`links`,
+//! `members::RecordedLinks`): where they differ from the member list's graph
+//! (`topology::member_list_graph`), which takes every member as reaching every other.
+//!
+//! Whom a member reaches it tells its neighbours (`swarm::reach`), but only them, and
+//! only for as long as their sessions last; the writers' trees are computed by every
+//! node alike from what every node has, and that is the document. So each member
+//! records its neighbours by the graph that it doesn't reach (`unreached`), and its
+//! neighbours it reaches that the graph doesn't link it to (`reached`: a ring that goes
+//! round a member it doesn't reach, a member behind NAT that reaches out to one the
+//! graph doesn't give it), each only once it has stayed so for a while: going down for
+//! `DOWN_FOR` keepalive intervals, up for `UP_FOR`, so a link that drops for a
+//! keepalive isn't written down, and one back is soon again. A link either end records
+//! as unreached is one that doesn't work; one either records as reached, one that does.
+//! Only neighbours (`topology::our_neighbours`, those we sync with) count as reached:
+//! not every member we have a session with (one that took us as its neighbour, or that
+//! was one before the member list was known), as changes pass on along a tree only
+//! where a sync opens. A member that went away is recorded as unreached by its neighbours (they
+//! can't tell it from a link that broke), and taken out once they reach it again.
+//!
+//! The record changes rarely: on a stable swarm, never; it is written only when a link
+//! has been down or up for its hold.
+
+use std::collections::{BTreeSet, HashMap};
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::{info, warn};
+
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::members::{self, RecordedLinks};
+use crate::swarm::{reach, topology};
+
+/// Keepalive intervals a link stays down before it is recorded so (`unreached`, or out
+/// of `reached`): as long as a silent peer takes to be held dormant (`dormant_after`).
+const DOWN_FOR: u32 = 4;
+/// Keepalive intervals a link stays up before it is recorded so (out of `unreached`,
+/// or into `reached`).
+const UP_FOR: u32 = 1;
+
+/// Since when each member's link has differed from our record.
+#[derive(Default)]
+pub(crate) struct Pending {
+    since: HashMap<String, Instant>,
+}
+
+/// With the peer worker's periodic pass (every 5 s): whom we reach now, against the
+/// member list's graph and our record; a difference that held for long enough goes into
+/// the record.
+pub(crate) async fn check(state: &AppState) {
+    if !state.is_ready().await {
+        return;
+    }
+    let me = state.local_id().to_owned();
+    let Some(graph) = topology::member_list_graph(state) else {
+        return;
+    };
+    let Some(ours) = graph.get(&me) else {
+        // Not listed yet.
+        return;
+    };
+    let now = Instant::now();
+    let reached = reach::reached_members(state, now).await;
+    let neighbours: BTreeSet<String> = topology::our_neighbours(state).into_iter().collect();
+    let links = RecordedLinks {
+        unreached: ours.difference(&reached).cloned().collect(),
+        reached: reached
+            .into_iter()
+            .filter(|member| {
+                neighbours.contains(member) && graph.contains_key(member) && !ours.contains(member)
+            })
+            .collect(),
+    };
+    let recorded = {
+        let replica = state.doc.read().await;
+        members::links(&replica.doc, &me)
+    };
+    let next = decide(
+        &mut state.working_links.lock_or_recover(),
+        &recorded,
+        &links,
+        now,
+        state.timing.keepalive,
+    );
+    if next == recorded {
+        return;
+    }
+    let (unreached, reached) = (next.unreached.len(), next.reached.len());
+    match state
+        .mutate_doc(move |doc| members::set_links(doc, &me, &next))
+        .await
+    {
+        Ok(()) => info!(
+            unreached,
+            reached, "Links that differ from the member list's graph recorded"
+        ),
+        Err(error) => warn!(error = %error, "Failed to record our links"),
+    }
+}
+
+/// The record to keep: `recorded`, but where a member's link has differed from it as
+/// `now_links` says since long enough before `now` (`pending`): down for `DOWN_FOR`
+/// keepalive intervals (newly unreached, or no longer reached), up for `UP_FOR` (reached
+/// again, or newly reached, or no longer our neighbour by the graph).
+fn decide(
+    pending: &mut Pending,
+    recorded: &RecordedLinks,
+    now_links: &RecordedLinks,
+    now: Instant,
+    keepalive: Duration,
+) -> RecordedLinks {
+    let members: BTreeSet<&String> = recorded
+        .unreached
+        .iter()
+        .chain(&recorded.reached)
+        .chain(&now_links.unreached)
+        .chain(&now_links.reached)
+        .collect();
+    pending.since.retain(|member, _| members.contains(member));
+    let mut next = recorded.clone();
+    for member in members {
+        let unreached = now_links.unreached.contains(member);
+        let reached = now_links.reached.contains(member);
+        let was_unreached = recorded.unreached.contains(member);
+        let was_reached = recorded.reached.contains(member);
+        if unreached == was_unreached && reached == was_reached {
+            pending.since.remove(member);
+            continue;
+        }
+        let down = (unreached && !was_unreached) || (was_reached && !reached);
+        let hold = keepalive * if down { DOWN_FOR } else { UP_FOR };
+        let since = *pending.since.entry(member.clone()).or_insert(now);
+        if now.saturating_duration_since(since) < hold {
+            continue;
+        }
+        pending.since.remove(member);
+        for (set, is) in [
+            (&mut next.unreached, unreached),
+            (&mut next.reached, reached),
+        ] {
+            if is {
+                set.insert(member.clone());
+            } else {
+                set.remove(member);
+            }
+        }
+    }
+    next
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn set(members: &[&str]) -> BTreeSet<String> {
+        members.iter().map(|member| (*member).to_owned()).collect()
+    }
+
+    fn links(unreached: &[&str], reached: &[&str]) -> RecordedLinks {
+        RecordedLinks {
+            unreached: set(unreached),
+            reached: set(reached),
+        }
+    }
+
+    /// A link goes into the record once down for `DOWN_FOR` keepalives, not for one; out
+    /// once up for `UP_FOR`; one reached the graph doesn't give, once up for `UP_FOR`.
+    #[test]
+    fn links_are_recorded_after_their_hold() {
+        let keepalive = Duration::from_secs(5);
+        let start = Instant::now();
+        let at = |secs: u64| start + Duration::from_secs(secs);
+        let mut pending = Pending::default();
+        let none = links(&[], &[]);
+
+        // Down for a keepalive, then up again: not recorded.
+        assert_eq!(
+            decide(&mut pending, &none, &links(&["b"], &[]), at(0), keepalive),
+            none
+        );
+        assert_eq!(
+            decide(&mut pending, &none, &links(&["b"], &[]), at(5), keepalive),
+            none
+        );
+        assert_eq!(decide(&mut pending, &none, &none, at(10), keepalive), none);
+        // Down again: from then on, `DOWN_FOR` keepalives.
+        assert_eq!(
+            decide(&mut pending, &none, &links(&["b"], &[]), at(15), keepalive),
+            none
+        );
+        assert_eq!(
+            decide(&mut pending, &none, &links(&["b"], &[]), at(30), keepalive),
+            none
+        );
+        let down = links(&["b"], &[]);
+        assert_eq!(decide(&mut pending, &none, &down, at(35), keepalive), down);
+
+        // Up again: out after `UP_FOR`.
+        assert_eq!(decide(&mut pending, &down, &none, at(40), keepalive), down);
+        assert_eq!(decide(&mut pending, &down, &none, at(45), keepalive), none);
+
+        // Reached beyond the graph: in after `UP_FOR`; gone: out after `DOWN_FOR`.
+        let beyond = links(&[], &["c"]);
+        assert_eq!(
+            decide(&mut pending, &none, &beyond, at(50), keepalive),
+            none
+        );
+        assert_eq!(
+            decide(&mut pending, &none, &beyond, at(55), keepalive),
+            beyond
+        );
+        assert_eq!(
+            decide(&mut pending, &beyond, &none, at(60), keepalive),
+            beyond
+        );
+        assert_eq!(
+            decide(&mut pending, &beyond, &none, at(80), keepalive),
+            none
+        );
+    }
+}