aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/working_links.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/working_links.rs')
-rw-r--r--src/swarm/working_links.rs222
1 files changed, 222 insertions, 0 deletions
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
+ );
+ }
+}