diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/mod.rs | 40 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 9 | ||||
| -rw-r--r-- | src/transport/wire.rs | 25 |
3 files changed, 72 insertions, 2 deletions
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(); |