diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/members.rs | 78 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer/worker.rs | 1 | ||||
| -rw-r--r-- | src/swarm/reach.rs | 42 | ||||
| -rw-r--r-- | src/swarm/working_links.rs | 222 |
5 files changed, 326 insertions, 18 deletions
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 + ); + } +} |