path: root/src/swarm/working_links.rs

//! 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>,
    /// 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<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 = {
        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<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
        );
    }
}