From 02989b6a22a7bbd2151d6a4ea4fe6dddc9f2995b Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 01:55:13 +0100 Subject: links: add failover --- src/swarm/links.rs | 143 +++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 129 insertions(+), 14 deletions(-) (limited to 'src/swarm/links.rs') 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, + /// How we ask each member (`split`). + routes: HashMap, +} + +/// 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, + asked: Instant, + arrived: bool, + failed: HashMap, 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, 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, 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, exclude: Option<&str>, + through_peers: bool, ) -> (Vec, HashMap>) { let mut direct = Vec::new(); let mut through: HashMap> = 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, Instant>, +) -> Option> { + 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, 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) { + 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 { +/// 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, Instant>, +) -> Option { 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 -- cgit v1.3.1