diff options
| author | tv <tv@krebsco.de> | 2026-11-06 14:29:42 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-11-06 14:29:42 +0100 |
| commit | d6f6c7aad904e650e6c7371dbd4776f6f0be2388 (patch) | |
| tree | c4cf94b5099e3ea31faddc7356459337127eb3f3 | |
| parent | 43305f5a0e2a87c68405989600b33e33c1ab70df (diff) | |
indirect probes
| -rw-r--r-- | docs/architecture.md | 2 | ||||
| -rw-r--r-- | docs/measurements.md | 3 | ||||
| -rw-r--r-- | docs/openapi.yaml | 5 | ||||
| -rw-r--r-- | docs/peers.md | 12 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 1 | ||||
| -rw-r--r-- | src/simtest/bench.rs | 9 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 6 | ||||
| -rw-r--r-- | src/swarm/working_links.rs | 362 | ||||
| -rw-r--r-- | src/transport/mod.rs | 40 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 9 | ||||
| -rw-r--r-- | src/transport/wire.rs | 25 |
11 files changed, 439 insertions, 35 deletions
diff --git a/docs/architecture.md b/docs/architecture.md index a63e36c..0f0bad7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -77,7 +77,7 @@ it switched to the right document without trusting anyone. ## Modules **[`transport/`](../src/transport/)** — [`mod.rs`](../src/transport/mod.rs) defines the message types (HELLO, HELLO_REPLY, sync -envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, LINKS/LINKS_REPLY) and the `Transport` trait; +envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, LINKS/LINKS_REPLY, PROBE/PROBE_REPLY) and the `Transport` trait; [`wire.rs`](../src/transport/wire.rs) is the wire format (a version byte, bumped with every change to the format, and a kind byte, then the message in MessagePack, derived from the types in [`mod.rs`](../src/transport/mod.rs) with codecs for hashes and IDs); [`mycelium/`](../src/transport/mycelium/) (with the `mycelium` feature) is the REST diff --git a/docs/measurements.md b/docs/measurements.md index f49f4ff..099bb40 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -179,7 +179,8 @@ clocks"); setting them is the operating system's job (NTP). Messages and payload bytes by direction (`sent`, `received`, `refused`) and kind (`hello`, `helloReply`, `sync`, `status`, `statusReply`, `blob`, `identity` for WHO and -ME, and `paths` for PING and PONG). The layers below the node count WHO and ME, and the +ME, `paths` for PING and PONG, `links` for LINKS and their replies, `probe` for PROBE +and PROBE_REPLY). The layers below the node count WHO and ME, and the paths layer's own PINGs and PONGs, under the address they went to or came from; a peer (`/peers`) has those of its addresses too, where it is known to be at them. Payload bytes are what a transport's message carries (the wire encoding): not the transport's framing (for Mycelium, its topic and the base64 of its diff --git a/docs/openapi.yaml b/docs/openapi.yaml index da89b6c..bd3186b 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1707,7 +1707,7 @@ components: ByKind: type: object - required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, links] + required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, links, probe] properties: hello: $ref: '#/components/schemas/Counter' @@ -1731,6 +1731,9 @@ components: links: $ref: '#/components/schemas/Counter' description: LINKS and LINKS_REPLY, asked for and answered for a map of the swarm (the dashboard). + probe: + $ref: '#/components/schemas/Counter' + description: PROBE and PROBE_REPLY, asking others whether they reach a member before recording a link to it as unreached. Traffic: type: object diff --git a/docs/peers.md b/docs/peers.md index dd3186f..79f2abb 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -366,9 +366,17 @@ its neighbours by the graph it doesn't reach (`unreached`), once down for four k intervals, and its neighbours it reaches that the graph doesn't link it to (`reached`: a ring going round a member it doesn't reach, a member reaching out from behind NAT), once up for one; each comes out again after the -other hold. On a stable swarm the entry has none. A member that went away is recorded +other hold; a link the other end records as unreached already isn't recorded again. +On a stable swarm the entry has none. 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. +out once they reach it again: so a node first asks up to three +of the member's own neighbours it reaches whether they reach it (a PROBE, answered from +what they know, contacting nobody: an active session that answers; a session that +doesn't, held dormant or unanswered HELLOs; or can't tell): if some don't and none does, the member is away and nothing is recorded +(asked again after another hold); otherwise, or with no answer within 10 s, the link +is recorded. A node that reaches nobody at all is the one away, and +records nothing. PROBEs are their own message, apart from the dashboard's +LINKS, which a node may leave unanswered. A writer's changes pass on along its tree ([`swarm/tree.rs`](../src/swarm/tree.rs)), the same on every node: built on the graph the member list gives (as if every member reached every other), less 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(); |
