3 files changed, 138 insertions, 14 deletions

diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index 29e460f..2c41f0b 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -66,6 +66,12 @@ pub(crate) struct NeighbourView {
     /// (`member_list_graph`): computed when first asked for, kept while the member list
     /// stays the same.
     whole: std::sync::OnceLock<Arc<BTreeMap<String, BTreeSet<String>>>>,
+    /// Where members recorded their links differ from that graph (`swarm::working_links`),
+    /// those with a record.
+    recorded: BTreeMap<String, members::RecordedLinks>,
+    /// That graph with the links members recorded (`working_graph`): computed when first
+    /// asked for, kept while the member list and the records stay the same.
+    working: std::sync::OnceLock<Arc<BTreeMap<String, BTreeSet<String>>>>,
 }
 
 /// Every member's neighbours as the member list gives them, as if every member reached
@@ -100,6 +106,59 @@ pub(crate) fn member_list_graph(
     })))
 }
 
+/// The member list's graph (`member_list_graph`) with the links members recorded
+/// (`swarm::working_links`): without those either end recorded as unreached, with those
+/// one recorded as reached (`with_recorded`). The same on every node that has the same
+/// document; what the writers' trees are built on (`swarm::tree`). None until the view
+/// is known.
+pub(crate) fn working_graph(state: &AppState) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
+    let whole = member_list_graph(state)?;
+    let view = state.neighbours.read_or_recover();
+    let view = view.as_ref()?;
+    Some(Arc::clone(view.working.get_or_init(|| {
+        if view.recorded.is_empty() {
+            whole
+        } else {
+            Arc::new(with_recorded(&whole, &view.recorded))
+        }
+    })))
+}
+
+/// `graph` with the links members recorded (`recorded`): those one recorded as reached
+/// added, between members `graph` has; then those either end recorded as unreached
+/// taken out (a link one end reaches and the other doesn't counts as down).
+pub(crate) fn with_recorded(
+    graph: &BTreeMap<String, BTreeSet<String>>,
+    recorded: &BTreeMap<String, members::RecordedLinks>,
+) -> BTreeMap<String, BTreeSet<String>> {
+    let mut graph = graph.clone();
+    for (member, links) in recorded {
+        for other in &links.reached {
+            if member != other && graph.contains_key(member) && graph.contains_key(other) {
+                graph
+                    .entry(member.clone())
+                    .or_default()
+                    .insert(other.clone());
+                graph
+                    .entry(other.clone())
+                    .or_default()
+                    .insert(member.clone());
+            }
+        }
+    }
+    for (member, links) in recorded {
+        for other in &links.unreached {
+            if let Some(neighbours) = graph.get_mut(member) {
+                neighbours.remove(other);
+            }
+            if let Some(neighbours) = graph.get_mut(other) {
+                neighbours.remove(member);
+            }
+        }
+    }
+    graph
+}
+
 /// Whether `peer` is in the member list, as the view was last computed from it; None
 /// while the view isn't known.
 pub(crate) fn listed(state: &AppState, peer: &str) -> Option<bool> {
@@ -382,7 +441,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
     if !state.is_ready().await {
         return None;
     }
-    let (current, stable, transports, alive, k) = {
+    let (current, stable, transports, alive, k, recorded) = {
         let replica = state.doc.read().await;
         members::apply(state, &replica.doc);
         let k = swarm_neighbours(&replica.doc, state.neighbours_k);
@@ -404,7 +463,12 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
             .iter()
             .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
             .collect();
-        (current, stable, transports, alive, k)
+        let recorded: BTreeMap<String, members::RecordedLinks> = current
+            .iter()
+            .map(|m| (m.clone(), members::links(&replica.doc, m)))
+            .filter(|(_, links)| *links != members::RecordedLinks::default())
+            .collect();
+        (current, stable, transports, alive, k, recorded)
     };
     if k != state.neighbours_k {
         info_once(&format!(
@@ -486,26 +550,33 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
                 && view.stable == stable
                 && view.transports == transports
                 && view.local_transports == local_transports
+                && view.recorded == recorded
         }) {
             return None;
         }
         view.as_ref().map(|view| {
-            // The member list's graph stays while the member list does.
-            let whole = if view.members == current
+            // The member list's graph stays while the member list does; with the
+            // records, while they do too.
+            let same_list = view.members == current
                 && view.k == k
                 && view.stable == stable
-                && view.transports == transports
-            {
+                && view.transports == transports;
+            let whole = if same_list {
                 view.whole.clone()
             } else {
                 Default::default()
             };
-            (view.neighbours.clone(), whole)
+            let working = if same_list && view.recorded == recorded {
+                view.working.clone()
+            } else {
+                Default::default()
+            };
+            (view.neighbours.clone(), whole, working)
         })
     };
-    let (previous, whole) = match previous {
-        Some((previous, whole)) => (Some(previous), whole),
-        None => (None, Default::default()),
+    let (previous, whole, working) = match previous {
+        Some((previous, whole, working)) => (Some(previous), whole, working),
+        None => (None, Default::default(), Default::default()),
     };
 
     // Before the first view everyone known counted as a neighbour.
@@ -567,6 +638,8 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         outbound,
         neighbours: computed,
         whole,
+        recorded,
+        working,
     });
     // New neighbours are kept alive and judged.
     state.peers_due.now();
diff --git a/src/swarm/topology/tests.rs b/src/swarm/topology/tests.rs
index 1b9c75d..ede3b6b 100644
--- a/src/swarm/topology/tests.rs
+++ b/src/swarm/topology/tests.rs
@@ -489,3 +489,54 @@ fn the_swarms_neighbours_count() {
     }
     assert!(affects_neighbours(&neighbours_path()));
 }
+
+/// The links members recorded: one either end recorded as unreached is out of the
+/// graph, both ways, even where the other end recorded it as reached; one recorded as
+/// reached is in, between members the graph has; the rest stays.
+#[test]
+fn recorded_links_change_the_graph() {
+    use crate::swarm::members::RecordedLinks;
+    let name = |n: &str| n.to_owned();
+    let set = |names: &[&str]| names.iter().map(|n| name(n)).collect::<BTreeSet<String>>();
+    let ring: BTreeMap<String, BTreeSet<String>> = [
+        ("a", set(&["b", "d"])),
+        ("b", set(&["a", "c"])),
+        ("c", set(&["b", "d"])),
+        ("d", set(&["c", "a"])),
+    ]
+    .into_iter()
+    .map(|(member, neighbours)| (name(member), neighbours))
+    .collect();
+    let recorded: BTreeMap<String, RecordedLinks> = [
+        (
+            name("a"),
+            RecordedLinks {
+                unreached: set(&["b"]),
+                reached: set(&["c", "stranger"]),
+            },
+        ),
+        (
+            name("b"),
+            RecordedLinks {
+                unreached: BTreeSet::new(),
+                reached: set(&["d"]),
+            },
+        ),
+        (
+            name("d"),
+            RecordedLinks {
+                unreached: set(&["b"]),
+                reached: BTreeSet::new(),
+            },
+        ),
+    ]
+    .into_iter()
+    .collect();
+    let changed = with_recorded(&ring, &recorded);
+    assert_eq!(changed[&name("a")], set(&["c", "d"]));
+    assert_eq!(changed[&name("b")], set(&["c"]));
+    assert_eq!(changed[&name("c")], set(&["a", "b", "d"]));
+    assert_eq!(changed[&name("d")], set(&["a", "c"]));
+    assert!(!changed.contains_key("stranger"));
+    assert_eq!(with_recorded(&ring, &BTreeMap::new()), ring);
+}
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index 9c9651e..b9491b3 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -2,9 +2,9 @@
 //! link that works, and as soon as the member before it has them.
 //!
 //! The tree is the same on every node, computed from what they share: the graph the
-//! member list gives (`topology::member_list_graph`: every member's neighbours, as if
-//! all reached all) and what each member tells its neighbours it reaches
-//! (`swarm::reach`). Every member's distance from the writer over that graph orders the
+//! member list gives, with the links members recorded as not working taken out and those
+//! it lacks that do added (`topology::working_graph`, `swarm::working_links`), and what
+//! each member tells its neighbours it reaches (`swarm::reach`). Every member's distance from the writer over that graph orders the
 //! members: by distance, then whether it reaches a member a hop closer, then pair key;
 //! a member's parent is the neighbour it reaches that comes before it, the closest to
 //! the writer first (a link much slower than its fastest to one counting a hop more),
@@ -53,7 +53,7 @@ pub(crate) struct Trees {
 /// The graph, and every member's distance from `writer` (by pair key) over it; None
 /// while the member list isn't known, or `writer` isn't in it.
 fn distances(state: &AppState, writer: u64) -> Option<(Arc<Graph>, Arc<HashMap<String, u32>>)> {
-    let graph = topology::member_list_graph(state)?;
+    let graph = topology::working_graph(state)?;
     let mut trees = state.trees.lock_or_recover();
     if !trees
         .graph