indirect probes

1 files changed, 337 insertions, 25 deletions

diff --git a/src/swarm/working_links.rs b/src/swarm/working_links.rs
index af476c4..873baf6 100644
--- a/src/swarm/working_links.rs
+++ b/src/swarm/working_links.rs
@@ -11,26 +11,36 @@
 //! 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.
+//! as unreached is one that doesn't work; one either records as reached, one that does;
+//! so a link the other end records as unreached already isn't recorded by this one.
 //! 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.
+//! 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. So a node first asks up to `PROBE_VIA` of
+//! the member's own neighbours that it reaches whether they reach it (a PROBE, as SWIM
+//! asks others before it suspects; answered from what they know, contacting nobody),
+//! and records the link unless some say they don't and none says it does: a member
+//! away, which its neighbours hold dormant, costs no writes, where each of them would
+//! write twice (going and coming back). Asked ones that can't tell, or don't answer,
+//! leave the link recorded. A node that reaches nobody at all is the
+//! one away, and records no link as unreached.
 //!
 //! 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::collections::{BTreeMap, BTreeSet, HashMap};
 use std::time::Duration;
 
 use tokio::time::Instant;
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
 
 use crate::locks::MutexExt;
 use crate::state::AppState;
 use crate::swarm::members::{self, RecordedLinks};
 use crate::swarm::{reach, topology};
+use crate::transport::{ParsedMessage, ProbeReply, ProbeRequest};
 
 /// 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`).
@@ -39,10 +49,61 @@ const DOWN_FOR: u32 = 4;
 /// or into `reached`).
 const UP_FOR: u32 = 1;
 
+/// Members asked about a member at once at most: of its
+/// neighbours by the graph that we reach, the ones with the lowest pair score with it,
+/// so askers spread out alike.
+const PROBE_VIA: usize = 3;
+/// How long the members asked have to answer (a PROBE_REPLY comes at once: a round
+/// trip, and losses).
+const PROBE_WAIT: Duration = Duration::from_secs(10);
+
+/// What we asked others about a member: when, with which nonce,
+/// whom, and their answers so far (`ProbeReply::reached`).
+struct Probe {
+    asked: Instant,
+    nonce: u64,
+    via: BTreeSet<String>,
+    answers: HashMap<String, Option<bool>>,
+}
+
+/// What the answers to a `Probe` say.
+#[derive(Debug, PartialEq, Eq)]
+enum Verdict {
+    /// Not all in yet, and not overdue.
+    Waiting,
+    /// One of those asked reaches the member: our link to it is what doesn't work.
+    Confirmed,
+    /// Some of those asked don't reach it either, and none does: it is away.
+    Away,
+}
+
+impl Probe {
+    /// One answer that it reaches the member settles it; otherwise once all answered,
+    /// or the rest are overdue: away if one said it doesn't reach it, else confirmed
+    /// (nobody could tell, or nobody answered: the link is recorded).
+    fn verdict(&self, now: Instant) -> Verdict {
+        if self.answers.values().any(|reached| *reached == Some(true)) {
+            return Verdict::Confirmed;
+        }
+        let all_in = self.via.iter().all(|peer| self.answers.contains_key(peer));
+        if !all_in && now.saturating_duration_since(self.asked) < PROBE_WAIT {
+            return Verdict::Waiting;
+        }
+        if self.answers.values().any(|reached| *reached == Some(false)) {
+            Verdict::Away
+        } else {
+            Verdict::Confirmed
+        }
+    }
+}
+
 /// Since when each member's link has differed from our record.
 #[derive(Default)]
 pub(crate) struct Pending {
     since: HashMap<String, Instant>,
+    /// What we asked others about each member, for links held long
+    /// enough that await confirmation.
+    probes: HashMap<String, Probe>,
     /// Checks done (`check`), for the benchmark to wait for one (`settled`).
     #[cfg(test)]
     checks: u64,
@@ -86,24 +147,103 @@ pub(crate) async fn check(state: &AppState) {
     let links = RecordedLinks {
         unreached: ours.difference(&reached).cloned().collect(),
         reached: reached
-            .into_iter()
+            .iter()
+            .cloned()
             .filter(|member| {
                 neighbours.contains(member) && graph.contains_key(member) && !ours.contains(member)
             })
             .collect(),
     };
-    let recorded = {
+    // Our record, and links the other end already records as unreached: either end's
+    // record takes a link out, so ours needs no copy (half the writes for a link that
+    // broke, and a link we can't confirm but the other end could is settled). One we
+    // record already stays: dropping it as the other end drops its own, both at once,
+    // would leave none.
+    let (recorded, links) = {
         let replica = state.doc.read().await;
-        members::links(&replica.doc, &me)
+        let recorded = members::links(&replica.doc, &me);
+        let mut links = links;
+        links.unreached.retain(|member| {
+            recorded.unreached.contains(member)
+                || !members::links(&replica.doc, member).unreached.contains(&me)
+        });
+        (recorded, links)
+    };
+    // Reaching nobody, we are the one away: no link of ours is
+    // recorded as unreached, which would go out, and be taken out, once we're back.
+    let mut links = links;
+    if reached.is_empty() {
+        links.unreached.clear();
+    }
+    // Links newly unreached that others confirm: asked, they
+    // reach the member, or can't tell; or there is nobody to ask.
+    let confirmed = {
+        let pending = state.working_links.lock_or_recover();
+        links
+            .unreached
+            .difference(&recorded.unreached)
+            .filter(|member| {
+                via(&graph, &reached, member).is_empty()
+                    || pending
+                        .probes
+                        .get(*member)
+                        .is_some_and(|probe| probe.verdict(now) == Verdict::Confirmed)
+            })
+            .cloned()
+            .collect::<BTreeSet<String>>()
     };
-    let next = {
+    let (next, to_probe) = {
         let mut pending = state.working_links.lock_or_recover();
         #[cfg(test)]
         {
             pending.checks += 1;
         }
-        decide(&mut pending, &recorded, &links, now, state.timing.keepalive)
+        let (next, awaiting) = decide(
+            &mut pending,
+            &recorded,
+            &links,
+            now,
+            state.timing.keepalive,
+            Some(&confirmed),
+        );
+        // Asked again once away for another hold (it may be back by then, or the link
+        // be all that's down).
+        let hold = state.timing.keepalive * DOWN_FOR;
+        pending.probes.retain(|member, _| awaiting.contains(member));
+        let mut to_probe = Vec::new();
+        for member in awaiting {
+            let due = pending.probes.get(&member).is_none_or(|probe| {
+                probe.verdict(now) == Verdict::Away
+                    && now.saturating_duration_since(probe.asked) >= PROBE_WAIT + hold
+            });
+            if due {
+                let probe = Probe {
+                    asked: now,
+                    nonce: crate::transport::paths::random_nonce(),
+                    via: via(&graph, &reached, &member).into_iter().collect(),
+                    answers: HashMap::new(),
+                };
+                to_probe.push((member.clone(), probe.nonce, probe.via.clone()));
+                pending.probes.insert(member, probe);
+            }
+        }
+        (next, to_probe)
     };
+    for (member, nonce, asked) in to_probe {
+        debug!(member = %member, via = ?asked, "Link down for its hold; asking whether others reach the member");
+        let (epoch, epoch_seq) = state.epoch_stamp();
+        let request = ParsedMessage::Probe(ProbeRequest {
+            epoch,
+            epoch_seq,
+            nonce,
+            member: member.clone(),
+        });
+        for peer in asked {
+            if let Err(error) = state.transport.send_message(&peer, &request).await {
+                debug!(peer = %peer, error = %format!("{error:#}"), "PROBE failed");
+            }
+        }
+    }
     if next == recorded {
         return;
     }
@@ -126,17 +266,89 @@ pub(crate) async fn check(state: &AppState) {
     }
 }
 
+/// A PROBE from `peer`: whether we reach the member it names, from what we know (no
+/// contact made): an active session that answers (`reach::reached_members`); a session
+/// that doesn't answer now (not yet held dormant: asks come after as long as that
+/// takes), held dormant, or HELLOs unanswered; else we can't tell. Only members are
+/// answered.
+pub(crate) async fn answer(state: &AppState, peer: &str, request: ProbeRequest) {
+    if topology::listed(state, peer) != Some(true) {
+        return;
+    }
+    let now = Instant::now();
+    let reached = if reach::reached_members(state, now)
+        .await
+        .contains(&request.member)
+    {
+        Some(true)
+    } else {
+        match state.get_peer_state(&request.member).await {
+            Some(member) => {
+                let member = member.lock().await;
+                (member.hello_state == crate::state::HelloState::Established
+                    || member.liveness.is_dormant()
+                    || member.is_unreachable())
+                .then_some(false)
+            }
+            None => None,
+        }
+    };
+    let (epoch, epoch_seq) = state.epoch_stamp();
+    let reply = ParsedMessage::ProbeReply(ProbeReply {
+        epoch,
+        epoch_seq,
+        nonce: request.nonce,
+        member: request.member,
+        reached,
+    });
+    if let Err(error) = state.transport.send_message(peer, &reply).await {
+        debug!(peer = %peer, error = %format!("{error:#}"), "PROBE_REPLY failed");
+    }
+}
+
+/// A PROBE_REPLY from `peer`: its answer kept, if it answers what we asked it.
+pub(crate) fn answered(state: &AppState, peer: &str, reply: ProbeReply) {
+    let mut pending = state.working_links.lock_or_recover();
+    if let Some(probe) = pending.probes.get_mut(&reply.member)
+        && probe.nonce == reply.nonce
+        && probe.via.contains(peer)
+    {
+        probe.answers.insert(peer.to_owned(), reply.reached);
+    }
+}
+
+/// The members to ask whether they reach `member`: its
+/// neighbours by the graph that we reach (the likeliest to have been in touch with
+/// it), up to `PROBE_VIA`, by the lowest pair score with it.
+fn via(
+    graph: &BTreeMap<String, BTreeSet<String>>,
+    reached: &BTreeSet<String>,
+    member: &str,
+) -> Vec<String> {
+    let theirs = graph.get(member);
+    let mut via: Vec<&String> = reached
+        .iter()
+        .filter(|peer| *peer != member && theirs.is_some_and(|theirs| theirs.contains(*peer)))
+        .collect();
+    via.sort_by_key(|peer| topology::pair_score(peer, member));
+    via.into_iter().take(PROBE_VIA).cloned().collect()
+}
+
 /// 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).
+/// again, or newly reached, or no longer our neighbour by the graph). A link newly
+/// unreached goes in only if `confirmed` has it (all do without): those held long
+/// enough that it hasn't are returned, to ask others about.
 fn decide(
     pending: &mut Pending,
     recorded: &RecordedLinks,
     now_links: &RecordedLinks,
     now: Instant,
     keepalive: Duration,
-) -> RecordedLinks {
+    confirmed: Option<&BTreeSet<String>>,
+) -> (RecordedLinks, BTreeSet<String>) {
+    let mut awaiting = BTreeSet::new();
     let members: BTreeSet<&String> = recorded
         .unreached
         .iter()
@@ -161,6 +373,13 @@ fn decide(
         if now.saturating_duration_since(since) < hold {
             continue;
         }
+        if unreached
+            && !was_unreached
+            && confirmed.is_some_and(|confirmed| !confirmed.contains(member))
+        {
+            awaiting.insert(member.clone());
+            continue;
+        }
         pending.since.remove(member);
         for (set, is) in [
             (&mut next.unreached, unreached),
@@ -173,7 +392,7 @@ fn decide(
             }
         }
     }
-    next
+    (next, awaiting)
 }
 
 #[cfg(test)]
@@ -203,46 +422,139 @@ mod tests {
 
         // Down for a keepalive, then up again: not recorded.
         assert_eq!(
-            decide(&mut pending, &none, &links(&["b"], &[]), at(0), keepalive),
+            decide(
+                &mut pending,
+                &none,
+                &links(&["b"], &[]),
+                at(0),
+                keepalive,
+                None
+            )
+            .0,
+            none
+        );
+        assert_eq!(
+            decide(
+                &mut pending,
+                &none,
+                &links(&["b"], &[]),
+                at(5),
+                keepalive,
+                None
+            )
+            .0,
             none
         );
         assert_eq!(
-            decide(&mut pending, &none, &links(&["b"], &[]), at(5), keepalive),
+            decide(&mut pending, &none, &none, at(10), keepalive, None).0,
             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),
+            decide(
+                &mut pending,
+                &none,
+                &links(&["b"], &[]),
+                at(15),
+                keepalive,
+                None
+            )
+            .0,
             none
         );
         assert_eq!(
-            decide(&mut pending, &none, &links(&["b"], &[]), at(30), keepalive),
+            decide(
+                &mut pending,
+                &none,
+                &links(&["b"], &[]),
+                at(30),
+                keepalive,
+                None
+            )
+            .0,
             none
         );
         let down = links(&["b"], &[]);
-        assert_eq!(decide(&mut pending, &none, &down, at(35), keepalive), down);
+        assert_eq!(
+            decide(&mut pending, &none, &down, at(35), keepalive, None).0,
+            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);
+        assert_eq!(
+            decide(&mut pending, &down, &none, at(40), keepalive, None).0,
+            down
+        );
+        assert_eq!(
+            decide(&mut pending, &down, &none, at(45), keepalive, None).0,
+            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),
+            decide(&mut pending, &none, &beyond, at(50), keepalive, None).0,
             none
         );
         assert_eq!(
-            decide(&mut pending, &none, &beyond, at(55), keepalive),
+            decide(&mut pending, &none, &beyond, at(55), keepalive, None).0,
             beyond
         );
         assert_eq!(
-            decide(&mut pending, &beyond, &none, at(60), keepalive),
+            decide(&mut pending, &beyond, &none, at(60), keepalive, None).0,
             beyond
         );
         assert_eq!(
-            decide(&mut pending, &beyond, &none, at(80), keepalive),
+            decide(&mut pending, &beyond, &none, at(80), keepalive, None).0,
+            none
+        );
+    }
+
+    /// With asks, a link down for its hold goes in only once
+    /// confirmed (one of those asked reaches the member); until then it is returned to
+    /// ask about, and stays pending. Other changes aren't asked about.
+    #[test]
+    fn unreached_links_wait_for_confirmation() {
+        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(&[], &[]);
+        let down = links(&["b"], &[]);
+        let nobody = BTreeSet::new();
+
+        assert_eq!(
+            decide(&mut pending, &none, &down, at(0), keepalive, Some(&nobody)),
+            (none.clone(), BTreeSet::new())
+        );
+        // Held long enough, not confirmed: asked about, not recorded.
+        assert_eq!(
+            decide(&mut pending, &none, &down, at(20), keepalive, Some(&nobody)),
+            (none.clone(), set(&["b"]))
+        );
+        assert_eq!(
+            decide(&mut pending, &none, &down, at(25), keepalive, Some(&nobody)),
+            (none.clone(), set(&["b"]))
+        );
+        // Confirmed: recorded.
+        assert_eq!(
+            decide(
+                &mut pending,
+                &none,
+                &down,
+                at(30),
+                keepalive,
+                Some(&set(&["b"]))
+            ),
+            (down.clone(), BTreeSet::new())
+        );
+        // Coming out needs no confirmation.
+        assert_eq!(
+            decide(&mut pending, &down, &none, at(31), keepalive, Some(&nobody)).0,
+            down
+        );
+        assert_eq!(
+            decide(&mut pending, &down, &none, at(36), keepalive, Some(&nobody)).0,
             none
         );
     }