//! 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, /// Checks done (`check`), for the benchmark to wait for one (`settled`). #[cfg(test)] checks: u64, /// Records written, for the soak (`written`). #[cfg(test)] written: u64, } /// How many times we wrote our record. #[cfg(test)] pub(crate) fn written(state: &AppState) -> u64 { state.working_links.lock_or_recover().written } /// How many checks were done, and whether none of our links differs from our record /// (nothing awaits its hold). #[cfg(test)] pub(crate) fn settled(state: &AppState) -> (u64, bool) { let pending = state.working_links.lock_or_recover(); (pending.checks, pending.since.is_empty()) } /// 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 = 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 = { let mut pending = state.working_links.lock_or_recover(); #[cfg(test)] { pending.checks += 1; } decide(&mut pending, &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(()) => { #[cfg(test)] { state.working_links.lock_or_recover().written += 1; } 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 { 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 ); } }