From 5994b49412c058079639585501a78e6458fdbb6f Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 19:54:09 +0100 Subject: HyParView (slim) --- src/transport/mod.rs | 39 +++++++++++++++++++++++++++++++++++++++ src/transport/traffic.rs | 12 ++++++++++++ src/transport/wire.rs | 32 ++++++++++++++++++++++++++++++-- 3 files changed, 81 insertions(+), 2 deletions(-) (limited to 'src/transport') diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 52b544a..8cbf412 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -251,6 +251,12 @@ pub(crate) enum ParsedMessage { Prune(TreeLink), /// Asks the recipient to send a writer's changes at once (`swarm::plumtree`). Graft(TreeLink), + /// Asks the recipient to be neighbours (`swarm::hyparview`). + Neighbour(NeighbourRequest), + /// Answers a NEIGHBOUR. + NeighbourReply(NeighbourAnswer), + /// Tells the recipient we are neighbours no more (`swarm::hyparview`). + Disconnect(NeighbourAnswer), } /// A node's proof that it holds the key of its node ID (`identity`), sending from its @@ -397,6 +403,9 @@ pub(crate) enum MessageKind { ProbeReply, Prune, Graft, + Neighbour, + NeighbourReply, + Disconnect, } impl ParsedMessage { @@ -418,6 +427,9 @@ impl ParsedMessage { ParsedMessage::ProbeReply(_) => MessageKind::ProbeReply, ParsedMessage::Prune(_) => MessageKind::Prune, ParsedMessage::Graft(_) => MessageKind::Graft, + ParsedMessage::Neighbour(_) => MessageKind::Neighbour, + ParsedMessage::NeighbourReply(_) => MessageKind::NeighbourReply, + ParsedMessage::Disconnect(_) => MessageKind::Disconnect, } } @@ -441,6 +453,10 @@ impl ParsedMessage { ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => { Some((link.epoch, link.epoch_seq)) } + ParsedMessage::Neighbour(request) => Some((request.epoch, request.epoch_seq)), + ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => { + Some((answer.epoch, answer.epoch_seq)) + } ParsedMessage::Who(_) | ParsedMessage::Me(_) | ParsedMessage::Ping(_) @@ -461,6 +477,29 @@ pub(crate) struct TreeLink { pub(crate) writer: u64, } +/// NEIGHBOUR: the sender asks the recipient to be neighbours (`swarm::hyparview`), as +/// HyParView's NEIGHBOR; `high` when the sender has none (then the recipient makes room +/// if it must). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct NeighbourRequest { + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub(crate) epoch_seq: u32, + pub(crate) high: bool, +} + +/// NEIGHBOUR_REPLY (whether the recipient of a NEIGHBOUR agreed) and DISCONNECT (never +/// agreed: neighbours no more) (`swarm::hyparview`). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct NeighbourAnswer { + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub(crate) epoch_seq: u32, + pub(crate) agreed: bool, +} + /// 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. diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 501c77b..291a76b 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -57,6 +57,8 @@ pub(crate) struct ByKind { pub(crate) probe: Counter, /// PRUNE and GRAFT (`swarm::plumtree`). pub(crate) plumtree: Counter, + /// NEIGHBOUR, NEIGHBOUR_REPLY and DISCONNECT (`swarm::hyparview`). + pub(crate) hyparview: Counter, } impl ByKind { @@ -73,6 +75,7 @@ impl ByKind { MessageKind::Links, MessageKind::Probe, MessageKind::Prune, + MessageKind::Neighbour, ] { let theirs = *other.get(kind); self.get_mut(kind).absorb(theirs); @@ -92,6 +95,9 @@ impl ByKind { MessageKind::Links | MessageKind::LinksReply => &self.links, MessageKind::Probe | MessageKind::ProbeReply => &self.probe, MessageKind::Prune | MessageKind::Graft => &self.plumtree, + MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => { + &self.hyparview + } } } @@ -108,6 +114,9 @@ impl ByKind { MessageKind::Links | MessageKind::LinksReply => &mut self.links, MessageKind::Probe | MessageKind::ProbeReply => &mut self.probe, MessageKind::Prune | MessageKind::Graft => &mut self.plumtree, + MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => { + &mut self.hyparview + } } } } @@ -302,6 +311,9 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) { MessageKind::ProbeReply => ("PROBE_REPLY", "probe"), MessageKind::Prune => ("PRUNE", "plumtree"), MessageKind::Graft => ("GRAFT", "plumtree"), + MessageKind::Neighbour => ("NEIGHBOUR", "hyparview"), + MessageKind::NeighbourReply => ("NEIGHBOUR_REPLY", "hyparview"), + MessageKind::Disconnect => ("DISCONNECT", "hyparview"), } } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 729acd3..0733d9c 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 = 37; +pub(crate) const VERSION: u8 = 38; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -35,6 +35,9 @@ fn tag(kind: MessageKind) -> u8 { MessageKind::ProbeReply => 13, MessageKind::Prune => 14, MessageKind::Graft => 15, + MessageKind::Neighbour => 16, + MessageKind::NeighbourReply => 17, + MessageKind::Disconnect => 18, } } @@ -94,6 +97,10 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result, WireError> { ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => { rmp_serde::encode::write(&mut payload, link)? } + ParsedMessage::Neighbour(request) => rmp_serde::encode::write(&mut payload, request)?, + ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => { + rmp_serde::encode::write(&mut payload, answer)? + } } Ok(payload) } @@ -171,6 +178,12 @@ pub(crate) fn decode(payload: &[u8]) -> Result { ParsedMessage::Prune(rmp_serde::from_slice(body)?) } else if is(MessageKind::Graft) { ParsedMessage::Graft(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Neighbour) { + ParsedMessage::Neighbour(rmp_serde::from_slice(body)?) + } else if is(MessageKind::NeighbourReply) { + ParsedMessage::NeighbourReply(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Disconnect) { + ParsedMessage::Disconnect(rmp_serde::from_slice(body)?) } else { return Err(WireError::UnknownKind(kind)); }) @@ -551,6 +564,21 @@ pub(crate) fn samples() -> Vec { epoch_seq: 1, writer: 5, }), + ParsedMessage::Neighbour(crate::transport::NeighbourRequest { + epoch, + epoch_seq: 1, + high: true, + }), + ParsedMessage::NeighbourReply(crate::transport::NeighbourAnswer { + epoch, + epoch_seq: 1, + agreed: true, + }), + ParsedMessage::Disconnect(crate::transport::NeighbourAnswer { + epoch, + epoch_seq: 1, + agreed: false, + }), ] } @@ -582,7 +610,7 @@ mod tests { #[test] fn every_kind_round_trips() { let samples = samples(); - assert_eq!(samples.len(), 16, "a kind is missing from `samples`"); + assert_eq!(samples.len(), 19, "a kind is missing from `samples`"); for message in samples { let bytes = encode(&message).unwrap(); let decoded = decode(&bytes).unwrap(); -- cgit v1.3.1