indirect probes

7 files changed, 422 insertions, 30 deletions

diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 94525cb..963dd0c 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -191,6 +191,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
             (direction, "identity", by.identity),
             (direction, "paths", by.paths),
             (direction, "links", by.links),
+            (direction, "probe", by.probe),
         ]
     };
     let mut series = Vec::new();
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 493936f..ac44685 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -422,7 +422,7 @@ impl Totals {
 }
 
 /// The kinds, in the order of the report.
-const KINDS: [&str; 9] = [
+const KINDS: [&str; 10] = [
     "SYNC",
     "STATUS",
     "STATUS_REPLY",
@@ -432,9 +432,10 @@ const KINDS: [&str; 9] = [
     "PING/PONG",
     "WHO/ME",
     "LINKS",
+    "PROBE",
 ];
 
-fn counters(by: &ByKind) -> [Counter; 9] {
+fn counters(by: &ByKind) -> [Counter; 10] {
     [
         by.sync,
         by.status,
@@ -445,10 +446,11 @@ fn counters(by: &ByKind) -> [Counter; 9] {
         by.paths,
         by.identity,
         by.links,
+        by.probe,
     ]
 }
 
-fn counters_mut(by: &mut ByKind) -> [&mut Counter; 9] {
+fn counters_mut(by: &mut ByKind) -> [&mut Counter; 10] {
     [
         &mut by.sync,
         &mut by.status,
@@ -459,6 +461,7 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 9] {
         &mut by.paths,
         &mut by.identity,
         &mut by.links,
+        &mut by.probe,
     ]
 }
 
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 9b701b5..7154357 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -248,6 +248,12 @@ pub(crate) async fn handle_message(
         transport::ParsedMessage::LinksReply(reply) => {
             crate::swarm::links::received(state, &peer, reply);
         }
+        transport::ParsedMessage::Probe(request) => {
+            crate::swarm::working_links::answer(state, &peer, request).await;
+        }
+        transport::ParsedMessage::ProbeReply(reply) => {
+            crate::swarm::working_links::answered(state, &peer, reply);
+        }
         transport::ParsedMessage::Who(_)
         | transport::ParsedMessage::Me(_)
         | transport::ParsedMessage::Ping(_)
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
         );
     }
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 00e07e5..743e713 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -243,6 +243,10 @@ pub(crate) enum ParsedMessage {
     Links(LinksRequest),
     /// Answers a LINKS.
     LinksReply(LinksReply),
+    /// Asks whether the recipient reaches a member (`swarm::working_links`).
+    Probe(ProbeRequest),
+    /// Answers a PROBE.
+    ProbeReply(ProbeReply),
 }
 
 /// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -385,6 +389,8 @@ pub(crate) enum MessageKind {
     Pong,
     Links,
     LinksReply,
+    Probe,
+    ProbeReply,
 }
 
 impl ParsedMessage {
@@ -402,6 +408,8 @@ impl ParsedMessage {
             ParsedMessage::Pong { .. } => MessageKind::Pong,
             ParsedMessage::Links(_) => MessageKind::Links,
             ParsedMessage::LinksReply(_) => MessageKind::LinksReply,
+            ParsedMessage::Probe(_) => MessageKind::Probe,
+            ParsedMessage::ProbeReply(_) => MessageKind::ProbeReply,
         }
     }
 
@@ -420,6 +428,8 @@ impl ParsedMessage {
             ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)),
             ParsedMessage::Links(request) => Some((request.epoch, request.epoch_seq)),
             ParsedMessage::LinksReply(reply) => Some((reply.epoch, reply.epoch_seq)),
+            ParsedMessage::Probe(request) => Some((request.epoch, request.epoch_seq)),
+            ParsedMessage::ProbeReply(reply) => Some((reply.epoch, reply.epoch_seq)),
             ParsedMessage::Who(_)
             | ParsedMessage::Me(_)
             | ParsedMessage::Ping(_)
@@ -428,6 +438,36 @@ impl ParsedMessage {
     }
 }
 
+/// PROBE: whether the recipient reaches `member` (`swarm::working_links`), as SWIM
+/// asks others before it suspects a member. Answered from what the recipient knows,
+/// without contacting the member.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct ProbeRequest {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// Repeated in the reply.
+    pub(crate) nonce: u64,
+    /// The member asked about (its node ID).
+    pub(crate) member: String,
+}
+
+/// PROBE_REPLY: whether the sender reaches the member asked about.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct ProbeReply {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// The request's.
+    pub(crate) nonce: u64,
+    pub(crate) member: String,
+    /// It does (an active session that answers); it doesn't (it holds the member
+    /// dormant, or its HELLOs went unanswered); None: it can't tell (no contact with it).
+    pub(crate) reached: Option<bool>,
+}
+
 /// LINKS: what a member is asked for a map of the swarm (`swarm::links`).
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub(crate) struct LinksRequest {
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index a34bf4b..2e4266c 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -53,6 +53,8 @@ pub(crate) struct ByKind {
     pub(crate) paths: Counter,
     /// LINKS and LINKS_REPLY (`swarm::links`).
     pub(crate) links: Counter,
+    /// PROBE and PROBE_REPLY (`swarm::working_links`).
+    pub(crate) probe: Counter,
 }
 
 impl ByKind {
@@ -67,6 +69,7 @@ impl ByKind {
             MessageKind::Who,
             MessageKind::Ping,
             MessageKind::Links,
+            MessageKind::Probe,
         ] {
             let theirs = *other.get(kind);
             self.get_mut(kind).absorb(theirs);
@@ -84,6 +87,7 @@ impl ByKind {
             MessageKind::Who | MessageKind::Me => &self.identity,
             MessageKind::Ping | MessageKind::Pong => &self.paths,
             MessageKind::Links | MessageKind::LinksReply => &self.links,
+            MessageKind::Probe | MessageKind::ProbeReply => &self.probe,
         }
     }
 
@@ -98,6 +102,7 @@ impl ByKind {
             MessageKind::Who | MessageKind::Me => &mut self.identity,
             MessageKind::Ping | MessageKind::Pong => &mut self.paths,
             MessageKind::Links | MessageKind::LinksReply => &mut self.links,
+            MessageKind::Probe | MessageKind::ProbeReply => &mut self.probe,
         }
     }
 }
@@ -191,6 +196,8 @@ fn related(peer: &str, message: &ParsedMessage) -> Option<String> {
         }
         ParsedMessage::Links(request) => format!("links {peer} {}", request.nonce),
         ParsedMessage::LinksReply(reply) => format!("links {peer} {}", reply.nonce),
+        ParsedMessage::Probe(request) => format!("probe {peer} {}", request.nonce),
+        ParsedMessage::ProbeReply(reply) => format!("probe {peer} {}", reply.nonce),
         ParsedMessage::Hello(hello) => format!("hello {peer} {}", Hex(&hello.run_id)),
         ParsedMessage::HelloReply(reply) => format!("hello {peer} {}", Hex(&reply.run_id)),
         ParsedMessage::Blob(blob) => format!("blob {}", blob.id),
@@ -286,6 +293,8 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) {
         MessageKind::Pong => ("PONG", "paths"),
         MessageKind::Links => ("LINKS", "links"),
         MessageKind::LinksReply => ("LINKS_REPLY", "links"),
+        MessageKind::Probe => ("PROBE", "probe"),
+        MessageKind::ProbeReply => ("PROBE_REPLY", "probe"),
     }
 }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index ef692b3..bf46fc9 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 35;
+pub(crate) const VERSION: u8 = 36;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -31,6 +31,8 @@ fn tag(kind: MessageKind) -> u8 {
         MessageKind::Pong => 9,
         MessageKind::Links => 10,
         MessageKind::LinksReply => 11,
+        MessageKind::Probe => 12,
+        MessageKind::ProbeReply => 13,
     }
 }
 
@@ -85,6 +87,8 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
         ParsedMessage::Pong { nonce, at } => rmp_serde::encode::write(&mut payload, &(nonce, at))?,
         ParsedMessage::Links(request) => rmp_serde::encode::write(&mut payload, request)?,
         ParsedMessage::LinksReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
+        ParsedMessage::Probe(request) => rmp_serde::encode::write(&mut payload, request)?,
+        ParsedMessage::ProbeReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
     }
     Ok(payload)
 }
@@ -154,6 +158,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
         ParsedMessage::Links(rmp_serde::from_slice(body)?)
     } else if is(MessageKind::LinksReply) {
         ParsedMessage::LinksReply(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Probe) {
+        ParsedMessage::Probe(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::ProbeReply) {
+        ParsedMessage::ProbeReply(rmp_serde::from_slice(body)?)
     } else {
         return Err(WireError::UnknownKind(kind));
     })
@@ -511,6 +519,19 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
                 links: Vec::new(),
             }],
         }),
+        ParsedMessage::Probe(crate::transport::ProbeRequest {
+            epoch,
+            epoch_seq: 1,
+            nonce: 4,
+            member: "q".into(),
+        }),
+        ParsedMessage::ProbeReply(crate::transport::ProbeReply {
+            epoch,
+            epoch_seq: 1,
+            nonce: 4,
+            member: "q".into(),
+            reached: None,
+        }),
     ]
 }
 
@@ -542,7 +563,7 @@ mod tests {
     #[test]
     fn every_kind_round_trips() {
         let samples = samples();
-        assert_eq!(samples.len(), 12, "a kind is missing from `samples`");
+        assert_eq!(samples.len(), 14, "a kind is missing from `samples`");
         for message in samples {
             let bytes = encode(&message).unwrap();
             let decoded = decode(&bytes).unwrap();