links: add failover

1 files changed, 129 insertions, 14 deletions

diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index 14c5cf6..c430d58 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -29,6 +29,11 @@
 //! `MIN_BATCH` at least). Such a
 //! subscription isn't ended by a message: it runs out, at every hop, unless the asker
 //! renews it, so one that is gone (or no longer looks) stops it within a lease.
+//!
+//! A way of asking a member that brings none of its links (`FAILED_AFTER`: we don't
+//! reach it after all, or the relay doesn't) is left out a while for that member, and
+//! the next ask (the renewal) goes the next way: so the old way runs out by itself, at
+//! no cost but the wait. Every hop does so for what it asks on.
 
 use std::collections::{HashMap, HashSet};
 use std::sync::Mutex;
@@ -80,6 +85,12 @@ const RELAY_BURST: f64 = 512.0;
 const RELAY_RATE: f64 = 32.0;
 /// Nodes a relayed LINKS goes through at most, the first included.
 const MAX_HOPS: u8 = 3;
+/// A way of asking a member (directly, or through a peer) that brought none of its
+/// links for this long after we asked has failed it (`split`): the next ask goes
+/// another way. More than a window (`MAX_BATCH`) along a few hops of slow transports.
+const FAILED_AFTER: Duration = Duration::from_secs(20);
+/// How long a way that failed a member is left out for that member.
+const FAILED_KEPT: Duration = Duration::from_secs(300);
 
 /// The requests out and the replies in, and when each asker was last answered.
 #[derive(Default)]
@@ -112,6 +123,18 @@ struct Inner {
     /// How many members each peer may still have us ask for it (`RELAY_BURST`,
     /// `RELAY_RATE`), as of when.
     relay_budget: HashMap<String, (f64, Instant)>,
+    /// How we ask each member (`split`).
+    routes: HashMap<String, Route>,
+}
+
+/// How we ask a member: the way we last did, since when, whether its links came since,
+/// and the ways that failed it lately (`FAILED_AFTER`), each since when.
+struct Route {
+    /// Through this peer; None: directly.
+    via: Option<String>,
+    asked: Instant,
+    arrived: bool,
+    failed: HashMap<Option<String>, Instant>,
 }
 
 /// One asker's relayed subscription to a member.
@@ -333,7 +356,7 @@ pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) {
 /// on (`DEFAULT_BATCH`).
 #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
 pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: Duration) {
-    let (direct, through) = split(state, members, None).await;
+    let (direct, through) = split(state, members, None, true).await;
     for member in direct {
         let state = state.clone();
         tokio::spawn(async move { ask(&state, &member, subscribe).await });
@@ -352,11 +375,15 @@ pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: D
 }
 
 /// `members` we can reach ourselves, and the others by the peer to ask them through
-/// (none for those no peer leads to); not `exclude` (who asked us).
+/// (none for those no peer leads to, nor without `through_peers`); not `exclude` (who
+/// asked us). A way that brought none of a member's links since we last asked it that
+/// way (`FAILED_AFTER`) is left out for a while (`FAILED_KEPT`): directly, the next
+/// relay, and so on; once every way failed, they are all tried again.
 async fn split(
     state: &AppState,
     members: Vec<String>,
     exclude: Option<&str>,
+    through_peers: bool,
 ) -> (Vec<String>, HashMap<String, Vec<String>>) {
     let mut direct = Vec::new();
     let mut through: HashMap<String, Vec<String>> = HashMap::new();
@@ -364,25 +391,111 @@ async fn split(
         if member == state.local_id() || Some(member.as_str()) == exclude {
             continue;
         }
-        if topology::reaches(state, &member).await {
-            direct.push(member);
-        } else if let Some(relay) = relay_for(state, &member, exclude).await {
-            through.entry(relay).or_default().push(member);
+        let failed = failed_ways(state, &member);
+        let mut via = way(state, &member, exclude, through_peers, &failed).await;
+        if via.is_none() && !failed.is_empty() {
+            forget_failed(state, &member);
+            via = way(state, &member, exclude, through_peers, &HashMap::new()).await;
+        }
+        let Some(via) = via else {
+            continue;
+        };
+        asked_by(state, &member, via.clone());
+        match via {
+            None => direct.push(member),
+            Some(relay) => through.entry(relay).or_default().push(member),
         }
     }
     (direct, through)
 }
 
+/// How to ask `member`, but the `failed` ways: Some(None) directly, Some(Some(peer))
+/// through `peer` (if `through_peers`); None if no way is left.
+async fn way(
+    state: &AppState,
+    member: &str,
+    exclude: Option<&str>,
+    through_peers: bool,
+    failed: &HashMap<Option<String>, Instant>,
+) -> Option<Option<String>> {
+    if !failed.contains_key(&None) && topology::reaches(state, member).await {
+        return Some(None);
+    }
+    if !through_peers {
+        return None;
+    }
+    relay_for(state, member, exclude, failed).await.map(Some)
+}
+
+/// The ways that failed `member` lately; the one we last asked it by among them if it
+/// brought nothing within `FAILED_AFTER`.
+fn failed_ways(state: &AppState, member: &str) -> HashMap<Option<String>, Instant> {
+    let now = Instant::now();
+    let mut inner = state.links.inner.lock_or_recover();
+    let Some(route) = inner.routes.get_mut(member) else {
+        return HashMap::new();
+    };
+    route
+        .failed
+        .retain(|_, since| now.saturating_duration_since(*since) < FAILED_KEPT);
+    if !route.arrived && now.saturating_duration_since(route.asked) >= FAILED_AFTER {
+        debug!(member = %member, via = ?route.via, "LINKS brought nothing that way; asking another");
+        route.failed.insert(route.via.clone(), now);
+    }
+    route.failed.clone()
+}
+
+/// Every way may be tried again for `member`.
+fn forget_failed(state: &AppState, member: &str) {
+    if let Some(route) = state.links.inner.lock_or_recover().routes.get_mut(member) {
+        route.failed.clear();
+    }
+}
+
+/// We ask `member` `via` now: the wait for its links starts over, unless we asked it so
+/// already and are still waiting.
+fn asked_by(state: &AppState, member: &str, via: Option<String>) {
+    let now = Instant::now();
+    let mut inner = state.links.inner.lock_or_recover();
+    if inner.routes.len() >= MAX_KEPT && !inner.routes.contains_key(member) {
+        inner.routes.clear();
+    }
+    let route = inner
+        .routes
+        .entry(member.to_owned())
+        .or_insert_with(|| Route {
+            via: via.clone(),
+            asked: now,
+            arrived: false,
+            failed: HashMap::new(),
+        });
+    if route.via != via || route.arrived {
+        route.via = via;
+        route.asked = now;
+        route.arrived = false;
+    }
+}
+
 /// The peer to ask `member` through, which we can't reach: an active member we reach
 /// (that answers us) that, by the member list, is on a transport `member` is on; else one on a transport
 /// we aren't on (it may know one that is). Of several, the one with the lowest pair score
-/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`.
-async fn relay_for(state: &AppState, member: &str, exclude: Option<&str>) -> Option<String> {
+/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`, nor one
+/// that failed `member` (`failed`).
+async fn relay_for(
+    state: &AppState,
+    member: &str,
+    exclude: Option<&str>,
+    failed: &HashMap<Option<String>, Instant>,
+) -> Option<String> {
     let theirs = topology::transports_of(state, member);
     let ours = topology::transports_of(state, state.local_id());
     let mut best: Option<(bool, u64, String)> = None;
     for (peer, peer_state) in state.peers_snapshot().await {
-        if peer == member || Some(peer.as_str()) == exclude || !state.transport.can_reach(&peer) {
+        if peer == member
+            || Some(peer.as_str()) == exclude
+            || failed.contains_key(&Some(peer.clone()))
+            || !state.transport.can_reach(&peer)
+        {
             continue;
         }
         {
@@ -625,7 +738,7 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
     for member in known {
         forward(state, &member);
     }
-    let (direct, through) = split(state, members, Some(asker)).await;
+    let (direct, through) = split(state, members, Some(asker), request.hops > 0).await;
     for member in direct {
         // Asked once while we subscribe anyway (our own dashboard, or another asker's):
         // what comes is passed on, and a LINKS with 0 would end that subscription.
@@ -641,10 +754,8 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
             ask(state, &member, lease).await;
         }
     }
-    if request.hops > 0 {
-        for (next, members) in through {
-            ask_via(state, &next, members, lease, rest, request.hops - 1).await;
-        }
+    for (next, members) in through {
+        ask_via(state, &next, members, lease, rest, request.hops - 1).await;
     }
 }
 
@@ -783,6 +894,10 @@ pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) {
                 inner.replies.clear();
             }
             let at = now.checked_sub(age).unwrap_or(now);
+            // Arrived, if older news: the way works.
+            if let Some(route) = inner.routes.get_mut(&member) {
+                route.arrived = true;
+            }
             // Not older news over newer.
             if inner
                 .replies